diff --git a/cop/development/trace_methods_cop.rb b/cop/development/trace_methods_cop.rb
index ec843a99c50..308033d8849 100644
--- a/cop/development/trace_methods_cop.rb
+++ b/cop/development/trace_methods_cop.rb
@@ -36,6 +36,8 @@ class TraceMethodsCop < RuboCop::Cop::Base
         :execute_query,
         :execute_query_lazy,
         :lex,
+        :object_loaded,
+        :objects,
         :parse,
         :resolve_type,
         :resolve_type_lazy,
@@ -70,7 +72,9 @@ def on_module(node)
             :lex, :analyze_query, :execute_query, :execute_query_lazy,
             # Only useful for isolated event tracking:
             :begin_dataloader, :end_dataloader,
-            :dataloader_fiber_exit, :dataloader_spawn_execution_fiber, :dataloader_spawn_source_fiber
+            :dataloader_fiber_exit, :dataloader_spawn_execution_fiber, :dataloader_spawn_source_fiber,
+            # Tracks object references, but not durations:
+            :objects, :object_loaded
           ]
           missing_defs.each do |missing_def|
             if all_defs.include?(:"begin_#{missing_def}") && all_defs.include?(:"end_#{missing_def}")
diff --git a/guides/_plugins/api_doc.rb b/guides/_plugins/api_doc.rb
index ab927cafe8c..8a665479568 100644
--- a/guides/_plugins/api_doc.rb
+++ b/guides/_plugins/api_doc.rb
@@ -131,9 +131,12 @@ def render(context)
         end
         text = h.gsub(/^#+ /, "")
         target = text.downcase
+          .gsub("🟡", "00emoji00")
+          .gsub("❌", "00emoji00")
           .gsub(/[^a-z0-9_]+/, "-")
           .sub(/-$/, "")
           .sub(/^-/, "")
+          .gsub("-00emoji00", "-")
 
         rendered_text = Kramdown::Document.new(text, auto_ids: false)
           .to_html
diff --git a/guides/execution/migration.md b/guides/execution/migration.md
index 498d3652016..fef35c5d76e 100644
--- a/guides/execution/migration.md
+++ b/guides/execution/migration.md
@@ -195,9 +195,23 @@ This is not supported because the new runtime doesn't actually produce `current_
 
 It is theoretically possible to support this but it will be a ton of work. If you use this for core runtime functions, please share your use case in a GitHub issue and we can investigate future options.
 
+### Scoped context ❌
+
+This is currently implemented with `current_path`. Another implementation is probably possible but not implemented yet. Please open an issue to discuss.
+
 ### `@defer` and `@stream` ❌
 
-This depends on `current_path` so isn't possible yet.
+`@defer` is supported with an implementation difference that _probably_ doesn't affect your application: previously, `@defer` worked by pausing and resuming the _same `GraphQL::Query` instance_. However, with `Execution::Next`, `@defer` takes a different approach. Instead, when a `GraphQL::Query` encounters `@defer`, it notes the location in the document and stops executing that branch. Later, when you request the deferred result, that branch of the query is resumed using a new instance of `GraphQL::Query::Partial`.
+
+This might matter if you're modifying `context` at runtime because those new instances _also_ have fresh `Query::Context` instances. The original query context _will_ get copied into the `@defer` branches using `Query::Context.new(**original_query.context.to_h)`, so any custom values will be available. But if you _assign new keys_ after the context is copied, those keys won't appear when running later `@defer`ed branches.
+
+To handle this, you can refactor how you accumulate data during execution. Instead of `||=`'ing into `context[...]` during execution, assign a new accumulator object _before_ starting the query, then call methods on that object to make any necessary state changes. That new object _will_ be copied into `@defer` partials, and since the object is shared between the different branches, any necessary state changes will still be "seen" everywhere.
+
+If this gives you trouble, please feel free to email me or open an issue on GitHub to discuss a migration strategy.
+
+##### GraphQL-Batch support
+
+When using `Execution::Next`, no custom code is required to support `graphql-batch`.
 
 ### ObjectCache ❌
 
@@ -242,12 +256,13 @@ Resolver classes are called.
 
 ### `raw_value` 🟡
 
-Supported but requires a manual opt-in at schema level. Support for this will probably get better somehow in a future version.
+Supported, but the `raw_value` call must be made on `context`, for example:
 
 ```ruby
-class MyAppSchema < GraphQL::Schema
-  uses_raw_value(true) # TODO This configuration will be improved in a future GraphQL-Ruby version
-  use GraphQL::Execution::Next
+field :values, SomeObjectType, resolve_static: true
+
+def self.values(context)
+  context.raw_value(...)
 end
 ```
 
diff --git a/lib/graphql.rb b/lib/graphql.rb
index bdc734178b2..110a40a54d7 100644
--- a/lib/graphql.rb
+++ b/lib/graphql.rb
@@ -21,9 +21,6 @@ def self.eager_load!
   class Error < StandardError
   end
 
-  class RuntimeError < Error
-  end
-
   # This error is raised when GraphQL-Ruby encounters a situation
   # that it *thought* would never happen. Please report this bug!
   class InvariantError < Error
@@ -122,6 +119,7 @@ class << self
   autoload :ParseError, "graphql/parse_error"
   autoload :Backtrace, "graphql/backtrace"
 
+  autoload :RuntimeError, "graphql/runtime_error"
   autoload :UnauthorizedError, "graphql/unauthorized_error"
   autoload :UnauthorizedEnumValueError, "graphql/unauthorized_enum_value_error"
   autoload :UnauthorizedFieldError, "graphql/unauthorized_field_error"
diff --git a/lib/graphql/analysis.rb b/lib/graphql/analysis.rb
index b90ebf8e255..36f14019f8f 100644
--- a/lib/graphql/analysis.rb
+++ b/lib/graphql/analysis.rb
@@ -40,13 +40,13 @@ def analyze_multiplex(multiplex, analyzers)
           end
         end
 
-        multiplex_results = multiplex_analyzers.map(&:result)
-        multiplex_errors = analysis_errors(multiplex_results)
 
+        multiplex_analyzers.map!(&:result)
+        multiplex_errors = analysis_errors(EmptyObjects::EMPTY_ARRAY, multiplex_analyzers)
         multiplex.queries.each_with_index do |query, idx|
-          query.analysis_errors = multiplex_errors + analysis_errors(query_results[idx])
+          query.analysis_errors = analysis_errors(multiplex_errors, query_results[idx])
         end
-        multiplex_results
+        multiplex_analyzers
       end
     end
 
@@ -55,13 +55,11 @@ def analyze_multiplex(multiplex, analyzers)
     # @return [Array<Any>] Results from those analyzers
     def analyze_query(query, analyzers, multiplex_analyzers: [])
       query.current_trace.analyze_query(query: query) do
-        query_analyzers = analyzers
-          .map { |analyzer| analyzer.new(query) }
-          .tap { _1.select!(&:analyze?) }
-
+        query_analyzers = analyzers.map { |analyzer| analyzer.new(query) }
+        query_analyzers.select!(&:analyze?)
         analyzers_to_run = query_analyzers + multiplex_analyzers
-        if !analyzers_to_run.empty?
 
+        if !analyzers_to_run.empty?
           analyzers_to_run.select!(&:visit?)
           if !analyzers_to_run.empty?
             visitor = GraphQL::Analysis::Visitor.new(
@@ -79,18 +77,27 @@ def analyze_query(query, analyzers, multiplex_analyzers: [])
 
           query_analyzers.map(&:result)
         else
-          []
+          EmptyObjects::EMPTY_ARRAY
         end
       end
     rescue TimeoutError => err
       [err]
     rescue GraphQL::UnauthorizedError, GraphQL::ExecutionError
       # This error was raised during analysis and will be returned the client before execution
-      []
+      EmptyObjects::EMPTY_ARRAY
     end
 
-    def analysis_errors(results)
-      results.flatten.tap { _1.select! { |r| r.is_a?(GraphQL::AnalysisError) } }
+    def analysis_errors(parent_errors, results)
+      if !results.empty?
+        results = results.flatten
+        results.select! { |r| r.is_a?(GraphQL::AnalysisError) }
+      end
+
+      if parent_errors.empty?
+        results
+      else
+        parent_errors + results
+      end
     end
   end
 end
diff --git a/lib/graphql/execution.rb b/lib/graphql/execution.rb
index 5088edb6047..edc2410ac87 100644
--- a/lib/graphql/execution.rb
+++ b/lib/graphql/execution.rb
@@ -1,10 +1,10 @@
 # frozen_string_literal: true
 require "graphql/execution/directive_checks"
+require "graphql/execution/next"
 require "graphql/execution/interpreter"
 require "graphql/execution/lazy"
 require "graphql/execution/lookahead"
 require "graphql/execution/multiplex"
-require "graphql/execution/next"
 require "graphql/execution/errors"
 
 module GraphQL
@@ -14,7 +14,7 @@ class Skip < GraphQL::RuntimeError
       attr_accessor :path
       def ast_nodes=(_ignored); end
 
-      def assign_graphql_result(query, result_data, key)
+      def finalize_graphql_result(query, result_data, key)
         result_data.delete(key)
       end
     end
diff --git a/lib/graphql/execution/interpreter/handles_raw_value.rb b/lib/graphql/execution/interpreter/handles_raw_value.rb
index 21f62eba964..c1cad5ed49f 100644
--- a/lib/graphql/execution/interpreter/handles_raw_value.rb
+++ b/lib/graphql/execution/interpreter/handles_raw_value.rb
@@ -5,6 +5,12 @@ module Execution
     class Interpreter
       # Wrapper for raw values
       class RawValue
+        include GraphQL::Execution::Next::Finalizer
+
+        def finalize_graphql_result(query, result_data, result_key)
+          result_data[result_key] = @object
+        end
+
         def initialize(obj = nil)
           @object = obj
         end
diff --git a/lib/graphql/execution/next.rb b/lib/graphql/execution/next.rb
index 45fdf699d5b..d21b07f5f71 100644
--- a/lib/graphql/execution/next.rb
+++ b/lib/graphql/execution/next.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 require "graphql/execution/next/prepare_object_step"
 require "graphql/execution/next/field_resolve_step"
+require "graphql/execution/next/finalize"
 require "graphql/execution/next/load_argument_step"
 require "graphql/execution/next/runner"
 require "graphql/execution/next/selections_step"
@@ -67,6 +68,22 @@ def self.run_all(schema, query_options, context: {}, max_complexity: schema.max_
         runner = Runner.new(multiplex, **schema.execution_next_options)
         runner.execute
       end
+
+      module Finalizer
+        attr_accessor :path
+        def finalize_graphql_result(query, result_data, result_key)
+          raise RequiredImplementationMissingError
+        end
+      end
+
+      module HaltExecution
+      end
+
+      module PostProcessor
+        def after_resolve(field_results)
+          raise RequiredImplementationMissingError, "#{self.class}#after_resolve should handle `field_results` and return a new value to use"
+        end
+      end
     end
   end
 end
diff --git a/lib/graphql/execution/next/field_resolve_step.rb b/lib/graphql/execution/next/field_resolve_step.rb
index adcd5a5b74c..70b7a68f876 100644
--- a/lib/graphql/execution/next/field_resolve_step.rb
+++ b/lib/graphql/execution/next/field_resolve_step.rb
@@ -22,12 +22,13 @@ def initialize(parent_type:, runner:, key:, selections_step:)
           @finish_extension_idx = nil
           @was_scoped = nil
           @pending_steps = nil
+          @post_processors = @directive_finalizers = nil
         end
 
         attr_reader :ast_node, :key, :parent_type, :selections_step, :runner,
           :field_definition, :object_is_authorized, :was_scoped, :field_results
 
-        attr_accessor :pending_steps, :arguments
+        attr_accessor :pending_steps, :arguments, :static_type
 
         def path
           @path ||= [*@selections_step.path, @key].freeze
@@ -171,7 +172,7 @@ def coerce_argument_value(arguments, arg_defn, arg_value, run_loads, target_keyw
             end
           end
 
-          if arg_value.is_a?(GraphQL::Error)
+          if arg_value.is_a?(GraphQL::RuntimeError)
             @arguments = arg_value
           elsif run_loads && arg_defn.loads && as_type.nil? && !arg_value.nil?
             # This is for legacy compat:
@@ -218,7 +219,7 @@ def coerce_argument_value(arguments, arg_defn, arg_value, run_loads, target_keyw
         def value
           query = @selections_step.query
           query.current_trace.begin_execute_field(@field_definition, @arguments, @field_results, query)
-          @field_results = sync(@field_results)
+          sync(@field_results)
           query.current_trace.end_execute_field(@field_definition, @arguments, @field_results, query, @field_results)
           @runner.add_step(self)
           true
@@ -279,7 +280,7 @@ def self.[](_key)
         def build_arguments
           query = @selections_step.query
           field_name = @ast_node.name
-          @field_definition = query.get_field(@parent_type, field_name) || raise("Invariant: no field found for #{@parent_type.to_type_signature}.#{ast_node.name}")
+          @field_definition = query.types.field(@parent_type, field_name) || raise("Invariant: no field found for #{@parent_type.to_type_signature}.#{ast_node.name}")
           arguments = coerce_arguments(@field_definition, @ast_node.arguments) # rubocop:disable Development/ContextIsPassedCop
           @arguments ||= arguments # may have already been set to an error
 
@@ -292,7 +293,7 @@ def build_arguments
         def execute_field
           objects = @selections_step.objects
           # TODO not as good because only one error?
-          if @arguments.is_a?(GraphQL::Error)
+          if @arguments.is_a?(GraphQL::RuntimeError)
             @field_results = Array.new(objects.size, @arguments)
             @object_is_authorized = AlwaysAuthorized
             build_results
@@ -337,7 +338,6 @@ def execute_field
             # TODO break this backwards compat somehow?
             objects = @selections_step.graphql_objects
           end
-
           if @runner.authorization && @runner.authorizes?(@field_definition, ctx)
             authorized_objects = []
             @object_is_authorized = objects.map { |o|
@@ -362,12 +362,60 @@ def execute_field
             authorized_objects = objects
             @object_is_authorized = AlwaysAuthorized
           end
-
           if @parent_type.default_relay? && authorized_objects.all? { |o| o.respond_to?(:was_authorized_by_scope_items?) && o.was_authorized_by_scope_items? }
             @was_scoped = true
           end
 
           query.current_trace.begin_execute_field(@field_definition, @arguments, authorized_objects, query)
+
+          if @runner.uses_runtime_directives
+            if @ast_nodes.nil? || @ast_nodes.size == 1
+              directives = if !@ast_node.directives.empty?
+                @ast_node.directives
+              else
+                nil
+              end
+            else
+              directives = nil
+              @ast_nodes.each do |n|
+                if (d = n.directives).any? # rubocop:disable Development/NoneWithoutBlockCop
+                  directives ||= []
+                  directives.concat(d)
+                end
+              end
+            end
+
+            if directives
+              directives.each do |dir_node|
+                if (dir_defn = @runner.runtime_directives[dir_node.name])
+                  # TODO: `coerce_arguments` modifies self, assuming it's field arguments. Extract to pure function for use
+                  # here and with fragments.
+                  dir_args = coerce_arguments(dir_defn, dir_node.arguments, false) # rubocop:disable Development/ContextIsPassedCop
+                  result = dir_defn.resolve_field(ast_nodes, @parent_type, field_definition, authorized_objects, dir_args, ctx)
+                  if !result.nil?
+                    if result.is_a?(Finalizer)
+                      result.path = path
+                      @directive_finalizers ||= []
+                      @directive_finalizers << result
+                    end
+
+                    if result.is_a?(PostProcessor)
+                      @post_processors ||= []
+                      @post_processors << result
+                    end
+
+                    if result.is_a?(HaltExecution)
+                      @directive_finalizers&.each { |f|
+                        @selections_step.results.each { |r|  @runner.add_finalizer(query, r, key, f) }
+                      }
+                      return
+                    end
+                  end
+                end
+              end
+            end
+          end
+
           has_extensions = @field_definition.extensions.size > 0
           if has_extensions
             @extended = GraphQL::Schema::Field::ExtendedState.new(@arguments, authorized_objects)
@@ -425,7 +473,7 @@ def finish_extensions
             case ext
             when Schema::Field::ConnectionExtension
               conns = ctx.schema.connections
-              @field_results = @field_results.map.each_with_index do |value, idx|
+              @field_results.map!.each_with_index do |value, idx|
                 object = @extended.object[idx]
                 conn = conns.populate_connection(@field_definition, object, value, @arguments, ctx)
                 if conn
@@ -437,7 +485,7 @@ def finish_extensions
               if @was_scoped.nil?
                 if (rt = @field_definition.type.unwrap).respond_to?(:scope_items)
                   @was_scoped = true
-                  @field_results = @field_results.map { |v| v.nil? ? v : rt.scope_items(v, ctx) }
+                  @field_results.map! { |v| v.nil? ? v : rt.scope_items(v, ctx) }
                 else
                   @was_scoped = false
                 end
@@ -461,6 +509,10 @@ def build_results
           return_type = @field_definition.type
           return_result_type = return_type.unwrap
 
+          @post_processors&.each do |post_processor|
+            @field_results = post_processor.after_resolve(@field_results)
+          end
+
           if return_result_type.kind.composite?
             @static_type = return_result_type
             if @ast_nodes
@@ -514,26 +566,40 @@ def build_results
                 field_result = nil
               end
               i += 1
-              result_h[@key] = if field_result.nil?
-                if return_type.non_null?
-                  add_non_null_error(false)
-                else
-                  nil
-                end
-              elsif field_result.is_a?(GraphQL::Error)
-                add_graphql_error(field_result)
-              else
-                # TODO `nil`s in [T!] types aren't handled
-                return_type.coerce_result(field_result, ctx)
-              end
+              finish_leaf_result(result_h, @key, field_result, return_type, ctx)
+            end
+          end
+        end
+
+        def finish_leaf_result(result_h, key, field_result, return_type, ctx)
+          final_field_result = if field_result.nil?
+            if return_type.non_null?
+              add_non_null_error(false)
+            else
+              nil
+            end
+          elsif field_result.is_a?(Finalizer)
+            if field_result.is_a?(GraphQL::RuntimeError)
+              add_graphql_error(field_result)
+            else
+              field_result.path = path
+              @runner.add_finalizer(ctx.query, result_h, key, field_result)
             end
+          else
+            # TODO `nil`s in [T!] types aren't handled
+            return_type.coerce_result(field_result, ctx)
           end
+
+          @directive_finalizers&.each { |f| @runner.add_finalizer(ctx.query, result_h, key, f) }
+          result_h[@key] = final_field_result
         end
 
         def enqueue_next_steps
           if !@all_next_results.empty?
             @all_next_objects.compact!
 
+            query = @selections_step.query
+            ctx = query.context
             if @static_type.kind.abstract?
               next_objects_by_type = Hash.new { |h, obj_t| h[obj_t] = [] }.compare_by_identity
               next_results_by_type = Hash.new { |h, obj_t| h[obj_t] = [] }.compare_by_identity
@@ -543,7 +609,7 @@ def enqueue_next_steps
                 if (object_type = @runner.runtime_type_at[result])
                   # OK
                 else
-                  object_type = @runner.resolve_type(@static_type, next_object, @selections_step.query)
+                  object_type = @runner.resolve_type(@static_type, next_object, query)
                   @runner.runtime_type_at[result] = object_type
                 end
                 next_objects_by_type[object_type] << next_object
@@ -551,6 +617,7 @@ def enqueue_next_steps
               end
 
               next_objects_by_type.each do |obj_type, next_objects|
+                query.current_trace.objects(obj_type, next_objects, ctx)
                 @runner.add_step(SelectionsStep.new(
                   path: path,
                   parent_type: obj_type,
@@ -558,10 +625,11 @@ def enqueue_next_steps
                   objects: next_objects,
                   results: next_results_by_type[obj_type],
                   runner: @runner,
-                  query: @selections_step.query,
+                  query: query,
                 ))
               end
             else
+              query.current_trace.objects(@static_type, @all_next_objects, ctx)
               @runner.add_step(SelectionsStep.new(
                 path: path,
                 parent_type: @static_type,
@@ -569,7 +637,7 @@ def enqueue_next_steps
                 objects: @all_next_objects,
                 results: @all_next_results,
                 runner: @runner,
-                query: @selections_step.query,
+                query: query,
               ))
             end
           end
@@ -596,8 +664,14 @@ def build_graphql_result(graphql_result, key, field_result, return_type, is_nn,
             else
               graphql_result[key] = nil
             end
-          elsif field_result.is_a?(GraphQL::Error)
-            graphql_result[key] = add_graphql_error(field_result)
+          elsif field_result.is_a?(Finalizer)
+            graphql_result[key] = if field_result.is_a?(GraphQL::RuntimeError)
+              add_graphql_error(field_result)
+            else
+              field_result.path = path
+              @runner.add_finalizer(@selections_step.query, graphql_result, key, field_result)
+              field_result
+            end
           elsif is_list
             if is_nn
               return_type = return_type.of_type
@@ -606,6 +680,7 @@ def build_graphql_result(graphql_result, key, field_result, return_type, is_nn,
             inner_type_nn = inner_type.non_null?
             inner_type_l = inner_type.list?
             list_result = graphql_result[key] = []
+            @directive_finalizers&.each { |f| @runner.add_finalizer(@selections_step.query, list_result, nil, f) }
             i = 0
             s = field_result.size
             while i < s
@@ -618,7 +693,6 @@ def build_graphql_result(graphql_result, key, field_result, return_type, is_nn,
                 ) && @runner.authorizes?(runtime_type, @selections_step.query.context)
               )))
             obj_step = PrepareObjectStep.new(
-              static_type: @static_type,
               object: field_result,
               runner: @runner,
               field_resolve_step: self,
@@ -633,8 +707,9 @@ def build_graphql_result(graphql_result, key, field_result, return_type, is_nn,
             ps << obj_step
             @runner.add_step(obj_step)
           else
-            next_result_h = {}
+            next_result_h = {}.compare_by_identity
             @all_next_results << next_result_h
+            @directive_finalizers&.each { |f| @runner.add_finalizer(@selections_step.query, next_result_h, nil, f) }
             @all_next_objects << field_result
             @runner.static_type_at[next_result_h] = @static_type
             graphql_result[key] = next_result_h
@@ -651,7 +726,11 @@ def resolve_batch(objects, context, args_hash)
               Array.new(objects.size, exec_err)
             end
           when :resolve_static
-            result = method_receiver.public_send(@field_definition.execution_next_mode_key, context, **args_hash)
+            result = begin
+              method_receiver.public_send(@field_definition.execution_next_mode_key, context, **args_hash)
+            rescue GraphQL::ExecutionError => err
+              err
+            end
             Array.new(objects.size, result)
           when :resolve_each
             objects.map do |o|
@@ -660,22 +739,20 @@ def resolve_batch(objects, context, args_hash)
               err
             end
           when :hash_key
-            objects.map { |o| o[@field_definition.execution_next_mode_key] }
+            k = @field_definition.execution_next_mode_key
+            objects.map { |o| o[k] }
           when :direct_send
-            if args_hash.empty?
-              objects.map do |o|
-                o.public_send(@field_definition.execution_next_mode_key)
-              rescue GraphQL::ExecutionError => err
-                err
-              rescue StandardError => stderr
-                begin
-                  @selections_step.query.handle_or_reraise(stderr)
-                rescue GraphQL::ExecutionError => ex_err
-                  ex_err
-                end
+            m = @field_definition.execution_next_mode_key
+            objects.map do |o|
+              o.public_send(m, **args_hash)
+            rescue GraphQL::ExecutionError => err
+              err
+            rescue StandardError => stderr
+              begin
+                @selections_step.query.handle_or_reraise(stderr)
+              rescue GraphQL::ExecutionError => ex_err
+                ex_err
               end
-            else
-              objects.map { |o| o.public_send(@field_definition.execution_next_mode_key, **args_hash) }
             end
           when :dig
             objects.map { |o| o.dig(*@field_definition.execution_next_mode_key) }
@@ -728,16 +805,6 @@ def resolve_batch(objects, context, args_hash)
           end
         end
       end
-
-      class RawValueFieldResolveStep < FieldResolveStep
-        def build_graphql_result(graphql_result, key, field_result, return_type, is_nn, is_list, is_from_array) # rubocop:disable Metrics/ParameterLists
-          if field_result.is_a?(Interpreter::RawValue)
-            graphql_result[key] = field_result.resolve
-          else
-            super
-          end
-        end
-      end
     end
   end
 end
diff --git a/lib/graphql/execution/next/finalize.rb b/lib/graphql/execution/next/finalize.rb
new file mode 100644
index 00000000000..61db6197319
--- /dev/null
+++ b/lib/graphql/execution/next/finalize.rb
@@ -0,0 +1,219 @@
+# frozen_string_literal: true
+module GraphQL
+  module Execution
+    module Next
+      class Finalize
+        def initialize(query, data, runner)
+          @query = query
+          @data = data
+          @static_type_at = runner.static_type_at
+          @runner = runner
+          @current_exec_path = query.path.dup
+          @current_result_path = query.path.dup
+          @finalizers = runner.finalizers ? runner.finalizers[query] : {}.compare_by_identity
+          @finalizers_count = 0
+          @finalizers.each do |key, values|
+            values.each do |key2, values2|
+              case values2
+              when Array
+                @finalizers_count += values2.size
+              else
+                @finalizers_count += 1
+              end
+            end
+          end
+
+          query.context.errors.each do |err|
+            err_path = err.path - @current_exec_path
+            key = err_path.pop
+            targets = [data]
+            while (part = err_path.shift)
+              targets.map! { |t| t[part] }
+              targets.flatten!
+            end
+
+            targets.each_with_index do |target, idx|
+              if target.is_a?(Hash)
+                if target[key].equal?(err)
+                  tf = @finalizers[target] ||= {}.compare_by_identity
+                  tf[key] = err
+                  @finalizers_count += 1
+                elsif (arr = target[key]).is_a?(Array)
+                  arr.each_with_index do |el, idx|
+                    if el.equal?(err)
+                      tf = @finalizers[arr] ||= {}.compare_by_identity
+                      tf[idx] = err
+                      @finalizers_count += 1
+                    end
+                  end
+                end
+              end
+            end
+          end
+        end
+
+        def run
+          if (selected_operation = @query.selected_operation) && @data
+            if @data.is_a?(Hash)
+              check_object_result(@data, @query.root_type, selected_operation.selections)
+            elsif @data.is_a?(Array)
+              check_list_result(@data, @query.root_type, selected_operation.selections)
+            elsif @data.is_a?(Finalizer)
+              dummy_data = {}
+              dummy_key = "__dummy"
+              @data.path = @query.path
+              @data.finalize_graphql_result(@query, dummy_data, dummy_key)
+              dummy_data[dummy_key]
+            else
+              raise ArgumentError, "Unexpected @data: #{@data.inspect}"
+            end
+          else
+            @data
+          end
+        end
+
+        private
+
+        def run_finalizers(result_path, finalizer_or_finalizers, result_data, result_key)
+          if finalizer_or_finalizers.is_a?(Array)
+            finalizer_or_finalizers.each { |f|
+              f.path = result_path
+              f.finalize_graphql_result(@query, result_data, result_key)
+            }
+            @finalizers_count -= finalizer_or_finalizers.size
+          else
+            f = finalizer_or_finalizers
+            f.path = result_path
+            f.finalize_graphql_result(@query, result_data, result_key)
+            @finalizers_count -= 1
+          end
+        end
+
+        def finalizers(result_value, key)
+          finalizers_for_value = @finalizers[result_value]
+          finalizers_for_value && finalizers_for_value[key]
+        end
+
+        def check_object_result(result_h, parent_type, ast_selections)
+          if (f = finalizers(result_h, nil))
+            run_finalizers(@current_result_path.dup, f, result_h, nil)
+            return result_h if @finalizers_count == 0
+          end
+
+          if parent_type.kind.abstract?
+            parent_type = @runner.runtime_type_at[result_h]
+          end
+
+          ast_selections.each do |ast_selection|
+            case ast_selection
+            when Language::Nodes::Field
+              key = ast_selection.alias || ast_selection.name
+              if (f = finalizers(result_h, key))
+                result_value = result_h[key]
+                run_finalizers(@current_result_path.dup << key, f, result_h, key)
+                new_result_value = result_h.key?(key) ? result_h[key] : :unassigned
+              end
+              next if !(f || result_h.key?(key))
+              begin
+                @current_exec_path << key
+                @current_result_path << key
+
+                field_defn = @query.context.types.field(parent_type, ast_selection.name) || raise("Invariant: No field found for #{static_type.to_type_signature}.#{ast_selection.name}")
+                result_type = field_defn.type
+                if (result_type_non_null = result_type.non_null?)
+                  result_type = result_type.of_type
+                end
+
+                if !f
+                  result_value = result_h[key]
+                  new_result_value = if result_type.list? && result_value
+                    check_list_result(result_value, result_type.of_type, ast_selection.selections)
+                  elsif !result_type.kind.leaf? && result_value
+                    check_object_result(result_value, result_type, ast_selection.selections)
+                  else
+                    result_value
+                  end
+                end
+
+                if new_result_value.nil? && result_type_non_null
+                  return nil
+                elsif :unassigned.equal?(new_result_value)
+                  # Do nothing
+                  break if @finalizers_count == 0
+                elsif !new_result_value.equal?(result_value)
+                  result_h[key] = new_result_value
+                  break if @finalizers_count == 0
+                end
+              ensure
+                @current_exec_path.pop
+                @current_result_path.pop
+              end
+            when Language::Nodes::InlineFragment
+              static_type_at_result = @static_type_at[result_h]
+              if static_type_at_result && (
+                  (t = ast_selection.type).nil? ||
+                  @runner.type_condition_applies?(@query.context, static_type_at_result, t.name)
+                )
+                result_h = check_object_result(result_h, parent_type, ast_selection.selections)
+              end
+            when Language::Nodes::FragmentSpread
+              fragment_defn = @query.document.definitions.find { |defn| defn.is_a?(Language::Nodes::FragmentDefinition) && defn.name == ast_selection.name }
+              static_type_at_result = @static_type_at[result_h]
+              if static_type_at_result && @runner.type_condition_applies?(@query.context, static_type_at_result, fragment_defn.type.name)
+                result_h = check_object_result(result_h, parent_type, fragment_defn.selections)
+              end
+            end
+          end
+
+          result_h
+        end
+
+        def check_list_result(result_arr, inner_type, ast_selections)
+          inner_type_non_null = false
+          if inner_type.non_null?
+            inner_type_non_null = true
+            inner_type = inner_type.of_type
+          end
+
+          new_invalid_null = false
+
+          if (f = finalizers(result_arr, nil))
+            run_finalizers(@current_result_path.dup, f, result_arr, nil)
+            return result_arr if @finalizers_count == 0
+          end
+
+          result_arr.each_with_index do |result_item, idx|
+            @current_result_path << idx
+            new_result = if (f = finalizers(result_arr, idx))
+              run_finalizers(@current_result_path.dup, f, result_arr, idx)
+              result_arr[idx]
+            elsif inner_type.list? && result_item
+              check_list_result(result_item, inner_type.of_type, ast_selections)
+            elsif !inner_type.kind.leaf? && result_item
+              check_object_result(result_item, inner_type, ast_selections)
+            else
+              result_item
+            end
+
+            if new_result.nil? && inner_type_non_null
+              new_invalid_null = true
+              result_arr[idx] = nil
+              break if @finalizers_count == 0
+            elsif !new_result.equal?(result_item)
+              result_arr[idx] = new_result
+              break if @finalizers_count == 0
+            end
+          ensure
+            @current_result_path.pop
+          end
+
+          if new_invalid_null
+            nil
+          else
+            result_arr
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/execution/next/load_argument_step.rb b/lib/graphql/execution/next/load_argument_step.rb
index 86d062d3ce9..f5deae248eb 100644
--- a/lib/graphql/execution/next/load_argument_step.rb
+++ b/lib/graphql/execution/next/load_argument_step.rb
@@ -45,10 +45,12 @@ def call
         private
 
         def assign_value
-          if @loaded_value.is_a?(GraphQL::Error)
+          if @loaded_value.is_a?(GraphQL::RuntimeError)
             @loaded_value.path = @field_resolve_step.path
             @field_resolve_step.arguments = @loaded_value
           else
+            query = @field_resolve_step.selections_step.query
+            query.current_trace.object_loaded(@argument_definition, @loaded_value, query.context)
             @arguments[@argument_key] = @loaded_value
           end
 
diff --git a/lib/graphql/execution/next/prepare_object_step.rb b/lib/graphql/execution/next/prepare_object_step.rb
index 5898e91d427..57e7e74ad46 100644
--- a/lib/graphql/execution/next/prepare_object_step.rb
+++ b/lib/graphql/execution/next/prepare_object_step.rb
@@ -3,8 +3,7 @@ module GraphQL
   module Execution
     module Next
       class PrepareObjectStep
-        def initialize(static_type:, object:, runner:, graphql_result:, key:, is_non_null:, field_resolve_step:, next_objects:, next_results:, is_from_array:)
-          @static_type = static_type
+        def initialize(object:, runner:, graphql_result:, key:, is_non_null:, field_resolve_step:, next_objects:, next_results:, is_from_array:)
           @object = object
           @runner = runner
           @field_resolve_step = field_resolve_step
@@ -28,9 +27,10 @@ def value
             query.current_trace.end_authorized(@resolved_type, @object, query.context, @authorized_value)
           elsif @resolved_type
             ctx = @field_resolve_step.selections_step.query.context
-            ctx.query.current_trace.begin_resolve_type(@static_type, @object, ctx)
+            st = @field_resolve_step.static_type
+            ctx.query.current_trace.begin_resolve_type(st, @object, ctx)
             @resolved_type, _ignored_value = @field_resolve_step.sync(@resolved_type)
-            ctx.query.current_trace.end_resolve_type(@static_type, @object, ctx, @resolved_type)
+            ctx.query.current_trace.end_resolve_type(st, @object, ctx, @resolved_type)
           end
           @runner.add_step(self)
         end
@@ -38,13 +38,14 @@ def value
         def call
           case @next_step
           when :resolve_type
-            if @static_type.kind.abstract?
+            static_type = @field_resolve_step.static_type
+            if static_type.kind.abstract?
               ctx = @field_resolve_step.selections_step.query.context
-              ctx.query.current_trace.begin_resolve_type(@static_type, @object, ctx)
-              @resolved_type, _ignored_value = @runner.schema.resolve_type(@static_type, @object, ctx)
-              ctx.query.current_trace.end_resolve_type(@static_type, @object, ctx, @resolved_type)
+              ctx.query.current_trace.begin_resolve_type(static_type, @object, ctx)
+              @resolved_type, _ignored_value = @runner.schema.resolve_type(static_type, @object, ctx)
+              ctx.query.current_trace.end_resolve_type(static_type, @object, ctx, @resolved_type)
             else
-              @resolved_type = @static_type
+              @resolved_type = static_type
             end
             if @runner.resolves_lazies && @runner.lazy?(@resolved_type)
               @next_step = :authorize
@@ -103,7 +104,7 @@ def create_result
               else
                 @graphql_result[@key] = @field_resolve_step.add_graphql_error(@authorization_error)
               end
-            rescue GraphQL::Error => err
+            rescue GraphQL::RuntimeError => err
               if @is_non_null
                 @graphql_result[@key] = @field_resolve_step.add_non_null_error(@is_from_array)
               else
@@ -118,7 +119,7 @@ def create_result
             @next_objects << @object
             @graphql_result[@key] = next_result_h
             @runner.runtime_type_at[next_result_h] = @resolved_type
-            @runner.static_type_at[next_result_h] = @static_type
+            @runner.static_type_at[next_result_h] = @field_resolve_step.static_type
           end
 
           @field_resolve_step.authorized_finished(self)
diff --git a/lib/graphql/execution/next/runner.rb b/lib/graphql/execution/next/runner.rb
index 24748842e1b..d0c63f158d9 100644
--- a/lib/graphql/execution/next/runner.rb
+++ b/lib/graphql/execution/next/runner.rb
@@ -9,11 +9,27 @@ def initialize(multiplex, authorization:)
           @steps_queue = []
           @runtime_type_at = {}.compare_by_identity
           @static_type_at = {}.compare_by_identity
+          @finalizers = nil
           @selected_operation = nil
           @dataloader = multiplex.context[:dataloader] ||= @schema.dataloader_class.new
           @resolves_lazies = @schema.resolves_lazies?
+
+          @runtime_directives = nil
+          @schema.directives.each do |name, dir_class|
+            if dir_class.runtime? && name != "include" && name != "skip"
+              @runtime_directives ||= {}
+              @runtime_directives[dir_class.graphql_name] = dir_class
+            end
+          end
+
+          if @runtime_directives.nil?
+            @uses_runtime_directives = false
+            @runtime_directives = EmptyObjects::EMPTY_HASH
+          else
+            @uses_runtime_directives = true
+          end
+
           @lazy_cache = resolves_lazies ? {}.compare_by_identity : nil
-          @field_resolve_step_class = @schema.uses_raw_value? ? RawValueFieldResolveStep : FieldResolveStep
           @authorization = authorization
           if @authorization
             @authorizes_cache = Hash.new do |h, query_context|
@@ -22,6 +38,8 @@ def initialize(multiplex, authorization:)
           end
         end
 
+        attr_reader :runtime_directives, :uses_runtime_directives, :finalizer_keys
+
         def resolve_type(type, object, query)
           query.current_trace.begin_resolve_type(type, object, query.context)
           resolved_type, _ignored_new_value = query.resolve_type(type, object)
@@ -43,7 +61,24 @@ def add_step(step)
           @dataloader.append_job(step)
         end
 
-        attr_reader :authorization, :steps_queue, :schema, :variables, :dataloader, :resolves_lazies, :authorizes, :static_type_at, :runtime_type_at
+        attr_reader :authorization, :steps_queue, :schema, :variables, :dataloader, :resolves_lazies, :authorizes, :static_type_at, :runtime_type_at, :finalizers
+
+        # @return [void]
+        def add_finalizer(query, result_value, key, finalizer)
+          @finalizers ||= {}.compare_by_identity
+          f_for_query = @finalizers[query] ||= {}.compare_by_identity
+          f_for_result = f_for_query[result_value] ||= {}.compare_by_identity
+          if (f = f_for_result[key])
+            if f.is_a?(Array)
+              f << finalizer
+            else
+              f_for_result[key] = [f, finalizer]
+            end
+          else
+            f_for_result[key] = finalizer
+          end
+          nil
+        end
 
         def execute
           Fiber[:__graphql_current_multiplex] = @multiplex
@@ -69,122 +104,39 @@ def execute
                 next
               end
 
-              selected_operation = query.document.definitions.first # TODO select named operation
-              data = {}
+              root_type = query.root_type
 
-              root_type = case selected_operation.operation_type
-              when nil, "query"
-                @schema.query
-              when "mutation"
-                @schema.mutation
-              when "subscription"
-                @schema.subscription
-              else
-                raise ArgumentError, "Unknown operation type: #{selected_operation.operation_type.inspect}"
+              if root_type.non_null?
+                root_type = root_type.of_type
               end
 
-              if self.authorization && authorizes?(root_type, query.context)
-                query.current_trace.begin_authorized(root_type, query.root_value, query.context)
-                auth_check = schema.sync_lazy(root_type.authorized?(query.root_value, query.context))
-                query.current_trace.end_authorized(root_type, query.root_value, query.context, auth_check)
-                root_value = if auth_check
-                  query.root_value
-                else
-                  begin
-                    auth_err = GraphQL::UnauthorizedError.new(object: query.root_value, type: root_type, context: query.context)
-                    new_val = schema.unauthorized_object(auth_err)
-                    if new_val
-                      auth_check = true
-                    end
-                    new_val
-                  rescue GraphQL::ExecutionError => ex_err
-                    # The old runtime didn't add path and ast_nodes to this
-                    query.context.add_error(ex_err)
-                    nil
-                  end
-                end
-
-                if !auth_check
-                  results << {}
-                  next
-                end
-              else
-                root_value = query.root_value
+              root_value = query.root_value
+              if resolves_lazies
+                root_value = schema.sync_lazy(root_value)
               end
 
-              results << { "data" => data }
-
-              case selected_operation.operation_type
-              when nil, "query"
-                isolated_steps[0] << SelectionsStep.new(
-                  parent_type: root_type,
-                  selections: selected_operation.selections,
-                  objects: [root_value],
-                  results: [data],
-                  path: EmptyObjects::EMPTY_ARRAY,
-                  runner: self,
-                  query: query,
-                )
-              when "mutation"
-                fields = {}
-                gather_selections(root_type, selected_operation.selections, nil, query, {}, into: fields)
-                fields.each_value do |field_resolve_step|
-                  isolated_steps << [SelectionsStep.new(
-                    parent_type: root_type,
-                    selections: field_resolve_step.ast_nodes || Array(field_resolve_step.ast_node),
-                    objects: [root_value],
-                    results: [data],
-                    path: EmptyObjects::EMPTY_ARRAY,
-                    runner: self,
-                    query: query,
-                  )]
-                end
-              when "subscription"
-                if !query.subscription_update?
-                  schema.subscriptions.initialize_subscriptions(query)
-                end
-                isolated_steps[0] << SelectionsStep.new(
-                  parent_type: root_type,
-                  selections: selected_operation.selections,
-                  objects: [root_value],
-                  results: [data],
-                  path: EmptyObjects::EMPTY_ARRAY,
-                  runner: self,
-                  query: query,
-                )
-              else
-                raise ArgumentError, "Unhandled operation type: #{operation.operation_type.inspect}"
-              end
-
-              @static_type_at[data] = root_type
-
-              # TODO This is stupid but makes multiplex_spec.rb pass
               trace.execute_query(query: query) do
+                begin_execute(isolated_steps, results, query, root_type, root_value)
               end
             end
 
-            while (next_isolated_steps = isolated_steps.shift)
-              next_isolated_steps.each do |step|
-                add_step(step)
-              end
-              @dataloader.run
-            end
-
-            # TODO This is stupid but makes multiplex_spec.rb pass
             trace.execute_query_lazy(query: nil, multiplex: @multiplex) do
+              while (next_isolated_steps = isolated_steps.shift)
+                next_isolated_steps.each do |step|
+                  add_step(step)
+                end
+                @dataloader.run
+              end
             end
 
             queries.each_with_index.map do |query, idx|
               result = results[idx]
-              if query.subscription?
-                @schema.subscriptions.finish_subscriptions(query)
-              end
 
-              fin_result = if query.context.errors.empty?
+              fin_result = if (!@finalizers&.key?(query) && query.context.errors.empty?) || !query.valid?
                 result
               else
                 data = result["data"]
-                data = propagate_errors(data, query)
+                data = Finalize.new(query, data, self).run
                 errors = []
                 query.context.errors.each do |err|
                   if err.respond_to?(:to_h)
@@ -207,16 +159,17 @@ def execute
           Fiber[:__graphql_current_multiplex] = nil
         end
 
-        def gather_selections(type_defn, ast_selections, selections_step, query, prototype_result, into:)
+        def gather_selections(type_defn, ast_selections, selections_step, query, all_selections, prototype_result, into:)
           ast_selections.each do |ast_selection|
             next if !directives_include?(query, ast_selection)
+
             case ast_selection
             when GraphQL::Language::Nodes::Field
               key = ast_selection.alias || ast_selection.name
               step = into[key] ||= begin
                 prototype_result[key] = nil
 
-                @field_resolve_step_class.new(
+                FieldResolveStep.new(
                   selections_step: selections_step,
                   key: key,
                   parent_type: type_defn,
@@ -227,13 +180,21 @@ def gather_selections(type_defn, ast_selections, selections_step, query, prototy
             when GraphQL::Language::Nodes::InlineFragment
               type_condition = ast_selection.type&.name
               if type_condition.nil? || type_condition_applies?(query.context, type_defn, type_condition)
-                gather_selections(type_defn, ast_selection.selections, selections_step, query, prototype_result, into: into)
+                if uses_runtime_directives && !ast_selection.directives.empty?
+                  all_selections << (into = { __node: ast_selection })
+                  all_selections << (prototype_result = {})
+                end
+                gather_selections(type_defn, ast_selection.selections, selections_step, query, all_selections, prototype_result, into: into)
               end
             when GraphQL::Language::Nodes::FragmentSpread
-              fragment_definition = query.document.definitions.find { |defn| defn.is_a?(GraphQL::Language::Nodes::FragmentDefinition) && defn.name == ast_selection.name }
+              fragment_definition = query.fragments[ast_selection.name]
               type_condition = fragment_definition.type.name
               if type_condition_applies?(query.context, type_defn, type_condition)
-                gather_selections(type_defn, fragment_definition.selections, selections_step, query, prototype_result, into: into)
+                if uses_runtime_directives && !ast_selection.directives.empty?
+                  all_selections << (into = { __node: ast_selection })
+                  all_selections << (prototype_result = {})
+                end
+                gather_selections(type_defn, fragment_definition.selections, selections_step, query, all_selections, prototype_result, into: into)
               end
             else
               raise ArgumentError, "Unsupported graphql selection node: #{ast_selection.class} (#{ast_selection.inspect})"
@@ -250,123 +211,148 @@ def lazy?(object)
           is_lazy
         end
 
-        private
-
-        def propagate_errors(data, query)
-          paths_to_check = query.context.errors.map(&:path)
-          paths_to_check.compact! # root-level auth errors currently come without a path
-          # TODO dry with above?
-          # This is also where a query-level "Step" would be used?
-          if (selected_operation = query.selected_operation)
-            root_type = case selected_operation.operation_type
-            when nil, "query"
-              query.schema.query
-            when "mutation"
-              query.schema.mutation
-            when "subscription"
-              query.schema.subscription
-            end
-            check_object_result(query, data, root_type, selected_operation.selections, [], [], paths_to_check)
+        def type_condition_applies?(context, concrete_type, type_name)
+          if type_name == concrete_type.graphql_name
+            true
+          else
+            abs_t = @schema.get_type(type_name, context)
+            p_types = @schema.possible_types(abs_t, context)
+            c_p_types = @schema.possible_types(concrete_type, context)
+            p_types.any? { |t| c_p_types.include?(t) }
           end
         end
 
-        def check_object_result(query, result_h, static_type, ast_selections, current_exec_path, current_result_path, paths_to_check)
-          current_path_len = current_exec_path.length
-          ast_selections.each do |ast_selection|
-            case ast_selection
-            when Language::Nodes::Field
-              begin
-                key = ast_selection.alias || ast_selection.name
-                current_exec_path << key
-                current_result_path << key
-                if paths_to_check.any? { |path_to_check| path_to_check[current_path_len] == key }
-                  result_value = result_h[key]
-                  field_defn = query.context.types.field(static_type, ast_selection.name)
-                  result_type = field_defn.type
-                  if (result_type_non_null = result_type.non_null?)
-                    result_type = result_type.of_type
-                  end
-
-                  new_result_value = if result_value.is_a?(GraphQL::Error)
-                    result_value.path = current_result_path.dup
-                    result_value.assign_graphql_result(query, result_h, key)
-                    result_h.key?(key) ? result_h[key] : :unassigned
-                  else
-                    if result_type.list?
-                      check_list_result(query, result_value, result_type.of_type, ast_selection.selections, current_exec_path, current_result_path, paths_to_check)
-                    elsif !result_type.kind.leaf?
-                      check_object_result(query, result_value, result_type, ast_selection.selections, current_exec_path, current_result_path, paths_to_check)
-                    else
-                      result_value
-                    end
-                  end
+        private
 
-                  if new_result_value.nil? && result_type_non_null
-                    return nil
-                  elsif :unassigned.equal?(new_result_value)
-                    # Do nothing
-                  elsif !new_result_value.equal?(result_value)
-                    result_h[key] = new_result_value
+        def begin_execute(isolated_steps, results, query, root_type, root_value)
+          data = {}
+          selected_operation = query.selected_operation
+          beginning_path = query.path
+
+          case root_type.kind.name
+          when "OBJECT"
+            if self.authorization && authorizes?(root_type, query.context)
+              query.current_trace.begin_authorized(root_type, root_value, query.context)
+              auth_check = schema.sync_lazy(root_type.authorized?(root_value, query.context))
+              query.current_trace.end_authorized(root_type, root_value, query.context, auth_check)
+              root_value = if auth_check
+                root_value
+              else
+                begin
+                  auth_err = GraphQL::UnauthorizedError.new(object: root_value, type: root_type, context: query.context)
+                  new_val = schema.unauthorized_object(auth_err)
+                  if new_val
+                    auth_check = true
                   end
+                  new_val
+                rescue GraphQL::ExecutionError => ex_err
+                  # The old runtime didn't add path and ast_nodes to this
+                  ex_err.path = beginning_path
+                  query.context.add_error(ex_err)
+                  nil
                 end
-              ensure
-                current_exec_path.pop
-                current_result_path.pop
-              end
-            when Language::Nodes::InlineFragment
-              static_type_at_result = @static_type_at[result_h]
-              if static_type_at_result && type_condition_applies?(query.context, static_type_at_result, ast_selection.type.name)
-                result_h = check_object_result(query, result_h, static_type, ast_selection.selections, current_exec_path, current_result_path, paths_to_check)
               end
-            when Language::Nodes::FragmentSpread
-              fragment_defn = query.document.definitions.find { |defn| defn.is_a?(Language::Nodes::FragmentDefinition) && defn.name == ast_selection.name }
-              static_type_at_result = @static_type_at[result_h]
-              if static_type_at_result && type_condition_applies?(query.context, static_type_at_result, fragment_defn.type.name)
-                result_h = check_object_result(query, result_h, static_type, fragment_defn.selections, current_exec_path, current_result_path, paths_to_check)
+
+              if !auth_check
+                results << {}
+                return
               end
             end
-          end
-
-          result_h
-        end
-
-        def check_list_result(query, result_arr, inner_type, ast_selections, current_exec_path, current_result_path, paths_to_check)
-          inner_type_non_null = false
-          if inner_type.non_null?
-            inner_type_non_null = true
-            inner_type = inner_type.of_type
-          end
 
-          new_invalid_null = false
-          result_arr.each_with_index do |result_item, idx|
-            current_result_path << idx
-            new_result = if result_item.is_a?(GraphQL::Error)
-              result_item.path = current_result_path.dup
-              result_item.assign_graphql_result(query, result_arr, idx)
-              result_arr[idx]
-            elsif inner_type.list?
-              check_list_result(query, result_item, inner_type.of_type, ast_selections, current_exec_path, current_result_path, paths_to_check)
-            elsif !inner_type.kind.leaf?
-              check_object_result(query, result_item, inner_type, ast_selections, current_exec_path, current_result_path, paths_to_check)
+            results << { "data" => data }
+            objects = [root_value]
+            query.current_trace.objects(root_type, objects, query.context)
+
+            if query.query?
+              isolated_steps[0] << SelectionsStep.new(
+                parent_type: root_type,
+                selections: query.selected_operation.selections,
+                objects: objects,
+                results: [data],
+                path: beginning_path,
+                runner: self,
+                query: query,
+              )
+            elsif query.mutation?
+              fields = {}
+              all_selections = [fields, (prototype_result = {})]
+              gather_selections(root_type, selected_operation.selections, nil, query, all_selections, prototype_result, into: fields)
+              if all_selections.length > 2
+                # TODO DRY with SelectionsStep with directive handling
+                raise "Directives on root mutation type not implemented yet"
+              end
+              fields.each_value do |field_resolve_step|
+                isolated_steps << [SelectionsStep.new(
+                  clobber: false, # `data` is being shared among several selections steps
+                  parent_type: root_type,
+                  selections: field_resolve_step.ast_nodes || Array(field_resolve_step.ast_node),
+                  objects: objects,
+                  results: [data],
+                  path: beginning_path,
+                  runner: self,
+                  query: query,
+                )]
+              end
+            elsif query.subscription?
+              if !query.subscription_update?
+                schema.subscriptions.initialize_subscriptions(query)
+                add_finalizer(query, data, nil, schema.subscriptions.finalizer)
+              end
+              isolated_steps[0] << SelectionsStep.new(
+                parent_type: root_type,
+                selections: selected_operation.selections,
+                objects: objects,
+                results: [data],
+                path: beginning_path,
+                runner: self,
+                query: query,
+              )
             else
-              result_item
+              raise ArgumentError, "Unknown operation type (not query, mutation or subscription): #{query.query_string}"
             end
-
-            if new_result.nil? && inner_type_non_null
-              new_invalid_null = true
-              result_arr[idx] = nil
-            elsif !new_result.equal?(result_item)
-              result_arr[idx] = new_result
+          when "UNION", "INTERFACE"
+            resolved_type = resolve_type(root_type, root_value, query)
+            if resolves_lazies
+              resolved_type = schema.sync_lazy(resolved_type)
             end
-          ensure
-            current_result_path.pop
-          end
-
-          if new_invalid_null
-            nil
+            objects = [root_value]
+            query.current_trace.objects(resolved_type, objects, query.context)
+            runtime_type_at[data] = resolved_type
+            results << { "data" => data }
+            isolated_steps[0] << SelectionsStep.new(
+              parent_type: resolved_type,
+              selections: query.selected_operation.selections,
+              objects: objects,
+              results: [data],
+              path: beginning_path,
+              runner: self,
+              query: query,
+            )
+          when "LIST"
+            inner_type = root_type.unwrap
+            case inner_type.kind.name
+            when "SCALAR", "ENUM"
+              results << run_isolated_scalar(root_type, query)
+            else
+              list_result = Array.new(root_value.size) { Hash.new.compare_by_identity }
+              results << { "data" => list_result }
+              isolated_steps[0] << SelectionsStep.new(
+                parent_type: inner_type,
+                selections: query.selected_operation.selections,
+                objects: root_value,
+                results: list_result,
+                path: beginning_path,
+                runner: self,
+                query: query,
+              )
+            end
+          when "SCALAR", "ENUM"
+            results << run_isolated_scalar(root_type, query)
           else
-            result_arr
+            raise "Unhandled root type kind: #{root_type.kind.name.inspect}"
           end
+
+          @static_type_at[data] = root_type
         end
 
         def dir_arg_value(query, arg_node)
@@ -395,15 +381,37 @@ def directives_include?(query, ast_selection)
           end
         end
 
-        def type_condition_applies?(context, concrete_type, type_name)
-          if type_name == concrete_type.graphql_name
-            true
-          else
-            abs_t = @schema.get_type(type_name, context)
-            p_types = @schema.possible_types(abs_t, context)
-            c_p_types = @schema.possible_types(concrete_type, context)
-            p_types.any? { |t| c_p_types.include?(t) }
+        def run_isolated_scalar(type, partial)
+          value = partial.root_value
+          dummy_path = partial.path.dup
+          key = dummy_path.pop
+          is_from_array = key.is_a?(Integer)
+
+          if lazy?(value)
+            value = @schema.sync_lazy(value)
           end
+          selections = partial.ast_nodes
+          dummy_ss = SelectionsStep.new(
+            parent_type: nil,
+            selections: selections,
+            objects: nil,
+            results: nil,
+            path: dummy_path,
+            runner: self,
+            query: partial,
+          )
+          dummy_frs = FieldResolveStep.new(
+            selections_step: dummy_ss,
+            key: key,
+            parent_type: nil,
+            runner: self,
+          )
+          dummy_frs.static_type = type
+          selections.each { |s| dummy_frs.append_selection(s) }
+
+          result = is_from_array ? [] : {}
+          dummy_frs.finish_leaf_result(result, key, value, type, partial.context)
+          { "data" => result[key] }
         end
       end
     end
diff --git a/lib/graphql/execution/next/selections_step.rb b/lib/graphql/execution/next/selections_step.rb
index 15b1bd813d2..b9382f694c9 100644
--- a/lib/graphql/execution/next/selections_step.rb
+++ b/lib/graphql/execution/next/selections_step.rb
@@ -3,7 +3,7 @@ module GraphQL
   module Execution
     module Next
       class SelectionsStep
-        def initialize(parent_type:, selections:, objects:, results:, runner:, query:, path:)
+        def initialize(parent_type:, selections:, objects:, results:, runner:, query:, path:, clobber: true)
           @path = path
           @parent_type = parent_type
           @selections = selections
@@ -12,6 +12,8 @@ def initialize(parent_type:, selections:, objects:, results:, runner:, query:, p
           @results = results
           @query = query
           @graphql_objects = nil
+          @all_selections = nil
+          @clobber = clobber
         end
 
         attr_reader :path, :query, :objects, :results
@@ -23,12 +25,72 @@ def graphql_objects
         end
 
         def call
-          grouped_selections = {}
-          prototype_result = @results.first
-          @runner.gather_selections(@parent_type, @selections, self, self.query, prototype_result, into: grouped_selections)
-          @results.each { |r| r.replace(prototype_result) }
-          grouped_selections.each_value do |frs|
-            @runner.add_step(frs)
+          @all_selections = [{}, (prototype_result = {})]
+          @runner.gather_selections(@parent_type, @selections, self, self.query, @all_selections, @all_selections[1], into: @all_selections[0])
+          continue_selections = []
+          i = 0
+          l = @all_selections.length
+          while i < l
+            grouped_selections = @all_selections[i]
+            selections_prototype_result = @all_selections[i + 1]
+            if (directives_owner = grouped_selections.delete(:__node))
+              directives = directives_owner.directives
+              continue_execution = true
+              directives.each do |dir_node|
+                dir_defn = @runner.runtime_directives[dir_node.name]
+                if dir_defn # not present for `skip` or `include`
+                  dummy_frs = FieldResolveStep.new(
+                    selections_step: self,
+                    key: nil,
+                    parent_type: @parent_type,
+                    runner: @runner,
+                  )
+                  dir_args = dummy_frs.coerce_arguments(dir_defn, dir_node.arguments, false) # rubocop:disable Development/ContextIsPassedCop
+                  result = case directives_owner
+                  when Language::Nodes::FragmentSpread
+                    dir_defn.resolve_fragment_spread(directives_owner, @parent_type, @objects, dir_args, self.query.context)
+                  when Language::Nodes::InlineFragment
+                    dir_defn.resolve_inline_fragment(directives_owner, @parent_type, @objects, dir_args, self.query.context)
+                  else
+                    raise ArgumentError, "Unhandled directive owner (#{directives_owner.class}): #{directives_owner.inspect}"
+                  end
+                  if result.is_a?(Finalizer)
+                    result.path = path
+                    @results.each do |r|
+                      @runner.add_finalizer(@query, r, nil, result)
+                    end
+                    if result.is_a?(HaltExecution)
+                      continue_execution = false
+                      break
+                    end
+                  end
+
+                  if continue_execution
+                    prototype_result.merge!(selections_prototype_result)
+                    grouped_selections.each_value { |v| continue_selections << v }
+                  end
+                else
+                  grouped_selections.each_value { |v| continue_selections << v }
+                end
+              end
+            else
+              grouped_selections.each_value { |v| continue_selections << v }
+            end
+
+            if @clobber
+              i2 = 0
+              l2 = @results.length
+              while i2 < l2
+                @results[i2].replace(prototype_result)
+                i2 += 1
+              end
+            end
+
+            continue_selections.each do |frs|
+              @runner.add_step(frs)
+            end
+
+            i += 2
           end
         end
       end
diff --git a/lib/graphql/execution_error.rb b/lib/graphql/execution_error.rb
index 7b26dee5fd0..dfb39ff932f 100644
--- a/lib/graphql/execution_error.rb
+++ b/lib/graphql/execution_error.rb
@@ -36,7 +36,7 @@ def initialize(message, ast_node: nil, ast_nodes: nil, options: nil, extensions:
       super(message)
     end
 
-    def assign_graphql_result(query, result_data, key)
+    def finalize_graphql_result(query, result_data, key)
       result_data[key] = nil
     end
 
diff --git a/lib/graphql/query.rb b/lib/graphql/query.rb
index 200c5200894..42414628099 100644
--- a/lib/graphql/query.rb
+++ b/lib/graphql/query.rb
@@ -159,6 +159,7 @@ def initialize(schema, query_string = nil, query: nil, document: nil, context: n
       @root_value = root_value
       @fragments = nil
       @operations = nil
+      @finalizers = @top_level_finalizers = nil
       @validate = validate
       self.static_validator = static_validator if static_validator
       context_tracers = (context ? context.fetch(:tracers, []) : [])
@@ -262,6 +263,10 @@ def operations
       with_prepared_ast { @operations }
     end
 
+    def path
+      EmptyObjects::EMPTY_ARRAY
+    end
+
     # Run subtree partials of this query and return their results.
     # Each partial is identified with a `path:` and `object:`
     # where the path references a field in the AST and the object will be treated
@@ -271,7 +276,11 @@ def operations
     # @return [Array<GraphQL::Query::Result>]
     def run_partials(partials_hashes)
       partials = partials_hashes.map { |partial_options| Partial.new(query: self, **partial_options) }
-      Execution::Interpreter.run_all(@schema, partials, context: @context)
+      if context[:__graphql_execute_next]
+        Execution::Next.run_all(@schema, partials, context: @context)
+      else
+        Execution::Interpreter.run_all(@schema, partials, context: @context)
+      end
     end
 
     # Get the result for this query, executing it once
diff --git a/lib/graphql/query/context.rb b/lib/graphql/query/context.rb
index da454f8c8f1..3896b714263 100644
--- a/lib/graphql/query/context.rb
+++ b/lib/graphql/query/context.rb
@@ -126,6 +126,12 @@ def add_error(error)
         nil
       end
 
+      # @param value [Object] Any object to be inserted directly into the final response
+      # @return [GraphQL::Execution::Interpreter::RawValue] Return this from the field
+      def raw_value(value)
+        GraphQL::Execution::Interpreter::RawValue.new(value)
+      end
+
       # @example Print the GraphQL backtrace during field resolution
       #   puts ctx.backtrace
       #
diff --git a/lib/graphql/query/partial.rb b/lib/graphql/query/partial.rb
index 31bd6abff6d..6e4c0e17555 100644
--- a/lib/graphql/query/partial.rb
+++ b/lib/graphql/query/partial.rb
@@ -32,6 +32,7 @@ def initialize(path: nil, object:, query:, context: nil, fragment_node: nil, typ
         @multiplex = nil
         @result_values = nil
         @result = nil
+        @finalizers = @top_level_finalizers = nil
 
         if fragment_node
           @ast_nodes = [fragment_node]
@@ -51,6 +52,10 @@ def leaf?
         @leaf
       end
 
+      def root_value
+        object
+      end
+
       attr_reader :context, :query, :ast_nodes, :root_type, :object, :field_definition, :path, :schema
 
       attr_accessor :multiplex, :result_values
@@ -90,10 +95,22 @@ def fragments
         @query.fragments
       end
 
+      def validate
+        @query.validate
+      end
+
       def valid?
         @query.valid?
       end
 
+      def query?
+        true
+      end
+
+      def run_partials(...)
+        @query.run_partials(...)
+      end
+
       def analyzers
         EmptyObjects::EMPTY_ARRAY
       end
@@ -107,7 +124,7 @@ def subscription?
       end
 
       def selected_operation
-        ast_nodes.first
+        Language::Nodes::OperationDefinition.new(selections: ast_nodes.flat_map(&:selections))
       end
 
       def static_errors
@@ -123,7 +140,6 @@ def selected_operation_name
       def set_type_info_from_path
         selections = [@query.selected_operation]
         type = @query.root_type
-        parent_type = nil
         field_defn = nil
 
         @path.each do |name_in_doc|
@@ -162,7 +178,6 @@ def set_type_info_from_path
           end
           field_name = next_selections.first.name
           field_defn = @schema.get_field(type, field_name, @query.context) || raise("Invariant: no field called #{field_name} on #{type.graphql_name}")
-          parent_type = type
           type = field_defn.type
           if type.non_null?
             type = type.of_type
diff --git a/lib/graphql/runtime_error.rb b/lib/graphql/runtime_error.rb
new file mode 100644
index 00000000000..110752d6cfd
--- /dev/null
+++ b/lib/graphql/runtime_error.rb
@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+module GraphQL
+  class RuntimeError < Error
+    include GraphQL::Execution::Next::Finalizer
+  end
+end
diff --git a/lib/graphql/schema.rb b/lib/graphql/schema.rb
index bd29ad0b5c1..e21a6ffa4f9 100644
--- a/lib/graphql/schema.rb
+++ b/lib/graphql/schema.rb
@@ -1362,14 +1362,6 @@ def lazy_resolve(lazy_class, value_method)
         lazy_methods.set(lazy_class, value_method)
       end
 
-      def uses_raw_value?
-        !!@uses_raw_value
-      end
-
-      def uses_raw_value(new_val)
-        @uses_raw_value = new_val
-      end
-
       def resolves_lazies?
         lazy_method_count = 0
         lazy_methods.each do |k, v|
@@ -1450,7 +1442,16 @@ def tracer(new_tracer, silence_deprecation_warning: false)
       end
 
       def tracers
-        find_inherited_value(:tracers, EMPTY_ARRAY) + own_tracers
+        inherited = find_inherited_value(:tracers, EMPTY_ARRAY)
+        if inherited.length > 0
+          if own_tracers.length > 0
+            inherited + own_tracers
+          else
+            inherited
+          end
+        else
+          own_tracers
+        end
       end
 
       # Mix `trace_mod` into this schema's `Trace` class so that its methods will be called at runtime.
@@ -1560,7 +1561,8 @@ def query_analyzer(new_analyzer)
       end
 
       def query_analyzers
-        find_inherited_value(:query_analyzers, EMPTY_ARRAY) + own_query_analyzers
+        inherited_qa = find_inherited_value(:query_analyzers, EMPTY_ARRAY)
+        inherited_qa.empty? ? own_query_analyzers : (inherited_qa + own_query_analyzers)
       end
 
       # @param new_analyzer [Class<GraphQL::Analysis::Analyzer>] An analyzer to run on multiplexes to this schema
diff --git a/lib/graphql/schema/directive.rb b/lib/graphql/schema/directive.rb
index 5689de73b59..22eb1d1ca65 100644
--- a/lib/graphql/schema/directive.rb
+++ b/lib/graphql/schema/directive.rb
@@ -31,17 +31,25 @@ def default_graphql_name
 
         def locations(*new_locations)
           if !new_locations.empty?
+            is_runtime = false
             new_locations.each do |new_loc|
-              if !LOCATIONS.include?(new_loc.to_sym)
+              loc_sym = new_loc.to_sym
+              if !LOCATIONS.include?(loc_sym)
                 raise ArgumentError, "#{self} (#{self.graphql_name}) has an invalid directive location: `locations #{new_loc}` "
               end
+              is_runtime ||= RUNTIME_LOCATIONS.include?(loc_sym)
             end
             @locations = new_locations
+            @is_runtime = is_runtime
           else
             @locations ||= (superclass.respond_to?(:locations) ? superclass.locations : [])
           end
         end
 
+        def runtime?
+          @is_runtime
+        end
+
         def default_directive(new_default_directive = nil)
           if new_default_directive != nil
             @default_directive = new_default_directive
@@ -104,8 +112,12 @@ def repeatable(new_value)
 
         def inherited(subclass)
           super
+          parent_class = self
           subclass.class_exec do
             @default_graphql_name ||= nil
+            @locations = parent_class.locations
+            @is_runtime = parent_class.runtime?
+            @repeatable = false
           end
         end
       end
@@ -177,13 +189,16 @@ def graphql_name
       end
 
       LOCATIONS = [
-        QUERY =                  :QUERY,
-        MUTATION =               :MUTATION,
-        SUBSCRIPTION =           :SUBSCRIPTION,
-        FIELD =                  :FIELD,
-        FRAGMENT_DEFINITION =    :FRAGMENT_DEFINITION,
-        FRAGMENT_SPREAD =        :FRAGMENT_SPREAD,
-        INLINE_FRAGMENT =        :INLINE_FRAGMENT,
+        *(RUNTIME_LOCATIONS = [
+          QUERY =                  :QUERY,
+          MUTATION =               :MUTATION,
+          SUBSCRIPTION =           :SUBSCRIPTION,
+          FIELD =                  :FIELD,
+          FRAGMENT_DEFINITION =    :FRAGMENT_DEFINITION,
+          FRAGMENT_SPREAD =        :FRAGMENT_SPREAD,
+          INLINE_FRAGMENT =        :INLINE_FRAGMENT,
+          VARIABLE_DEFINITION =    :VARIABLE_DEFINITION,
+        ]),
         SCHEMA =                 :SCHEMA,
         SCALAR =                 :SCALAR,
         OBJECT =                 :OBJECT,
@@ -195,7 +210,6 @@ def graphql_name
         ENUM_VALUE =             :ENUM_VALUE,
         INPUT_OBJECT =           :INPUT_OBJECT,
         INPUT_FIELD_DEFINITION = :INPUT_FIELD_DEFINITION,
-        VARIABLE_DEFINITION =    :VARIABLE_DEFINITION,
       ]
 
       DEFAULT_DEPRECATION_REASON = 'No longer supported'
diff --git a/lib/graphql/schema/interface.rb b/lib/graphql/schema/interface.rb
index 160d51693a2..1c11e1e0aca 100644
--- a/lib/graphql/schema/interface.rb
+++ b/lib/graphql/schema/interface.rb
@@ -33,6 +33,26 @@ def definition_methods(&block)
           self::DefinitionMethods.module_exec(&block)
         end
 
+        # Instance methods defined in this block will become class methods on objects that implement this interface.
+        # Use it to implement `resolve_each:`, `resolve_batch:`, and `resolve_static:` fields.
+        # @example
+        #   field :thing, String, resolve_static: true
+        #
+        #   resolver_methods do
+        #     def thing
+        #       Somehow.get.thing
+        #     end
+        #   end
+        def resolver_methods(&block)
+          if !defined?(@_resolver_methods)
+            resolver_methods_module = Module.new
+            @_resolver_methods = resolver_methods_module
+            const_set(:ResolverMethods, resolver_methods_module)
+            extend(self::ResolverMethods)
+          end
+          self::ResolverMethods.module_exec(&block)
+        end
+
         # @see {Schema::Warden} hides interfaces without visible implementations
         def visible?(context)
           true
@@ -79,6 +99,12 @@ def included(child_class)
             if !backtrace_line
               raise "Attach interfaces using `implements(#{self})`, not `include(#{self})`"
             end
+
+            child_class.ancestors.reverse_each do |ancestor|
+              if ancestor.const_defined?(:ResolverMethods)
+                child_class.extend(ancestor::ResolverMethods)
+              end
+            end
           end
 
           super
diff --git a/lib/graphql/schema/list.rb b/lib/graphql/schema/list.rb
index e5764c9e50d..56e0928f58f 100644
--- a/lib/graphql/schema/list.rb
+++ b/lib/graphql/schema/list.rb
@@ -22,6 +22,10 @@ def to_type_signature
         @type_signature ||= -"[#{@of_type.to_type_signature}]"
       end
 
+      def authorizes?(ctx)
+        of_type.authorizes?(ctx)
+      end
+
       # This is for introspection, where it's expected the name will be `null`
       def graphql_name
         nil
diff --git a/lib/graphql/schema/member/base_dsl_methods.rb b/lib/graphql/schema/member/base_dsl_methods.rb
index 53ff00aade2..ce91290004e 100644
--- a/lib/graphql/schema/member/base_dsl_methods.rb
+++ b/lib/graphql/schema/member/base_dsl_methods.rb
@@ -26,16 +26,6 @@ def graphql_name(new_name = nil)
           end
         end
 
-        # Just a convenience method to point out that people should use graphql_name instead
-        def name(new_name = nil)
-          return super() if new_name.nil?
-
-          fail(
-            "The new name override method is `graphql_name`, not `name`. Usage: "\
-            "graphql_name \"#{new_name}\""
-          )
-        end
-
         # Call this method to provide a new description; OR
         # call it without an argument to get the description
         # @param new_description [String]
diff --git a/lib/graphql/subscriptions.rb b/lib/graphql/subscriptions.rb
index 005bf176bde..f119ce41c62 100644
--- a/lib/graphql/subscriptions.rb
+++ b/lib/graphql/subscriptions.rb
@@ -259,6 +259,21 @@ def finish_subscriptions(query)
       nil
     end
 
+    def finalizer
+      Finalizer.new(self)
+    end
+
+    class Finalizer
+      include Execution::Next::Finalizer
+      def initialize(subscriptions)
+        @subscriptions = subscriptions
+      end
+
+      def finalize_graphql_result(query, result_data, result_key)
+        @subscriptions.finish_subscriptions(query)
+      end
+    end
+
     private
 
     # Recursively normalize `args` as belonging to `arg_owner`:
diff --git a/lib/graphql/tracing/trace.rb b/lib/graphql/tracing/trace.rb
index 4017fdfe4ca..54490fcddf9 100644
--- a/lib/graphql/tracing/trace.rb
+++ b/lib/graphql/tracing/trace.rb
@@ -98,6 +98,12 @@ def authorized(query:, type:, object:)
         yield
       end
 
+      def objects(type, object, context)
+      end
+
+      def object_loaded(argument_definition, object, context)
+      end
+
       # A call to `.authorized?` is starting
       # @param type [Class<GraphQL::Schema::Object>]
       # @param object [Object]
diff --git a/lib/graphql/unauthorized_error.rb b/lib/graphql/unauthorized_error.rb
index 71ee783328f..56cf79b2b43 100644
--- a/lib/graphql/unauthorized_error.rb
+++ b/lib/graphql/unauthorized_error.rb
@@ -30,7 +30,7 @@ def initialize(message = nil, object: nil, type: nil, context: nil)
 
     attr_accessor :path, :ast_nodes
 
-    def assign_graphql_result(query, result_data, key)
+    def finalize_graphql_result(query, result_data, key)
       result_data[key] = nil
     end
   end
diff --git a/spec/graphql/authorization_spec.rb b/spec/graphql/authorization_spec.rb
index f0b16bd1ede..9c5232a41d9 100644
--- a/spec/graphql/authorization_spec.rb
+++ b/spec/graphql/authorization_spec.rb
@@ -345,6 +345,8 @@ class Nothing < GraphQL::Schema::Directive
       def self.visible?(ctx)
         !!ctx[:show_nothing_directive]
       end
+
+      def self.resolve_field(...); end
     end
 
     class Schema < GraphQL::Schema
@@ -621,7 +623,11 @@ def auth_execute(*args, **kwargs)
     it "halts on unauthorized mutations" do
       query = "mutation { doUnauthorizedStuff(input: {}) { __typename } }"
       res = auth_execute(query, context: { unauthorized_mutation: true })
-      assert_nil res["data"].fetch("doUnauthorizedStuff")
+      if TESTING_EXEC_NEXT
+        assert_equal({}, res["data"])
+      else
+        assert_nil res["data"].fetch("doUnauthorizedStuff")
+      end
       assert_raises GraphQL::RequiredImplementationMissingError do
         auth_execute(query)
       end
@@ -906,7 +912,7 @@ def auth_execute(*args, **kwargs)
         unauth_res = auth_execute(query, context: { query_unauthorized: true })
 
         assert_equal({
-          "errors" => [{"message"=>"Unauthorized Query: nil"}],
+          "errors" => [(TESTING_EXEC_NEXT ? {"message"=>"Unauthorized Query: nil", "path" => [] } : {"message"=>"Unauthorized Query: nil"})],
           "data" => nil,
         }, unauth_res.to_h)
       end
diff --git a/spec/graphql/directive_spec.rb b/spec/graphql/directive_spec.rb
index 8f1b78e3ab2..f12831a5b3c 100644
--- a/spec/graphql/directive_spec.rb
+++ b/spec/graphql/directive_spec.rb
@@ -3,7 +3,13 @@
 
 describe "GraphQL::Directive" do
   let(:variables) { {"t" => true, "f" => false} }
-  let(:result) { Dummy::Schema.execute(query_string, variables: variables) }
+  let(:result) {
+    if TESTING_EXEC_NEXT
+      Dummy::Schema.execute_next(query_string, variables: variables)
+    else
+      Dummy::Schema.execute(query_string, variables: variables)
+    end
+  }
   describe "on fields" do
     let(:query_string) { %|query directives($t: Boolean!, $f: Boolean!) {
       cheese(id: 1) {
diff --git a/spec/graphql/execution/interpreter_spec.rb b/spec/graphql/execution/interpreter_spec.rb
index c534dcc811b..9476c3d1333 100644
--- a/spec/graphql/execution/interpreter_spec.rb
+++ b/spec/graphql/execution/interpreter_spec.rb
@@ -296,7 +296,6 @@ class Schema < GraphQL::Schema
       query(Query)
       mutation(Mutation)
       lazy_resolve(Box, :value)
-      uses_raw_value(true)
       use GraphQL::Schema::AlwaysVisible
       use(GraphQL::Execution::Next) if TESTING_EXEC_NEXT
 
diff --git a/spec/graphql/execution/batching/errors_spec.rb b/spec/graphql/execution/next/finalize_spec.rb
similarity index 99%
rename from spec/graphql/execution/batching/errors_spec.rb
rename to spec/graphql/execution/next/finalize_spec.rb
index b9bc3f58c40..e7e69edc0c7 100644
--- a/spec/graphql/execution/batching/errors_spec.rb
+++ b/spec/graphql/execution/next/finalize_spec.rb
@@ -518,13 +518,6 @@ def test_multiple_offenses_for_null_position_report_all_instances
       ],
     }
     expected = {
-      "data" => {
-        "test" => [
-          { "req" => "yes", "opt" => nil },
-          { "req" => "yes", "opt" => nil },
-          { "req" => "yes", "opt" => nil },
-        ],
-      },
       "errors" => [{
         "message" => "Not okay!",
         "locations" => [{ "line" => 1, "column" => 14 }],
@@ -534,9 +527,16 @@ def test_multiple_offenses_for_null_position_report_all_instances
         "locations" => [{ "line" => 1, "column" => 14 }],
         "path" => ["test", 2, "opt"],
       }],
+      "data" => {
+        "test" => [
+          { "req" => "yes", "opt" => nil },
+          { "req" => "yes", "opt" => nil },
+          { "req" => "yes", "opt" => nil },
+        ],
+      },
     }
 
-    assert_equal expected, exec_test(schema, "{ test { req opt } }", source)
+    assert_graphql_equal expected, exec_test(schema, "{ test { req opt } }", source)
   end
 
   def test_multiple_offenses_for_non_null_position_without_intersecting_propagation_report_all_instances
diff --git a/spec/graphql/execution/next_spec.rb b/spec/graphql/execution/next_spec.rb
index 91389d2bc8d..3b5e6b49242 100644
--- a/spec/graphql/execution/next_spec.rb
+++ b/spec/graphql/execution/next_spec.rb
@@ -342,7 +342,7 @@ def run_next(...)
     assert_graphql_equal expected_result, result
   end
 
-  it "propages nested nulls in objects in lists" do
+  it "propagates nested nulls in objects in lists" do
     NextExecutionSchema::DATA << OpenStruct.new(
       name: nil,
       species: [OpenStruct.new(name: "Artichoke")]
@@ -381,6 +381,6 @@ def run_next(...)
         ]
       },
     }
-    assert_equal expected_result, result
+    assert_graphql_equal expected_result, result
   end
 end
diff --git a/spec/graphql/query/partial_spec.rb b/spec/graphql/query/partial_spec.rb
index 68132069aa3..ce81526a865 100644
--- a/spec/graphql/query/partial_spec.rb
+++ b/spec/graphql/query/partial_spec.rb
@@ -52,13 +52,21 @@ class Farm < GraphQL::Schema::Object
       implements Entity
       field :name, String
       field :products, [FarmProduct]
-      field :error, Int
+      field :error, Int, resolve_static: true
 
-      def error
+      def self.error(context)
         raise GraphQL::ExecutionError, "This is a field error"
       end
 
-      field :neighboring_farm, Farm
+      def error
+        self.class.error(context)
+      end
+
+      field :neighboring_farm, Farm, resolve_batch: true
+
+      def self.neighboring_farm(objects, context)
+        context.dataload_all(FarmSource, objects.map(&:neighboring_farm_id))
+      end
 
       def neighboring_farm
         dataloader.with(FarmSource).load(object.neighboring_farm_id)
@@ -66,11 +74,17 @@ def neighboring_farm
     end
 
     class UpcasedFarm < GraphQL::Schema::Object
-      field :name, String
+      field :name, String, resolve_each: true
 
-      def name
+      graphql_name "UpcasedFarm"
+
+      def self.name(object, context)
         object[:name].upcase
       end
+
+      def name
+        self.class.name(object, context)
+      end
     end
 
     class Market < GraphQL::Schema::Object
@@ -85,57 +99,87 @@ class Thing < GraphQL::Schema::Union
     class Query < GraphQL::Schema::Object
       field :farms, [Farm], fallback_value: Database::FARMS.values
 
-      field :farm, Farm do
+      field :farm, Farm, resolve_static: true do
         argument :id, ID, loads: Farm, as: :farm
       end
 
-      def farm(farm:)
+      def self.farm(context, farm:)
         farm
       end
 
+      def farm(farm:)
+        self.class.farm(context, farm: farm)
+      end
+
       field :farm_names, [String], fallback_value: Database::FARMS.each_value.map(&:name)
 
-      field :query, Query, fallback_value: true
+      field :query, Query, fallback_value: true, resolve_static: true
+
+      def self.query(_ctx)
+        true
+      end
 
-      field :thing, Thing
+      field :thing, Thing, resolve_static: true
+
+      def self.thing(context)
+        Database.get("1")
+      end
 
       def thing
+        self.class.thing(context)
+      end
+
+      field :entity, Entity, resolve_static: true
+      def self.entity(context)
         Database.get("1")
       end
 
-      field :entity, Entity
-      def entity; Database.get("1"); end
+      def entity
+        self.class.entity(context)
+      end
 
-      field :read_context, String do
+      field :read_context, String, resolve_static: true do
         argument :key, String
       end
 
-      def read_context(key:)
+      def self.read_context(context, key:)
         -> { context[key].to_s }
       end
 
-      field :current_path, [String]
+      def read_context(key:)
+        self.class.read_context(context, key: key)
+      end
+
+      field :current_path, [String], resolve_legacy_instance_method: true
       def current_path
         context.current_path
       end
 
-      field :current_values, [String]
-      def current_values
+      field :current_values, [String], resolve_static: true
+      def self.current_values(context)
         [
           GraphQL::Current.operation_name,
-          GraphQL::Current.field.path,
+          GraphQL::Current.field&.path,
           GraphQL::Current.dataloader_source_class.inspect,
         ]
       end
+
+      def current_values
+        self.class.current_values(context)
+      end
     end
 
     class Mutation < GraphQL::Schema::Object
-      field :update_farm, Farm do
+      field :update_farm, Farm, resolve_static: true do
         argument :name, String
       end
 
+      def self.update_farm(context, name:)
+        OpenStruct.new({ name: name })
+      end
+
       def update_farm(name:)
-        { name: name }
+        self.class.update_farm(context, name: name)
       end
     end
 
@@ -151,6 +195,7 @@ def self.resolve_type(abs_type, object, ctx)
     end
 
     use GraphQL::Dataloader
+    use GraphQL::Execution::Next
     lazy_resolve Proc, :call
   end
 
@@ -160,6 +205,9 @@ def self.resolve_type(abs_type, object, ctx)
 
   def run_partials(string, partial_configs, **query_kwargs)
     query = GraphQL::Query.new(PartialSchema, string, **query_kwargs)
+    if TESTING_EXEC_NEXT
+      query.context[:__graphql_execute_next] = true
+    end
     query.run_partials(partial_configs)
   end
 
@@ -172,7 +220,7 @@ def run_partials(string, partial_configs, **query_kwargs)
     results = run_partials(str, [
       { path: ["farm1"], object: PartialSchema::Database::FARMS["1"] },
       { path: ["farm2"], object: OpenStruct.new(name: "Injected Farm") },
-      { path: ["farms", 0], object: { name: "Kestrel Hollow", products: [:__MEAT__, "EGGS"]} },
+      { path: ["farms", 0], object: OpenStruct.new({ name: "Kestrel Hollow", products: [:__MEAT__, "EGGS"]}) },
     ])
 
     assert_equal [
@@ -198,8 +246,8 @@ def run_partials(string, partial_configs, **query_kwargs)
     fragment_node = document.definitions.first.selections.first.selections.first
     other_fragment_node = fragment_node.selections[1]
     results = run_partials(str, [
-      { fragment_node: fragment_node, type: PartialSchema::Farm, object: { name: "Belair Farm" } },
-      { fragment_node: other_fragment_node, type: PartialSchema::UpcasedFarm, object: { name: "Free Union Grass Farm" } }
+      { fragment_node: fragment_node, type: PartialSchema::Farm, object: OpenStruct.new({ name: "Belair Farm" }) },
+      { fragment_node: other_fragment_node, type: PartialSchema::UpcasedFarm, object: OpenStruct.new({ name: "Free Union Grass Farm" }) }
     ])
     assert_equal({ "name" => "Belair Farm", "n2" => "Belair Farm" }, results[0]["data"])
     assert_equal({ "n2" => "FREE UNION GRASS FARM" }, results[1]["data"])
@@ -215,15 +263,16 @@ def run_partials(string, partial_configs, **query_kwargs)
     }"
 
     node = GraphQL.parse(str).definitions.last
-    results = run_partials(str, [{ fragment_node: node, type: PartialSchema::Farm, object: { name: "Clovertop Creamery" } }])
+    results = run_partials(str, [{ fragment_node: node, type: PartialSchema::Farm, object: OpenStruct.new({ name: "Clovertop Creamery" }) }])
     assert_equal({ "farmName" => "Clovertop Creamery" }, results[0]["data"])
   end
 
   it "works with GraphQL::Current" do
     res = run_partials("query CheckCurrentValues { query { currentValues } }", [path: ["query"], object: nil])
-    assert_equal ["CheckCurrentValues", "Query.currentValues", "nil"], res[0]["data"]["currentValues"]
+    assert_equal ["CheckCurrentValues", TESTING_EXEC_NEXT ? nil : "Query.currentValues", "nil"], res[0]["data"]["currentValues"]
   end
 
+
   it "returns errors if they occur" do
     str = "{
       farm1: farm(id: \"1\") { error }
@@ -299,7 +348,7 @@ def run_partials(string, partial_configs, **query_kwargs)
 
   it "runs arrays and returns useful metadata in the result" do
     str = "{ farms { name } }"
-    results = run_partials(str, [{ path: ["farms"], object: [{ name: "Twenty Paces" }, { name: "Spring Creek Blooms" }]}])
+    results = run_partials(str, [{ path: ["farms"], object: [OpenStruct.new({ name: "Twenty Paces" }), OpenStruct.new({ name: "Spring Creek Blooms" })]}])
     result = results.first
     assert_equal [{ "name" => "Twenty Paces" }, { "name" => "Spring Creek Blooms" }], result["data"]
     assert_equal ["farms"], result.path
@@ -319,7 +368,7 @@ def run_partials(string, partial_configs, **query_kwargs)
     assert_equal "Twenty Paces", results[0]["data"]
     assert_equal "Caromont", results[1]["data"]
     assert_equal({
-      "errors" => [{"message" => "Boom!", "locations" => [{"line" => 1, "column" => 11}], "path" => ["query", "farmNames", 2, "farmNames"]}],
+      "errors" => [{"message" => "Boom!", "locations" => [{"line" => 1, "column" => 11}], "path" => ["query", "farmNames", 2, TESTING_EXEC_NEXT ? nil : "farmNames"].compact}],
       "data" => nil
     }, results[2])
   end
@@ -387,8 +436,8 @@ def run_partials(string, partial_configs, **query_kwargs)
     }"
 
     results = run_partials(str, [
-      { path: ["thing"], object: { name: "Whisper Hill" } },
-      { path: ["thing"], object: { is_market: true, name: "Crozet Farmers Market", is_year_round: false } },
+      { path: ["thing"], object: OpenStruct.new({ name: "Whisper Hill" }) },
+      { path: ["thing"], object: OpenStruct.new({ is_market: true, name: "Crozet Farmers Market", is_year_round: false })},
     ])
 
     assert_equal({ "name" => "Whisper Hill" }, results[0]["data"])
@@ -404,8 +453,8 @@ def run_partials(string, partial_configs, **query_kwargs)
     }"
 
     results = run_partials(str, [
-      { path: ["entity"], object: { name: "Whisper Hill" } },
-      { path: ["entity"], object: { is_market: true, name: "Crozet Farmers Market" } },
+      { path: ["entity"], object: OpenStruct.new({ name: "Whisper Hill" }) },
+      { path: ["entity"], object: OpenStruct.new({ is_market: true, name: "Crozet Farmers Market" }) },
     ])
 
     assert_equal({ "name" => "Whisper Hill", "__typename" => "Farm" }, results[0]["data"])
@@ -452,11 +501,14 @@ def run_partials(string, partial_configs, **query_kwargs)
       { path: ["q1", "q2", "query", "currentPath"], object: ["injected", "path"] },
     ])
 
-    assert_equal({"q1" => { "q2" => { "query" => { "currentPath" => ["q1", "q2", "query", "currentPath"] } } } }, results[0]["data"])
+    # No current path in exec-next
+    expected_path = TESTING_EXEC_NEXT ? nil : ["q1", "q2", "query", "currentPath"]
+
+    assert_equal({"q1" => { "q2" => { "query" => { "currentPath" => expected_path } } } }, results[0]["data"])
     assert_equal [], results[0].partial.path
-    assert_equal({"query" => {"currentPath" => ["q1", "q2", "query", "currentPath"]}}, results[1]["data"])
+    assert_equal({"query" => {"currentPath" => expected_path } }, results[1]["data"])
     assert_equal ["q1", "q2"], results[1].partial.path
-    assert_equal({ "currentPath" => ["q1", "q2", "query", "currentPath"] }, results[2]["data"])
+    assert_equal({ "currentPath" => expected_path }, results[2]["data"])
     assert_equal ["q1", "q2", "query"], results[2].partial.path
     assert_equal(["injected", "path"], results[3]["data"])
     assert_equal ["q1", "q2", "query", "currentPath"], results[3].partial.path
@@ -466,7 +518,7 @@ def run_partials(string, partial_configs, **query_kwargs)
     str = "mutation { updateFarm(name: \"Brawndo Acres\") { name } }"
     results = run_partials(str, [
       { path: [], object: nil },
-      { path: ["updateFarm"], object: { name: "Georgetown Farm" } },
+      { path: ["updateFarm"], object: OpenStruct.new({ name: "Georgetown Farm" }) },
       { path: ["updateFarm", "name"], object: "Notta Farm" },
     ])
 
@@ -484,7 +536,7 @@ def run_partials(string, partial_configs, **query_kwargs)
     }"
 
     results = run_partials(str, [
-      { path: ["entity"], object: { name: GraphQL::ExecutionError.new("Boom!") } },
+      { path: ["entity"], object: OpenStruct.new({ name: GraphQL::ExecutionError.new("Boom!") }) },
       { path: ["entity", "name"], object: GraphQL::ExecutionError.new("Bang!") },
       { path: ["entity", "name"], object: -> { GraphQL::ExecutionError.new("Blorp!") } },
     ])
@@ -494,11 +546,11 @@ def run_partials(string, partial_configs, **query_kwargs)
       "data" => { "name" => nil, "__typename" => "Farm" }
     }, results[0])
     assert_equal({
-      "errors" => [{"message" => "Bang!", "locations" => [{"line" => 3, "column" => 9}], "path" => ["entity", "name", "name"]}],
+      "errors" => [{"message" => "Bang!", "locations" => [{"line" => 3, "column" => 9}], "path" => ["entity", "name", TESTING_EXEC_NEXT ? nil : "name"].compact}],
       "data" => nil
     }, results[1])
     assert_equal({
-      "errors" => [{"message" => "Blorp!", "locations" => [{"line" => 3, "column" => 9}], "path" => ["entity", "name", "name"]}],
+      "errors" => [{"message" => "Blorp!", "locations" => [{"line" => 3, "column" => 9}], "path" => ["entity", "name",  TESTING_EXEC_NEXT ? nil : "name" ].compact}],
       "data" => nil
     }, results[2])
   end
diff --git a/spec/graphql/schema/directive_spec.rb b/spec/graphql/schema/directive_spec.rb
index 99573325459..f6c6c4de356 100644
--- a/spec/graphql/schema/directive_spec.rb
+++ b/spec/graphql/schema/directive_spec.rb
@@ -141,35 +141,98 @@ def self.resolve(obj, args, ctx)
         ctx[:count_fields][path] << field_count
         nil # this does nothing
       end
+
+      class FieldCount
+        include GraphQL::Execution::Next::Finalizer
+
+        def initialize(ast_node)
+          @ast_node = ast_node
+        end
+
+        def finalize_graphql_result(query, result_data, result_key)
+          counts = query.context[:count_fields] ||= {}
+          count = case @ast_node
+          when GraphQL::Language::Nodes::Field
+            1
+          when GraphQL::Language::Nodes::InlineFragment
+            @ast_node.selections.size
+          when GraphQL::Language::Nodes::FragmentSpread
+            frag = query.fragments[@ast_node.name]
+            frag.selections.size
+          else
+            raise ArgumentError, "Unexpected ast_node: #{ast_node.inspect}"
+          end
+          counts[path] ||= [count]
+        end
+      end
+
+      def self.resolve_field(ast_nodes, parent_type, field_defn, objects, arguments, context)
+        FieldCount.new(ast_nodes.first)
+      end
+
+      def self.resolve_inline_fragment(ast_node, parent_type, objects, _args, context)
+        FieldCount.new(ast_node)
+      end
+
+      def self.resolve_fragment_spread(ast_node, parent_type, objects, _args, context)
+        FieldCount.new(ast_node)
+      end
     end
 
     class Thing < GraphQL::Schema::Object
-      field :name, String, null: false
+      field :name, String, null: false, hash_key: :name
     end
 
     module HasThings
       include GraphQL::Schema::Interface
-      field :thing, Thing, null: false, extras: [:ast_node]
+      field :thing, Thing, null: false, extras: [:ast_node], resolve_static: true
 
       def thing(ast_node:)
-        context[:name_resolved_count] ||= 0
-        context[:name_resolved_count] += 1
-        { name: ast_node.alias || ast_node.name }
+        self.class.thing(context, ast_node: ast_node)
+      end
+
+      resolver_methods do
+        def thing(context, ast_node:)
+          context[:name_resolved_count] ||= 0
+          context[:name_resolved_count] += 1
+          { name: ast_node.alias || ast_node.name }
+        end
       end
 
-      field :lazy_thing, Thing, null: false, extras: [:ast_node]
+      field :lazy_thing, Thing, null: false, extras: [:ast_node], resolve_static: true
+
       def lazy_thing(ast_node:)
-        -> { thing(ast_node: ast_node) }
+        self.class.lazy_thing(context, ast_node: ast_node)
+      end
+
+      resolver_methods do
+        def lazy_thing(context, ast_node:)
+          -> { self.thing(context, ast_node: ast_node) }
+        end
       end
 
-      field :dataloaded_thing, Thing, null: false, extras: [:ast_node]
+      field :dataloaded_thing, Thing, null: false, extras: [:ast_node], resolve_static: true
+
       def dataloaded_thing(ast_node:)
-        dataloader.with(ThingSource).load(ast_node.alias || ast_node.name)
+        self.class.dataloaded_thing(context, ast_node: ast_node)
+      end
+
+      resolver_methods do
+        def dataloaded_thing(context, ast_node:)
+          context.dataload(ThingSource, ast_node.alias || ast_node.name)
+        end
       end
 
-      field :lazy_things, [Thing], extras: [:ast_node]
+      field :lazy_things, [Thing], extras: [:ast_node], resolve_static: true
+
       def lazy_things(ast_node:)
-        -> { [thing(ast_node: ast_node), thing(ast_node: ast_node)]}
+        self.class.lazy_things(context, ast_node: ast_node)
+      end
+
+      resolver_methods do
+        def lazy_things(context, ast_node:)
+          -> { [self.thing(context, ast_node: ast_node), self.thing(context, ast_node: ast_node)]}
+        end
       end
     end
 
@@ -190,6 +253,15 @@ class Schema < GraphQL::Schema
       directive(CountFields)
       lazy_resolve(Proc, :call)
       use GraphQL::Dataloader
+      use GraphQL::Execution::Next
+    end
+  end
+
+  def exec_query(...)
+    if TESTING_EXEC_NEXT
+      RuntimeDirectiveTest::Schema.execute_next(...)
+    else
+      RuntimeDirectiveTest::Schema.execute(...)
     end
   end
 
@@ -226,7 +298,7 @@ class Schema < GraphQL::Schema
       }
       GRAPHQL
 
-      res = RuntimeDirectiveTest::Schema.execute(query_str)
+      res = exec_query(query_str)
       expected_data = {
         "t1" => {
           "t1n" => "t1",
@@ -268,7 +340,7 @@ class Schema < GraphQL::Schema
         }
       }
       GRAPHQL
-      res = RuntimeDirectiveTest::Schema.execute(query_str)
+      res = exec_query(query_str)
       expected_data = { "t1" => { "name" => "t1"}, "t2" => { "name" => "t2" }, "t3" => { "name" => "t3" } }
       assert_graphql_equal expected_data, res["data"]
 
@@ -288,7 +360,7 @@ class Schema < GraphQL::Schema
         }
       }
       "
-      res = RuntimeDirectiveTest::Schema.execute(query_str, context: { backtrace: true })
+      res = exec_query(query_str, context: { backtrace: true })
       assert_equal 2, res["data"]["lazyThings"].size
     end
   end
diff --git a/spec/graphql/tracing/trace_spec.rb b/spec/graphql/tracing/trace_spec.rb
index 2a54210dd10..b6fe4a8029d 100644
--- a/spec/graphql/tracing/trace_spec.rb
+++ b/spec/graphql/tracing/trace_spec.rb
@@ -8,4 +8,99 @@
     superable_methods_source = superable_methods.map { |m| "        #{m.inspect},\n" }.join
     assert_includes trace_source, superable_methods_source
   end
+
+
+  describe "object hooks" do
+    class ObjectHooksSchema < GraphQL::Schema
+      module Node
+        include GraphQL::Schema::Interface
+        field :name, String
+      end
+
+      class Thing < GraphQL::Schema::Object
+        implements Node
+      end
+
+      class Query < GraphQL::Schema::Object
+        field :things, [Thing], resolve_static: true
+
+        def self.things(context)
+          [OpenStruct.new(name: "Thing One"), OpenStruct.new(name: "Thing Two")]
+        end
+
+        field :thing, Thing, resolve_static: true do
+          argument :id, ID, loads: Thing, as: :thing
+        end
+
+        def self.thing(context, thing:)
+          thing
+        end
+
+        field :thing_name, String, resolve_static: true do
+          argument :thing_id, ID, loads: Thing
+        end
+
+        def self.thing_name(context, thing:)
+          thing.name
+        end
+
+        field :node, Node, resolve_static: true do
+          argument :id, ID, loads: Node, as: :node
+        end
+
+        def self.node(context, node:)
+          node
+        end
+      end
+
+      query(Query)
+      use GraphQL::Execution::Next
+
+      def self.object_from_id(id, ctx)
+        OpenStruct.new(name: "Thing ##{id}")
+      end
+
+      def self.resolve_type(abs_type, obj, ctx)
+        Thing
+      end
+
+      module LogTrace
+        def objects(type, objects, context)
+          context[:log] ||= []
+          context[:log] << "#{objects.size} objects as #{type.graphql_name}"
+          super
+        end
+
+        def object_loaded(argument_definition, object, context)
+          context[:log] ||= []
+          context[:log] << "#{argument_definition.path} loaded #{object.class}"
+          super
+        end
+      end
+
+      trace_with(LogTrace)
+    end
+
+    it "calls hooks with errors encountered during execution" do
+      res = ObjectHooksSchema.execute_next("{ things { name } }")
+      assert_equal ["Thing One", "Thing Two"], res["data"]["things"].map { |t| t["name"] }
+      assert_equal ["1 objects as Query", "2 objects as Thing"], res.context[:log]
+
+      res = ObjectHooksSchema.execute_next("{ thing(id: \"5\") { name } }")
+      assert_equal "Thing #5", res["data"]["thing"]["name"]
+      assert_equal ["1 objects as Query", "Query.thing.id loaded OpenStruct", "1 objects as Thing"], res.context[:log]
+
+      res = ObjectHooksSchema.execute_next("{ thingName(thingId: \"77\") }")
+      assert_equal "Thing #77", res["data"]["thingName"]
+      assert_equal ["1 objects as Query", "Query.thingName.thingId loaded OpenStruct"], res.context[:log]
+
+      res = ObjectHooksSchema.execute_next("{ node(id: \"33\") { name } }")
+      assert_equal "Thing #33", res["data"]["node"]["name"]
+      assert_equal ["1 objects as Query", "Query.node.id loaded OpenStruct", "1 objects as Thing"], res.context[:log]
+
+      partial_res = res.query.run_partials([{ path: ["node"], object: OpenStruct.new(name: "Injected thing"), context: { log: [] } }])
+      assert_equal "Injected thing", partial_res[0]["data"]["name"]
+      assert_equal ["1 objects as Thing"], partial_res[0].context[:log]
+    end
+  end
 end
diff --git a/spec/support/dummy/schema.rb b/spec/support/dummy/schema.rb
index 0fbcb6c7f10..de083b2e85c 100644
--- a/spec/support/dummy/schema.rb
+++ b/spec/support/dummy/schema.rb
@@ -550,6 +550,7 @@ class Schema < GraphQL::Schema
     orphan_types Honey
     trace_with GraphQL::Tracing::CallLegacyTracers
     directives(DirectiveForVariableDefinition)
+    use GraphQL::Execution::Next
 
     rescue_from(NoSuchDairyError) { |err| raise GraphQL::ExecutionError, err.message  }