From 6a7ba6188d98d8cfcdaf5b0314cf8b6ef70ad6b4 Mon Sep 17 00:00:00 2001
From: Greg MacWilliam <greg.macwilliam@shopify.com>
Date: Sun, 21 Jun 2026 20:54:52 -0400
Subject: [PATCH] add incremental @defer support.

---
 README.md                                     |  56 +-
 Rakefile                                      |   2 +-
 lib/graphql/breadth_exec.rb                   |   1 +
 lib/graphql/breadth_exec/errors.rb            |   1 -
 lib/graphql/breadth_exec/executor.rb          | 229 +++--
 .../breadth_exec/executor/error_formatter.rb  |   3 -
 .../breadth_exec/executor/execution_field.rb  |  21 +-
 .../executor/execution_planner.rb             | 140 ++-
 .../executor/execution_promise.rb             | 347 ++++---
 .../breadth_exec/executor/execution_scope.rb  |  32 +-
 .../breadth_exec/executor/lazy_element.rb     |   2 +-
 lib/graphql/breadth_exec/field_resolvers.rb   |  31 +-
 .../breadth_exec/has_breadth_resolver.rb      |   6 +-
 lib/graphql/breadth_exec/incremental.rb       |  18 +
 .../breadth_exec/incremental/context.rb       | 187 ++++
 .../breadth_exec/incremental/defer_usage.rb   |  22 +
 .../incremental/deferred_delivery.rb          |  42 +
 .../incremental/deferred_execution_scope.rb   |  67 ++
 .../breadth_exec/incremental/partitioner.rb   |  84 ++
 .../breadth_exec/incremental/publisher.rb     |  65 ++
 .../breadth_exec/incremental/result.rb        |  38 +
 .../breadth_exec/incremental/selection.rb     |  22 +
 lib/graphql/breadth_exec/lazy_loader.rb       |  10 +-
 .../subscription_response_stream.rb           |   1 +
 lib/graphql/breadth_exec/tracer.rb            |  40 +-
 sorbet/config                                 |   2 +
 sorbet/rbi/shims/graphql.rbi                  |  46 +
 .../executor/execution_promise_test.rb        |  14 +-
 .../breadth_exec/executor/incremental_test.rb | 934 ++++++++++++++++++
 .../executor/subscriptions_test.rb            |  56 +-
 .../breadth_exec/executor/tracers_test.rb     |  72 +-
 .../breadth_exec/field_resolvers_test.rb      |   4 +-
 .../breadth_exec/incremental/context_test.rb  |  51 +
 .../incremental/publisher_test.rb             |  58 ++
 34 files changed, 2382 insertions(+), 322 deletions(-)
 create mode 100644 lib/graphql/breadth_exec/incremental.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/context.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/defer_usage.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/deferred_delivery.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/deferred_execution_scope.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/partitioner.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/publisher.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/result.rb
 create mode 100644 lib/graphql/breadth_exec/incremental/selection.rb
 create mode 100644 test/graphql/breadth_exec/executor/incremental_test.rb
 create mode 100644 test/graphql/breadth_exec/incremental/context_test.rb
 create mode 100644 test/graphql/breadth_exec/incremental/publisher_test.rb

diff --git a/README.md b/README.md
index eedef8a..cf0a200 100644
--- a/README.md
+++ b/README.md
@@ -2,13 +2,34 @@
 
 _**The core algorithm backing Shopify's _GraphQL Cardinal_ engine.** Learn more about the breadth-first GraphQL design advantages in the [blog post](https://shopify.engineering/faster-breadth-first-graphql-execution). For a typescipt port, see [graphql-breadth-js](https://github.com/gmac/graphql-breadth-js)._
 
+* Runs field executions breadth-first, layer by layer (versus depth-first, tree by tree).
+* Individual resolvers are implicitly batched.
+* Lazy resolvers sharing I/O bind entire object sets to a single promise.
+* Processes via queuing rather than recursion.
+
+```
+ruby 3.2.1 (2023-02-08 revision 31819e82c8) +YJIT [arm64-darwin23]
+
+Non-lazy comparison:
+graphql-breadth_exec: 1000 x 3 scalars:     1257.1 i/s
+graphql-ruby resolve_batch: 1000 x 3 scalars:      773.2 i/s - 1.63x  slower
+graphql-ruby classic: 1000 x 3 scalars:      102.0 i/s - 12.32x  slower
+
+Lazy comparison:
+graphql-breadth_exec LazyLoader: 1000 x 1 lazy scalar:     2469.5 i/s
+graphql-ruby execute_next + dataloader: 1000 x 1 lazy scalar:      533.1 i/s - 4.63x  slower
+graphql-ruby execute_next + graphql-batch: 1000 x 1 lazy scalar:      291.7 i/s - 8.47x  slower
+graphql-ruby graphql-batch: 1000 x 1 lazy scalar:      178.7 i/s - 13.82x  slower
+graphql-ruby dataloader: 1000 x 1 lazy scalar:      114.6 i/s - 21.56x  slower
+```
+
 # Support
 
 The execution algorithm is proven at scale in production. This implementation still has some limitations:
 
 * Uses GraphQL Ruby schemas.
 * Currently no built-in validation or analysis, do it ahead of time.
-* Currently no defer or stream.
+* Currently no stream.
 * Supports input validations, but NOT input transformations (ie: "prepare" hooks).
 
 # Usage
@@ -656,9 +677,29 @@ end
 
 This architecture makes cascading resolvers run repeatedly on every field in a subtree, rather than just once at the top of the owning field's subtree. This pattern is more granular and generally safer for isolation and parallelism, though has more resolver churn than a typical depth traversal so should be used accordingly.
 
+## Incremental results (`@defer`)
+
+Query and mutation operations that may contain `@defer` should use `incremental_result`. This always returns a `GraphQL::BreadthExec::Incremental::Result`, even when the operation has no active deferred work:
+
+```ruby
+result = executor.incremental_result
+
+deliver(result.initial_result)
+
+if result.incremental?
+  result.subsequent_results.each do |payload|
+    deliver(payload)
+  end
+end
+```
+
+When no deferred work is active, `initial_result` is the normal GraphQL result hash and `incremental?` is false. When deferred work is active, `initial_result` includes pending records and `hasNext`, and `subsequent_results` yields later incremental payloads.
+
+The basic and incremental entry points are intentionally strict. Call either `result` OR `incremental_result` for a query or mutation executor depending on the request's support for incremental delivery (ex: multi-part and SSE requests); switching entry points after execution has started raises an implementation error.
+
 ## Subscriptions
 
-Query and mutation execution use `result`, which always returns a normal GraphQL result hash. If a controller accepts subscription operations, use `result_or_subscribe` instead. For query and mutation operations it returns the same result hash as `result`; for subscription operations it returns a `GraphQL::BreadthExec::SubscriptionResponseStream` on successful source setup, or a normal GraphQL result hash for public setup errors. Calling `result` for a subscription operation raises an implementation error.
+Query and mutation execution use `result`, which always returns a normal GraphQL result hash. Subscription operations use `subscribe`, which returns a `GraphQL::BreadthExec::SubscriptionResponseStream` on successful source setup, or a normal GraphQL result hash for public setup errors. Each entry point is strict about its operation type, matching the `execute` / `subscribe` split in graphql-js: calling `result` (or `incremental_result`) for a subscription operation raises an implementation error, and calling `subscribe` for a query or mutation operation raises an implementation error. A controller that accepts both inspects the operation type and dispatches accordingly:
 
 ```ruby
 executor = GraphQL::BreadthExec::Executor.new(
@@ -668,14 +709,13 @@ executor = GraphQL::BreadthExec::Executor.new(
   context: { ... },
 )
 
-response = executor.result_or_subscribe
-
-if response.is_a?(GraphQL::BreadthExec::SubscriptionResponseStream)
-  response.each do |event_result|
+if executor.subscription?
+  stream = executor.subscribe
+  stream.each do |event_result|
     deliver(event_result)
   end
 else
-  deliver(response)
+  deliver(executor.result)
 end
 ```
 
@@ -726,7 +766,7 @@ executor = GraphQL::BreadthExec::Executor.new(
   context: { write_value_events: write_value_events },
 )
 
-stream = executor.result_or_subscribe
+stream = executor.subscribe
 stream.each do |event_result|
   # {"data"=>{"onWriteValue"=>{"value"=>"first"}}}
   # {"data"=>{"onWriteValue"=>{"value"=>"second"}}}
diff --git a/Rakefile b/Rakefile
index 4a27b10..672fbc4 100644
--- a/Rakefile
+++ b/Rakefile
@@ -56,7 +56,7 @@ namespace :benchmark do
   end
 
   desc "Benchmark one lazy scalar field across many objects"
-  task :lazy_field_batch do
+  task :resolve_lazy_batch do
     prepare_benchmark
     GraphQLBenchmark.benchmark_lazy_field_batch
   end
diff --git a/lib/graphql/breadth_exec.rb b/lib/graphql/breadth_exec.rb
index 945c74c..39e7527 100644
--- a/lib/graphql/breadth_exec.rb
+++ b/lib/graphql/breadth_exec.rb
@@ -48,4 +48,5 @@ class ExecutionField; end
 require_relative "breadth_exec/has_breadth_resolver"
 require_relative "breadth_exec/introspection"
 require_relative "breadth_exec/executor"
+require_relative "breadth_exec/incremental"
 require_relative "breadth_exec/version"
diff --git a/lib/graphql/breadth_exec/errors.rb b/lib/graphql/breadth_exec/errors.rb
index 5894cf2..1ea305b 100644
--- a/lib/graphql/breadth_exec/errors.rb
+++ b/lib/graphql/breadth_exec/errors.rb
@@ -6,7 +6,6 @@
 module GraphQL
   module BreadthExec
     class Error < StandardError; end
-    class DocumentError < Error; end
     class ImplementationError < Error; end
     class MethodNotImplementedError < Error; end
 
diff --git a/lib/graphql/breadth_exec/executor.rb b/lib/graphql/breadth_exec/executor.rb
index 36d381e..b808fd9 100644
--- a/lib/graphql/breadth_exec/executor.rb
+++ b/lib/graphql/breadth_exec/executor.rb
@@ -16,7 +16,7 @@ module GraphQL
   module BreadthExec
     class Executor
       #: type type_resolver = ^(untyped, GraphQL::Query::Context) -> singleton(GraphQL::Schema::Member)?
-      #: type resolver = FieldResolver[untyped] | DirectiveResolver | type_resolver
+      #: type resolver = FieldResolver | DirectiveResolver | type_resolver
 
       EMPTY_ARRAY = [].freeze
       EMPTY_OBJECT = {}.freeze
@@ -44,6 +44,9 @@ class Executor
       #: ExecutionPlanner
       attr_reader :planner
 
+      #: Incremental::Context?
+      attr_reader :incremental
+
       #: Hash[untyped, singleton(GraphQL::Schema::Object)]?
       attr_reader :abstract_result_types
 
@@ -71,7 +74,7 @@ def initialize(schema, document, resolvers: EMPTY_OBJECT, root_object: nil, vari
         @root_object = root_object
         @context_value = context
         @tracers = tracers
-        @result = {}
+        @result = nil
         @data = {}
         @errors = []
         @exec_queue = []
@@ -80,10 +83,26 @@ def initialize(schema, document, resolvers: EMPTY_OBJECT, root_object: nil, vari
         @abstract_result_types = nil
         @loader_cache = nil
         @paths = nil
+        @incremental = nil
         @executed = false
         @aborted = false
       end
 
+      #: -> bool
+      def query?
+        @query.selected_operation.operation_type == ExecutionPlanner::QUERY_OPERATION
+      end
+
+      #: -> bool
+      def mutation?
+        @query.selected_operation.operation_type == ExecutionPlanner::MUTATION_OPERATION
+      end
+
+      #: -> bool
+      def subscription?
+        @query.selected_operation.operation_type == ExecutionPlanner::SUBSCRIPTION_OPERATION
+      end
+
       #: -> Hash[String, GraphQL::Language::Nodes::FragmentDefinition]
       def fragments
         @query.fragments
@@ -94,6 +113,11 @@ def variables
         @input.variables
       end
 
+      #: -> extensions_hash
+      def response_extensions
+        @context.response_extensions
+      end
+
       #: -> PathFormatter
       def paths
         @paths ||= PathFormatter.new
@@ -101,26 +125,42 @@ def paths
 
       #: -> graphql_result
       def result
-        operation = @query.selected_operation
-        if operation.operation_type == ExecutionPlanner::SUBSCRIPTION_OPERATION
-          raise ImplementationError, "Use result_or_subscribe for subscription operations"
-        end
+        raise ImplementationError, "Use subscribe for subscription operations" if subscription?
+        raise ImplementationError, "Cannot call result after incremental_result" if incremental?
 
         return @result if executed?
 
-        execute
+        @result = execute
+      end
+
+      #: -> Incremental::Result
+      def incremental_result
+        raise ImplementationError, "Use subscribe for subscription operations" if subscription?
+
+        return @result if @result.is_a?(Incremental::Result)
+        raise ImplementationError, "Cannot call incremental_result after result" if executed?
+
+        @incremental = Incremental::Context.new(self, data: @data)
+        initial_result = execute
+        subsequent_results = EMPTY_ARRAY
+        pending_deliveries = @incremental.prepare_pending
+
+        unless pending_deliveries.empty?
+          initial_result["pending"] = @incremental.pending_payloads(pending_deliveries)
+          initial_result["hasNext"] = true
+          subsequent_results = Enumerator.new { execute_next_incremental_result(_1) }
+        end
+
+        @result = Incremental::Result.new(initial_result: initial_result, subsequent_results: subsequent_results)
       end
 
       #: -> (SubscriptionResponseStream | graphql_result)
-      def result_or_subscribe
+      def subscribe
+        raise ImplementationError, "Only allowed for subscription operations" unless subscription?
+
         return @result if executed?
 
-        operation = @query.selected_operation
-        @result = if operation.operation_type == ExecutionPlanner::SUBSCRIPTION_OPERATION
-          subscribe(operation)
-        else
-          execute
-        end
+        @result = execute_subscription
       end
 
       #: (untyped) -> graphql_result
@@ -162,19 +202,21 @@ def handle_or_reraise(original_error, exec_field: nil)
             error
           rescue Exception => exception
             @raised_exception = exception
-            @tracers.each { _1.on_exception(exception, @context, exec_field:) }
+            unless @tracers.empty?
+              @tracers.each { _1.on_exception(exception, @context, exec_field:) }
+            end
             raise exception
           end
           ExecutionError.from(handled_error, exec_field:, cause: original_error)
         end
       end
 
-      #: (ExecutionError, ?untyped?, ?exec_field: ExecutionField[untyped]?) -> ExecutionError
+      #: (ExecutionError, ?untyped?, ?exec_field: ExecutionField[untyped]?) -> StandardError
       def add_error(error, result = nil, exec_field: nil)
         error = exec_field ? ExecutionError.from(error, exec_field:) : error
 
         if exec_field
-          current_type = exec_field.type
+          current_type = exec_field.type #: untyped
           current_type = current_type.of_type while current_type.list? || (current_type.non_null? && current_type.of_type.list?)
           error = invalidate_non_null_value(exec_field, current_type, error) if current_type.non_null?
         end
@@ -183,11 +225,6 @@ def add_error(error, result = nil, exec_field: nil)
         error
       end
 
-      #: -> Integer
-      def error_count
-        @invalidated_results.size
-      end
-
       #: -> ErrorFormatter
       def error_result_formatter
         @error_result_formatter ||= ErrorFormatter.new(
@@ -207,6 +244,11 @@ def executed?
         @executed
       end
 
+      #: -> bool
+      def incremental?
+        !!@incremental
+      end
+
       protected
 
       #: -> graphql_result
@@ -249,18 +291,7 @@ def execute
 
             @planner.plan_scopes(root_scopes).each do |exec_scope|
               @exec_queue << exec_scope
-
-              # Process eager scopes, then process resulting lazy fields
-              # Work is always dequeued for processing to maintain a clean queue state
-              until aborted? || (@exec_queue.empty? && @lazy_queue.empty?)
-                if !@exec_queue.empty?
-                  exec_scope = @exec_queue.shift #: as !nil
-                  execute_scope(exec_scope)
-                else
-                  lazy_elements = @lazy_queue.shift(@lazy_queue.length)
-                  execute_lazy(lazy_elements)
-                end
-              end
+              run!
             end
 
             unless @tracers.empty?
@@ -306,6 +337,23 @@ def execute
 
       private
 
+      #: (?Array[ExecutionScope]) -> void
+      def run!(exec_scopes = EMPTY_ARRAY)
+        @exec_queue.concat(exec_scopes) unless exec_scopes.empty?
+
+        # Process eager scopes, then process resulting lazy fields.
+        # Work is always dequeued for processing to maintain a clean queue state.
+        until aborted? || (@exec_queue.empty? && @lazy_queue.empty?)
+          if !@exec_queue.empty?
+            exec_scope = @exec_queue.shift #: as !nil
+            execute_scope(exec_scope)
+          else
+            lazy_elements = @lazy_queue.shift(@lazy_queue.length)
+            execute_lazy(lazy_elements)
+          end
+        end
+      end
+
       #: (
       #|   Array[ExecutionDirective],
       #|   ?current_field: ExecutionField[untyped]?,
@@ -313,7 +361,7 @@ def execute
       #| ) { () -> untyped } -> untyped
       def execute_with_directives(exec_directives, current_field: nil, at: 0, &block)
         while at < exec_directives.size
-          exec_directive = exec_directives[at]
+          exec_directive = exec_directives[at] #: as !nil
 
           if exec_directive.resolver.applies?(exec_directive, current_field)
             exec_directive.validate!
@@ -371,6 +419,10 @@ def execute_scope(exec_scope)
           return
         end
 
+        unless @tracers.empty?
+          @tracers.each { _1.before_scope(exec_scope, @context) }
+        end
+
         exec_scope.fields.each_value do |exec_field|
           execute_field(exec_field) unless exec_field.scope.aborted?
         end
@@ -406,7 +458,7 @@ def execute_lazy(lazy_elements)
         pending_loaders = (@loader_cache || EMPTY_OBJECT).each_value.reject { _1.promised.empty? }
 
         if pending_loaders.empty?
-          raise ImplementationError, "Lazy #{lazy_elements.first&.inspect} produced a promise without a loader"
+          raise ImplementationError, "Lazy #{lazy_elements.first} produced a promise without a loader"
         end
 
         pending_loaders.each do |loader|
@@ -420,7 +472,12 @@ def execute_lazy(lazy_elements)
             next
           end
 
+          lazy_start_time = EMPTY_TIMER
           begin
+            unless @tracers.empty?
+              lazy_start_time = timer
+              @tracers.each { _1.before_lazy_set(loader, loader_elements, @context) }
+            end
             loader.execute!(@context)
           rescue StandardError => e
             handled_error = handle_or_reraise(e)
@@ -435,6 +492,11 @@ def execute_lazy(lazy_elements)
                 element.abort!
               end
             end
+          ensure
+            unless @tracers.empty?
+              duration = timer(lazy_start_time)
+              @tracers.each { _1.after_lazy_set(loader, loader_elements, @context, duration:) }
+            end
           end
         end
 
@@ -512,7 +574,7 @@ def promise_resolved?(promise, element:)
         if promise.rejected?
           reason = promise.reason
           unless reason.is_a?(StandardError)
-            reason = UnknownLazyRejectionError.new("Lazy #{element.inspect} was rejected for an unknown reason: #{reason}")
+            reason = UnknownLazyRejectionError.new("Lazy #{element} was rejected for an unknown reason: #{reason}")
           end
 
           exec_field = case element
@@ -545,12 +607,14 @@ def execute_field(exec_field)
           return
         end
 
-        parent_type = exec_field.scope.parent_type
-        field_name = exec_field.name
         parent_objects = exec_field.scope.objects
 
+        resolve_start_time = EMPTY_TIMER
         begin
-          @tracers.each { _1.before_resolve_field(parent_type, field_name, parent_objects.length, @context) }
+          unless @tracers.empty?
+            resolve_start_time = timer
+            @tracers.each { _1.before_resolve_field(exec_field, @context) }
+          end
           exec_field.lazy_state_executing!
           exec_field.validate!
           pre_authorized = @authorization.authorized_field?(exec_field, @context)
@@ -571,7 +635,10 @@ def execute_field(exec_field)
           @errors << error if error.base_error?
           exec_field.result = Array.new(parent_objects.length, error)
         ensure
-          @tracers.each { _1.after_resolve_field(parent_type, field_name, parent_objects.length, @context) }
+          unless @tracers.empty?
+            duration = timer(resolve_start_time)
+            @tracers.each { _1.after_resolve_field(exec_field, @context, duration:) }
+          end
         end
 
         if exec_field.lazy_result?
@@ -784,7 +851,11 @@ def propagate_null!(exec_field)
         if highest_nulled_depth.zero?
           # Mark the entire executor as aborted when non-null propagation hits the top.
           # This prevents subsequent isolated root scopes (mutations) from running.
-          @aborted = true
+          if (deferred_root = exec_field.scope.deferred_root)
+            deferred_root.abort!
+          else
+            @aborted = true
+          end
         elsif highest_list_depth.negative? || highest_nulled_depth <= highest_list_depth
           # Abort all non-null ancestor scopes that meet or exceed the highest-level list.
           # (all lists must be completely invalidated, or else remain alive).
@@ -848,6 +919,9 @@ def build_abstract_scopes(exec_field, return_type, next_objects, next_results)
 
         scopes = []
         abstract_scope = AbstractExecutionScope.new(parent_type: return_type, parent_field: exec_field, scopes: scopes)
+        unless @tracers.empty?
+          @tracers.each { _1.before_abstract_scope(abstract_scope, @context) }
+        end
         next_objects_by_type.each do |impl_type, impl_type_objects|
           scopes << ExecutionScope.new(
             executor: self,
@@ -855,6 +929,7 @@ def build_abstract_scopes(exec_field, return_type, next_objects, next_results)
             parent_field: exec_field,
             parent_type: impl_type,
             selections: exec_field.selections,
+            deferred: exec_field.scope.deferred?,
             objects: impl_type_objects,
             results: next_results_by_type[impl_type],
           )
@@ -863,9 +938,10 @@ def build_abstract_scopes(exec_field, return_type, next_objects, next_results)
         @exec_queue.concat(@planner.plan_scopes(scopes))
       end
 
-      #: (GraphQL::Language::Nodes::OperationDefinition) -> (SubscriptionResponseStream | graphql_result)
-      def subscribe(operation)
+      #: -> (SubscriptionResponseStream | graphql_result)
+      def execute_subscription
         @executed = true
+        operation = @query.selected_operation
 
         unless @context.errors.empty?
           return build_result(errors: @context.errors.map(&:to_h))
@@ -909,8 +985,55 @@ def subscribe(operation)
         end
       end
 
+      #: (Enumerator::Yielder) -> void
+      def execute_next_incremental_result(yielder)
+        pending_payloads = []
+        incremental_payloads = []
+        completed_deliveries = []
+        completed_errors_by_delivery = {}.compare_by_identity
+
+        loop do
+          ready_scopes = @incremental.ready_scopes
+          break if ready_scopes.empty?
+
+          initial_error_count = @invalidated_results.size
+          run!(@planner.plan_scopes(ready_scopes))
+          has_errors = @invalidated_results.size > initial_error_count
+
+          ready_scopes.each do |exec_scope|
+            deliveries = @incremental.deliveries_for(exec_scope)
+            deliveries.each do |index, path, deferred_deliveries|
+              data = exec_scope.results[index]
+              errors = EMPTY_ARRAY
+              if has_errors
+                data, errors = error_result_formatter.format_object(exec_scope.parent_type, exec_scope.selections, data, path)
+              end
+
+              if data.nil? && !errors.empty?
+                deferred_deliveries.each { (completed_errors_by_delivery[_1] ||= []).concat(errors) }
+              else
+                incremental_payloads << @incremental.incremental_payload(deferred_deliveries, path, data, errors:)
+              end
+              completed_deliveries.concat(deferred_deliveries)
+            end
+
+            exec_scope.executed = true
+            pending_payloads.concat(@incremental.pending_payloads(@incremental.prepare_pending))
+          end
+        end
+
+        completed_payloads = @incremental.completed_payloads(completed_deliveries, errors_by_delivery: completed_errors_by_delivery)
+        payload = { "hasNext" => false }
+        payload["pending"] = pending_payloads unless pending_payloads.empty?
+        payload["incremental"] = incremental_payloads unless incremental_payloads.empty?
+        payload["completed"] = completed_payloads unless completed_payloads.empty?
+        yielder << payload
+      end
+
       #: (?data: Util::NilLike | graphql_result | nil, ?errors: Array[error_hash]) -> graphql_result
       def build_result(data: UNDEFINED, errors: EMPTY_ARRAY)
+        result = {}
+
         # Truncate errors but preserve GraphQL-Ruby's validation error limit message when present.
         # For execution errors, add our own message to inform truncation.
         if @max_reported_errors && errors.size > @max_reported_errors
@@ -921,23 +1044,19 @@ def build_result(data: UNDEFINED, errors: EMPTY_ARRAY)
         end
 
         # install errors first to surface at the top
-        @result["errors"] = errors unless errors.empty?
+        result["errors"] = errors unless errors.empty?
 
         # only install data when execution has run and generated (possibly partial) results
-        @result["data"] = data unless data.equal?(UNDEFINED)
+        result["data"] = data unless data.equal?(UNDEFINED)
 
-        # collect extensions that have already been added to the result by tracers,
-        # and check for extensions defined through `@context.response_extensions`.
-        # extensions always get re-added to the result as the final key
-        result_extensions = @result.delete("extensions")
+        # Check for extensions defined through `@context.response_extensions`.
+        # Extensions always get added to the result as the final key.
         has_context_extensions = @context.namespace?(:__query_result_extensions__)
-        if result_extensions || has_context_extensions
-          result_extensions ||= {}
-          result_extensions.merge!(@context.namespace(:__query_result_extensions__)) if has_context_extensions
-          @result["extensions"] = Util.deep_copy(result_extensions, stringify_keys: true)
+        if has_context_extensions
+          result["extensions"] = Util.deep_copy(response_extensions, stringify_keys: true)
         end
 
-        @result
+        result
       end
 
       #: (ExecutionError, ?exec_field: ExecutionField[untyped]?) -> Array[error_hash]
diff --git a/lib/graphql/breadth_exec/executor/error_formatter.rb b/lib/graphql/breadth_exec/executor/error_formatter.rb
index 1fc6c95..37bbef7 100644
--- a/lib/graphql/breadth_exec/executor/error_formatter.rb
+++ b/lib/graphql/breadth_exec/executor/error_formatter.rb
@@ -108,9 +108,6 @@ def propagate_object_scope_errors(raw_object, parent_type, selections, state)
 
             result = propagate_object_scope_errors(raw_object, fragment_type, fragment.selections, state)
             return nil if result.nil?
-
-          else
-            raise DocumentError.new("Invalid selection node type")
           end
         end
 
diff --git a/lib/graphql/breadth_exec/executor/execution_field.rb b/lib/graphql/breadth_exec/executor/execution_field.rb
index bc545ff..edffb65 100644
--- a/lib/graphql/breadth_exec/executor/execution_field.rb
+++ b/lib/graphql/breadth_exec/executor/execution_field.rb
@@ -23,12 +23,16 @@ class ExecutionField
       #: Array[ExecutionDirective]
       attr_reader :directives
 
-      #: FieldResolver[untyped]
+      #: FieldResolver
       attr_reader :resolver
 
       #: Array[GraphQL::Language::Nodes::Field]
       attr_reader :nodes
 
+      # Only populated during incremental (`@defer`) execution; nil on the hot path.
+      #: Array[Incremental::Selection]?
+      attr_reader :incremental_selections
+
       #: graphql_arguments
       attr_reader :arguments
 
@@ -46,10 +50,11 @@ class ExecutionField
       #|   nodes: Array[GraphQL::Language::Nodes::Field],
       #|   scope: ExecutionScope,
       #|   definition: GraphQL::Schema::Field,
-      #|   resolver: FieldResolver[untyped],
+      #|   resolver: FieldResolver,
       #|   ?directives: Array[ExecutionDirective],
+      #|   ?incremental_selections: Array[Incremental::Selection]?,
       #| ) -> void
-      def initialize(key, nodes:, scope:, definition:, resolver:, directives: EMPTY_ARRAY)
+      def initialize(key, nodes:, scope:, definition:, resolver:, directives: EMPTY_ARRAY, incremental_selections: nil)
         super()
         @key = key.freeze
         @scope = scope
@@ -62,6 +67,7 @@ def initialize(key, nodes:, scope:, definition:, resolver:, directives: EMPTY_AR
         @result = nil
         @arguments, @argument_errors = executor.input.coerce_argument_values(@definition, @nodes.first)
         @mutable_arguments = nil
+        @incremental_selections = incremental_selections
         @path = nil
         @schema_path = nil
       end
@@ -138,7 +144,7 @@ def allows_lazy?
 
       #: () -> Array[String]
       def path
-        @path ||= (@scope ? [*@scope.path, @key] : []).freeze
+        @path ||= [*@scope.path, @key].freeze
       end
 
       #: (Integer) -> error_path
@@ -168,7 +174,12 @@ def map_objects_with_index(&block)
 
       #: [T] (T) -> Array[T]
       def resolve_all(value)
-        value = handle_or_reraise(value) if value.is_a?(StandardError)
+        value = case value
+        when StandardError
+          handle_or_reraise(value)
+        else
+          value
+        end
         Array.new(objects.length, value)
       end
 
diff --git a/lib/graphql/breadth_exec/executor/execution_planner.rb b/lib/graphql/breadth_exec/executor/execution_planner.rb
index 67e0715..cb1e740 100644
--- a/lib/graphql/breadth_exec/executor/execution_planner.rb
+++ b/lib/graphql/breadth_exec/executor/execution_planner.rb
@@ -8,7 +8,7 @@ class ExecutionPlanner
       MUTATION_OPERATION = "mutation".freeze
       SUBSCRIPTION_OPERATION = "subscription".freeze
       TYPENAME_FIELD = "__typename".freeze
-      ENGINE_RUNTIME_DIRECTIVES = ["skip", "include"].freeze
+      ENGINE_RUNTIME_DIRECTIVES = ["skip", "include", "defer", "stream"].freeze
 
       class << self
         #: (String, GraphQL::Query::Context) -> singleton(GraphQL::Schema::Object)
@@ -93,12 +93,14 @@ def root_scopes_for_operation(operation, root_object:, result:)
               results: [result],
             ),
           ]
+        else
+          raise ImplementationError, "Unsupported operation type: #{operation.operation_type}"
         end
       end
 
       #: (Array[ExecutionScope]) -> Array[ExecutionScope]
       def plan_scopes(scopes)
-        scopes.reject! { _1.objects.empty? }
+        scopes.delete_if { _1.objects.empty? }
         scopes.freeze
         return scopes if scopes.empty?
 
@@ -114,6 +116,7 @@ def plan_scopes(scopes)
 
       private
 
+      # Fast path: just build AST nodes grouped by key without defer overhead.
       #: (
       #|   singleton(GraphQL::Schema::Member) parent_type,
       #|   Array[selection_node] selections,
@@ -138,8 +141,38 @@ def selections_grouped_by_key(parent_type, selections, map: Hash.new { |h, k| h[
             if parent_type_possible?(fragment_type, parent_type)
               selections_grouped_by_key(parent_type, fragment.selections, map: map)
             end
-          else
-            raise DocumentError.new("Invalid selection node type")
+          end
+        end
+
+        map
+      end
+
+      # Incremental path: wraps each field node in an incremental selection carrying the enclosing `@defer` usage.
+      #: (
+      #|   singleton(GraphQL::Schema::Member) parent_type,
+      #|   Array[selection_node] selections,
+      #|   ?map: Hash[String, Array[Incremental::Selection]],
+      #|   ?defer_usage: Incremental::DeferUsage?,
+      #| ) -> Hash[String, Array[Incremental::Selection]]
+      def incremental_selections_grouped_by_key(parent_type, selections, map: Hash.new { |h, k| h[k] = [] }, defer_usage: nil)
+        types = @context.types
+        selections.each do |node|
+          next if node_skipped?(node)
+
+          case node
+          when GraphQL::Language::Nodes::Field
+            map[(node.alias || node.name).freeze] << Incremental::Selection.new(node, defer_usage:)
+          when GraphQL::Language::Nodes::InlineFragment
+            fragment_type = node.type ? types.type(node.type.name) : parent_type
+            if parent_type_possible?(fragment_type, parent_type)
+              incremental_selections_grouped_by_key(parent_type, node.selections, map: map, defer_usage: defer_usage_for(node, defer_usage) || defer_usage)
+            end
+          when GraphQL::Language::Nodes::FragmentSpread
+            fragment = @executor.fragments.fetch(node.name)
+            fragment_type = types.type(fragment.type.name)
+            if parent_type_possible?(fragment_type, parent_type)
+              incremental_selections_grouped_by_key(parent_type, fragment.selections, map: map, defer_usage: defer_usage_for(node, defer_usage) || defer_usage)
+            end
           end
         end
 
@@ -152,22 +185,23 @@ def node_skipped?(node)
 
         node.directives.any? do |directive|
           if directive.name == "skip"
-            if_argument?(directive.arguments.first)
+            argument_value(directive.arguments.first)
           elsif directive.name == "include"
-            !if_argument?(directive.arguments.first)
-          else
+            !argument_value(directive.arguments.first)
+          elsif !ENGINE_RUNTIME_DIRECTIVES.include?(directive.name)
             @has_runtime_directives = true
             false
           end
         end
       end
 
-      #: (untyped) -> bool
-      def if_argument?(bool_arg)
-        if bool_arg.value.is_a?(GraphQL::Language::Nodes::VariableIdentifier)
-          @executor.variables[bool_arg.value.name] || @executor.variables[bool_arg.value.name.to_sym]
+      #: (untyped) -> untyped
+      def argument_value(arg)
+        value = arg.value
+        if value.is_a?(GraphQL::Language::Nodes::VariableIdentifier)
+          @executor.variables[value.name]
         else
-          bool_arg.value
+          value
         end
       end
 
@@ -179,11 +213,41 @@ def parent_type_possible?(fragment_type, parent_type)
         fragment_type == parent_type || @context.types.possible_types(fragment_type).include?(parent_type)
       end
 
+      #: (GraphQL::Language::Nodes::InlineFragment | GraphQL::Language::Nodes::FragmentSpread, Incremental::DeferUsage?) -> Incremental::DeferUsage?
+      def defer_usage_for(node, parent)
+        return nil if node.directives.empty?
+
+        directive = node.directives.find { _1.name == "defer" }
+        return nil unless directive
+
+        condition = directive.arguments.find { _1.name == "if" }
+        return nil if condition && argument_value(condition) == false
+
+        label_arg = directive.arguments.find { _1.name == "label" }
+        label = label_arg ? argument_value(label_arg) : nil
+        label = nil unless label.is_a?(String)
+
+        Incremental::DeferUsage.new(label, parent:)
+      end
+
       #: (
       #|   ExecutionScope,
-      #|   ?ordered_fields: Array[ExecutionField[untyped]],
+      #|   ?Array[ExecutionField[untyped]],
       #| ) -> Array[ExecutionField[untyped]]
       def build_execution_tree(exec_scope, ordered_fields = [])
+        if @executor.incremental?
+          build_incremental_execution_fields(exec_scope, ordered_fields)
+        else
+          build_execution_fields(exec_scope, ordered_fields)
+        end
+
+        exec_scope.fields.freeze
+        ordered_fields
+      end
+
+      # Fast path for building basic scope fields without incremental info (majority use case)
+      #: (ExecutionScope, Array[ExecutionField[untyped]]) -> void
+      def build_execution_fields(exec_scope, ordered_fields)
         @has_runtime_directives = false
         selections_by_key = selections_grouped_by_key(exec_scope.parent_type, exec_scope.selections)
         has_runtime_directives = @has_runtime_directives
@@ -192,9 +256,47 @@ def build_execution_tree(exec_scope, ordered_fields = [])
           exec_field = build_execution_field(key, nodes, exec_scope, has_runtime_directives)
           add_execution_field_branch(exec_field, ordered_fields)
         end
+      end
 
-        exec_scope.fields.freeze
-        ordered_fields
+      # Incremental path for partitioning fields into deferred delivery groups, and registering deferred scopes.
+      #: (ExecutionScope, Array[ExecutionField[untyped]]) -> void
+      def build_incremental_execution_fields(exec_scope, ordered_fields)
+        @has_runtime_directives = false
+
+        parent_usages = EMPTY_SET
+        selections_by_key = if exec_scope.is_a?(Incremental::DeferredExecutionScope)
+          parent_usages = exec_scope.defer_usages
+          exec_scope.field_selections
+        elsif (parent_field = exec_scope.parent_field)
+          map = Hash.new { |h, k| h[k] = [] }
+          parent_incremental_selections = parent_field.incremental_selections #: as !nil
+          parent_incremental_selections.each do |incremental_selection|
+            incremental_selections_grouped_by_key(
+              exec_scope.parent_type,
+              incremental_selection.node.selections,
+              map: map,
+              defer_usage: incremental_selection.defer_usage,
+            )
+          end
+
+          parent_usages = Incremental::Partitioner.defer_usages_for(parent_incremental_selections)
+          map
+        else
+          incremental_selections_grouped_by_key(exec_scope.parent_type, exec_scope.selections)
+        end
+
+        has_runtime_directives = @has_runtime_directives
+        base_selections, selections_by_defer_usage = Incremental::Partitioner.partition(selections_by_key, parent_usages:)
+
+        selections_by_defer_usage.each do |defer_usage, incremental_selections|
+          incremental = @executor.incremental #: as !nil
+          incremental.register_deferred_scope(exec_scope, incremental_selections, defer_usage)
+        end
+
+        base_selections.each do |key, incremental_selections|
+          exec_field = build_execution_field(key, incremental_selections.map(&:node), exec_scope, has_runtime_directives, incremental_selections:)
+          add_execution_field_branch(exec_field, ordered_fields)
+        end
       end
 
       #: (ExecutionField[untyped], Array[ExecutionField[untyped]]) -> void
@@ -210,6 +312,7 @@ def add_execution_field_branch(exec_field, ordered_fields)
           parent_type: return_type,
           parent_field: exec_field,
           selections: exec_field.selections,
+          deferred: exec_field.scope.deferred?,
           objects: [],
           results: [],
         )
@@ -222,9 +325,10 @@ def add_execution_field_branch(exec_field, ordered_fields)
       #|   Array[GraphQL::Language::Nodes::Field] nodes,
       #|   ExecutionScope exec_scope,
       #|   ?bool has_runtime_directives,
+      #|   ?incremental_selections: Array[Incremental::Selection]?,
       #| ) -> ExecutionField[untyped]
-      def build_execution_field(key, nodes, exec_scope, has_runtime_directives = false)
-        first_node = nodes.first
+      def build_execution_field(key, nodes, exec_scope, has_runtime_directives = false, incremental_selections: nil)
+        first_node = nodes.first #: as !nil
         node_name = first_node.name
 
         definition = @context.types.field(exec_scope.parent_type, node_name)
@@ -266,13 +370,13 @@ def build_execution_field(key, nodes, exec_scope, has_runtime_directives = false
           definition: definition,
           resolver: resolver,
           directives: directives,
+          incremental_selections: incremental_selections,
         )
       end
 
       #: (GraphQL::Language::Nodes::Directive, ?depth: Integer) -> ExecutionDirective
       def build_execution_directive(node, depth: 0)
         definition = @schema.directives[node.name]
-        raise DocumentError.new("No directive definition for '@#{node.name}'") unless definition
 
         resolver = if definition.respond_to?(:breadth_resolver) && definition.breadth_resolver
           definition.breadth_resolver
diff --git a/lib/graphql/breadth_exec/executor/execution_promise.rb b/lib/graphql/breadth_exec/executor/execution_promise.rb
index fea68a5..0d491ff 100644
--- a/lib/graphql/breadth_exec/executor/execution_promise.rb
+++ b/lib/graphql/breadth_exec/executor/execution_promise.rb
@@ -3,193 +3,230 @@
 
 module GraphQL
   module BreadthExec
-    class ExecutionPromise
-      PENDING = :pending
-      FULFILLED = :fulfilled
-      REJECTED = :rejected
-
-      #: Array[ExecutionPromise]?
-      attr_reader :registry
-
-      #: (?registry: Array[ExecutionPromise]?) ?{ (Proc, Proc) -> void } -> void
-      def initialize(registry: nil, &executor)
-        @state = PENDING
-        @value = nil
-        @reason = nil
-        @observers = []
-        @registry = nil
-        with_registry(registry) if registry
-
-        if executor
-          begin
-            executor.call(method(:resolve), method(:reject))
-          rescue StandardError => e
-            reject(e)
+    class Executor
+      class ExecutionPromise
+        class Deferred
+          #: Executor::ExecutionPromise
+          attr_reader :promise
+
+          #: ^(untyped) -> untyped
+          attr_reader :resolver
+
+          #: (?registry: Array[ExecutionPromise]?) -> void
+          def initialize(registry: nil)
+            @resolver = nil
+            @promise = ExecutionPromise.new(registry:) { |resolve, _| @resolver = resolve }
           end
         end
-      end
 
-      #: (Array[ExecutionPromise]) -> ExecutionPromise
-      def self.all(promises)
-        raise ArgumentError, "promises cannot be empty" if promises.empty?
-
-        new do |resolve, reject|
-          results = Array.new(promises.length)
-          completed = 0
-          promises.each_with_index do |promise, index|
-            promise.then(
-              ->(value) {
-                results[index] = value
-                completed += 1
-                resolve.call(results) if completed == promises.length
-              },
-              reject,
-            )
+        PENDING = :pending
+        FULFILLED = :fulfilled
+        REJECTED = :rejected
+
+        RAISE_REASON = ->(reason) { raise reason }
+        IDENTITY = ->(value) { value }
+
+        class << self
+          #: (Array[ExecutionPromise]) -> ExecutionPromise
+          def all(promises)
+            raise ArgumentError, "promises cannot be empty" if promises.empty?
+
+            ExecutionPromise.new do |resolve, reject|
+              results = Array.new(promises.length)
+              completed = 0
+              total = promises.length
+
+              promises.each_with_index do |promise, index|
+                promise.then(
+                  ->(value) {
+                    results[index] = value
+                    completed += 1
+                    resolve.call(results) if completed == total
+                  },
+                  reject,
+                )
+              end
+            end
           end
         end
-      end
-
-      #: (Array[ExecutionPromise]) -> ExecutionPromise
-      def with_registry(registry)
-        @registry = registry
-        registry << self unless registry.include?(self)
-        self
-      end
-
-      #: (
-      #|   ?(^(untyped value) -> untyped | Proc)? on_fulfilled,
-      #|   ?(^(untyped) -> void | Proc)? on_rejected
-      #| ) ?{ (untyped) -> untyped } -> ExecutionPromise
-      def then(on_fulfilled = nil, on_rejected = nil, &block)
-        raise ArgumentError, "Either on_fulfilled or block is required" unless on_fulfilled || on_rejected || block_given?
-        raise ArgumentError, "Exactly one of on_fulfilled or block is required" if on_fulfilled && block_given?
 
-        next_promise = self.class.new(registry: @registry)
-        fulfilled_handler = on_fulfilled || block
-        rejected_handler = on_rejected || ->(reason) { raise reason }
+        attr_reader :registry
+
+        #: (?registry: Array[ExecutionPromise]?) ?{ (Proc, Proc) -> void } -> void
+        def initialize(registry: nil, &block)
+          @state = PENDING
+          @value = nil #: untyped
+          @reason = nil #: Exception?
+          @observers = nil #: Array[untyped]?
+
+          with_registry(registry) if registry
+
+          if block_given?
+            begin
+              yield(
+                ->(value) { resolve(value) },
+                ->(reason) { reject(reason) }
+              )
+            rescue => init_error
+              reject(init_error)
+            end
+          end
+        end
 
-        if resolved?
-          next_promise.send(:dispatch_fulfilled, @value, fulfilled_handler)
-        elsif rejected?
-          next_promise.send(:dispatch_rejected, @reason, rejected_handler)
-        else
-          @observers << [next_promise, fulfilled_handler, rejected_handler]
+        #: (Array[ExecutionPromise]) -> ExecutionPromise
+        def with_registry(registry)
+          @registry = registry
+          @registry << self unless registry.include?(self)
+          self
         end
 
-        next_promise
-      end
+        #: (
+        #|   ?(^(untyped value) -> untyped | Proc)? on_fulfilled,
+        #|   ?(^(Exception) -> void | Proc)? on_rejected
+        #| ) ?{ (untyped) -> untyped } -> ExecutionPromise
+        def then(on_fulfilled = nil, on_rejected = nil, &block)
+          raise ArgumentError, "Either on_fulfilled or block is required" unless on_fulfilled || block_given?
+          raise ArgumentError, "Exactly one of on_fulfilled or block is required" if on_fulfilled && block_given?
+
+          handler = on_fulfilled || block
+          next_promise = ExecutionPromise.new(registry: @registry)
+
+          case @state
+          when FULFILLED
+            next_promise.dispatch_fulfilled(@value, handler)
+          when REJECTED
+            next_promise.dispatch_rejected(@reason, on_rejected || RAISE_REASON)
+          else
+            subscribe(next_promise, handler, on_rejected || RAISE_REASON)
+          end
 
-      #: (?(^(untyped) -> void | Proc)? on_rejected) ?{ (untyped) -> untyped } -> ExecutionPromise
-      def catch(on_rejected = nil, &block)
-        self.then(->(value) { value }, on_rejected || block)
-      end
+          next_promise
+        end
 
-      #: () -> bool
-      def resolved?
-        @state == FULFILLED
-      end
+        def catch(on_rejected = nil, &block)
+          self.then(IDENTITY, on_rejected || block)
+        end
 
-      #: () -> bool
-      def rejected?
-        @state == REJECTED
-      end
+        #: () -> bool
+        def resolved?
+          @state == FULFILLED
+        end
 
-      #: () -> bool
-      def pending?
-        @state == PENDING
-      end
+        #: () -> bool
+        def rejected?
+          @state == REJECTED
+        end
 
-      #: () -> untyped?
-      def value
-        @value if resolved?
-      end
+        #: () -> bool
+        def pending?
+          @state == PENDING
+        end
 
-      #: () -> untyped?
-      def reason
-        @reason if rejected?
-      end
+        #: () -> untyped?
+        def value
+          @value if resolved?
+        end
 
-      private
+        #: () -> untyped?
+        def reason
+          @reason if rejected?
+        end
 
-      #: (untyped) -> void
-      def resolve(value)
-        return unless pending?
+        protected
 
-        if value.is_a?(ExecutionPromise)
-          if value.equal?(self)
-            reject(StandardError.new("A promise cannot resolve to itself"))
-          elsif value.resolved?
-            resolve(value.value)
-          elsif value.rejected?
-            reject(value.reason)
+        #: (untyped, Proc?) -> void
+        def dispatch_fulfilled(value, handler)
+          if handler
+            settle_from_handler(value, handler)
           else
-            value.then(method(:resolve), method(:reject))
+            resolve(value)
           end
-          return
         end
 
-        @state = FULFILLED
-        @value = value
-        notify_fulfilled
-      end
-
-      #: (untyped) -> void
-      def reject(reason)
-        return unless pending?
-
-        @state = REJECTED
-        @reason = reason
-        notify_rejected
-      end
+        #: (untyped, Proc?) -> void
+        def dispatch_rejected(reason, handler)
+          if handler
+            settle_from_handler(reason, handler)
+          else
+            reject(reason)
+          end
+        end
 
-      #: (untyped, Proc?) -> void
-      def dispatch_fulfilled(value, handler)
-        handler ? settle_from_handler(value, handler) : resolve(value)
-      end
+        #: (ExecutionPromise, Proc?, Proc?) -> void
+        def subscribe(observer, on_fulfilled, on_rejected)
+          if @observers
+            @observers.push(observer, on_fulfilled, on_rejected)
+          else
+            @observers = [observer, on_fulfilled, on_rejected]
+          end
+        end
 
-      #: (untyped, Proc?) -> void
-      def dispatch_rejected(reason, handler)
-        handler ? settle_from_handler(reason, handler) : reject(reason)
-      end
+        private
+
+        #: (untyped) -> void
+        def resolve(result)
+          return unless @state == PENDING
+
+          if result.is_a?(ExecutionPromise)
+            if result.equal?(self)
+              reject(ArgumentError.new("A promise cannot resolve to itself"))
+              return
+            end
+
+            if result.resolved?
+              resolve(result.value)
+            elsif result.rejected?
+              reject(result.reason)
+            else
+              result.subscribe(self, nil, nil)
+            end
+          else
+            @state = FULFILLED
+            @value = result
+            notify_fulfilled
+          end
+        end
 
-      #: (untyped, Proc) -> void
-      def settle_from_handler(input, handler)
-        resolve(handler.call(input))
-      rescue StandardError => e
-        reject(e)
-      end
+        #: (untyped) -> void
+        def reject(reason)
+          return unless @state == PENDING
 
-      #: () -> void
-      def notify_fulfilled
-        observers = @observers
-        @observers = []
-        observers.each do |promise, on_fulfilled, _on_rejected|
-          promise.send(:dispatch_fulfilled, @value, on_fulfilled)
+          @state = REJECTED
+          @reason = reason
+          notify_rejected
         end
-      end
 
-      #: () -> void
-      def notify_rejected
-        observers = @observers
-        @observers = []
-        observers.each do |promise, _on_fulfilled, on_rejected|
-          promise.send(:dispatch_rejected, @reason, on_rejected)
+        #: (untyped, Proc) -> void
+        def settle_from_handler(input, handler)
+          resolve(handler.call(input))
+        rescue => error
+          reject(error)
         end
-      end
-    end
 
-    class Deferred
-      #: ExecutionPromise
-      attr_reader :promise
+        #: () -> void
+        def notify_fulfilled
+          return unless @observers
 
-      #: ^(untyped) -> untyped
-      attr_reader :resolver
+          observers = @observers
+          @observers = nil
+          i = 0
+          while i < observers.length
+            observers[i].dispatch_fulfilled(@value, observers[i + 1])
+            i += 3
+          end
+        end
 
-      #: (?registry: Array[ExecutionPromise]?) -> void
-      def initialize(registry: nil)
-        @resolver = nil
-        @promise = ExecutionPromise.new(registry: registry) do |resolve, _reject|
-          @resolver = resolve
+        #: () -> void
+        def notify_rejected
+          return unless @observers
+
+          observers = @observers
+          @observers = nil
+          i = 0
+          while i < observers.length
+            observers[i].dispatch_rejected(@reason, observers[i + 2])
+            i += 3
+          end
         end
       end
     end
diff --git a/lib/graphql/breadth_exec/executor/execution_scope.rb b/lib/graphql/breadth_exec/executor/execution_scope.rb
index 7f15036..f81364d 100644
--- a/lib/graphql/breadth_exec/executor/execution_scope.rb
+++ b/lib/graphql/breadth_exec/executor/execution_scope.rb
@@ -50,6 +50,7 @@ class ExecutionScope
       #|   ?parent_field: ExecutionField[untyped]?,
       #|   ?path: Array[String],
       #|   ?parent: ExecutionScope?,
+      #|   ?deferred: bool,
       #| ) -> void
       def initialize(
         executor:,
@@ -60,7 +61,8 @@ def initialize(
         abstraction: nil,
         parent_field: nil,
         path: [],
-        parent: nil
+        parent: nil,
+        deferred: false
       )
         super()
         @executor = executor
@@ -73,6 +75,7 @@ def initialize(
         @path = (parent_field ? parent_field.path : path).freeze
         @parent = parent_field ? parent_field.scope : parent
         @fields = {}
+        @deferred = deferred
         @executed = false
         @root = nil
         @planning_root = nil
@@ -101,7 +104,11 @@ def depth
 
       #: -> Array[String]
       def schema_path
-        parent_field ? parent_field.schema_path : EMPTY_ARRAY
+        if (field = parent_field)
+          field.schema_path
+        else
+          EMPTY_ARRAY
+        end
       end
 
       #: (Integer) -> error_path
@@ -124,11 +131,30 @@ def aborted?
         @aborted
       end
 
+      #: -> bool
+      def deferred?
+        @deferred
+      end
+
+      #: -> ExecutionScope?
+      def deferred_root
+        return self if deferred?
+
+        exec_scope = @parent
+        while exec_scope
+          return exec_scope if exec_scope.deferred?
+
+          exec_scope = exec_scope.parent
+        end
+
+        nil
+      end
+
       #: -> bool
       def aborted_subtree?
         return true if @aborted
 
-        exec_field = parent_field
+        exec_field = parent_field #: ExecutionField[untyped]?
         while exec_field
           if exec_field.scope.aborted?
             abort!
diff --git a/lib/graphql/breadth_exec/executor/lazy_element.rb b/lib/graphql/breadth_exec/executor/lazy_element.rb
index 623dd47..84f52fd 100644
--- a/lib/graphql/breadth_exec/executor/lazy_element.rb
+++ b/lib/graphql/breadth_exec/executor/lazy_element.rb
@@ -56,7 +56,7 @@ def preload(loader_class, args: nil, keys: nil)
           loader.load(element: self, keys: keys).with_registry(preload_promises)
         else
           @lazy_preloads ||= {}
-          deferred = (@lazy_preloads[[loader_class, args]] ||= Deferred.new(registry: preload_promises))
+          deferred = (@lazy_preloads[[loader_class, args]] ||= ExecutionPromise::Deferred.new(registry: preload_promises))
           deferred.promise
         end
       end
diff --git a/lib/graphql/breadth_exec/field_resolvers.rb b/lib/graphql/breadth_exec/field_resolvers.rb
index 64e8b07..b39e359 100644
--- a/lib/graphql/breadth_exec/field_resolvers.rb
+++ b/lib/graphql/breadth_exec/field_resolvers.rb
@@ -3,34 +3,32 @@
 
 module GraphQL
   module BreadthExec
-    #: [ContextType < GraphQL::Query::Context]
     class FieldResolver
-      #: (Executor::ExecutionField[untyped], ContextType) -> void
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> void
       def plan(_exec_field, _ctx)
         nil
       end
 
-      #: (Executor::ExecutionField[untyped], ContextType) -> (Array[untyped] | ExecutionPromise)
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> (Array[untyped] | Executor::ExecutionPromise)
       def resolve(exec_field, ctx)
         raise NotImplementedError, "FieldResolver#resolve must be implemented."
       end
 
-      #: (Array[untyped] | ExecutionPromise) { (Array[untyped]) -> Array[untyped] } -> (Array[untyped] | ExecutionPromise)
-      def handle_resolved(result)
-        if result.is_a?(ExecutionPromise)
-          result.then { |values| yield(values) }
+      #: (Array[untyped] | Executor::ExecutionPromise) { (Array[untyped]) -> Array[untyped] } -> (Array[untyped] | Executor::ExecutionPromise)
+      def handle_resolved(result, &block)
+        if result.is_a?(Executor::ExecutionPromise)
+          result.then { |values| block.call(values) }
         else
-          yield(result)
+          block.call(result)
         end
       end
 
-      #: (Executor::ExecutionField[untyped], ContextType) -> Enumerable
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> Enumerable
       def subscribe(_exec_field, _ctx)
         raise NotImplementedError, "FieldResolver#subscribe must be implemented."
       end
     end
 
-    #: [ContextType = GraphQL::Query::Context]
     class HashKeyResolver < FieldResolver
       #: String | Symbol
       attr_reader :key
@@ -40,23 +38,22 @@ def initialize(key)
         @key = key
       end
 
-      #: (Executor::ExecutionField[untyped], ContextType) -> Array[untyped]
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> Array[untyped]
       def resolve(exec_field, _ctx)
         exec_field.map_objects { _1[@key] }
       end
     end
 
-    #: [ContextType = GraphQL::Query::Context]
     class MethodResolver < FieldResolver
       #: type method_name = String | Symbol
 
-      #: (method_name, *method_name, ?fallback: untyped) -> void
+      #: (*method_name, ?fallback: untyped) -> void
       def initialize(*names, fallback: nil)
         @names = names
         @fallback = fallback
       end
 
-      #: (Executor::ExecutionField[untyped], ContextType) -> Array[untyped]
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> Array[untyped]
       def resolve(exec_field, _ctx)
         exec_field.map_objects do |obj|
           @names.reduce(obj) do |memo, name|
@@ -69,15 +66,13 @@ def resolve(exec_field, _ctx)
       end
     end
 
-    #: [ContextType = GraphQL::Query::Context]
     class SelfResolver < FieldResolver
-      #: (Executor::ExecutionField[untyped], ContextType) -> Array[untyped]
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> Array[untyped]
       def resolve(exec_field, _ctx)
         exec_field.map_objects(&:itself)
       end
     end
 
-    #: [ContextType = GraphQL::Query::Context]
     class ValueResolver < FieldResolver
       #: untyped
       attr_reader :value
@@ -87,7 +82,7 @@ def initialize(value)
         @value = value
       end
 
-      #: (Executor::ExecutionField[untyped], ContextType) -> Array[untyped]
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> Array[untyped]
       def resolve(exec_field, _ctx)
         exec_field.resolve_all(@value)
       end
diff --git a/lib/graphql/breadth_exec/has_breadth_resolver.rb b/lib/graphql/breadth_exec/has_breadth_resolver.rb
index 18a534b..a6bbf84 100644
--- a/lib/graphql/breadth_exec/has_breadth_resolver.rb
+++ b/lib/graphql/breadth_exec/has_breadth_resolver.rb
@@ -21,7 +21,7 @@ def extended(_base)
           end
         end
 
-        #: FieldResolver[untyped]?
+        #: FieldResolver?
         attr_reader :breadth_resolver
 
         #: -> (String | Symbol)
@@ -29,7 +29,7 @@ def original_name
           super
         end
 
-        #: (Symbol | FieldResolver[untyped] | singleton(FieldResolver) | nil) -> void
+        #: (Symbol | FieldResolver | singleton(FieldResolver) | nil) -> void
         def breadth_resolver=(value)
           @breadth_resolver = case value
           when Symbol
@@ -73,7 +73,7 @@ def extended(base)
         #: DirectiveResolver?
         attr_accessor :breadth_resolver
 
-        #: (Class) -> void
+        #: (untyped) -> void
         def inherited(base)
           super
           base.breadth_resolver = breadth_resolver
diff --git a/lib/graphql/breadth_exec/incremental.rb b/lib/graphql/breadth_exec/incremental.rb
new file mode 100644
index 0000000..7905d45
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental.rb
@@ -0,0 +1,18 @@
+# typed: true
+# frozen_string_literal: true
+
+require_relative "incremental/defer_usage"
+require_relative "incremental/deferred_delivery"
+require_relative "incremental/deferred_execution_scope"
+require_relative "incremental/partitioner"
+require_relative "incremental/selection"
+require_relative "incremental/result"
+require_relative "incremental/publisher"
+require_relative "incremental/context"
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/context.rb b/lib/graphql/breadth_exec/incremental/context.rb
new file mode 100644
index 0000000..a075664
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/context.rb
@@ -0,0 +1,187 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class Context
+        #: Executor
+        attr_reader :executor
+
+        #: Publisher
+        attr_reader :publisher
+
+        #: (
+        #|   Executor executor,
+        #|   data: graphql_result,
+        #| ) -> void
+        def initialize(executor, data:)
+          @executor = executor
+          @data = data
+          @publisher = Publisher.new
+          @deferred_scopes = []
+          @pending_deliveries = []
+          @announced_deliveries = {}.compare_by_identity
+          @completed_deliveries = {}.compare_by_identity
+          @deliveries_by_usage = {}.compare_by_identity
+        end
+
+        #: (Executor::ExecutionScope, Hash[String, Array[Selection]], Set[DeferUsage]) -> void
+        def register_deferred_scope(base_scope, field_selections, defer_usages)
+          @deferred_scopes << DeferredExecutionScope.new(
+            base_scope:,
+            field_selections:,
+            defer_usages:,
+          )
+        end
+
+        #: -> bool
+        def active?
+          true
+        end
+
+        #: -> bool
+        def deferred?
+          @deferred_scopes.any?
+        end
+
+        #: -> Array[DeferredDelivery]
+        def prepare_pending
+          @deferred_scopes.each do |deferred_scope|
+            next if deferred_scope.announced? || !deferred_scope.ready?
+
+            pending_deliveries_for(deferred_scope)
+            deferred_scope.announced = true
+          end
+
+          @pending_deliveries.uniq!
+          pending = @pending_deliveries
+          @pending_deliveries = []
+          pending
+        end
+
+        #: -> Array[DeferredExecutionScope]
+        def ready_scopes
+          @deferred_scopes.select { _1.announced? && !_1.executed? && _1.ready? }.each(&:prepare!)
+        end
+
+        #: (DeferredExecutionScope) -> Array[[Integer, error_path, Array[DeferredDelivery]]]
+        def deliveries_for(deferred_scope)
+          deliveries = []
+          index = 0
+          while index < deferred_scope.base_scope.objects.length
+            path = deferred_scope.base_scope.object_path(index)
+            unless deferred_path_nulled?(path)
+              deferred_deliveries = deferred_scope.defer_usages.map { delivery_for(_1, path) }
+              deliveries << [index, path, deferred_deliveries]
+            end
+            index += 1
+          end
+          deliveries
+        end
+
+        #: (Array[DeferredDelivery]) -> Array[graphql_result]
+        def pending_payloads(deliveries)
+          @publisher.pending(deliveries)
+        end
+
+        #: (Array[DeferredDelivery], error_path, graphql_result, ?errors: Array[error_hash]) -> graphql_result
+        def incremental_payload(deliveries, path, data, errors: EMPTY_ARRAY)
+          @publisher.incremental(deliveries, path, data, errors:)
+        end
+
+        #: (Array[DeferredDelivery], ?errors_by_delivery: Hash[DeferredDelivery, Array[error_hash]]) -> Array[graphql_result]
+        def completed_payloads(deliveries, errors_by_delivery: EMPTY_OBJECT)
+          deliveries.uniq.filter_map do |delivery|
+            next if @completed_deliveries[delivery]
+
+            @completed_deliveries[delivery] = true
+            @publisher.completed(delivery, errors: errors_by_delivery[delivery] || EMPTY_ARRAY)
+          end
+        end
+
+        private
+
+        #: (DeferredExecutionScope) -> Array[DeferredDelivery]
+        def pending_deliveries_for(deferred_scope)
+          pending = []
+          index = 0
+          while index < deferred_scope.base_scope.objects.length
+            path = deferred_scope.base_scope.object_path(index)
+            unless deferred_path_nulled?(path)
+              deferred_scope.defer_usages.each do |defer_usage|
+                delivery = delivery_for(defer_usage, path)
+                unless @completed_deliveries[delivery] || @announced_deliveries[delivery]
+                  @announced_deliveries[delivery] = true
+                  pending << delivery
+                  @pending_deliveries << delivery
+                end
+              end
+            end
+            index += 1
+          end
+
+          pending
+        end
+
+        #: (DeferUsage, error_path) -> DeferredDelivery
+        def delivery_for(defer_usage, path)
+          existing = nearest_delivery_for(defer_usage, path)
+          return existing if existing
+
+          parent = if (parent_usage = defer_usage.parent)
+            nearest_delivery_for(parent_usage, path)
+          end
+
+          delivery = DeferredDelivery.new(path.dup, defer_usage.label, parent:)
+          (@deliveries_by_usage[defer_usage] ||= []) << delivery
+          delivery
+        end
+
+        #: (DeferUsage, error_path) -> DeferredDelivery?
+        def nearest_delivery_for(defer_usage, path)
+          deliveries = @deliveries_by_usage[defer_usage]
+          return nil unless deliveries
+
+          deliveries
+            .select { _1.path_prefix_of?(path) }
+            .max_by { _1.path.length }
+        end
+
+        # True when the formatted initial result null-bubbled away the object at `path`
+        # (e.g. a non-null child error nulled a nullable list element). Deferred execution rooted
+        # at such a path must not be announced or delivered: there is no live object to patch.
+        #: (error_path) -> bool
+        def deferred_path_nulled?(path)
+          return false if path.empty?
+
+          path_nulled_in?(@data, path, 0)
+        end
+
+        # Walks `path` from `index` into the formatted result `value`, returning true once a
+        # *present-but-nil* slot is reached. A missing Hash key or out-of-range Array index
+        # returns false.
+        #: (untyped, error_path, Integer) -> bool
+        def path_nulled_in?(value, path, index)
+          return false if index == path.length
+
+          segment = path[index]
+          child = case value
+          when Hash
+            return false unless value.key?(segment)
+            value[segment]
+          when Array
+            return false unless segment.is_a?(Integer) && segment >= 0 && segment < value.length
+            value[segment]
+          else
+            return false
+          end
+
+          return true if child.nil?
+
+          path_nulled_in?(child, path, index + 1)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/defer_usage.rb b/lib/graphql/breadth_exec/incremental/defer_usage.rb
new file mode 100644
index 0000000..13fc4c6
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/defer_usage.rb
@@ -0,0 +1,22 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class DeferUsage
+        #: String?
+        attr_reader :label
+
+        #: DeferUsage?
+        attr_reader :parent
+
+        #: (?String?, ?parent: DeferUsage?) -> void
+        def initialize(label = nil, parent: nil)
+          @label = label
+          @parent = parent
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/deferred_delivery.rb b/lib/graphql/breadth_exec/incremental/deferred_delivery.rb
new file mode 100644
index 0000000..508f5ee
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/deferred_delivery.rb
@@ -0,0 +1,42 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class DeferredDelivery
+        #: error_path
+        attr_reader :path
+
+        #: String?
+        attr_reader :label
+
+        #: DeferredDelivery?
+        attr_reader :parent
+
+        #: (error_path, ?String?, ?parent: DeferredDelivery?) -> void
+        def initialize(path, label = nil, parent: nil)
+          @path = path.freeze
+          @label = label
+          @parent = parent
+        end
+
+        # True when this delivery's path is a prefix of (or equal to) `path`,
+        # i.e. `path` falls at or below this delivery in the result tree.
+        #: (error_path) -> bool
+        def path_prefix_of?(path)
+          return false if @path.length > path.length
+
+          i = 0
+          while i < @path.length
+            return false unless @path[i] == path[i]
+
+            i += 1
+          end
+
+          true
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/deferred_execution_scope.rb b/lib/graphql/breadth_exec/incremental/deferred_execution_scope.rb
new file mode 100644
index 0000000..eae7a27
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/deferred_execution_scope.rb
@@ -0,0 +1,67 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class DeferredExecutionScope < Executor::ExecutionScope
+        #: Executor::ExecutionScope
+        attr_reader :base_scope
+
+        #: Hash[String, Array[Incremental::Selection]]
+        attr_reader :field_selections
+
+        #: Set[DeferUsage]
+        attr_reader :defer_usages
+
+        #: bool
+        attr_writer :announced
+
+        #: (
+        #|   base_scope: Executor::ExecutionScope,
+        #|   field_selections: Hash[String, Array[Incremental::Selection]],
+        #|   defer_usages: Set[DeferUsage],
+        #| ) -> void
+        def initialize(base_scope:, field_selections:, defer_usages:)
+          @base_scope = base_scope
+          @field_selections = field_selections
+          @defer_usages = defer_usages
+          @announced = false
+
+          super(
+            executor: base_scope.executor,
+            parent_type: base_scope.parent_type,
+            selections: field_selections.each_value.flat_map { |incremental_selections| incremental_selections.map(&:node) }.freeze,
+            objects: EMPTY_ARRAY,
+            results: EMPTY_ARRAY,
+            abstraction: base_scope.abstraction,
+            parent_field: base_scope.parent_field,
+            path: base_scope.path,
+            parent: base_scope.parent,
+            deferred: true,
+          )
+        end
+
+        #: -> bool
+        def ready?
+          @base_scope.has_authorized_objects? && @base_scope.executed? && !@base_scope.aborted?
+        end
+
+        #: -> bool
+        def announced?
+          @announced
+        end
+
+        #: -> DeferredExecutionScope
+        def prepare!
+          if @results.empty?
+            @objects = @base_scope.objects
+            @results = Array.new(@base_scope.objects.size) { {} }.freeze
+          end
+
+          self
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/partitioner.rb b/lib/graphql/breadth_exec/incremental/partitioner.rb
new file mode 100644
index 0000000..2518d2c
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/partitioner.rb
@@ -0,0 +1,84 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class Partitioner
+        class << self
+          #: (
+          #|   Hash[String, Array[Incremental::Selection]] field_selections,
+          #|   parent_usages: Set[DeferUsage],
+          #| ) -> [Hash[String, Array[Incremental::Selection]], Hash[Set[DeferUsage], Hash[String, Array[Incremental::Selection]]]]
+          def partition(field_selections, parent_usages:)
+            base_field_selections = {}
+            field_selections_by_defer_usage = {}
+
+            field_selections.each do |response_key, incremental_selections|
+              filtered_defer_usage_set = defer_usages_for(incremental_selections)
+
+              if same_set?(filtered_defer_usage_set, parent_usages)
+                base_field_selections[response_key] = incremental_selections
+                next
+              end
+
+              deferred_field_selections = field_selections_for_defer_usage(field_selections_by_defer_usage, filtered_defer_usage_set)
+              deferred_field_selections[response_key] = incremental_selections
+            end
+
+            [base_field_selections, field_selections_by_defer_usage]
+          end
+
+          #: (Array[Incremental::Selection]) -> Set[DeferUsage]
+          def defer_usages_for(incremental_selections)
+            filtered = Set.new.compare_by_identity
+
+            incremental_selections.each do |incremental_selection|
+              defer_usage = incremental_selection.defer_usage
+              return EMPTY_SET unless defer_usage
+
+              filtered << defer_usage
+            end
+
+            filtered.each do |defer_usage|
+              parent = defer_usage.parent
+              while parent
+                if filtered.include?(parent)
+                  filtered.delete(defer_usage)
+                  break
+                end
+
+                parent = parent.parent
+              end
+            end
+
+            filtered.freeze
+          end
+
+          private
+
+          #: (
+          #|   Hash[Set[DeferUsage], Hash[String, Array[Incremental::Selection]]],
+          #|   Set[DeferUsage],
+          #| ) -> Hash[String, Array[Incremental::Selection]]
+          def field_selections_for_defer_usage(field_selections_by_defer_usage, defer_usage_set)
+            existing = field_selections_by_defer_usage.find { |set, _| same_set?(set, defer_usage_set) }
+            return existing[1] if existing
+
+            field_selections = {}
+            field_selections_by_defer_usage[defer_usage_set] = field_selections
+            field_selections
+          end
+
+          #: (Set[DeferUsage], Set[DeferUsage]) -> bool
+          def same_set?(left, right)
+            return true if left.equal?(right)
+            return false unless left.size == right.size
+
+            left.all? { right.include?(_1) }
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/publisher.rb b/lib/graphql/breadth_exec/incremental/publisher.rb
new file mode 100644
index 0000000..ca3c2ba
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/publisher.rb
@@ -0,0 +1,65 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class Publisher
+        def initialize
+          @ids = {}.compare_by_identity
+          @next_id = 0
+        end
+
+        #: (Array[DeferredDelivery]) -> Array[graphql_result]
+        def pending(deliveries)
+          deliveries.map do |delivery|
+            result = {
+              "id" => id_for(delivery),
+              "path" => delivery.path,
+            }
+            result["label"] = delivery.label if delivery.label
+            result
+          end
+        end
+
+        #: (Array[DeferredDelivery], error_path, graphql_result, ?errors: Array[error_hash]) -> graphql_result
+        def incremental(deliveries, path, data, errors: EMPTY_ARRAY)
+          delivery = best_delivery_for(deliveries, path)
+          result = {
+            "data" => data,
+            "id" => id_for(delivery),
+          }
+
+          sub_path = path.drop(delivery.path.length)
+          result["subPath"] = sub_path unless sub_path.empty?
+          result["errors"] = errors unless errors.empty?
+          result
+        end
+
+        #: (DeferredDelivery, ?errors: Array[error_hash]) -> graphql_result
+        def completed(delivery, errors: EMPTY_ARRAY)
+          result = { "id" => id_for(delivery) }
+          result["errors"] = errors unless errors.empty?
+          @ids.delete(delivery)
+          result
+        end
+
+        private
+
+        #: (DeferredDelivery) -> String
+        def id_for(delivery)
+          @ids[delivery] ||= begin
+            id = @next_id.to_s
+            @next_id += 1
+            id
+          end
+        end
+
+        #: (Array[DeferredDelivery], error_path) -> DeferredDelivery
+        def best_delivery_for(deliveries, path)
+          deliveries.max_by { |delivery| delivery.path_prefix_of?(path) ? delivery.path.length : -1 } #: as !nil
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/result.rb b/lib/graphql/breadth_exec/incremental/result.rb
new file mode 100644
index 0000000..151e02a
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/result.rb
@@ -0,0 +1,38 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class Result
+        #: graphql_result
+        attr_reader :initial_result
+
+        #: Enumerable
+        attr_reader :subsequent_results
+
+        #: (initial_result: graphql_result, subsequent_results: Enumerable) -> void
+        def initialize(initial_result:, subsequent_results:)
+          @initial_result = initial_result
+          @subsequent_results = subsequent_results
+          @incremental = subsequent_results != EMPTY_ARRAY
+        end
+
+        #: -> bool
+        def incremental?
+          @incremental
+        end
+
+        #: -> graphql_result
+        def to_h
+          return @initial_result unless incremental?
+
+          {
+            "initialResult" => @initial_result,
+            "subsequentResults" => @subsequent_results.to_a,
+          }
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/incremental/selection.rb b/lib/graphql/breadth_exec/incremental/selection.rb
new file mode 100644
index 0000000..7745f81
--- /dev/null
+++ b/lib/graphql/breadth_exec/incremental/selection.rb
@@ -0,0 +1,22 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module BreadthExec
+    module Incremental
+      class Selection
+        #: GraphQL::Language::Nodes::Field
+        attr_reader :node
+
+        #: DeferUsage?
+        attr_reader :defer_usage
+
+        #: (GraphQL::Language::Nodes::Field, ?defer_usage: DeferUsage?) -> void
+        def initialize(node, defer_usage: nil)
+          @node = node
+          @defer_usage = defer_usage
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth_exec/lazy_loader.rb b/lib/graphql/breadth_exec/lazy_loader.rb
index 5013dc5..5f6204b 100644
--- a/lib/graphql/breadth_exec/lazy_loader.rb
+++ b/lib/graphql/breadth_exec/lazy_loader.rb
@@ -18,7 +18,7 @@ class LazyFulfillment
         #: Hash[untyped, untyped]?
         attr_reader :eager_values
 
-        #: ExecutionPromise
+        #: Executor::ExecutionPromise
         attr_reader :promise
 
         #: (
@@ -26,7 +26,7 @@ class LazyFulfillment
         #|   keys: Array[untyped],
         #|   identities: Array[untyped],
         #|   ?eager_values: Hash[untyped, untyped]?,
-        #|   ?pre_deferred: Deferred?,
+        #|   ?pre_deferred: Executor::ExecutionPromise::Deferred?,
         #| ) -> void
         def initialize(element:, keys:, identities:, eager_values: nil, pre_deferred: nil)
           @element = element
@@ -34,7 +34,7 @@ def initialize(element:, keys:, identities:, eager_values: nil, pre_deferred: ni
           @identities = identities
           @eager_values = eager_values
           @resolver = pre_deferred&.resolver
-          @promise = pre_deferred&.promise || ExecutionPromise.new { |resolve, _reject| @resolver = resolve }
+          @promise = pre_deferred&.promise || Executor::ExecutionPromise.new { |resolve, _reject| @resolver = resolve }
         end
 
         #: (untyped) -> void
@@ -101,8 +101,8 @@ def fulfill_identity(identity, result)
       #|   keys: Array[untyped],
       #|   ?eager_values: Hash[untyped, untyped]?,
       #|   ?load_nil_keys: bool,
-      #|   ?pre_deferred: Deferred?,
-      #| ) -> ExecutionPromise
+      #|   ?pre_deferred: Executor::ExecutionPromise::Deferred?,
+      #| ) -> Executor::ExecutionPromise
       def load(element:, keys:, eager_values: nil, load_nil_keys: false, pre_deferred: nil)
         eager_values = nil if eager_values&.empty?
         compact = !load_nil_keys
diff --git a/lib/graphql/breadth_exec/subscription_response_stream.rb b/lib/graphql/breadth_exec/subscription_response_stream.rb
index 379eef5..a35e1e7 100644
--- a/lib/graphql/breadth_exec/subscription_response_stream.rb
+++ b/lib/graphql/breadth_exec/subscription_response_stream.rb
@@ -18,6 +18,7 @@ def initialize(executor:, source_stream:)
         @source_stream = source_stream
       end
 
+      # @override
       #: () ?{ (graphql_result) -> void } -> Enumerator?
       def each(&block)
         return enum_for(:each) unless block
diff --git a/lib/graphql/breadth_exec/tracer.rb b/lib/graphql/breadth_exec/tracer.rb
index ce55338..35a8ce2 100644
--- a/lib/graphql/breadth_exec/tracer.rb
+++ b/lib/graphql/breadth_exec/tracer.rb
@@ -5,43 +5,59 @@ module GraphQL
   module BreadthExec
     class Tracer
       #: (Executor, GraphQL::Query::Context) -> void
-      def start(*)
+      def start(executor, context)
       end
 
       #: (Executor, GraphQL::Query::Context, duration: Float) -> void
-      def finish(*, duration:)
+      def finish(executor, context, duration:)
       end
 
       #: (Executor, GraphQL::Query::Context) -> void
-      def before_execute(*)
+      def before_execute(executor, context)
       end
 
       #: (Executor, GraphQL::Query::Context, duration: Float) -> void
-      def after_execute(*, duration:)
+      def after_execute(executor, context, duration:)
       end
 
       #: (Executor, GraphQL::Query::Context) -> void
-      def before_format_errors(*)
+      def before_format_errors(executor, context)
       end
 
-      #: (singleton(GraphQL::Schema::Member), String, Integer, GraphQL::Query::Context) -> void
-      def before_resolve_field(*)
+      #: (Executor::AbstractExecutionScope, GraphQL::Query::Context) -> void
+      def before_abstract_scope(abstract_scope, context)
       end
 
-      #: (singleton(GraphQL::Schema::Member), String, Integer, GraphQL::Query::Context, ?duration: Float?) -> Float?
-      def after_resolve_field(*args, duration: nil)
+      #: (Executor::ExecutionScope, GraphQL::Query::Context) -> void
+      def before_scope(exec_scope, context)
       end
 
       #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> void
-      def before_build_field_result(*)
+      def before_resolve_field(exec_field, context)
       end
 
       #: (Executor::ExecutionField[untyped], GraphQL::Query::Context, duration: Float) -> void
-      def after_build_field_result(*, duration:)
+      def after_resolve_field(exec_field, context, duration:)
+      end
+
+      #: (LazyLoader[untyped], Array[Executor::LazyElement], GraphQL::Query::Context) -> void
+      def before_lazy_set(loader, elements, context)
+      end
+
+      #: (LazyLoader[untyped], Array[Executor::LazyElement], GraphQL::Query::Context, duration: Float) -> void
+      def after_lazy_set(loader, elements, context, duration:)
+      end
+
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context) -> void
+      def before_build_field_result(exec_field, context)
+      end
+
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context, duration: Float) -> void
+      def after_build_field_result(exec_field, context, duration:)
       end
 
       #: (Exception, GraphQL::Query::Context, ?exec_field: Executor::ExecutionField[untyped]?) -> void
-      def on_exception(*, exec_field: nil)
+      def on_exception(exception, context, exec_field: nil)
       end
     end
   end
diff --git a/sorbet/config b/sorbet/config
index a01b9dc..e2628a7 100644
--- a/sorbet/config
+++ b/sorbet/config
@@ -1,5 +1,7 @@
 --dir
 lib
+--parser=prism
+--enable-experimental-rbs-comments
 --dir
 sorbet/rbi
 --ignore=/test/
diff --git a/sorbet/rbi/shims/graphql.rbi b/sorbet/rbi/shims/graphql.rbi
index d3b97a2..52998e6 100644
--- a/sorbet/rbi/shims/graphql.rbi
+++ b/sorbet/rbi/shims/graphql.rbi
@@ -105,6 +105,12 @@ module GraphQL
         def name; end
       end
 
+      class VariableDefinition < AbstractNode
+        def default_value; end
+        def name; end
+        def type; end
+      end
+
       class NullValue < AbstractNode; end
 
       class WrapperType < AbstractNode
@@ -120,6 +126,11 @@ module GraphQL
         def arguments; end
       end
 
+      class Argument < AbstractNode
+        def name; end
+        def value; end
+      end
+
       class Enum < AbstractNode
         def name; end
       end
@@ -127,6 +138,12 @@ module GraphQL
   end
 
   class Schema
+    def self.directives; end
+    def self.query; end
+    def self.lazy?(value); end
+    def self.sync_lazy(value); end
+    def self.type_error(error, context); end
+
     def directives; end
     def query; end
     def lazy?(value); end
@@ -134,6 +151,17 @@ module GraphQL
     def type_error(error, context); end
 
     class Member
+      def self.graphql_name; end
+      def self.kind; end
+      def self.unwrap; end
+      def self.list?; end
+      def self.non_null?; end
+      def self.of_type; end
+      def self.to_type_signature; end
+      def self.to_list_type; end
+      def self.to_non_null_type; end
+      def self.const_get(name); end
+
       def graphql_name; end
       def kind; end
       def unwrap; end
@@ -141,13 +169,23 @@ module GraphQL
       def non_null?; end
       def of_type; end
       def to_type_signature; end
+      def to_list_type; end
+      def to_non_null_type; end
       def const_get(name); end
     end
 
     class Object < Member
+      def self.introspection?; end
+
       def introspection?; end
     end
 
+    class InputObject < Member
+      def self.directives; end
+
+      def directives; end
+    end
+
     class Field
       def graphql_name; end
       def name; end
@@ -158,6 +196,14 @@ module GraphQL
       def introspection?; end
     end
 
+    class Argument
+      def default_value; end
+      def default_value?; end
+      def graphql_name; end
+      def keyword; end
+      def type; end
+    end
+
     class Directive
       def graphql_name; end
     end
diff --git a/test/graphql/breadth_exec/executor/execution_promise_test.rb b/test/graphql/breadth_exec/executor/execution_promise_test.rb
index 6ca5c6b..2bc4a5b 100644
--- a/test/graphql/breadth_exec/executor/execution_promise_test.rb
+++ b/test/graphql/breadth_exec/executor/execution_promise_test.rb
@@ -4,7 +4,7 @@
 
 class GraphQL::BreadthExec::Executor::ExecutionPromiseTest < Minitest::Test
   def test_then_dispatches_synchronously
-    promise = GraphQL::BreadthExec::ExecutionPromise.new
+    promise = GraphQL::BreadthExec::Executor::ExecutionPromise.new
     seen = nil
 
     chained = promise.then { |value| seen = value.upcase }
@@ -16,8 +16,8 @@ def test_then_dispatches_synchronously
   end
 
   def test_then_adopts_nested_execution_promise
-    inner = GraphQL::BreadthExec::ExecutionPromise.new
-    outer = GraphQL::BreadthExec::ExecutionPromise.new { |resolve, _reject| resolve.call(inner) }
+    inner = GraphQL::BreadthExec::Executor::ExecutionPromise.new
+    outer = GraphQL::BreadthExec::Executor::ExecutionPromise.new { |resolve, _reject| resolve.call(inner) }
 
     refute_predicate outer, :resolved?
     inner.send(:resolve, "done")
@@ -28,16 +28,16 @@ def test_then_adopts_nested_execution_promise
 
   def test_registry_tracks_chained_promises
     registry = []
-    promise = GraphQL::BreadthExec::ExecutionPromise.new(registry: registry)
+    promise = GraphQL::BreadthExec::Executor::ExecutionPromise.new(registry: registry)
     chained = promise.then { |value| value }
 
     assert_equal [promise, chained], registry
   end
 
   def test_all_resolves_in_order
-    first = GraphQL::BreadthExec::ExecutionPromise.new
-    second = GraphQL::BreadthExec::ExecutionPromise.new
-    all = GraphQL::BreadthExec::ExecutionPromise.all([first, second])
+    first = GraphQL::BreadthExec::Executor::ExecutionPromise.new
+    second = GraphQL::BreadthExec::Executor::ExecutionPromise.new
+    all = GraphQL::BreadthExec::Executor::ExecutionPromise.all([first, second])
 
     second.send(:resolve, "b")
     first.send(:resolve, "a")
diff --git a/test/graphql/breadth_exec/executor/incremental_test.rb b/test/graphql/breadth_exec/executor/incremental_test.rb
new file mode 100644
index 0000000..22a13ab
--- /dev/null
+++ b/test/graphql/breadth_exec/executor/incremental_test.rb
@@ -0,0 +1,934 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class GraphQL::BreadthExec::Executor::IncrementalTest < Minitest::Test
+  class BatchTrackingLoader < GraphQL::BreadthExec::LazyLoader
+    class << self
+      attr_accessor :perform_keys
+    end
+
+    self.perform_keys = []
+
+    def map?
+      true
+    end
+
+    def perform_map(keys, _ctx)
+      self.class.perform_keys << keys.dup
+      keys
+    end
+  end
+
+  class LazyHashResolver < GraphQL::BreadthExec::FieldResolver
+    def initialize(key)
+      @key = key
+    end
+
+    def resolve(exec_field, _ctx)
+      exec_field.lazy(loader_class: BatchTrackingLoader, keys: exec_field.objects.map { _1[@key] })
+    end
+  end
+
+  SOURCE = {
+    "products" => {
+      "nodes" => [{
+        "id" => "gid://shopify/Product/1",
+        "title" => "Banana",
+        "must" => "yes",
+        "variants" => {
+          "nodes" => [{
+            "id" => "gid://shopify/Variant/1",
+            "title" => "Small Banana",
+          }],
+        },
+      }, {
+        "id" => "gid://shopify/Product/2",
+        "title" => "Apple",
+        "must" => "yes",
+        "variants" => {
+          "nodes" => [{
+            "id" => "gid://shopify/Variant/2",
+            "title" => "Small Apple",
+          }],
+        },
+      }],
+    },
+  }.freeze
+
+  def test_incremental_result_returns_normal_result_without_defer
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes {
+          id
+          title
+        }
+      }
+    }|).incremental_result
+
+    expected = {
+      "data" => {
+        "products" => {
+          "nodes" => [
+            { "id" => "gid://shopify/Product/1", "title" => "Banana" },
+            { "id" => "gid://shopify/Product/2", "title" => "Apple" },
+          ],
+        },
+      },
+    }
+
+    assert_instance_of GraphQL::BreadthExec::Incremental::Result, result
+    refute result.incremental?
+    assert_equal expected, result.initial_result
+    assert_equal [], result.subsequent_results.to_a
+    assert_equal expected, result.to_h
+  end
+
+  def test_incremental_result_raises_after_result
+    executor = build_executor(%|{
+      products(first: 2) {
+        nodes { id }
+      }
+    }|)
+
+    executor.result
+    error = assert_raises(GraphQL::BreadthExec::ImplementationError) do
+      executor.incremental_result
+    end
+
+    assert_equal "Cannot call incremental_result after result", error.message
+  end
+
+  def test_result_raises_after_incremental_result
+    executor = build_executor(%|{
+      products(first: 2) {
+        nodes { id }
+      }
+    }|)
+
+    executor.incremental_result
+    error = assert_raises(GraphQL::BreadthExec::ImplementationError) do
+      executor.result
+    end
+
+    assert_equal "Cannot call result after incremental_result", error.message
+  end
+
+  def test_incremental_result_returns_same_result_on_repeat_call
+    executor = build_executor(%|{
+      products(first: 2) {
+        nodes { id }
+      }
+    }|)
+
+    result = executor.incremental_result
+    assert_same result, executor.incremental_result
+  end
+
+  def test_incremental_result_does_not_defer_when_if_is_false
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer(if: false) { title }
+        }
+      }
+    }|).incremental_result
+
+    expected = {
+      "data" => {
+        "products" => {
+          "nodes" => [
+            { "id" => "gid://shopify/Product/1", "title" => "Banana" },
+            { "id" => "gid://shopify/Product/2", "title" => "Apple" },
+          ],
+        },
+      },
+    }
+
+    assert_instance_of GraphQL::BreadthExec::Incremental::Result, result
+    refute result.incremental?
+    assert_equal expected, result.initial_result
+    assert_equal [], result.subsequent_results.to_a
+    assert_equal(
+      expected,
+      result.to_h,
+    )
+  end
+
+  def test_incremental_result_defers_when_if_is_null
+    result = build_executor(%|query($shouldDefer: Boolean) {
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer(if: $shouldDefer) { title }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes", 0] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{ "data" => { "title" => "Banana" }, "id" => "0" }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_defers_fragment_for_each_list_object
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes {
+          id
+          ... @defer { title }
+        }
+      }
+    }|).incremental_result
+
+    assert_instance_of GraphQL::BreadthExec::Incremental::Result, result
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [
+              { "id" => "gid://shopify/Product/1" },
+              { "id" => "gid://shopify/Product/2" },
+            ],
+          },
+        },
+        "pending" => [
+          { "id" => "0", "path" => ["products", "nodes", 0] },
+          { "id" => "1", "path" => ["products", "nodes", 1] },
+        ],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+
+    assert_equal(
+      [{
+        "incremental" => [
+          { "data" => { "title" => "Banana" }, "id" => "0" },
+          { "data" => { "title" => "Apple" }, "id" => "1" },
+        ],
+        "completed" => [
+          { "id" => "0" },
+          { "id" => "1" },
+        ],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_includes_defer_label
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer(label: "ProductTitle") { title }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      [{ "id" => "0", "path" => ["products", "nodes", 0], "label" => "ProductTitle" }],
+      result.initial_result.fetch("pending"),
+    )
+  end
+
+  def test_incremental_result_treats_null_label_as_no_label
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer(label: null) { title }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      [{ "id" => "0", "path" => ["products", "nodes", 0] }],
+      result.initial_result.fetch("pending"),
+    )
+  end
+
+  def test_incremental_result_deduplicates_fragment_also_selected_without_defer
+    [
+      %|query ProductTitleQuery {
+        products(first: 1) {
+          nodes {
+            ...Title @defer(label: "DeferredTitle")
+            ...Title
+          }
+        }
+      }
+
+      fragment Title on Product {
+        title
+      }|,
+      %|query ProductTitleQuery {
+        products(first: 1) {
+          nodes {
+            ...Title
+            ...Title @defer(label: "DeferredTitle")
+          }
+        }
+      }
+
+      fragment Title on Product {
+        title
+      }|,
+    ].each do |document|
+      result = build_executor(document, source: one_product_source).incremental_result
+
+      expected = {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "title" => "Banana" }],
+          },
+        },
+      }
+
+      refute result.incremental?
+      assert_equal expected, result.initial_result
+      assert_equal [], result.subsequent_results.to_a
+    end
+  end
+
+  def test_incremental_result_defers_inline_fragment
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... on Product @defer(label: "InlineTitle") {
+            title
+          }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes", 0], "label" => "InlineTitle" }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{ "data" => { "title" => "Banana" }, "id" => "0" }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_does_not_emit_empty_defer_fragments
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          ... @defer {
+            title @skip(if: true)
+          }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    expected = {
+      "data" => {
+        "products" => {
+          "nodes" => [{}],
+        },
+      },
+    }
+
+    refute result.incremental?
+    assert_equal expected, result.initial_result
+    assert_equal [], result.subsequent_results.to_a
+  end
+
+  def test_incremental_result_emits_children_of_empty_defer_fragments
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          ... @defer {
+            ... @defer {
+              title
+            }
+          }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{}],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes", 0] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{ "data" => { "title" => "Banana" }, "id" => "0" }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_deduplicates_nested_defers_on_the_same_object
+    result = build_executor(%|query ProductTitleQuery {
+      products(first: 1) {
+        nodes {
+          ... @defer {
+            ...ProductTitle
+            ... @defer {
+              ...ProductTitle
+              ... @defer {
+                ...ProductTitle
+              }
+            }
+          }
+        }
+      }
+    }
+
+    fragment ProductTitle on Product {
+      id
+      title
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{}],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes", 0] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "data" => {
+            "id" => "gid://shopify/Product/1",
+            "title" => "Banana",
+          },
+          "id" => "0",
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_defers_top_level_fragment
+    result = build_executor(%|{
+      ... @defer(label: "Top") {
+        products(first: 1) {
+          nodes { title }
+        }
+      }
+    }|).incremental_result
+
+    assert_equal(
+      {
+        "data" => {},
+        "pending" => [{ "id" => "0", "path" => [], "label" => "Top" }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "data" => {
+            "products" => {
+              "nodes" => [
+                { "title" => "Banana" },
+                { "title" => "Apple" },
+              ],
+            },
+          },
+          "id" => "0",
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_formats_errors_in_top_level_deferred_fragment
+    source = Marshal.load(Marshal.dump(one_product_source))
+    source["products"]["nodes"][0]["title"] = GraphQL::ExecutionError.new("No title")
+
+    result = build_executor(%|{
+      ... @defer(label: "Top") {
+        products(first: 1) {
+          nodes { title }
+        }
+      }
+    }|, source: source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {},
+        "pending" => [{ "id" => "0", "path" => [], "label" => "Top" }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "data" => {
+            "products" => {
+              "nodes" => [{ "title" => nil }],
+            },
+          },
+          "id" => "0",
+          "errors" => [{
+            "message" => "No title",
+            "locations" => [{ "line" => 4, "column" => 19 }],
+            "path" => ["products", "nodes", 0, "title"],
+          }],
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_separately_emits_fragments_with_different_labels
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          ... @defer(label: "DeferredId") { id }
+          ... @defer(label: "DeferredTitle") { title }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{}],
+          },
+        },
+        "pending" => [
+          { "id" => "0", "path" => ["products", "nodes", 0], "label" => "DeferredId" },
+          { "id" => "1", "path" => ["products", "nodes", 0], "label" => "DeferredTitle" },
+        ],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [
+          { "data" => { "id" => "gid://shopify/Product/1" }, "id" => "0" },
+          { "data" => { "title" => "Banana" }, "id" => "1" },
+        ],
+        "completed" => [{ "id" => "0" }, { "id" => "1" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_uses_sub_path_for_nested_payloads
+    result = build_executor(%|{
+      ... @defer(label: "DeferredId") {
+        products(first: 1) {
+          nodes { id }
+        }
+      }
+      ... @defer(label: "DeferredTitle") {
+        products(first: 1) {
+          nodes { title }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {},
+        "pending" => [
+          { "id" => "0", "path" => [], "label" => "DeferredId" },
+          { "id" => "1", "path" => [], "label" => "DeferredTitle" },
+        ],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [
+          {
+            "data" => {
+              "products" => {
+                "nodes" => [{}],
+              },
+            },
+            "id" => "0",
+          },
+          {
+            "data" => { "id" => "gid://shopify/Product/1" },
+            "id" => "0",
+            "subPath" => ["products", "nodes", 0],
+          },
+          {
+            "data" => { "title" => "Banana" },
+            "id" => "1",
+            "subPath" => ["products", "nodes", 0],
+          },
+        ],
+        "completed" => [{ "id" => "0" }, { "id" => "1" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_formats_nullable_errors_in_deferred_payload
+    source = Marshal.load(Marshal.dump(SOURCE))
+    source["products"]["nodes"][0]["title"] = GraphQL::ExecutionError.new("Oops!")
+
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer { title }
+        }
+      }
+    }|, source: source).incremental_result
+
+    assert_equal(
+      [{
+        "incremental" => [
+          {
+            "data" => { "title" => nil },
+            "id" => "0",
+            "errors" => [{
+              "message" => "Oops!",
+              "locations" => [{ "line" => 5, "column" => 24 }],
+              "path" => ["products", "nodes", 0, "title"],
+            }],
+          },
+          { "data" => { "title" => "Apple" }, "id" => "1" },
+        ],
+        "completed" => [
+          { "id" => "0" },
+          { "id" => "1" },
+        ],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_supports_nested_defer
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer(label: "Outer") {
+            title
+            variants(first: 1) {
+              nodes {
+                id
+                ... @defer(label: "Inner") { title }
+              }
+            }
+          }
+        }
+      }
+    }|, source: one_product_source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes", 0], "label" => "Outer" }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "pending" => [{
+          "id" => "1",
+          "path" => ["products", "nodes", 0, "variants", "nodes", 0],
+          "label" => "Inner",
+        }],
+        "incremental" => [
+          {
+            "data" => {
+              "title" => "Banana",
+              "variants" => {
+                "nodes" => [{ "id" => "gid://shopify/Variant/1" }],
+              },
+            },
+            "id" => "0",
+          },
+          { "data" => { "title" => "Small Banana" }, "id" => "1" },
+        ],
+        "completed" => [
+          { "id" => "0" },
+          { "id" => "1" },
+        ],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_cancels_nested_defer_when_parent_deferred_fragment_fails
+    source = Marshal.load(Marshal.dump(SOURCE))
+    source["products"]["nodes"] = [source["products"]["nodes"].first]
+    source["products"]["nodes"][0]["must"] = nil
+
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer(label: "Outer") {
+            must
+            variants(first: 1) {
+              nodes {
+                id
+                ... @defer(label: "Inner") { title }
+              }
+            }
+          }
+        }
+      }
+    }|, source: source).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes", 0], "label" => "Outer" }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "completed" => [{
+          "id" => "0",
+          "errors" => [{
+            "message" => "Cannot return null for non-nullable field Product.must",
+            "locations" => [{ "line" => 6, "column" => 13 }],
+            "path" => ["products", "nodes", 0, "must"],
+            "extensions" => { "code" => "INVALID_NULL" },
+          }],
+        }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_completes_deferred_group_with_non_null_errors
+    source = Marshal.load(Marshal.dump(SOURCE))
+    source["products"]["nodes"][0]["must"] = nil
+    source["products"]["nodes"] = [source["products"]["nodes"][0]]
+
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          id
+          ... @defer { must }
+        }
+      }
+    }|, source: source).incremental_result
+
+    assert_equal(
+      [{
+        "completed" => [{
+          "id" => "0",
+          "errors" => [{
+            "message" => "Cannot return null for non-nullable field Product.must",
+            "locations" => [{ "line" => 5, "column" => 24 }],
+            "path" => ["products", "nodes", 0, "must"],
+            "extensions" => { "code" => "INVALID_NULL" },
+          }],
+        }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_cancels_deferred_payload_for_null_bubbled_parent
+    source = Marshal.load(Marshal.dump(SOURCE))
+    source["products"]["nodes"] = [source["products"]["nodes"].first]
+    source["products"]["nodes"][0]["must"] = nil
+
+    result = build_executor(%|{
+      products(first: 1) {
+        nodes {
+          must
+          ... @defer { title }
+        }
+      }
+    }|, source: source).incremental_result
+
+    assert_equal(
+      { "products" => nil },
+      result.initial_result.fetch("data"),
+    )
+    refute result.initial_result.key?("pending")
+    refute result.initial_result.key?("hasNext")
+    assert_equal(
+      [{
+        "message" => "Cannot return null for non-nullable field Product.must",
+        "locations" => [{ "line" => 4, "column" => 11 }],
+        "path" => ["products", "nodes", 0, "must"],
+        "extensions" => { "code" => "INVALID_NULL" },
+      }],
+      result.initial_result.fetch("errors"),
+    )
+    assert_equal [], result.subsequent_results.to_a
+  end
+
+  def test_incremental_result_filters_deferred_payload_for_null_bubbled_list_element
+    # `nodes: [Node]!` has NULLABLE elements. Element 0's non-null `must` returns nil, so the
+    # element null-bubbles to null in the initial result while the list and its siblings survive
+    # (no scope abort). A deferred fragment is attached to that now-dead element: it must NOT be
+    # announced as pending or delivered as incremental — there is no live object to patch into.
+    source = {
+      "nodes" => [
+        { "__typename__" => "Product", "id" => "1", "must" => nil, "title" => "Banana" },
+        { "__typename__" => "Product", "id" => "2", "must" => "yes", "title" => "Apple" },
+      ],
+    }
+
+    result = build_executor(%|{
+      nodes(ids: ["1", "2"]) {
+        ... on Product {
+          id
+          must
+          ... @defer { title }
+        }
+      }
+    }|, source: source).incremental_result
+
+    assert_instance_of GraphQL::BreadthExec::Incremental::Result, result
+
+    initial = result.initial_result
+    assert_equal(
+      { "nodes" => [nil, { "id" => "2", "must" => "yes" }] },
+      initial.fetch("data"),
+    )
+    # only the surviving element (index 1) is announced; the null-bubbled element 0 is filtered out.
+    assert_equal(
+      [{ "id" => "0", "path" => ["nodes", 1] }],
+      initial.fetch("pending"),
+    )
+    assert_equal(true, initial.fetch("hasNext"))
+    assert_equal(["nodes", 0, "must"], initial.fetch("errors").first.fetch("path"))
+
+    # the dead element delivers nothing; the surviving element delivers and completes normally.
+    assert_equal(
+      [{
+        "incremental" => [{ "data" => { "title" => "Apple" }, "id" => "0" }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_ready_deferred_executions_batch_lazy_work
+    source = Marshal.load(Marshal.dump(SOURCE))
+    source["products"]["nodes"][0]["maybe"] = "Yellow"
+    source["products"]["nodes"][1]["maybe"] = "Red"
+
+    resolvers = BREADTH_RESOLVERS.merge(
+      "Product" => BREADTH_RESOLVERS.fetch("Product").merge(
+        "title" => LazyHashResolver.new("title"),
+        "maybe" => LazyHashResolver.new("maybe"),
+      ),
+    )
+
+    BatchTrackingLoader.perform_keys = []
+
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes {
+          id
+          ... @defer(label: "Title") { title }
+          ... @defer(label: "Maybe") { maybe }
+        }
+      }
+    }|, source:, resolvers:).incremental_result
+
+    result.subsequent_results.to_a
+
+    assert_equal(
+      [["Banana", "Apple", "Yellow", "Red"]],
+      BatchTrackingLoader.perform_keys,
+    )
+  end
+
+  private
+
+  def build_executor(document, source: SOURCE, resolvers: BREADTH_RESOLVERS)
+    GraphQL::BreadthExec::Executor.new(
+      SCHEMA,
+      GraphQL.parse(document),
+      resolvers: resolvers,
+      root_object: source,
+    )
+  end
+
+  def one_product_source
+    {
+      "products" => {
+        "nodes" => [SOURCE.fetch("products").fetch("nodes").first],
+      },
+    }
+  end
+end
diff --git a/test/graphql/breadth_exec/executor/subscriptions_test.rb b/test/graphql/breadth_exec/executor/subscriptions_test.rb
index 6877f15..71281d3 100644
--- a/test/graphql/breadth_exec/executor/subscriptions_test.rb
+++ b/test/graphql/breadth_exec/executor/subscriptions_test.rb
@@ -63,14 +63,30 @@ def test_result_raises_for_subscription_operation
       subscription_executor(%|subscription { onWriteValue { value } }|).result
     end
 
-    assert_equal "Use result_or_subscribe for subscription operations", error.message
+    assert_equal "Use subscribe for subscription operations", error.message
   end
 
-  def test_result_or_subscribe_returns_subscription_response_stream_for_subscription_operation
+  def test_incremental_result_raises_for_subscription_operation
+    error = assert_raises(GraphQL::BreadthExec::ImplementationError) do
+      subscription_executor(%|subscription { onWriteValue { value } }|).incremental_result
+    end
+
+    assert_equal "Use subscribe for subscription operations", error.message
+  end
+
+  def test_subscribe_raises_for_query_operation
+    error = assert_raises(GraphQL::BreadthExec::ImplementationError) do
+      subscription_executor(%|{ noop }|).subscribe
+    end
+
+    assert_equal "Only allowed for subscription operations", error.message
+  end
+
+  def test_subscribe_returns_subscription_response_stream_for_subscription_operation
     result = subscription_executor(
       %|subscription { onWriteValue { value } }|,
       root_object: { "events" => [{ "value" => "direct" }] },
-    ).result_or_subscribe
+    ).subscribe
 
     assert_instance_of GraphQL::BreadthExec::SubscriptionResponseStream, result
     assert_equal [{ "data" => { "onWriteValue" => { "value" => "direct" } } }], result.to_a
@@ -79,7 +95,7 @@ def test_result_or_subscribe_returns_subscription_response_stream_for_subscripti
   def test_subscribe_reports_graphql_execution_error_from_source_resolver
     resolver = SourceResolver.new(subscribe: ->(_exec_field, _ctx) { raise GraphQL::ExecutionError, "Cannot subscribe" })
 
-    result = subscription_executor(%|subscription { onWriteValue { value } }|, source_resolver: resolver).result_or_subscribe
+    result = subscription_executor(%|subscription { onWriteValue { value } }|, source_resolver: resolver).subscribe
 
     assert_equal "Cannot subscribe", result.dig("errors", 0, "message")
     assert_equal ["onWriteValue"], result.dig("errors", 0, "path")
@@ -90,7 +106,7 @@ def test_subscribe_raises_for_plain_field_resolver_without_subscription_hook
       subscription_executor(
         %|subscription { onWriteValue { value } }|,
         source_resolver: GraphQL::BreadthExec::SelfResolver.new,
-      ).result_or_subscribe
+      ).subscribe
     end
 
     assert_equal "FieldResolver#subscribe must be implemented.", error.message
@@ -100,17 +116,17 @@ def test_subscribe_raises_for_non_enumerable_source
     resolver = SourceResolver.new(subscribe: ->(_exec_field, _ctx) { Object.new })
 
     error = assert_raises(GraphQL::BreadthExec::ImplementationError) do
-      subscription_executor(%|subscription { onWriteValue { value } }|, source_resolver: resolver).result_or_subscribe
+      subscription_executor(%|subscription { onWriteValue { value } }|, source_resolver: resolver).subscribe
     end
 
     assert_equal "Subscription source must return an Enumerable", error.message
   end
 
   def test_subscribe_raises_for_lazy_source_setup
-    resolver = SourceResolver.new(subscribe: ->(_exec_field, _ctx) { GraphQL::BreadthExec::ExecutionPromise.new })
+    resolver = SourceResolver.new(subscribe: ->(_exec_field, _ctx) { GraphQL::BreadthExec::Executor::ExecutionPromise.new })
 
     error = assert_raises(GraphQL::BreadthExec::ImplementationError) do
-      subscription_executor(%|subscription { onWriteValue { value } }|, source_resolver: resolver).result_or_subscribe
+      subscription_executor(%|subscription { onWriteValue { value } }|, source_resolver: resolver).subscribe
     end
 
     assert_equal "Subscription source must return an Enumerable", error.message
@@ -120,7 +136,7 @@ def test_stream_yields_one_result_for_one_source_event
     stream = subscription_executor(
       %|subscription { onWriteValue { value } }|,
       root_object: { "events" => [{ "value" => "first" }] },
-    ).result_or_subscribe
+    ).subscribe
 
     assert_instance_of GraphQL::BreadthExec::SubscriptionResponseStream, stream
     assert_equal(
@@ -139,7 +155,7 @@ def test_stream_preserves_multiple_event_order
           { "value" => "third" },
         ],
       },
-    ).result_or_subscribe
+    ).subscribe
 
     assert_equal(
       [
@@ -165,7 +181,7 @@ def test_stream_yields_events_published_after_subscription_result_is_created
     stream = subscription_executor(
       %|subscription { onWriteValue { value } }|,
       root_object: { "events" => source_stream },
-    ).result_or_subscribe
+    ).subscribe
 
     deliveries = Queue.new
     stream_errors = Queue.new
@@ -208,7 +224,7 @@ def test_event_execution_uses_subscription_root_resolver
       source_resolver: resolver,
       root_object: { "events" => [{ "payload" => { "value" => "from resolver" } }] },
       context: { test_context: "ctx" },
-    ).result_or_subscribe
+    ).subscribe
 
     assert_equal [{ "data" => { "onWriteValue" => { "value" => "from resolver" } } }], stream.to_a
     assert_equal [[{}, "ctx"]], resolver.subscribe_calls
@@ -219,7 +235,7 @@ def test_event_execution_supports_lazy_fields
     stream = subscription_executor(
       %|subscription { onWriteValue { lazyValue } }|,
       root_object: { "events" => [{ "lazyValue" => "loaded" }] },
-    ).result_or_subscribe
+    ).subscribe
 
     assert_equal(
       [{ "data" => { "onWriteValue" => { "lazyValue" => "loaded" } } }],
@@ -231,7 +247,7 @@ def test_event_execution_errors_are_returned_in_result
     stream = subscription_executor(
       %|subscription { onWriteValue { error } }|,
       root_object: { "events" => [{ "value" => "bad" }] },
-    ).result_or_subscribe
+    ).subscribe
 
     assert_equal(
       [{
@@ -255,7 +271,7 @@ def test_source_stream_errors_propagate_to_consumer
     stream = subscription_executor(
       %|subscription { onWriteValue { value } }|,
       root_object: { "events" => source_stream },
-    ).result_or_subscribe
+    ).subscribe
 
     responses = []
     error = assert_raises(RuntimeError) do
@@ -266,21 +282,21 @@ def test_source_stream_errors_propagate_to_consumer
     assert_equal [{ "data" => { "onWriteValue" => { "value" => "before error" } } }], responses
   end
 
-  def test_result_or_subscribe_returns_same_stream_when_called_twice
+  def test_subscribe_returns_same_stream_when_called_twice
     executor = subscription_executor(%|subscription { onWriteValue { value } }|)
 
-    stream = executor.result_or_subscribe
+    stream = executor.subscribe
 
-    assert_same stream, executor.result_or_subscribe
+    assert_same stream, executor.subscribe
   end
 
   def test_result_raises_after_subscription_response_stream_was_created
     executor = subscription_executor(%|subscription { onWriteValue { value } }|)
 
-    executor.result_or_subscribe
+    executor.subscribe
     error = assert_raises(GraphQL::BreadthExec::ImplementationError) { executor.result }
 
-    assert_equal "Use result_or_subscribe for subscription operations", error.message
+    assert_equal "Use subscribe for subscription operations", error.message
   end
 
   private
diff --git a/test/graphql/breadth_exec/executor/tracers_test.rb b/test/graphql/breadth_exec/executor/tracers_test.rb
index 9d9b2db..f0c213e 100644
--- a/test/graphql/breadth_exec/executor/tracers_test.rb
+++ b/test/graphql/breadth_exec/executor/tracers_test.rb
@@ -9,12 +9,25 @@ def resolve(_exec_field, _context)
     end
   end
 
+  class ExtensionTracer < GraphQL::BreadthExec::Tracer
+    def before_execute(executor, _context)
+      executor.response_extensions[:trace] = { started: true }
+    end
+
+    def after_execute(executor, _context, duration:)
+      executor.response_extensions[:trace][:finished] = true
+    end
+  end
+
   class TestTracer < GraphQL::BreadthExec::Tracer
     attr_reader :start_calls, :finish_calls, :finish_durations
     attr_reader :before_execute_calls, :after_execute_calls, :execute_durations
     attr_reader :before_format_errors_calls
-    attr_reader :before_resolve_field_calls, :after_resolve_field_calls
+    attr_reader :before_scope_calls
+    attr_reader :before_resolve_field_calls, :after_resolve_field_calls, :resolve_durations
     attr_reader :before_build_field_result_calls, :after_build_field_result_calls, :build_field_result_durations
+    attr_reader :before_lazy_set_calls, :after_lazy_set_calls, :lazy_set_durations
+    attr_reader :before_abstract_scope_calls
     attr_reader :on_exception_calls
 
     def initialize
@@ -26,11 +39,17 @@ def initialize
       @after_execute_calls = []
       @execute_durations = []
       @before_format_errors_calls = []
+      @before_scope_calls = []
       @before_resolve_field_calls = []
       @after_resolve_field_calls = []
+      @resolve_durations = []
       @before_build_field_result_calls = []
       @after_build_field_result_calls = []
       @build_field_result_durations = []
+      @before_lazy_set_calls = []
+      @after_lazy_set_calls = []
+      @lazy_set_durations = []
+      @before_abstract_scope_calls = []
       @on_exception_calls = []
     end
 
@@ -56,12 +75,30 @@ def before_format_errors(executor, _context)
       @before_format_errors_calls << executor
     end
 
-    def before_resolve_field(parent_type, field_name, objects_count, _context)
-      @before_resolve_field_calls << [parent_type.graphql_name, field_name, objects_count]
+    def before_scope(exec_scope, _context)
+      @before_scope_calls << exec_scope.parent_type.graphql_name
+    end
+
+    def before_resolve_field(exec_field, _context)
+      @before_resolve_field_calls << [exec_field.parent_type.graphql_name, exec_field.name, exec_field.objects.length]
+    end
+
+    def after_resolve_field(exec_field, _context, duration:)
+      @after_resolve_field_calls << [exec_field.parent_type.graphql_name, exec_field.name, exec_field.objects.length]
+      @resolve_durations << duration
+    end
+
+    def before_lazy_set(loader, elements, _context)
+      @before_lazy_set_calls << [loader.class, elements.map(&:class)]
+    end
+
+    def after_lazy_set(loader, elements, _context, duration:)
+      @after_lazy_set_calls << [loader.class, elements.map(&:class)]
+      @lazy_set_durations << duration
     end
 
-    def after_resolve_field(parent_type, field_name, objects_count, _context)
-      @after_resolve_field_calls << [parent_type.graphql_name, field_name, objects_count]
+    def before_abstract_scope(abstract_scope, _context)
+      @before_abstract_scope_calls << abstract_scope.parent_type.graphql_name
     end
 
     def before_build_field_result(exec_field, _context)
@@ -119,6 +156,17 @@ def test_calls_field_resolve_hooks_with_parent_type_field_name_and_object_count
       @tracer.before_resolve_field_calls,
     )
     assert_equal @tracer.before_resolve_field_calls, @tracer.after_resolve_field_calls
+    assert @tracer.resolve_durations.all? { _1.is_a?(Float) && _1 >= 0 }
+  end
+
+  def test_calls_scope_hooks
+    result = breadth_exec(@document, @source, tracers: [@tracer])
+
+    assert_equal @source, result["data"]
+    assert_equal(
+      ["Query", "ProductConnection", "Product"],
+      @tracer.before_scope_calls,
+    )
   end
 
   def test_calls_field_result_build_hooks
@@ -144,6 +192,20 @@ def test_calls_before_format_errors
     assert @tracer.before_format_errors_calls.all? { _1.is_a?(GraphQL::BreadthExec::Executor) }
   end
 
+  def test_tracers_can_install_response_extensions
+    result = breadth_exec(@document, @source, tracers: [ExtensionTracer.new])
+
+    assert_equal(
+      {
+        "trace" => {
+          "started" => true,
+          "finished" => true,
+        },
+      },
+      result["extensions"],
+    )
+  end
+
   def test_calls_on_exception_for_unhandled_errors
     resolvers = BREADTH_RESOLVERS.merge(
       "Query" => BREADTH_RESOLVERS.fetch("Query").merge(
diff --git a/test/graphql/breadth_exec/field_resolvers_test.rb b/test/graphql/breadth_exec/field_resolvers_test.rb
index d0c59bf..cf20d98 100644
--- a/test/graphql/breadth_exec/field_resolvers_test.rb
+++ b/test/graphql/breadth_exec/field_resolvers_test.rb
@@ -115,11 +115,11 @@ def test_handle_resolved_yields_array_directly
 
   def test_handle_resolved_wraps_promise_with_then
     resolver = HandleResolvedTestResolver.new
-    promise = GraphQL::BreadthExec::ExecutionPromise.new
+    promise = GraphQL::BreadthExec::Executor::ExecutionPromise.new
 
     result = resolver.handle_resolved(promise) { |values| values.map(&:upcase) }
 
-    assert_instance_of GraphQL::BreadthExec::ExecutionPromise, result
+    assert_instance_of GraphQL::BreadthExec::Executor::ExecutionPromise, result
   end
 
   private
diff --git a/test/graphql/breadth_exec/incremental/context_test.rb b/test/graphql/breadth_exec/incremental/context_test.rb
new file mode 100644
index 0000000..10b86fe
--- /dev/null
+++ b/test/graphql/breadth_exec/incremental/context_test.rb
@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class GraphQL::BreadthExec::Incremental::ContextTest < Minitest::Test
+  DeferredDelivery = GraphQL::BreadthExec::Incremental::DeferredDelivery
+  DeferredExecutionScope = GraphQL::BreadthExec::Incremental::DeferredExecutionScope
+  ExecutionScope = GraphQL::BreadthExec::Executor::ExecutionScope
+  IncrementalContext = GraphQL::BreadthExec::Incremental::Context
+
+  def test_completed_payloads_deduplicate_deliveries
+    context = IncrementalContext.new(nil, data: {})
+    delivery = DeferredDelivery.new(["hero"])
+
+    assert_equal(
+      [{ "id" => "0", "errors" => [{ "message" => "bad" }] }],
+      context.completed_payloads(
+        [delivery, delivery],
+        errors_by_delivery: { delivery => [{ "message" => "bad" }] },
+      ),
+    )
+    assert_equal [], context.completed_payloads([delivery])
+  end
+
+  def test_deferred_execution_scope_is_not_ready_when_base_scope_aborted
+    executor = GraphQL::BreadthExec::Executor.new(
+      SCHEMA,
+      GraphQL.parse("{ noResolver }"),
+      resolvers: BREADTH_RESOLVERS,
+      root_object: {},
+    )
+    base_scope = ExecutionScope.new(
+      executor: executor,
+      parent_type: SCHEMA.query,
+      selections: [],
+      objects: [{}].freeze,
+      results: [{}].freeze,
+    )
+    base_scope.executed = true
+
+    deferred_scope = DeferredExecutionScope.new(
+      base_scope: base_scope,
+      field_selections: {},
+      defer_usages: Set.new,
+    )
+
+    assert deferred_scope.ready?
+    base_scope.abort!
+    refute deferred_scope.ready?
+  end
+end
diff --git a/test/graphql/breadth_exec/incremental/publisher_test.rb b/test/graphql/breadth_exec/incremental/publisher_test.rb
new file mode 100644
index 0000000..1c4cf26
--- /dev/null
+++ b/test/graphql/breadth_exec/incremental/publisher_test.rb
@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class GraphQL::BreadthExec::Incremental::PublisherTest < Minitest::Test
+  DeferredDelivery = GraphQL::BreadthExec::Incremental::DeferredDelivery
+  Publisher = GraphQL::BreadthExec::Incremental::Publisher
+
+  def test_pending_assigns_stable_ids_and_includes_label
+    publisher = Publisher.new
+    delivery = DeferredDelivery.new(["hero"], "HeroFields")
+
+    expected = [{ "id" => "0", "path" => ["hero"], "label" => "HeroFields" }]
+
+    assert_equal expected, publisher.pending([delivery])
+    assert_equal expected, publisher.pending([delivery])
+  end
+
+  def test_incremental_payload_uses_deepest_matching_delivery
+    publisher = Publisher.new
+    parent = DeferredDelivery.new(["hero"], "HeroFields")
+    child = DeferredDelivery.new(["hero", "friends", 0], "FriendFields", parent: parent)
+
+    publisher.pending([parent, child])
+
+    assert_equal(
+      {
+        "data" => { "name" => "Han" },
+        "id" => "1",
+        "subPath" => ["profile"],
+      },
+      publisher.incremental([parent, child], ["hero", "friends", 0, "profile"], { "name" => "Han" }),
+    )
+    assert_equal(
+      {
+        "data" => { "appearsIn" => ["NEWHOPE"] },
+        "id" => "0",
+        "subPath" => ["appearsIn"],
+      },
+      publisher.incremental([parent, child], ["hero", "appearsIn"], { "appearsIn" => ["NEWHOPE"] }),
+    )
+  end
+
+  def test_completed_includes_errors_and_allocates_new_ids_after_completion
+    publisher = Publisher.new
+    delivery = DeferredDelivery.new(["hero"])
+
+    assert_equal [{ "id" => "0", "path" => ["hero"] }], publisher.pending([delivery])
+    assert_equal(
+      {
+        "id" => "0",
+        "errors" => [{ "message" => "bad" }],
+      },
+      publisher.completed(delivery, errors: [{ "message" => "bad" }]),
+    )
+    assert_equal [{ "id" => "1", "path" => ["hero"] }], publisher.pending([delivery])
+  end
+end