diff --git a/lib/graphql/execution/interpreter/runtime.rb b/lib/graphql/execution/interpreter/runtime.rb
index a2d0f389f5..3b86cff011 100644
--- a/lib/graphql/execution/interpreter/runtime.rb
+++ b/lib/graphql/execution/interpreter/runtime.rb
@@ -1,6 +1,14 @@
 # frozen_string_literal: true
 require "graphql/execution/interpreter/runtime/graphql_result"
 
+#####
+# Next thoughts
+#
+# - `continue_field` is probably a step of its own -- that method can somehow be factored out
+# - It seems like Dataloader/Lazy will fit in at the queue level, so the flow would be:
+#   - Run jobs from queue
+#   - Then, run dataloader/lazies
+#   - Repeat
 module GraphQL
   module Execution
     class Interpreter
@@ -26,6 +34,65 @@ def current_object
             :current_arguments, :current_field, :was_authorized_by_scope_items
         end
 
+        class RunQueue
+          def initialize(runtime:)
+            @runtime = runtime
+            @current_flush = []
+            @dataloader = runtime.dataloader
+            @lazies_at_depth = runtime.lazies_at_depth
+            @running_eagerly = false
+          end
+
+          def append_step(step)
+            @current_flush << step
+          end
+
+          def complete(eager: false)
+            # p [self.class, __method__, eager, caller(1,1).first, @current_flush.size]
+            prev_eagerly = @running_eagerly
+            @running_eagerly = eager
+            while (fl = @current_flush) && fl.any?
+              @current_flush = []
+              steps_to_rerun_after_lazy = []
+              while fl.any?
+                while (step = fl.shift)
+                  step_finished = false
+                  while !step_finished
+                    # p [:run_step, step.inspect_step]
+                    step_result = step.run_step
+                    step_finished = step.step_finished?
+                    if !step_finished && @runtime.lazy?(step_result)
+                      # p [:lazy, step_result.class, step.depth]
+                      @lazies_at_depth[step.depth] << step
+                      steps_to_rerun_after_lazy << step
+                      step_finished = true # we'll come back around to it
+                    end
+                  end
+
+                  if @running_eagerly && @current_flush.any?
+                    # This is for mutations. If a mutation parent field enqueues any child fields,
+                    # we need to run those before running other mutation parent fields.
+                    fl.unshift(*@current_flush)
+                    @current_flush.clear
+                  end
+                end
+
+                if @current_flush.any?
+                  fl.concat(@current_flush)
+                  @current_flush.clear
+                else
+                  fl.concat(steps_to_rerun_after_lazy)
+                  steps_to_rerun_after_lazy.clear
+                  @dataloader.run
+                  Interpreter::Resolve.resolve_each_depth(@lazies_at_depth, @dataloader)
+                end
+              end
+            end
+          ensure
+            @running_eagerly = prev_eagerly
+          end
+        end
+
         # @return [GraphQL::Query]
         attr_reader :query
 
@@ -35,6 +102,10 @@ def current_object
         # @return [GraphQL::Query::Context]
         attr_reader :context
 
+        attr_reader :dataloader, :current_trace, :lazies_at_depth
+
+        attr_accessor :run_queue
+
         def initialize(query:, lazies_at_depth:)
           @query = query
           @current_trace = query.current_trace
@@ -54,9 +125,11 @@ def initialize(query:, lazies_at_depth:)
           end
           # { Class => Boolean }
           @lazy_cache = {}.compare_by_identity
+          @run_queue = RunQueue.new(runtime: self)
         end
 
         def final_result
+          # TODO can `graphql_result_data` be set to `nil` when `.wrap` fails?
           @response.respond_to?(:graphql_result_data) ? @response.graphql_result_data : @response
         end
 
@@ -64,6 +137,31 @@ def inspect
           "#<#{self.class.name} response=#{@response.inspect}>"
         end
 
+        class OperationDirectivesStep
+          def initialize(runtime, object, method_to_call, directives, next_step)
+            @runtime = runtime
+            @object = object
+            @method_to_call = method_to_call
+            @directives = directives
+            @next_step = next_step
+          end
+
+          def run_step
+            @runtime.call_method_on_directives(@method_to_call, @object, @directives) do
+              @runtime.run_queue.append_step(@next_step)
+              @next_step
+            end
+          end
+
+          def step_finished?
+            true
+          end
+
+          def inspect_step
+            "#{self.class.name.split("::").last}##{object_id}(#{@directives ? @directives.map(&:name).join(", ") : nil}) => #{@next_step.inspect_step}"
+          end
+        end
+
         # @return [void]
         def run_eager
           root_type = query.root_type
@@ -86,38 +184,18 @@ def run_eager
           object = schema.sync_lazy(object) # TODO test query partial with lazy root object
           runtime_state = get_current_runtime_state
           case root_type.kind.name
-          when "OBJECT"
-            object_proxy = root_type.wrap(object, context)
-            object_proxy = schema.sync_lazy(object_proxy)
-            if object_proxy.nil?
-              @response = nil
-            else
-              @response = GraphQLResultHash.new(nil, root_type, object_proxy, nil, false, selections, is_eager, ast_node, nil, nil)
-              @response.base_path = base_path
-              runtime_state.current_result = @response
-              call_method_on_directives(:resolve, object, ast_node.directives) do
-                each_gathered_selections(@response) do |selections, is_selection_array, ordered_result_keys|
-                  @response.ordered_result_keys ||= ordered_result_keys
-                  if is_selection_array
-                    selection_response = GraphQLResultHash.new(nil, root_type, object_proxy, nil, false, selections, is_eager, ast_node, nil, nil)
-                    selection_response.ordered_result_keys = ordered_result_keys
-                    final_response = @response
-                  else
-                    selection_response = @response
-                    final_response = nil
-                  end
+          when "OBJECT", "UNION", "INTERFACE"
+            # TODO: use `nil` for top-level result when `.wrap` returns `nil`
+            @response = GraphQLResultHash.new(self, nil, root_type, object, nil, false, selections, is_eager, ast_node, nil, nil)
+            @response.base_path = base_path
 
-                  @dataloader.append_job {
-                    evaluate_selections(
-                      selections,
-                      selection_response,
-                      final_response,
-                      nil,
-                    )
-                  }
-                end
-              end
+            runtime_state.current_result = @response
+            next_step = if !ast_node.directives.empty?
+              OperationDirectivesStep.new(self, object, :resolve, ast_node.directives, @response)
+            else
+              @response
             end
+            @run_queue.append_step(next_step)
           when "LIST"
             inner_type = root_type.unwrap
             case inner_type.kind.name
@@ -125,40 +203,26 @@ def run_eager
               result_name = ast_node.alias || ast_node.name
               field_defn = query.field_definition
               owner_type = field_defn.owner
-              selection_result = GraphQLResultHash.new(nil, owner_type, nil, nil, false, EmptyObjects::EMPTY_ARRAY, false, ast_node, nil, nil)
+              selection_result = GraphQLResultHash.new(self, nil, owner_type, nil, nil, false, EmptyObjects::EMPTY_ARRAY, false, ast_node, nil, nil)
               selection_result.base_path = base_path
               selection_result.ordered_result_keys = [result_name]
-              runtime_state = get_current_runtime_state
               runtime_state.current_result = selection_result
               runtime_state.current_result_name = result_name
               continue_value = continue_value(object, field_defn, false, ast_node, result_name, selection_result)
               if HALT != continue_value
-                continue_field(continue_value, owner_type, field_defn, root_type, ast_node, nil, false, nil, nil, result_name, selection_result, false, runtime_state) # rubocop:disable Metrics/ParameterLists
+                continue_field(continue_value, field_defn, root_type, ast_node, nil, false, nil, result_name, selection_result, false, runtime_state) # rubocop:disable Metrics/ParameterLists
               end
               @response = selection_result[result_name]
             else
-              @response = GraphQLResultArray.new(nil, root_type, nil, nil, false, selections, false, ast_node, nil, nil)
+              @response = GraphQLResultArray.new(self, nil, root_type, object, nil, false, selections, false, ast_node, nil, nil)
               @response.base_path = base_path
-              idx = nil
-              object.each do |inner_value|
-                idx ||= 0
-                this_idx = idx
-                idx += 1
-                @dataloader.append_job do
-                  runtime_state.current_result_name = this_idx
-                  runtime_state.current_result = @response
-                  continue_field(
-                    inner_value, root_type, nil, inner_type, nil, @response.graphql_selections, false, object_proxy,
-                    nil, this_idx, @response, false, runtime_state
-                  )
-                end
-              end
+              @run_queue.append_step(@response)
             end
           when "SCALAR", "ENUM"
             result_name = ast_node.alias || ast_node.name
             field_defn = query.field_definition
             owner_type = field_defn.owner
-            selection_result = GraphQLResultHash.new(nil, owner_type, nil, nil, false, EmptyObjects::EMPTY_ARRAY, false, ast_node, nil, nil)
+            selection_result = GraphQLResultHash.new(self, nil, owner_type, nil, nil, false, EmptyObjects::EMPTY_ARRAY, false, ast_node, nil, nil)
             selection_result.ordered_result_keys = [result_name]
             selection_result.base_path = base_path
             runtime_state = get_current_runtime_state
@@ -166,34 +230,13 @@ def run_eager
             runtime_state.current_result_name = result_name
             continue_value = continue_value(object, field_defn, false, ast_node, result_name, selection_result)
             if HALT != continue_value
-              continue_field(continue_value, owner_type, field_defn, query.root_type, ast_node, nil, false, nil, nil, result_name, selection_result, false, runtime_state) # rubocop:disable Metrics/ParameterLists
+              continue_field(continue_value, field_defn, query.root_type, ast_node, nil, false, nil, result_name, selection_result, false, runtime_state) # rubocop:disable Metrics/ParameterLists
             end
             @response = selection_result[result_name]
-          when "UNION", "INTERFACE"
-            resolved_type, _resolved_obj = resolve_type(root_type, object)
-            resolved_type = schema.sync_lazy(resolved_type)
-            object_proxy = resolved_type.wrap(object, context)
-            object_proxy = schema.sync_lazy(object_proxy)
-            @response = GraphQLResultHash.new(nil, resolved_type, object_proxy, nil, false, selections, false, query.ast_nodes.first, nil, nil)
-            @response.base_path = base_path
-            each_gathered_selections(@response) do |selections, is_selection_array, ordered_result_keys|
-              @response.ordered_result_keys ||= ordered_result_keys
-              if is_selection_array == true
-                raise "This isn't supported yet"
-              end
-
-              @dataloader.append_job {
-                evaluate_selections(
-                  selections,
-                  @response,
-                  nil,
-                  runtime_state,
-                )
-              }
-            end
           else
             raise "Invariant: unsupported type kind for partial execution: #{root_type.kind.inspect} (#{root_type})"
           end
+          @run_queue.complete
           nil
         end
 
@@ -285,211 +328,215 @@ def gather_selections(owner_object, owner_type, selections, selections_to_run, s
           selections_to_run || selections_by_name
         end
 
-        NO_ARGS = GraphQL::EmptyObjects::EMPTY_HASH
-
-        # @return [void]
-        def evaluate_selections(gathered_selections, selections_result, target_result, runtime_state) # rubocop:disable Metrics/ParameterLists
-          runtime_state ||= get_current_runtime_state
-          runtime_state.current_result_name = nil
-          runtime_state.current_result = selections_result
-          # This is a less-frequent case; use a fast check since it's often not there.
-          if (directives = gathered_selections[:graphql_directives])
-            gathered_selections.delete(:graphql_directives)
-          end
-
-          call_method_on_directives(:resolve, selections_result.graphql_application_value, directives) do
-            gathered_selections.each do |result_name, field_ast_nodes_or_ast_node|
-              # Field resolution may pause the fiber,
-              # so it wouldn't get to the `Resolve` call that happens below.
-              # So instead trigger a run from this outer context.
-              if selections_result.graphql_is_eager
-                @dataloader.clear_cache
-                @dataloader.run_isolated {
-                  evaluate_selection(
-                    result_name, field_ast_nodes_or_ast_node, selections_result
-                  )
-                  @dataloader.clear_cache
-                }
-              else
-                @dataloader.append_job {
-                  evaluate_selection(
-                    result_name, field_ast_nodes_or_ast_node, selections_result
-                  )
-                }
-              end
+        class FieldResolveStep
+          def initialize(runtime, field, ast_node, ast_nodes, result_name, selection_result)
+            @runtime = runtime
+            @field = field
+            @object = selection_result.graphql_application_value
+            if @field.dynamic_introspection
+              @object = field.owner.wrap(@object, @runtime.context)
             end
-            if target_result
-              selections_result.merge_into(target_result)
-            end
-            selections_result
+            @ast_node = ast_node
+            @ast_nodes = ast_nodes
+            @result_name = result_name
+            @selection_result = selection_result
+            @next_selections = nil
+            @step = :inspect_ast
           end
-        end
 
-        # @return [void]
-        def evaluate_selection(result_name, field_ast_nodes_or_ast_node, selections_result) # rubocop:disable Metrics/ParameterLists
-          return if selections_result.graphql_dead
-          # As a performance optimization, the hash key will be a `Node` if
-          # there's only one selection of the field. But if there are multiple
-          # selections of the field, it will be an Array of nodes
-          if field_ast_nodes_or_ast_node.is_a?(Array)
-            field_ast_nodes = field_ast_nodes_or_ast_node
-            ast_node = field_ast_nodes.first
-          else
-            field_ast_nodes = nil
-            ast_node = field_ast_nodes_or_ast_node
+          attr_reader :selection_result
+
+          def inspect_step
+            "#{self.class.name.split("::").last}##{object_id}/#@step(#{@field.path} @ #{@selection_result.path.join(".")}.#{@result_name})"
           end
-          field_name = ast_node.name
-          owner_type = selections_result.graphql_result_type
-          field_defn = query.types.field(owner_type, field_name)
 
-          # Set this before calling `run_with_directives`, so that the directive can have the latest path
-          runtime_state = get_current_runtime_state
-          runtime_state.current_field = field_defn
-          runtime_state.current_result = selections_result
-          runtime_state.current_result_name = result_name
+          def step_finished?
+            @step == :finished
+          end
 
-          owner_object = selections_result.graphql_application_value
-          if field_defn.dynamic_introspection
-            owner_object = field_defn.owner.wrap(owner_object, context)
+          def depth
+            @selection_result.depth + 1
           end
 
-          if !field_defn.any_arguments?
-            resolved_arguments = GraphQL::Execution::Interpreter::Arguments::EMPTY
-            if field_defn.extras.size == 0
-              evaluate_selection_with_resolved_keyword_args(
-                NO_ARGS, resolved_arguments, field_defn, ast_node, field_ast_nodes, owner_object, result_name, selections_result, runtime_state
-              )
-            else
-              evaluate_selection_with_args(resolved_arguments, field_defn, ast_node, field_ast_nodes, owner_object, result_name, selections_result, runtime_state)
-            end
-          else
-            @query.arguments_cache.dataload_for(ast_node, field_defn, owner_object) do |resolved_arguments|
-              runtime_state = get_current_runtime_state # This might be in a different fiber
-              evaluate_selection_with_args(resolved_arguments, field_defn, ast_node, field_ast_nodes, owner_object, result_name, selections_result, runtime_state)
+          attr_accessor :result
+
+          def value # Lazy API
+            @result = begin
+              @runtime.schema.sync_lazy(@result)
+            rescue GraphQL::ExecutionError => err
+              err
+            rescue StandardError => err
+              begin
+                @runtime.query.handle_or_reraise(err)
+              rescue GraphQL::ExecutionError => ex_err
+                ex_err
+              end
             end
           end
-        end
 
-        def evaluate_selection_with_args(arguments, field_defn, ast_node, field_ast_nodes, object, result_name, selection_result, runtime_state)  # rubocop:disable Metrics/ParameterLists
-          after_lazy(arguments, field: field_defn, ast_node: ast_node, owner_object: object, arguments: arguments, result_name: result_name, result: selection_result, runtime_state: runtime_state) do |resolved_arguments, runtime_state|
-            if resolved_arguments.is_a?(GraphQL::ExecutionError) || resolved_arguments.is_a?(GraphQL::UnauthorizedError)
-              return_type_non_null = field_defn.type.non_null?
-              continue_value(resolved_arguments, field_defn, return_type_non_null, ast_node, result_name, selection_result)
-              next
+          def run_step
+            if @selection_result.graphql_dead
+              @step = :finished
+              return nil
             end
-
-            kwarg_arguments = if field_defn.extras.empty?
-              if resolved_arguments.empty?
-                # We can avoid allocating the `{ Symbol => Object }` hash in this case
-                NO_ARGS
+            case @step
+            when :inspect_ast
+              # Optimize for the case that field is selected only once
+              if @ast_nodes.nil? || @ast_nodes.size == 1
+                @next_selections = @ast_node.selections
+                directives = @ast_node.directives
               else
-                resolved_arguments.keyword_arguments
+                @next_selections = []
+                directives = []
+                @ast_nodes.each { |f|
+                  @next_selections.concat(f.selections)
+                  directives.concat(f.directives)
+                }
               end
-            else
-              # Bundle up the extras, then make a new arguments instance
-              # that includes the extras, too.
-              extra_args = {}
-              field_defn.extras.each do |extra|
-                case extra
-                when :ast_node
-                  extra_args[:ast_node] = ast_node
-                when :execution_errors
-                  extra_args[:execution_errors] = ExecutionErrors.new(context, ast_node, current_path)
-                when :path
-                  extra_args[:path] = current_path
-                when :lookahead
-                  if !field_ast_nodes
-                    field_ast_nodes = [ast_node]
-                  end
 
-                  extra_args[:lookahead] = Execution::Lookahead.new(
-                    query: query,
-                    ast_nodes: field_ast_nodes,
-                    field: field_defn,
-                  )
-                when :argument_details
-                  # Use this flag to tell Interpreter::Arguments to add itself
-                  # to the keyword args hash _before_ freezing everything.
-                  extra_args[:argument_details] = :__arguments_add_self
-                when :parent
-                  parent_result = selection_result.graphql_parent
-                  extra_args[:parent] = parent_result&.graphql_application_value&.object
+              if directives.any?
+                @step = :finished # some way to detect whether the block below is called or not
+                @runtime.call_method_on_directives(:resolve, @object, directives) do
+                  @step = :load_arguments
+                  self # TODO what kind of compatibility is possible here?
+                end
+              else
+                # TODO some way to continue without this step
+                @step = :load_arguments
+              end
+            when :load_arguments
+              if !@field.any_arguments?
+                @resolved_arguments = GraphQL::Execution::Interpreter::Arguments::EMPTY
+                if @field.extras.size == 0
+                  @kwarg_arguments = EmptyObjects::EMPTY_HASH
+                  @step = :call_field_resolver # kwargs are already ready -- they're empty
                 else
-                  extra_args[extra] = field_defn.fetch_extra(extra, context)
+                  @step = :prepare_kwarg_arguments
                 end
+                nil
+              else
+                @step = :prepare_kwarg_arguments
+                @runtime.query.arguments_cache.dataload_for(@ast_node, @field, @object) do |resolved_arguments|
+                  @result = resolved_arguments
+                end
+                @result
               end
-              if !extra_args.empty?
-                resolved_arguments = resolved_arguments.merge_extras(extra_args)
+            when :prepare_kwarg_arguments
+              if @resolved_arguments.nil? && @result.nil?
+                @runtime.dataloader.run
+              end
+              @resolved_arguments ||= @result
+              @result = nil
+              if @resolved_arguments.is_a?(GraphQL::ExecutionError) || @resolved_arguments.is_a?(GraphQL::UnauthorizedError)
+                return_type_non_null = @field.type.non_null?
+                @runtime.continue_value(@resolved_arguments, @field, return_type_non_null, @ast_node, @result_name, @selection_result)
+                @step = :finished
+                return
               end
-              resolved_arguments.keyword_arguments
-            end
-
-            evaluate_selection_with_resolved_keyword_args(kwarg_arguments, resolved_arguments, field_defn, ast_node, field_ast_nodes, object, result_name, selection_result, runtime_state)
-          end
-        end
 
-        def evaluate_selection_with_resolved_keyword_args(kwarg_arguments, resolved_arguments, field_defn, ast_node, field_ast_nodes, object, result_name, selection_result, runtime_state)  # rubocop:disable Metrics/ParameterLists
-          runtime_state.current_field = field_defn
-          runtime_state.current_arguments = resolved_arguments
-          runtime_state.current_result_name = result_name
-          runtime_state.current_result = selection_result
-          # Optimize for the case that field is selected only once
-          if field_ast_nodes.nil? || field_ast_nodes.size == 1
-            next_selections = ast_node.selections
-            directives = ast_node.directives
-          else
-            next_selections = []
-            directives = []
-            field_ast_nodes.each { |f|
-              next_selections.concat(f.selections)
-              directives.concat(f.directives)
-            }
-          end
+              @kwarg_arguments = if @field.extras.empty?
+                if @resolved_arguments.empty?
+                  # We can avoid allocating the `{ Symbol => Object }` hash in this case
+                  EmptyObjects::EMPTY_HASH
+                else
+                  @resolved_arguments.keyword_arguments
+                end
+              else
+                # Bundle up the extras, then make a new arguments instance
+                # that includes the extras, too.
+                extra_args = {}
+                @field.extras.each do |extra|
+                  case extra
+                  when :ast_node
+                    extra_args[:ast_node] = @ast_node
+                  when :execution_errors
+                    extra_args[:execution_errors] = ExecutionErrors.new(@runtime.context, @ast_node, @runtime.current_path)
+                  when :path
+                    extra_args[:path] = @runtime.current_path
+                  when :lookahead
+                    if !@field_ast_nodes
+                      @field_ast_nodes = [@ast_node]
+                    end
 
-          field_result = call_method_on_directives(:resolve, object, directives) do
-            if !directives.empty?
-              # This might be executed in a different context; reset this info
-              runtime_state = get_current_runtime_state
-              runtime_state.current_field = field_defn
-              runtime_state.current_arguments = resolved_arguments
-              runtime_state.current_result_name = result_name
-              runtime_state.current_result = selection_result
-            end
-            # Actually call the field resolver and capture the result
-            app_result = begin
-              @current_trace.begin_execute_field(field_defn, object, kwarg_arguments, query)
-              @current_trace.execute_field(field: field_defn, ast_node: ast_node, query: query, object: object, arguments: kwarg_arguments) do
-                field_defn.resolve(object, kwarg_arguments, context)
+                    extra_args[:lookahead] = Execution::Lookahead.new(
+                      query: @runtime.query,
+                      ast_nodes: @field_ast_nodes,
+                      field: @field,
+                    )
+                  when :argument_details
+                    # Use this flag to tell Interpreter::Arguments to add itself
+                    # to the keyword args hash _before_ freezing everything.
+                    extra_args[:argument_details] = :__arguments_add_self
+                  when :parent
+                    parent_result = @selection_result.graphql_parent
+                    if parent_result.is_a?(GraphQL::Execution::Interpreter::Runtime::GraphQLResultArray)
+                      parent_result = parent_result.graphql_parent
+                    end
+                    parent_value = parent_result&.graphql_application_value&.object
+                    extra_args[:parent] = parent_value
+                  else
+                    extra_args[extra] = @field.fetch_extra(extra, @runtime.context)
+                  end
+                end
+                if !extra_args.empty?
+                  @resolved_arguments = @resolved_arguments.merge_extras(extra_args)
+                end
+                @resolved_arguments.keyword_arguments
               end
-            rescue GraphQL::ExecutionError => err
-              err
-            rescue StandardError => err
-              begin
-                query.handle_or_reraise(err)
-              rescue GraphQL::ExecutionError => ex_err
-                ex_err
+              @step = :call_field_resolver
+              nil
+            when :call_field_resolver
+              # if !directives.empty?
+                # This might be executed in a different context; reset this info
+              runtime_state = @runtime.get_current_runtime_state
+              runtime_state.current_field = @field
+              runtime_state.current_arguments = @resolved_arguments
+              runtime_state.current_result_name = @result_name
+              runtime_state.current_result = @selection_result
+              # end
+
+              # Actually call the field resolver and capture the result
+              query = @runtime.query
+              app_result = begin
+                @runtime.current_trace.begin_execute_field(@field, @object, @kwarg_arguments, query)
+                @runtime.current_trace.execute_field(field: @field, ast_node: @ast_node, query: query, object: @object, arguments: @kwarg_arguments) do
+                  @field.resolve(@object, @kwarg_arguments, query.context)
+                end
+              rescue GraphQL::ExecutionError => err
+                err
+              rescue StandardError => err
+                begin
+                  query.handle_or_reraise(err)
+                rescue GraphQL::ExecutionError => ex_err
+                  ex_err
+                end
               end
-            end
-            @current_trace.end_execute_field(field_defn, object, kwarg_arguments, query, app_result)
-            after_lazy(app_result, field: field_defn, ast_node: ast_node, owner_object: object, arguments: resolved_arguments, result_name: result_name, result: selection_result, runtime_state: runtime_state) do |inner_result, runtime_state|
-              owner_type = selection_result.graphql_result_type
-              return_type = field_defn.type
-              continue_value = continue_value(inner_result, field_defn, return_type.non_null?, ast_node, result_name, selection_result)
-              if HALT != continue_value
+              @runtime.current_trace.end_execute_field(@field, @object, @kwarg_arguments, query, app_result)
+              @step = :handle_resolved_value
+              @result = app_result
+            when :handle_resolved_value
+              runtime_state = @runtime.get_current_runtime_state
+              runtime_state.current_field = @field
+              runtime_state.current_arguments = @resolved_arguments
+              runtime_state.current_result_name = @result_name
+              runtime_state.current_result = @selection_result
+              return_type = @field.type
+              @result = @runtime.continue_value(@result, @field, return_type.non_null?, @ast_node, @result_name, @selection_result)
+
+              if !HALT.equal?(@result)
+                runtime_state = @runtime.get_current_runtime_state
                 was_scoped = runtime_state.was_authorized_by_scope_items
                 runtime_state.was_authorized_by_scope_items = nil
-                continue_field(continue_value, owner_type, field_defn, return_type, ast_node, next_selections, false, object, resolved_arguments, result_name, selection_result, was_scoped, runtime_state)
+                @runtime.continue_field(@result, @field, return_type, @ast_node, @next_selections, false, @resolved_arguments, @result_name, @selection_result, was_scoped, runtime_state)
               else
                 nil
               end
+              @step = :finished
+              nil
+            else
+              raise "Invariant: unexpected #{self.class} step: #{@step.inspect} (#{inspect_step})"
             end
           end
-          # If this field is a root mutation field, immediately resolve
-          # all of its child fields before moving on to the next root mutation field.
-          # (Subselections of this mutation will still be resolved level-by-level.)
-          if selection_result.graphql_is_eager
-            Interpreter::Resolve.resolve_all([field_result], @dataloader)
-          end
         end
 
         def set_result(selection_result, result_name, value, is_child_result, is_non_null)
@@ -654,7 +701,7 @@ def continue_value(value, field, is_non_null, ast_node, result_name, selection_r
         # Location information from `path` and `ast_node`.
         #
         # @return [Lazy, Array, Hash, Object] Lazy, Array, and Hash are all traversed to resolve lazy values later
-        def continue_field(value, owner_type, field, current_type, ast_node, next_selections, is_non_null, owner_object, arguments, result_name, selection_result, was_scoped, runtime_state) # rubocop:disable Metrics/ParameterLists
+        def continue_field(value, field, current_type, ast_node, next_selections, is_non_null, arguments, result_name, selection_result, was_scoped, runtime_state) # rubocop:disable Metrics/ParameterLists
           if current_type.non_null?
             current_type = current_type.of_type
             is_non_null = true
@@ -671,136 +718,87 @@ def continue_field(value, owner_type, field, current_type, ast_node, next_select
             end
             set_result(selection_result, result_name, r, false, is_non_null)
             r
-          when "UNION", "INTERFACE"
-            resolved_type_or_lazy = begin
-                                      resolve_type(current_type, value)
-                                    rescue GraphQL::ExecutionError, GraphQL::UnauthorizedError => ex_err
-                                      return continue_value(ex_err, field, is_non_null, ast_node, result_name, selection_result)
-                                    rescue StandardError => err
-                                      begin
-                                        query.handle_or_reraise(err)
-                                      rescue GraphQL::ExecutionError => ex_err
-                                        return continue_value(ex_err, field, is_non_null, ast_node, result_name, selection_result)
-                                      end
-                                    end
-            after_lazy(resolved_type_or_lazy, ast_node: ast_node, field: field, owner_object: owner_object, arguments: arguments, trace: false, result_name: result_name, result: selection_result, runtime_state: runtime_state) do |resolved_type_result, runtime_state|
-              if resolved_type_result.is_a?(Array) && resolved_type_result.length == 2
-                resolved_type, resolved_value = resolved_type_result
-              else
-                resolved_type = resolved_type_result
-                resolved_value = value
-              end
-
-              possible_types = query.types.possible_types(current_type)
-              if !possible_types.include?(resolved_type)
-                parent_type = field.owner_type
-                err_class = current_type::UnresolvedTypeError
-                type_error = err_class.new(resolved_value, field, parent_type, resolved_type, possible_types)
-                schema.type_error(type_error, context)
-                set_result(selection_result, result_name, nil, false, is_non_null)
-                nil
-              else
-                continue_field(resolved_value, owner_type, field, resolved_type, ast_node, next_selections, is_non_null, owner_object, arguments, result_name, selection_result, was_scoped, runtime_state)
-              end
-            end
-          when "OBJECT"
-            object_proxy = begin
-              was_scoped ? current_type.wrap_scoped(value, context) : current_type.wrap(value, context)
-            rescue GraphQL::ExecutionError => err
-              err
-            end
-            after_lazy(object_proxy, ast_node: ast_node, field: field, owner_object: owner_object, arguments: arguments, trace: false, result_name: result_name, result: selection_result, runtime_state: runtime_state) do |inner_object, runtime_state|
-              continue_value = continue_value(inner_object, field, is_non_null, ast_node, result_name, selection_result)
-              if HALT != continue_value
-                response_hash = GraphQLResultHash.new(result_name, current_type, continue_value, selection_result, is_non_null, next_selections, false, ast_node, arguments, field)
-                set_result(selection_result, result_name, response_hash, true, is_non_null)
-                each_gathered_selections(response_hash) do |selections, is_selection_array, ordered_result_keys|
-                  response_hash.ordered_result_keys ||= ordered_result_keys
-                  if is_selection_array
-                    this_result = GraphQLResultHash.new(result_name, current_type, continue_value, selection_result, is_non_null, selections, false, ast_node, arguments, field)
-                    this_result.ordered_result_keys = ordered_result_keys
-                    final_result = response_hash
-                  else
-                    this_result = response_hash
-                    final_result = nil
-                  end
-
-                  evaluate_selections(
-                    selections,
-                    this_result,
-                    final_result,
-                    runtime_state,
-                  )
-                end
-              end
-            end
+          when "OBJECT", "UNION", "INTERFACE"
+            response_hash = GraphQLResultHash.new(self, result_name, current_type, value, selection_result, is_non_null, next_selections, false, ast_node, arguments, field)
+            response_hash.was_scoped = was_scoped
+            @run_queue.append_step response_hash
           when "LIST"
-            inner_type = current_type.of_type
-            # This is true for objects, unions, and interfaces
-            use_dataloader_job = !inner_type.unwrap.kind.input?
-            inner_type_non_null = inner_type.non_null?
-            response_list = GraphQLResultArray.new(result_name, current_type, owner_object, selection_result, is_non_null, next_selections, false, ast_node, arguments, field)
+            response_list = GraphQLResultArray.new(self, result_name, current_type, value, selection_result, is_non_null, next_selections, false, ast_node, arguments, field)
             set_result(selection_result, result_name, response_list, true, is_non_null)
-            idx = nil
-            list_value = begin
-              begin
-                value.each do |inner_value|
-                  idx ||= 0
-                  this_idx = idx
-                  idx += 1
-                  if use_dataloader_job
-                    @dataloader.append_job do
-                      resolve_list_item(inner_value, inner_type, inner_type_non_null, ast_node, field, owner_object, arguments, this_idx, response_list, owner_type, was_scoped, runtime_state)
-                    end
-                  else
-                    resolve_list_item(inner_value, inner_type, inner_type_non_null, ast_node, field, owner_object, arguments, this_idx, response_list, owner_type, was_scoped, runtime_state)
-                  end
-                end
+            @run_queue.append_step(response_list)
+            response_list # TODO smell this is used because its returned by `yield` inside a directive
+          else
+            raise "Invariant: Unhandled type kind #{current_type.kind} (#{current_type})"
+          end
+        end
 
-                response_list
-              rescue NoMethodError => err
-                # Ruby 2.2 doesn't have NoMethodError#receiver, can't check that one in this case. (It's been EOL since 2017.)
-                if err.name == :each && (err.respond_to?(:receiver) ? err.receiver == value : true)
-                  # This happens when the GraphQL schema doesn't match the implementation. Help the dev debug.
-                  raise ListResultFailedError.new(value: value, field: field, path: current_path)
-                else
-                  # This was some other NoMethodError -- let it bubble to reveal the real error.
-                  raise
-                end
-              rescue GraphQL::ExecutionError, GraphQL::UnauthorizedError => ex_err
-                ex_err
-              rescue StandardError => err
-                begin
-                  query.handle_or_reraise(err)
-                rescue GraphQL::ExecutionError => ex_err
-                  ex_err
-                end
-              end
+        class ListItemStep
+          def initialize(runtime, list_result, index, item_value)
+            @runtime = runtime
+            @list_result = list_result
+            @index = index
+            @item_value = item_value
+            @step = :check_directives
+          end
+
+          def step_finished?
+            @step == :finished
+          end
+
+          def inspect_step
+            "#{self.class.name.split("::").last}##{object_id}@#{@index}"
+          end
+
+          def value # Lazy API
+            @item_value = begin
+              @runtime.schema.sync_lazy(@item_value)
+            rescue GraphQL::ExecutionError => err
+              err
             rescue StandardError => err
               begin
-                query.handle_or_reraise(err)
+                @runtime.query.handle_or_reraise(err)
               rescue GraphQL::ExecutionError => ex_err
                 ex_err
               end
             end
-            # Detect whether this error came while calling `.each` (before `idx` is set) or while running list *items* (after `idx` is set)
-            error_is_non_null = idx.nil? ? is_non_null : inner_type.non_null?
-            continue_value(list_value, field, error_is_non_null, ast_node, result_name, selection_result)
-          else
-            raise "Invariant: Unhandled type kind #{current_type.kind} (#{current_type})"
           end
-        end
 
-        def resolve_list_item(inner_value, inner_type, inner_type_non_null, ast_node, field, owner_object, arguments, this_idx, response_list, owner_type, was_scoped, runtime_state) # rubocop:disable Metrics/ParameterLists
-          runtime_state.current_result_name = this_idx
-          runtime_state.current_result = response_list
-          call_method_on_directives(:resolve_each, owner_object, ast_node.directives) do
-            # This will update `response_list` with the lazy
-            after_lazy(inner_value, ast_node: ast_node, field: field, owner_object: owner_object, arguments: arguments, result_name: this_idx, result: response_list, runtime_state: runtime_state) do |inner_inner_value, runtime_state|
-              continue_value = continue_value(inner_inner_value, field, inner_type_non_null, ast_node, this_idx, response_list)
-              if HALT != continue_value
-                continue_field(continue_value, owner_type, field, inner_type, ast_node, response_list.graphql_selections, false, owner_object, arguments, this_idx, response_list, was_scoped, runtime_state)
+          def depth
+            @list_result.depth + 1
+          end
+
+          def run_step
+            case @step
+            when :check_directives
+              if (dirs = @list_result.ast_node.directives).any?
+                @step = :finished
+              runtime_state = @runtime.get_current_runtime_state
+              runtime_state.current_result_name = @index
+              runtime_state.current_result = @list_result
+                @runtime.call_method_on_directives(:resolve_each, @list_result.graphql_application_value, dirs) do
+                  @step = :check_lazy_item
+                end
+              else
+                @step = :check_lazy_item
               end
+            when :check_lazy_item
+              @step = :handle_item
+              if @runtime.lazy?(@item_value)
+                @item_value
+              else
+                nil
+              end
+            when :handle_item
+              item_type = @list_result.graphql_result_type.of_type
+              item_type_non_null = item_type.non_null?
+              continue_value = @runtime.continue_value(@item_value, @list_result.graphql_field, item_type_non_null, @list_result.ast_node, @index, @list_result)
+              if !HALT.equal?(continue_value)
+                was_scoped = false # TODO!!
+                @runtime.continue_field(continue_value, @list_result.graphql_field, item_type, @list_result.ast_node, @list_result.graphql_selections, false, @list_result.graphql_arguments, @index, @list_result, was_scoped, @runtime.get_current_runtime_state)
+              end
+              @step = :finished
+            else
+              raise "Invariant: unexpected step: #{inspect_step}"
             end
           end
         end
diff --git a/lib/graphql/execution/interpreter/runtime/graphql_result.rb b/lib/graphql/execution/interpreter/runtime/graphql_result.rb
index fdba7911bb..034cd6c499 100644
--- a/lib/graphql/execution/interpreter/runtime/graphql_result.rb
+++ b/lib/graphql/execution/interpreter/runtime/graphql_result.rb
@@ -5,7 +5,8 @@ module Execution
     class Interpreter
       class Runtime
         module GraphQLResult
-          def initialize(result_name, result_type, application_value, parent_result, is_non_null_in_parent, selections, is_eager, ast_node, graphql_arguments, graphql_field) # rubocop:disable Metrics/ParameterLists
+          def initialize(runtime_instance, result_name, result_type, application_value, parent_result, is_non_null_in_parent, selections, is_eager, ast_node, graphql_arguments, graphql_field) # rubocop:disable Metrics/ParameterLists
+            @runtime = runtime_instance
             @ast_node = ast_node
             @graphql_arguments = graphql_arguments
             @graphql_field = graphql_field
@@ -22,6 +23,7 @@ def initialize(result_name, result_type, application_value, parent_result, is_no
             @graphql_selections = selections
             @graphql_is_eager = is_eager
             @base_path = nil
+            @graphql_depth = nil
           end
 
           # TODO test full path in Partial
@@ -51,13 +53,265 @@ def build_path(path_array)
         end
 
         class GraphQLResultHash
-          def initialize(_result_name, _result_type, _application_value, _parent_result, _is_non_null_in_parent, _selections, _is_eager, _ast_node, _graphql_arguments, graphql_field) # rubocop:disable Metrics/ParameterLists
+
+          def initialize(_runtime_inst, _result_name, _result_type, _application_value, _parent_result, _is_non_null_in_parent, _selections, _is_eager, _ast_node, _graphql_arguments, graphql_field) # rubocop:disable Metrics/ParameterLists
             super
             @graphql_result_data = {}
             @ordered_result_keys = nil
+            @target_result = nil
+            @was_scoped = nil
+            @resolve_type_result = nil
+            @authorized_check_result = nil
+            if @graphql_result_type.kind.object?
+              @step = :authorize_application_value
+            else
+              @step = :resolve_abstract_type
+            end
+          end
+
+          def set_step(new_step)
+            @step = new_step
+          end
+
+          def depth
+            @graphql_depth ||= begin
+              parent_depth = @graphql_parent ? @graphql_parent.depth : 0
+              parent_depth + 1
+            end
+          end
+
+          def inspect_step
+            "#{self.class.name.split("::").last}##{object_id}(#{@graphql_result_type.to_type_signature} @ #{path.join(".")}})"
           end
 
-          attr_accessor :ordered_result_keys
+          def step_finished?
+            @step == :finished
+          end
+
+          def value
+            if @resolve_type_result
+              @resolve_type_result = @runtime.schema.sync_lazy(@resolve_type_result)
+            elsif @authorized_check_result
+              @runtime.current_trace.begin_authorized(@graphql_result_type, @graphql_application_value, @runtime.context)
+              @runtime.current_trace.authorized_lazy(query: @runtime.query, type: @graphql_result_type, object: @graphql_application_value) do
+                @authorized_check_result = @runtime.schema.sync_lazy(@authorized_check_result)
+                @runtime.current_trace.end_authorized(@graphql_result_type, @graphql_application_value, @runtime.context, @authorized_check_result)
+              end
+            else
+              @graphql_application_value = @runtime.schema.sync_lazy(@graphql_application_value)
+            end
+          end
+
+          def resolve_abstract_type
+            current_type = @graphql_result_type
+            value = @graphql_application_value
+            @resolve_type_result = begin
+              @runtime.resolve_type(current_type, value)
+            rescue GraphQL::ExecutionError, GraphQL::UnauthorizedError => ex_err
+              return @runtime.continue_value(ex_err, @graphql_field, @graphql_is_non_null_in_parent, @ast_node, @graphql_result_name, @graphql_parent)
+            rescue StandardError => err
+              begin
+                @runtime.query.handle_or_reraise(err)
+              rescue GraphQL::ExecutionError => ex_err
+                return @runtime.continue_value(ex_err, @graphql_field, @graphql_is_non_null_in_parent, @ast_node, @graphql_result_name, @graphql_parent)
+              end
+            end
+          end
+
+          def handle_resolved_type
+            if @resolve_type_result.is_a?(Array) && @resolve_type_result.length == 2
+              resolved_type, resolved_value = @resolve_type_result
+            else
+              resolved_type = @resolve_type_result
+              resolved_value = value
+            end
+            @resolve_type_result = nil
+            current_type = @graphql_result_type
+            possible_types = @runtime.query.types.possible_types(current_type)
+            if !possible_types.include?(resolved_type)
+              field = @graphql_field
+              parent_type = field.owner_type
+              err_class = current_type::UnresolvedTypeError
+              type_error = err_class.new(resolved_value, field, parent_type, resolved_type, possible_types)
+              @runtime.schema.type_error(type_error, @runtime.context)
+              @runtime.set_result(self, @result_name, nil, false, is_non_null)
+              nil
+            else
+              @graphql_result_type = resolved_type
+            end
+          end
+
+          def wrap_application_value
+            @graphql_application_value = begin
+              value = @graphql_application_value
+              context = @runtime.context
+              @was_scoped ? @graphql_result_type.wrap_scoped(value, context) : @graphql_result_type.wrap(value, context)
+            rescue GraphQL::ExecutionError => err
+              err
+            end
+            @step = :handle_wrapped_application_value
+            @graphql_application_value
+          end
+
+          def authorize_application_value
+            @runtime.current_trace.begin_authorized(@graphql_result_type, @graphql_application_value, @runtime.context)
+            begin
+              @authorized_check_result = @runtime.current_trace.authorized(query: @runtime.query, type: @graphql_result_type, object: @graphql_application_value) do
+                begin
+                  @graphql_result_type.authorized?(@graphql_application_value, @runtime.context)
+                rescue GraphQL::UnauthorizedError => err
+                  @runtime.schema.unauthorized_object(err)
+                rescue StandardError => err
+                  @runtime.query.handle_or_reraise(err)
+                end
+              end
+            ensure
+              @step = :handle_authorized_application_value
+              @runtime.current_trace.end_authorized(@graphql_result_type, @graphql_application_value, @runtime.context, @authorized_check_result)
+            end
+          end
+
+          def run_step
+            case @step
+            when :resolve_abstract_type
+              @step = :handle_resolved_type
+              resolve_abstract_type
+            when :handle_resolved_type
+              handle_resolved_type
+              authorize_application_value
+            when :authorize_application_value
+              # TODO skip if scoped
+              authorize_application_value
+            when :handle_authorized_application_value
+              # TODO doesn't actually support `.wrap`
+              if @authorized_check_result
+                # TODO don't call `scoped_new here`
+                @graphql_application_value = @graphql_result_type.scoped_new(@graphql_application_value, @runtime.context)
+              else
+                # It failed the authorization check, so go to the schema's authorized object hook
+                err = GraphQL::UnauthorizedError.new(object: @graphql_application_value, type: @graphql_result_type, context: @runtime.context)
+                # If a new value was returned, wrap that instead of the original value
+                begin
+                  new_obj = @runtime.schema.unauthorized_object(err)
+                  if new_obj
+                    # TODO don't do this work-around
+                    @graphql_application_value = @graphql_result_type.scoped_new(new_obj, @runtime.context)
+                  else
+                    @graphql_application_value = nil
+                  end
+                end
+              end
+              @authorized_check_result = nil
+              @step = :handle_wrapped_application_value
+            when :handle_wrapped_application_value
+              @graphql_application_value = @runtime.continue_value(@graphql_application_value, @graphql_field, @graphql_is_non_null_in_parent, @ast_node, @graphql_result_name, @graphql_parent)
+              if HALT.equal?(@graphql_application_value)
+                @step = :finished
+                return
+              elsif @graphql_parent
+                @runtime.set_result(@graphql_parent, @graphql_result_name, self, true, @graphql_is_non_null_in_parent)
+              end
+              @step = :run_selections
+            when :run_selections
+              @runtime.each_gathered_selections(self) do |gathered_selections, is_selection_array, ordered_result_keys|
+                @ordered_result_keys ||= ordered_result_keys
+                if is_selection_array
+                  selections_result = GraphQLResultHash.new(
+                    @runtime,
+                    @graphql_result_name,
+                    @graphql_result_type,
+                    @graphql_application_value,
+                    @graphql_parent,
+                    @graphql_is_non_null_in_parent,
+                    gathered_selections,
+                    @graphql_is_eager,
+                    @ast_node,
+                    @graphql_arguments,
+                    @graphql_field)
+                  selections_result.target_result = self
+                  selections_result.ordered_result_keys = ordered_result_keys
+                  selections_result.set_step :run_selection_directives
+                  @runtime.run_queue.append_step(selections_result)
+                  @step = :finished # Continuing from others now -- could actually reuse this instance for the first one tho
+                else
+                  selections_result = self
+                  @target_result = nil
+                  @graphql_selections = gathered_selections
+                  # TODO extract these substeps out into methods, call that method directly
+                  @step = :run_selection_directives
+                end
+                runtime_state = @runtime.get_current_runtime_state
+                runtime_state.current_result_name = nil
+                runtime_state.current_result = selections_result
+                nil
+              end
+            when :run_selection_directives
+              if (directives = @graphql_selections[:graphql_directives])
+                @graphql_selections.delete(:graphql_directives)
+                @step = :finished # some way to detect whether the block below is called or not
+                @runtime.call_method_on_directives(:resolve, @graphql_application_value, directives) do
+                  @step = :call_each_field
+                end
+              else
+                # TODO some way to continue without this step
+                @step = :call_each_field
+              end
+            when :call_each_field
+              @graphql_selections.each do |result_name, field_ast_nodes_or_ast_node|
+                # Field resolution may pause the fiber,
+                # so it wouldn't get to the `Resolve` call that happens below.
+                # So instead trigger a run from this outer context.
+                if @graphql_is_eager
+                  prev_queue = @runtime.run_queue
+                  @runtime.run_queue = RunQueue.new(runtime: @runtime)
+                  @runtime.dataloader.clear_cache
+                  @runtime.dataloader.run_isolated {
+                    evaluate_selection(
+                      result_name, field_ast_nodes_or_ast_node
+                    )
+                    @runtime.dataloader.clear_cache
+                  }
+                  @runtime.run_queue.complete(eager: true)
+                  @runtime.run_queue = prev_queue
+                else
+                  @runtime.dataloader.append_job {
+                    evaluate_selection(
+                      result_name, field_ast_nodes_or_ast_node
+                    )
+                  }
+                end
+              end
+
+              if @target_result
+                self.merge_into(@target_result)
+              end
+              @step = :finished
+            else
+              raise "Invariant: invalid state for #{self.class}: #{@step}"
+            end
+          end
+
+          def evaluate_selection(result_name, field_ast_nodes_or_ast_node) # rubocop:disable Metrics/ParameterLists
+            return if @graphql_dead
+
+            # Don't create the list of nodes if there's only one node
+            if field_ast_nodes_or_ast_node.is_a?(Array)
+              field_ast_nodes = field_ast_nodes_or_ast_node
+              ast_node = field_ast_nodes.first
+            else
+              field_ast_nodes = nil
+              ast_node = field_ast_nodes_or_ast_node
+            end
+
+            field_name = ast_node.name
+            owner_type = @graphql_result_type
+            field_defn = @runtime.query.types.field(owner_type, field_name)
+
+            resolve_field_step = FieldResolveStep.new(@runtime, field_defn, ast_node, field_ast_nodes, result_name, self)
+            @runtime.run_queue.append_step(resolve_field_step)
+          end
+
+          attr_accessor :ordered_result_keys, :target_result, :was_scoped
 
           include GraphQLResult
 
@@ -162,11 +416,75 @@ def fix_result_order
         class GraphQLResultArray
           include GraphQLResult
 
-          def initialize(_result_name, _result_type, _application_value, _parent_result, _is_non_null_in_parent, _selections, _is_eager, _ast_node, _graphql_arguments, graphql_field) # rubocop:disable Metrics/ParameterLists
+          def initialize(_runtime_inst, _result_name, _result_type, _application_value, _parent_result, _is_non_null_in_parent, _selections, _is_eager, _ast_node, _graphql_arguments, graphql_field) # rubocop:disable Metrics/ParameterLists
             super
             @graphql_result_data = []
           end
 
+          def inspect_step
+            "#{self.class.name.split("::").last}##{object_id}(#{@graphql_result_type.to_type_signature} @ #{path.join(".")})"
+          end
+
+          def depth
+            @graphql_depth ||= @graphql_parent.depth + 1
+          end
+
+          def step_finished?
+            true
+          end
+
+          def run_step
+            current_type = @graphql_result_type
+            inner_type = current_type.of_type
+            # This is true for objects, unions, and interfaces
+            # use_dataloader_job = !inner_type.unwrap.kind.input?
+            idx = nil
+            list_value = begin
+              begin
+                @graphql_application_value.each do |inner_value|
+                  idx ||= 0
+                  this_idx = idx
+                  idx += 1
+                  list_item_step = ListItemStep.new(
+                    @runtime,
+                    self,
+                    this_idx,
+                    inner_value,
+                  )
+                  @runtime.run_queue.append_step(list_item_step)
+                end
+
+                self
+              rescue NoMethodError => err
+                # Ruby 2.2 doesn't have NoMethodError#receiver, can't check that one in this case. (It's been EOL since 2017.)
+                if err.name == :each && (err.respond_to?(:receiver) ? err.receiver == @graphql_application_value : true)
+                  # This happens when the GraphQL schema doesn't match the implementation. Help the dev debug.
+                  raise ListResultFailedError.new(value: @graphql_application_value, field: @graphql_field, path: @runtime.current_path)
+                else
+                  # This was some other NoMethodError -- let it bubble to reveal the real error.
+                  raise
+                end
+              rescue GraphQL::ExecutionError, GraphQL::UnauthorizedError => ex_err
+                ex_err
+              rescue StandardError => err
+                begin
+                  @runtime.query.handle_or_reraise(err)
+                rescue GraphQL::ExecutionError => ex_err
+                  ex_err
+                end
+              end
+            rescue StandardError => err
+              begin
+                @runtime.query.handle_or_reraise(err)
+              rescue GraphQL::ExecutionError => ex_err
+                ex_err
+              end
+            end
+            # Detect whether this error came while calling `.each` (before `idx` is set) or while running list *items* (after `idx` is set)
+            error_is_non_null = idx.nil? ? @graphql_is_non_null_in_parent : inner_type.non_null?
+            @runtime.continue_value(list_value, @graphql_field, error_is_non_null, @ast_node, @graphql_result_name, @graphql_parent)
+          end
+
           def graphql_skip_at(index)
             # Mark this index as dead. It's tricky because some indices may already be storing
             # `Lazy`s. So the runtime is still holding indexes _before_ skipping,
diff --git a/lib/graphql/schema/object.rb b/lib/graphql/schema/object.rb
index e1f5a99f09..ad337f21de 100644
--- a/lib/graphql/schema/object.rb
+++ b/lib/graphql/schema/object.rb
@@ -43,6 +43,8 @@ def wrap_scoped(object, context)
           scoped_new(object, context)
         end
 
+        # TODO Runtime calls to `.wrap` have been removed, maybe clean this up
+
         # This is called by the runtime to return an object to call methods on.
         def wrap(object, context)
           authorized_new(object, context)
diff --git a/spec/graphql/schema/directive_spec.rb b/spec/graphql/schema/directive_spec.rb
index 6b1d7b1faf..f423f20963 100644
--- a/spec/graphql/schema/directive_spec.rb
+++ b/spec/graphql/schema/directive_spec.rb
@@ -337,9 +337,12 @@ def self.resolve_each(object, args, context)
           end
         end
 
-        def self.resolve(obj, args, ctx)
+        def self.resolve(object, arguments, context)
           value = yield
-          value.values.compact!
+          # Previously, `yield` returned a finished value. But it doesn't anymore.
+          runtime_instance = context.namespace(:interpreter_runtime)[:runtime]
+          runtime_instance.run_queue.complete
+          value.selection_result.values.compact!
           value
         end
       end