From aa19f117b3bdbaeaf4641fa31233f243548cf521 Mon Sep 17 00:00:00 2001
From: Greg MacWilliam <greg.macwilliam@shopify.com>
Date: Mon, 22 Jun 2026 23:37:31 -0400
Subject: [PATCH 1/4] support incremental @stream

---
 lib/graphql/breadth/executor.rb               | 224 ++++++++++++++++--
 .../breadth/executor/execution_planner.rb     |  28 +++
 .../breadth/executor/path_formatter.rb        |   7 +-
 lib/graphql/breadth/incremental.rb            |   3 +
 lib/graphql/breadth/incremental/context.rb    |  36 ++-
 lib/graphql/breadth/incremental/publisher.rb  |  16 +-
 .../breadth/incremental/stream_delivery.rb    |  22 ++
 .../incremental/stream_execution_scope.rb     |  82 +++++++
 .../breadth/incremental/stream_usage.rb       |  22 ++
 .../breadth/executor/incremental_test.rb      | 137 +++++++++++
 10 files changed, 551 insertions(+), 26 deletions(-)
 create mode 100644 lib/graphql/breadth/incremental/stream_delivery.rb
 create mode 100644 lib/graphql/breadth/incremental/stream_execution_scope.rb
 create mode 100644 lib/graphql/breadth/incremental/stream_usage.rb

diff --git a/lib/graphql/breadth/executor.rb b/lib/graphql/breadth/executor.rb
index 801529f..2ef07d5 100644
--- a/lib/graphql/breadth/executor.rb
+++ b/lib/graphql/breadth/executor.rb
@@ -676,6 +676,7 @@ def build_field_result(exec_field, resolved_objects)
         field_key = exec_field.key
         field_type = exec_field.type
         return_type = field_type.unwrap
+        stream_usage = active_stream_usage_for(exec_field, field_type)
 
         if resolved_objects.length != parent_objects.length
           handle_or_reraise(ResultCountMismatchError.new(
@@ -693,7 +694,11 @@ def build_field_result(exec_field, resolved_objects)
           i = 0
           while i < resolved_objects.length
             object = resolved_objects[i]
-            parent_results[i][field_key] = build_and_flatmap_composite_result(exec_field, field_type, object, next_objects, next_results)
+            parent_results[i][field_key] = if stream_usage
+              build_streaming_composite_result(exec_field, field_type, object, next_objects, next_results, stream_usage, i)
+            else
+              build_and_flatmap_composite_result(exec_field, field_type, object, next_objects, next_results)
+            end
             i += 1
           end
 
@@ -712,7 +717,11 @@ def build_field_result(exec_field, resolved_objects)
           i = 0
           while i < resolved_objects.length
             val = resolved_objects[i]
-            parent_results[i][field_key] = build_leaf_result(exec_field, field_type, val)
+            parent_results[i][field_key] = if stream_usage
+              build_streaming_leaf_result(exec_field, field_type, val, stream_usage, i)
+            else
+              build_leaf_result(exec_field, field_type, val)
+            end
             i += 1
           end
         end
@@ -727,6 +736,150 @@ def build_field_result(exec_field, resolved_objects)
         end
       end
 
+      #: (ExecutionField[untyped], singleton(GraphQL::Schema::Member)) -> Incremental::StreamUsage?
+      def active_stream_usage_for(exec_field, field_type)
+        return nil unless incremental?
+        return nil unless Util.unwrap_non_null(field_type).list?
+
+        @planner.stream_usage_for(exec_field)
+      end
+
+      #: (
+      #|   ExecutionField[untyped] exec_field,
+      #|   untyped current_type,
+      #|   untyped object,
+      #|   Array[untyped] next_objects,
+      #|   Array[Hash[String, untyped]] next_results,
+      #|   Incremental::StreamUsage stream_usage,
+      #|   Integer object_index,
+      #| ) -> untyped
+      def build_streaming_composite_result(exec_field, current_type, object, next_objects, next_results, stream_usage, object_index)
+        return build_and_flatmap_composite_result(exec_field, current_type, object, next_objects, next_results) if object.nil? || object.is_a?(StandardError)
+        return build_and_flatmap_composite_result(exec_field, current_type, object, next_objects, next_results) unless Util.unwrap_non_null(current_type).list?
+
+        objects = list_items_for_stream(exec_field, object)
+        initial_count = stream_usage.initial_count
+        item_type = Util.unwrap_non_null(current_type).of_type
+
+        initial_items = objects.take(initial_count)
+        initial_result = initial_items.map do |src|
+          build_and_flatmap_composite_result(exec_field, item_type, src, next_objects, next_results)
+        end
+
+        register_composite_stream_scope(exec_field, item_type, stream_usage, object_index, objects.drop(initial_count), initial_count) if initial_count <= objects.length
+
+        initial_result
+      end
+
+      #: (
+      #|   ExecutionField[untyped] exec_field,
+      #|   untyped current_type,
+      #|   untyped val,
+      #|   Incremental::StreamUsage stream_usage,
+      #|   Integer object_index,
+      #| ) -> untyped
+      def build_streaming_leaf_result(exec_field, current_type, val, stream_usage, object_index)
+        return build_leaf_result(exec_field, current_type, val) if val.nil? || val.is_a?(StandardError)
+        return build_leaf_result(exec_field, current_type, val) unless Util.unwrap_non_null(current_type).list?
+
+        values = list_items_for_stream(exec_field, val)
+        initial_count = stream_usage.initial_count
+        item_type = Util.unwrap_non_null(current_type).of_type
+
+        initial_result = values.take(initial_count).map { build_leaf_result(exec_field, item_type, _1) }
+        if initial_count <= values.length
+          stream_items = values.drop(initial_count).map { build_leaf_result(exec_field, item_type, _1) }
+          register_stream_scope(
+            exec_field,
+            exec_field.scope.parent_type,
+            EMPTY_ARRAY,
+            EMPTY_ARRAY,
+            EMPTY_ARRAY,
+            stream_items,
+            EMPTY_ARRAY,
+            stream_usage,
+            object_index,
+            initial_count,
+          )
+        end
+
+        initial_result
+      end
+
+      #: (ExecutionField[untyped], untyped) -> Array[untyped]
+      def list_items_for_stream(exec_field, value)
+        return value if value.is_a?(Array)
+        return value.to_a if value.is_a?(Enumerable)
+
+        raise InvalidListResultError.new(exec_field:, result_type: value.class)
+      end
+
+      #: (
+      #|   ExecutionField[untyped] exec_field,
+      #|   untyped item_type,
+      #|   Incremental::StreamUsage stream_usage,
+      #|   Integer object_index,
+      #|   Array[untyped] tail_objects,
+      #|   Integer initial_count,
+      #| ) -> void
+      def register_composite_stream_scope(exec_field, item_type, stream_usage, object_index, tail_objects, initial_count)
+        stream_items = []
+        stream_objects = []
+        stream_results = []
+        object_paths = []
+        stream_path = exec_field.object_path(object_index)
+
+        tail_objects.each_with_index do |src, index|
+          before_results = stream_results.length
+          stream_items << build_and_flatmap_composite_result(exec_field, item_type, src, stream_objects, stream_results)
+          while before_results < stream_results.length
+            object_paths[before_results] = [*stream_path, initial_count + index]
+            before_results += 1
+          end
+        end
+
+        register_stream_scope(
+          exec_field,
+          item_type.unwrap,
+          exec_field.selections,
+          stream_objects,
+          stream_results,
+          stream_items,
+          object_paths,
+          stream_usage,
+          object_index,
+          initial_count,
+        )
+      end
+
+      #: (
+      #|   ExecutionField[untyped] exec_field,
+      #|   singleton(GraphQL::Schema::Object) parent_type,
+      #|   Array[selection_node] selections,
+      #|   Array[untyped] objects,
+      #|   Array[untyped] results,
+      #|   Array[untyped] items,
+      #|   Array[error_path] object_paths,
+      #|   Incremental::StreamUsage stream_usage,
+      #|   Integer object_index,
+      #|   Integer initial_count,
+      #| ) -> void
+      def register_stream_scope(exec_field, parent_type, selections, objects, results, items, object_paths, stream_usage, object_index, initial_count)
+        stream_path = exec_field.object_path(object_index)
+        stream_delivery = Incremental::StreamDelivery.new(stream_path, stream_usage.label)
+        @incremental.register_stream_scope(Incremental::StreamExecutionScope.new(
+          parent_field: exec_field,
+          parent_type:,
+          selections:,
+          objects:,
+          results:,
+          items:,
+          object_paths:,
+          delivery: stream_delivery,
+          initial_index: initial_count,
+        ))
+      end
+
       #: (
       #|   ExecutionField[untyped] exec_field,
       #|   untyped current_type,
@@ -997,24 +1150,31 @@ def execute_next_incremental_result(yielder)
           break if ready_scopes.empty?
 
           initial_error_count = @invalidated_results.size
-          run!(@planner.plan_scopes(ready_scopes))
+          scopes_to_execute = ready_scopes.reject { _1.is_a?(Incremental::StreamExecutionScope) && _1.objects.empty? }
+          run!(@planner.plan_scopes(scopes_to_execute)) unless scopes_to_execute.empty?
           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:)
+            if exec_scope.is_a?(Incremental::StreamExecutionScope)
+              items, stream_errors = stream_items_for(exec_scope)
+              incremental_payloads << @incremental.stream_payload(exec_scope.delivery, items, errors: stream_errors) unless items.empty?
+              completed_deliveries << exec_scope.delivery
+            else
+              deliveries = @incremental.deliveries_for(exec_scope)
+              deliveries.each do |index, path, deferred_deliveries|
+                data = exec_scope.results[index]
+                deferred_errors = EMPTY_ARRAY
+                if has_errors
+                  data, deferred_errors = error_result_formatter.format_object(exec_scope.parent_type, exec_scope.selections, data, path)
+                end
+
+                if data.nil? && !deferred_errors.empty?
+                  deferred_deliveries.each { (completed_errors_by_delivery[_1] ||= []).concat(deferred_errors) }
+                else
+                  incremental_payloads << @incremental.incremental_payload(deferred_deliveries, path, data, errors: deferred_errors)
+                end
+                completed_deliveries.concat(deferred_deliveries)
               end
-              completed_deliveries.concat(deferred_deliveries)
             end
 
             exec_scope.executed = true
@@ -1030,6 +1190,38 @@ def execute_next_incremental_result(yielder)
         yielder << payload
       end
 
+      #: (Incremental::StreamExecutionScope) -> [Array[untyped], Array[error_hash]]
+      def stream_items_for(exec_scope)
+        items = []
+        errors = []
+
+        exec_scope.items.each_with_index do |item, index|
+          path = exec_scope.item_path(index)
+          if (err = @invalidated_results[item])
+            append_stream_item_errors(err, errors, path)
+            items << nil
+          elsif !exec_scope.selections.empty?
+            data, item_errors = error_result_formatter.format_object(exec_scope.parent_type, exec_scope.selections, item, path)
+            errors.concat(item_errors)
+            items << data
+          else
+            items << item
+          end
+        end
+
+        [items, errors]
+      end
+
+      #: (ExecutionError, Array[error_hash], error_path) -> void
+      def append_stream_item_errors(error, errors, path)
+        error.each do |err|
+          next if err.equal?(UNREPORTED_ERROR)
+
+          errors << err.to_h.tap { _1["path"] = path }
+          @context.errors << err.cause if err.cause
+        end
+      end
+
       #: (?data: Util::NilLike | graphql_result | nil, ?errors: Array[error_hash]) -> graphql_result
       def build_result(data: UNDEFINED, errors: EMPTY_ARRAY)
         result = {}
diff --git a/lib/graphql/breadth/executor/execution_planner.rb b/lib/graphql/breadth/executor/execution_planner.rb
index 665019d..f10c0cc 100644
--- a/lib/graphql/breadth/executor/execution_planner.rb
+++ b/lib/graphql/breadth/executor/execution_planner.rb
@@ -55,6 +55,32 @@ def root_directives_for_operation(operation)
           operation.directives.map { |node| build_execution_directive(node, depth: 0) }
         end
 
+        #: (ExecutionField[untyped]) -> Incremental::StreamUsage?
+        def stream_usage_for(exec_field)
+          return nil unless @executor.incremental?
+
+          node = exec_field.nodes.first #: as !nil
+          return nil if node.directives.empty?
+
+          directive = node.directives.find { _1.name == "stream" }
+          return nil unless directive
+
+          condition = directive.arguments.find { _1.name == "if" }
+          return nil if condition && argument_value(condition) == false
+
+          initial_count_arg = directive.arguments.find { _1.name == "initialCount" }
+          initial_count = initial_count_arg ? argument_value(initial_count_arg) : 0
+          unless initial_count.is_a?(Integer) && initial_count >= 0
+            raise ExecutionError.new("initialCount must be a positive integer", exec_field:)
+          end
+
+          label_arg = directive.arguments.find { _1.name == "label" }
+          label = label_arg ? argument_value(label_arg) : nil
+          label = nil unless label.is_a?(String)
+
+          Incremental::StreamUsage.new(label, initial_count:)
+        end
+
         #: (
         #|   GraphQL::Language::Nodes::OperationDefinition,
         #|   root_object: untyped,
@@ -268,6 +294,8 @@ def build_incremental_execution_fields(exec_scope, ordered_fields)
           selections_by_key = if exec_scope.is_a?(Incremental::DeferredExecutionScope)
             parent_usages = exec_scope.defer_usages
             exec_scope.field_selections
+          elsif exec_scope.is_a?(Incremental::StreamExecutionScope)
+            incremental_selections_grouped_by_key(exec_scope.parent_type, exec_scope.selections)
           elsif (parent_field = exec_scope.parent_field)
             map = Hash.new { |h, k| h[k] = [] }
             parent_incremental_selections = parent_field.incremental_selections #: as !nil
diff --git a/lib/graphql/breadth/executor/path_formatter.rb b/lib/graphql/breadth/executor/path_formatter.rb
index ba7fe1b..a8ced8e 100644
--- a/lib/graphql/breadth/executor/path_formatter.rb
+++ b/lib/graphql/breadth/executor/path_formatter.rb
@@ -29,11 +29,16 @@ def initialize
 
         #: (Executor::ExecutionScope, Integer) -> error_path
         def object_path(exec_scope, index)
-          current_path = []
+          current_path = [] #: error_path
 
           current_scope = exec_scope #: Executor::ExecutionScope?
           breadth_index = index
           while current_scope
+            if current_scope.is_a?(Incremental::StreamExecutionScope)
+              current_scope.stream_item_path(breadth_index).reverse_each { current_path.prepend(_1) }
+              break
+            end
+
             # index the scope unless it has already been done
             scope_indices = @indices_by_scope[current_scope]
             index_scope(current_scope, scope_indices) if scope_indices.empty?
diff --git a/lib/graphql/breadth/incremental.rb b/lib/graphql/breadth/incremental.rb
index aefd681..ca40e1a 100644
--- a/lib/graphql/breadth/incremental.rb
+++ b/lib/graphql/breadth/incremental.rb
@@ -4,6 +4,9 @@
 require_relative "incremental/defer_usage"
 require_relative "incremental/deferred_delivery"
 require_relative "incremental/deferred_execution_scope"
+require_relative "incremental/stream_usage"
+require_relative "incremental/stream_delivery"
+require_relative "incremental/stream_execution_scope"
 require_relative "incremental/partitioner"
 require_relative "incremental/selection"
 require_relative "incremental/result"
diff --git a/lib/graphql/breadth/incremental/context.rb b/lib/graphql/breadth/incremental/context.rb
index c334b38..5f3fd9e 100644
--- a/lib/graphql/breadth/incremental/context.rb
+++ b/lib/graphql/breadth/incremental/context.rb
@@ -20,6 +20,7 @@ def initialize(executor, data:)
           @data = data
           @publisher = Publisher.new
           @deferred_scopes = []
+          @stream_scopes = []
           @pending_deliveries = []
           @announced_deliveries = {}.compare_by_identity
           @completed_deliveries = {}.compare_by_identity
@@ -35,6 +36,11 @@ def register_deferred_scope(base_scope, field_selections, defer_usages)
           )
         end
 
+        #: (StreamExecutionScope) -> void
+        def register_stream_scope(stream_scope)
+          @stream_scopes << stream_scope
+        end
+
         #: -> bool
         def active?
           true
@@ -42,10 +48,10 @@ def active?
 
         #: -> bool
         def deferred?
-          @deferred_scopes.any?
+          @deferred_scopes.any? || @stream_scopes.any?
         end
 
-        #: -> Array[DeferredDelivery]
+        #: -> Array[DeferredDelivery | StreamDelivery]
         def prepare_pending
           @deferred_scopes.each do |deferred_scope|
             next if deferred_scope.announced? || !deferred_scope.ready?
@@ -53,6 +59,16 @@ def prepare_pending
             pending_deliveries_for(deferred_scope)
             deferred_scope.announced = true
           end
+          @stream_scopes.each do |stream_scope|
+            next if stream_scope.announced? || !stream_scope.ready? || deferred_path_nulled?(stream_scope.delivery.path)
+
+            delivery = stream_scope.delivery
+            unless @completed_deliveries[delivery] || @announced_deliveries[delivery]
+              @announced_deliveries[delivery] = true
+              @pending_deliveries << delivery
+            end
+            stream_scope.announced = true
+          end
 
           @pending_deliveries.uniq!
           pending = @pending_deliveries
@@ -60,9 +76,12 @@ def prepare_pending
           pending
         end
 
-        #: -> Array[DeferredExecutionScope]
+        #: -> Array[DeferredExecutionScope | StreamExecutionScope]
         def ready_scopes
-          @deferred_scopes.select { _1.announced? && !_1.executed? && _1.ready? }.each(&:prepare!)
+          (
+            @deferred_scopes.select { _1.announced? && !_1.executed? && _1.ready? } +
+            @stream_scopes.select { _1.announced? && !_1.executed? && _1.ready? && !deferred_path_nulled?(_1.delivery.path) }
+          ).each(&:prepare!)
         end
 
         #: (DeferredExecutionScope) -> Array[[Integer, error_path, Array[DeferredDelivery]]]
@@ -80,7 +99,7 @@ def deliveries_for(deferred_scope)
           deliveries
         end
 
-        #: (Array[DeferredDelivery]) -> Array[graphql_result]
+        #: (Array[DeferredDelivery | StreamDelivery]) -> Array[graphql_result]
         def pending_payloads(deliveries)
           @publisher.pending(deliveries)
         end
@@ -90,7 +109,12 @@ 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]
+        #: (StreamDelivery, Array[untyped], ?errors: Array[error_hash]) -> graphql_result
+        def stream_payload(delivery, items, errors: EMPTY_ARRAY)
+          @publisher.stream(delivery, items, errors:)
+        end
+
+        #: (Array[DeferredDelivery | StreamDelivery], ?errors_by_delivery: Hash[DeferredDelivery | StreamDelivery, 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]
diff --git a/lib/graphql/breadth/incremental/publisher.rb b/lib/graphql/breadth/incremental/publisher.rb
index a66254f..39faa3b 100644
--- a/lib/graphql/breadth/incremental/publisher.rb
+++ b/lib/graphql/breadth/incremental/publisher.rb
@@ -10,7 +10,7 @@ def initialize
           @next_id = 0
         end
 
-        #: (Array[DeferredDelivery]) -> Array[graphql_result]
+        #: (Array[DeferredDelivery | StreamDelivery]) -> Array[graphql_result]
         def pending(deliveries)
           deliveries.map do |delivery|
             result = {
@@ -36,7 +36,17 @@ def incremental(deliveries, path, data, errors: EMPTY_ARRAY)
           result
         end
 
-        #: (DeferredDelivery, ?errors: Array[error_hash]) -> graphql_result
+        #: (StreamDelivery, Array[untyped], ?errors: Array[error_hash]) -> graphql_result
+        def stream(delivery, items, errors: EMPTY_ARRAY)
+          result = {
+            "items" => items,
+            "id" => id_for(delivery),
+          }
+          result["errors"] = errors unless errors.empty?
+          result
+        end
+
+        #: (DeferredDelivery | StreamDelivery, ?errors: Array[error_hash]) -> graphql_result
         def completed(delivery, errors: EMPTY_ARRAY)
           result = { "id" => id_for(delivery) }
           result["errors"] = errors unless errors.empty?
@@ -46,7 +56,7 @@ def completed(delivery, errors: EMPTY_ARRAY)
 
         private
 
-        #: (DeferredDelivery) -> String
+        #: (DeferredDelivery | StreamDelivery) -> String
         def id_for(delivery)
           @ids[delivery] ||= begin
             id = @next_id.to_s
diff --git a/lib/graphql/breadth/incremental/stream_delivery.rb b/lib/graphql/breadth/incremental/stream_delivery.rb
new file mode 100644
index 0000000..5c81581
--- /dev/null
+++ b/lib/graphql/breadth/incremental/stream_delivery.rb
@@ -0,0 +1,22 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    module Incremental
+      class StreamDelivery
+        #: error_path
+        attr_reader :path
+
+        #: String?
+        attr_reader :label
+
+        #: (error_path, ?String?) -> void
+        def initialize(path, label = nil)
+          @path = path.freeze
+          @label = label
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth/incremental/stream_execution_scope.rb b/lib/graphql/breadth/incremental/stream_execution_scope.rb
new file mode 100644
index 0000000..dc26fe1
--- /dev/null
+++ b/lib/graphql/breadth/incremental/stream_execution_scope.rb
@@ -0,0 +1,82 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    module Incremental
+      class StreamExecutionScope < Executor::ExecutionScope
+        #: StreamDelivery
+        attr_reader :delivery
+
+        #: Array[untyped]
+        attr_reader :items
+
+        #: Integer
+        attr_reader :initial_index
+
+        #: bool
+        attr_writer :announced
+
+        #: (
+        #|   parent_field: Executor::ExecutionField[untyped],
+        #|   parent_type: singleton(GraphQL::Schema::Object),
+        #|   selections: Array[selection_node],
+        #|   objects: Array[untyped],
+        #|   results: Array[untyped],
+        #|   items: Array[untyped],
+        #|   object_paths: Array[error_path],
+        #|   delivery: StreamDelivery,
+        #|   initial_index: Integer,
+        #| ) -> void
+        def initialize(parent_field:, parent_type:, selections:, objects:, results:, items:, object_paths:, delivery:, initial_index:)
+          @delivery = delivery
+          @items = items
+          @object_paths = object_paths
+          @initial_index = initial_index
+          @announced = false
+
+          super(
+            executor: parent_field.executor,
+            parent_type:,
+            selections:,
+            objects:,
+            results:,
+            parent_field:,
+            deferred: true,
+          )
+        end
+
+        #: -> bool
+        def ready?
+          field = parent_field #: as !nil
+          field.scope.executed? && !field.scope.aborted?
+        end
+
+        #: -> bool
+        def announced?
+          @announced
+        end
+
+        #: -> StreamExecutionScope
+        def prepare!
+          self
+        end
+
+        #: (Integer) -> error_path
+        def item_path(index)
+          [*@delivery.path, @initial_index + index]
+        end
+
+        #: (Integer) -> error_path
+        def stream_item_path(index)
+          @object_paths[index] || item_path(index)
+        end
+
+        #: (Integer) -> error_path
+        def object_path(index)
+          stream_item_path(index)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth/incremental/stream_usage.rb b/lib/graphql/breadth/incremental/stream_usage.rb
new file mode 100644
index 0000000..bde145c
--- /dev/null
+++ b/lib/graphql/breadth/incremental/stream_usage.rb
@@ -0,0 +1,22 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    module Incremental
+      class StreamUsage
+        #: String?
+        attr_reader :label
+
+        #: Integer
+        attr_reader :initial_count
+
+        #: (?String?, initial_count: Integer) -> void
+        def initialize(label = nil, initial_count:)
+          @label = label
+          @initial_count = initial_count
+        end
+      end
+    end
+  end
+end
diff --git a/test/graphql/breadth/executor/incremental_test.rb b/test/graphql/breadth/executor/incremental_test.rb
index db0497e..96eea76 100644
--- a/test/graphql/breadth/executor/incremental_test.rb
+++ b/test/graphql/breadth/executor/incremental_test.rb
@@ -913,6 +913,143 @@ def test_ready_deferred_executions_batch_lazy_work
     )
   end
 
+  def test_incremental_result_streams_list_field
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream(initialCount: 1) {
+          id
+          title
+        }
+      }
+    }|).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1", "title" => "Banana" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "items" => [{ "id" => "gid://shopify/Product/2", "title" => "Apple" }],
+          "id" => "0",
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_streams_with_default_initial_count
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream {
+          id
+        }
+      }
+    }|).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "items" => [
+            { "id" => "gid://shopify/Product/1" },
+            { "id" => "gid://shopify/Product/2" },
+          ],
+          "id" => "0",
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+  end
+
+  def test_incremental_result_includes_stream_label
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream(initialCount: 1, label: "ProductNodes") {
+          id
+        }
+      }
+    }|).incremental_result
+
+    assert_equal(
+      [{ "id" => "0", "path" => ["products", "nodes"], "label" => "ProductNodes" }],
+      result.initial_result.fetch("pending"),
+    )
+  end
+
+  def test_incremental_result_does_not_stream_when_if_is_false
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream(initialCount: 0, if: false) {
+          id
+        }
+      }
+    }|).incremental_result
+
+    expected = {
+      "data" => {
+        "products" => {
+          "nodes" => [
+            { "id" => "gid://shopify/Product/1" },
+            { "id" => "gid://shopify/Product/2" },
+          ],
+        },
+      },
+    }
+
+    refute result.incremental?
+    assert_equal expected, result.initial_result
+    assert_equal [], result.subsequent_results.to_a
+  end
+
+  def test_ready_stream_executions_batch_lazy_work
+    resolvers = BREADTH_RESOLVERS.merge(
+      "Product" => BREADTH_RESOLVERS.fetch("Product").merge(
+        "title" => LazyHashResolver.new("title"),
+      ),
+    )
+
+    BatchTrackingLoader.perform_keys = []
+
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream {
+          title
+        }
+      }
+    }|, resolvers:).incremental_result
+
+    result.subsequent_results.to_a
+
+    assert_equal(
+      [["Banana", "Apple"]],
+      BatchTrackingLoader.perform_keys,
+    )
+  end
+
   private
 
   def build_executor(document, source: SOURCE, resolvers: BREADTH_RESOLVERS)

From 2d400f28a4624fb8adb42644dd610abc3bd65d14 Mon Sep 17 00:00:00 2001
From: Greg MacWilliam <greg.macwilliam@shopify.com>
Date: Mon, 22 Jun 2026 23:59:12 -0400
Subject: [PATCH 2/4] changes to the stream model

---
 lib/graphql/breadth.rb                        |   1 +
 lib/graphql/breadth/executor.rb               | 124 ++++++++++++-----
 lib/graphql/breadth/field_resolvers.rb        |   5 +
 .../incremental/stream_execution_scope.rb     |  13 +-
 lib/graphql/breadth/stream_source.rb          |  20 +++
 .../breadth/executor/incremental_test.rb      | 126 ++++++++++++++++++
 6 files changed, 258 insertions(+), 31 deletions(-)
 create mode 100644 lib/graphql/breadth/stream_source.rb

diff --git a/lib/graphql/breadth.rb b/lib/graphql/breadth.rb
index 8fae326..5bb5459 100644
--- a/lib/graphql/breadth.rb
+++ b/lib/graphql/breadth.rb
@@ -42,6 +42,7 @@ class ExecutionField; end
 require_relative "breadth/executor/execution_promise"
 require_relative "breadth/lazy_loader"
 require_relative "breadth/tracer"
+require_relative "breadth/stream_source"
 require_relative "breadth/field_resolvers"
 require_relative "breadth/directive_resolvers"
 require_relative "breadth/subscription_response_stream"
diff --git a/lib/graphql/breadth/executor.rb b/lib/graphql/breadth/executor.rb
index 2ef07d5..f7a4942 100644
--- a/lib/graphql/breadth/executor.rb
+++ b/lib/graphql/breadth/executor.rb
@@ -619,15 +619,16 @@ def execute_field(exec_field)
           exec_field.validate!
           pre_authorized = @authorization.authorized_field?(exec_field, @context)
           pre_authorized &&= @authorization.authorized_type?(exec_field.type.unwrap, @context, exec_field: exec_field)
+          stream_usage = pre_authorized ? active_stream_usage_for(exec_field, exec_field.type) : nil
 
           # each branch must assign `exec_field.result` to make it available in the final ensure block
           if !pre_authorized
             exec_field.result = exec_field.resolve_all(FieldAuthorizationError.new(exec_field: exec_field))
           elsif exec_field.directives.empty?
-            exec_field.result = exec_field.resolver.resolve(exec_field, @context)
+            exec_field.result = resolve_field(exec_field, stream_usage)
           else
             execute_with_directives(exec_field.directives, current_field: exec_field) do
-              exec_field.result = exec_field.resolver.resolve(exec_field, @context)
+              exec_field.result = resolve_field(exec_field, stream_usage)
             end
           end
         rescue StandardError => e
@@ -649,6 +650,15 @@ def execute_field(exec_field)
         end
       end
 
+      #: (ExecutionField[untyped], Incremental::StreamUsage?) -> (Array[untyped] | ExecutionPromise)
+      def resolve_field(exec_field, stream_usage)
+        if stream_usage
+          exec_field.resolver.resolve_stream(exec_field, @context, initial_count: stream_usage.initial_count)
+        else
+          exec_field.resolver.resolve(exec_field, @context)
+        end
+      end
+
       #: (ExecutionField[untyped]) -> void
       def build_field_placeholder(exec_field)
         field_key = exec_field.key
@@ -757,16 +767,24 @@ def build_streaming_composite_result(exec_field, current_type, object, next_obje
         return build_and_flatmap_composite_result(exec_field, current_type, object, next_objects, next_results) if object.nil? || object.is_a?(StandardError)
         return build_and_flatmap_composite_result(exec_field, current_type, object, next_objects, next_results) unless Util.unwrap_non_null(current_type).list?
 
-        objects = list_items_for_stream(exec_field, object)
         initial_count = stream_usage.initial_count
         item_type = Util.unwrap_non_null(current_type).of_type
+        initial_items, remaining_items, register_stream = split_stream_items(exec_field, object, initial_count)
 
-        initial_items = objects.take(initial_count)
         initial_result = initial_items.map do |src|
           build_and_flatmap_composite_result(exec_field, item_type, src, next_objects, next_results)
         end
 
-        register_composite_stream_scope(exec_field, item_type, stream_usage, object_index, objects.drop(initial_count), initial_count) if initial_count <= objects.length
+        if register_stream
+          register_composite_stream_scope(
+            exec_field,
+            item_type,
+            stream_usage,
+            object_index,
+            remaining_items,
+            initial_items.length,
+          )
+        end
 
         initial_result
       end
@@ -782,62 +800,89 @@ def build_streaming_leaf_result(exec_field, current_type, val, stream_usage, obj
         return build_leaf_result(exec_field, current_type, val) if val.nil? || val.is_a?(StandardError)
         return build_leaf_result(exec_field, current_type, val) unless Util.unwrap_non_null(current_type).list?
 
-        values = list_items_for_stream(exec_field, val)
         initial_count = stream_usage.initial_count
         item_type = Util.unwrap_non_null(current_type).of_type
+        initial_items, remaining_items, register_stream = split_stream_items(exec_field, val, initial_count)
 
-        initial_result = values.take(initial_count).map { build_leaf_result(exec_field, item_type, _1) }
-        if initial_count <= values.length
-          stream_items = values.drop(initial_count).map { build_leaf_result(exec_field, item_type, _1) }
+        initial_result = initial_items.map { build_leaf_result(exec_field, item_type, _1) }
+        if register_stream
           register_stream_scope(
             exec_field,
             exec_field.scope.parent_type,
             EMPTY_ARRAY,
-            EMPTY_ARRAY,
-            EMPTY_ARRAY,
-            stream_items,
-            EMPTY_ARRAY,
+            [],
+            [],
+            [],
+            [],
             stream_usage,
             object_index,
-            initial_count,
+            initial_items.length,
+            prepare: ->(stream_scope) { prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items) },
           )
         end
 
         initial_result
       end
 
-      #: (ExecutionField[untyped], untyped) -> Array[untyped]
-      def list_items_for_stream(exec_field, value)
-        return value if value.is_a?(Array)
-        return value.to_a if value.is_a?(Enumerable)
+      #: (ExecutionField[untyped], untyped, Integer) -> [Array[untyped], untyped, bool]
+      def split_stream_items(exec_field, value, initial_count)
+        case value
+        when StreamSource
+          initial_items = value.initial_items
+          remaining_items = value.remaining_items
+          return [initial_items, remaining_items, !remaining_items.nil?]
+        when Array
+          return [value.take(initial_count), value.drop(initial_count), initial_count <= value.length]
+        when Enumerable
+          initial_items = []
+          iterator = value.each #: as Enumerator
+          while initial_items.length < initial_count
+            begin
+              initial_items << iterator.next
+            rescue StopIteration
+              return [initial_items, iterator, false]
+            end
+          end
+          return [initial_items, iterator, true]
+        end
 
         raise InvalidListResultError.new(exec_field:, result_type: value.class)
       end
 
+      #: (ExecutionField[untyped], untyped) { (untyped, Integer) -> void } -> void
+      def each_stream_item(exec_field, items, &block)
+        if items.is_a?(Array)
+          items.each_with_index { |item, index| yield item, index }
+        elsif items.respond_to?(:next)
+          index = 0
+          loop do
+            yield items.next, index
+            index += 1
+          end
+        elsif items.is_a?(Enumerable)
+          items.each_with_index { |item, index| yield item, index }
+        else
+          raise InvalidListResultError.new(exec_field:, result_type: items.class)
+        end
+      rescue StopIteration
+        nil
+      end
+
       #: (
       #|   ExecutionField[untyped] exec_field,
       #|   untyped item_type,
       #|   Incremental::StreamUsage stream_usage,
       #|   Integer object_index,
-      #|   Array[untyped] tail_objects,
+      #|   untyped remaining_items,
       #|   Integer initial_count,
       #| ) -> void
-      def register_composite_stream_scope(exec_field, item_type, stream_usage, object_index, tail_objects, initial_count)
+      def register_composite_stream_scope(exec_field, item_type, stream_usage, object_index, remaining_items, initial_count)
         stream_items = []
         stream_objects = []
         stream_results = []
         object_paths = []
         stream_path = exec_field.object_path(object_index)
 
-        tail_objects.each_with_index do |src, index|
-          before_results = stream_results.length
-          stream_items << build_and_flatmap_composite_result(exec_field, item_type, src, stream_objects, stream_results)
-          while before_results < stream_results.length
-            object_paths[before_results] = [*stream_path, initial_count + index]
-            before_results += 1
-          end
-        end
-
         register_stream_scope(
           exec_field,
           item_type.unwrap,
@@ -849,9 +894,26 @@ def register_composite_stream_scope(exec_field, item_type, stream_usage, object_
           stream_usage,
           object_index,
           initial_count,
+          prepare: ->(stream_scope) do
+            each_stream_item(exec_field, remaining_items) do |src, index|
+              before_results = stream_results.length
+              stream_items << build_and_flatmap_composite_result(exec_field, item_type, src, stream_objects, stream_results)
+              while before_results < stream_results.length
+                object_paths[before_results] = [*stream_path, initial_count + index]
+                before_results += 1
+              end
+            end
+          end,
         )
       end
 
+      #: (Incremental::StreamExecutionScope, ExecutionField[untyped], untyped, untyped) -> void
+      def prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items)
+        each_stream_item(exec_field, remaining_items) do |val, _index|
+          stream_scope.items << build_leaf_result(exec_field, item_type, val)
+        end
+      end
+
       #: (
       #|   ExecutionField[untyped] exec_field,
       #|   singleton(GraphQL::Schema::Object) parent_type,
@@ -863,8 +925,9 @@ def register_composite_stream_scope(exec_field, item_type, stream_usage, object_
       #|   Incremental::StreamUsage stream_usage,
       #|   Integer object_index,
       #|   Integer initial_count,
+      #|   ?prepare: Proc?,
       #| ) -> void
-      def register_stream_scope(exec_field, parent_type, selections, objects, results, items, object_paths, stream_usage, object_index, initial_count)
+      def register_stream_scope(exec_field, parent_type, selections, objects, results, items, object_paths, stream_usage, object_index, initial_count, prepare: nil)
         stream_path = exec_field.object_path(object_index)
         stream_delivery = Incremental::StreamDelivery.new(stream_path, stream_usage.label)
         @incremental.register_stream_scope(Incremental::StreamExecutionScope.new(
@@ -877,6 +940,7 @@ def register_stream_scope(exec_field, parent_type, selections, objects, results,
           object_paths:,
           delivery: stream_delivery,
           initial_index: initial_count,
+          prepare:,
         ))
       end
 
diff --git a/lib/graphql/breadth/field_resolvers.rb b/lib/graphql/breadth/field_resolvers.rb
index d32f88e..7d58ba1 100644
--- a/lib/graphql/breadth/field_resolvers.rb
+++ b/lib/graphql/breadth/field_resolvers.rb
@@ -14,6 +14,11 @@ def resolve(exec_field, ctx)
         raise NotImplementedError, "FieldResolver#resolve must be implemented."
       end
 
+      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context, initial_count: Integer) -> (Array[untyped] | Executor::ExecutionPromise)
+      def resolve_stream(exec_field, ctx, initial_count:)
+        resolve(exec_field, ctx)
+      end
+
       #: (Array[untyped] | Executor::ExecutionPromise) { (Array[untyped]) -> Array[untyped] } -> (Array[untyped] | Executor::ExecutionPromise)
       def handle_resolved(result, &block)
         if result.is_a?(Executor::ExecutionPromise)
diff --git a/lib/graphql/breadth/incremental/stream_execution_scope.rb b/lib/graphql/breadth/incremental/stream_execution_scope.rb
index dc26fe1..d79321d 100644
--- a/lib/graphql/breadth/incremental/stream_execution_scope.rb
+++ b/lib/graphql/breadth/incremental/stream_execution_scope.rb
@@ -17,6 +17,9 @@ class StreamExecutionScope < Executor::ExecutionScope
         #: bool
         attr_writer :announced
 
+        #: bool
+        attr_reader :prepared
+
         #: (
         #|   parent_field: Executor::ExecutionField[untyped],
         #|   parent_type: singleton(GraphQL::Schema::Object),
@@ -27,12 +30,15 @@ class StreamExecutionScope < Executor::ExecutionScope
         #|   object_paths: Array[error_path],
         #|   delivery: StreamDelivery,
         #|   initial_index: Integer,
+        #|   ?prepare: Proc?,
         #| ) -> void
-        def initialize(parent_field:, parent_type:, selections:, objects:, results:, items:, object_paths:, delivery:, initial_index:)
+        def initialize(parent_field:, parent_type:, selections:, objects:, results:, items:, object_paths:, delivery:, initial_index:, prepare: nil)
           @delivery = delivery
           @items = items
           @object_paths = object_paths
           @initial_index = initial_index
+          @prepare = prepare
+          @prepared = false
           @announced = false
 
           super(
@@ -59,6 +65,11 @@ def announced?
 
         #: -> StreamExecutionScope
         def prepare!
+          unless @prepared
+            @prepare&.call(self)
+            @prepared = true
+          end
+
           self
         end
 
diff --git a/lib/graphql/breadth/stream_source.rb b/lib/graphql/breadth/stream_source.rb
new file mode 100644
index 0000000..f2eeb1b
--- /dev/null
+++ b/lib/graphql/breadth/stream_source.rb
@@ -0,0 +1,20 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    class StreamSource
+      #: Array[untyped]
+      attr_reader :initial_items
+
+      #: untyped
+      attr_reader :remaining_items
+
+      #: (initial_items: Array[untyped], remaining_items: untyped) -> void
+      def initialize(initial_items:, remaining_items:)
+        @initial_items = initial_items
+        @remaining_items = remaining_items
+      end
+    end
+  end
+end
diff --git a/test/graphql/breadth/executor/incremental_test.rb b/test/graphql/breadth/executor/incremental_test.rb
index 96eea76..a18081a 100644
--- a/test/graphql/breadth/executor/incremental_test.rb
+++ b/test/graphql/breadth/executor/incremental_test.rb
@@ -30,6 +30,29 @@ def resolve(exec_field, _ctx)
     end
   end
 
+  class StreamSourceNodesResolver < GraphQL::Breadth::FieldResolver
+    attr_reader :calls
+
+    def initialize(initial_items:, remaining_items:)
+      @initial_items = initial_items
+      @remaining_items = remaining_items
+      @calls = []
+    end
+
+    def resolve(exec_field, _ctx)
+      @calls << [:resolve]
+      exec_field.resolve_all(@initial_items + @remaining_items.to_a)
+    end
+
+    def resolve_stream(exec_field, _ctx, initial_count:)
+      @calls << [:resolve_stream, initial_count]
+      exec_field.resolve_all(GraphQL::Breadth::StreamSource.new(
+        initial_items: @initial_items,
+        remaining_items: @remaining_items,
+      ))
+    end
+  end
+
   SOURCE = {
     "products" => {
       "nodes" => [{
@@ -948,6 +971,109 @@ def test_incremental_result_streams_list_field
     )
   end
 
+  def test_incremental_result_streams_enumerable_list_without_materializing_tail
+    pulled_ids = []
+    nodes = SOURCE.fetch("products").fetch("nodes")
+    source = {
+      "products" => {
+        "nodes" => Enumerator.new do |yielder|
+          nodes.each do |node|
+            pulled_ids << node.fetch("id")
+            yielder << node
+          end
+        end,
+      },
+    }
+
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream(initialCount: 1) {
+          id
+        }
+      }
+    }|, source:).incremental_result
+
+    assert_equal ["gid://shopify/Product/1"], pulled_ids
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "items" => [{ "id" => "gid://shopify/Product/2" }],
+          "id" => "0",
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+    assert_equal ["gid://shopify/Product/1", "gid://shopify/Product/2"], pulled_ids
+  end
+
+  def test_incremental_result_resolves_stream_source_for_streamed_list_field
+    pulled_remaining_ids = []
+    nodes = SOURCE.fetch("products").fetch("nodes")
+    remaining_items = Enumerator.new do |yielder|
+      node = nodes.fetch(1)
+      pulled_remaining_ids << node.fetch("id")
+      yielder << node
+    end
+    nodes_resolver = StreamSourceNodesResolver.new(
+      initial_items: [nodes.fetch(0)],
+      remaining_items:,
+    )
+    resolvers = BREADTH_RESOLVERS.merge(
+      "ProductConnection" => BREADTH_RESOLVERS.fetch("ProductConnection").merge(
+        "nodes" => nodes_resolver,
+      ),
+    )
+
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream(initialCount: 1) {
+          id
+        }
+      }
+    }|, resolvers:).incremental_result
+
+    assert_equal [[:resolve_stream, 1]], nodes_resolver.calls
+    assert_equal [], pulled_remaining_ids
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+          },
+        },
+        "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "items" => [{ "id" => "gid://shopify/Product/2" }],
+          "id" => "0",
+        }],
+        "completed" => [{ "id" => "0" }],
+        "hasNext" => false,
+      }],
+      result.subsequent_results.to_a,
+    )
+    assert_equal ["gid://shopify/Product/2"], pulled_remaining_ids
+  end
+
   def test_incremental_result_streams_with_default_initial_count
     result = build_executor(%|{
       products(first: 2) {

From acac751d5267f7019a25f799630540ede2b17984 Mon Sep 17 00:00:00 2001
From: Greg MacWilliam <greg.macwilliam@shopify.com>
Date: Tue, 23 Jun 2026 12:08:07 -0400
Subject: [PATCH 3/4] craziness.

---
 README.md                                     |  99 ++++-
 lib/graphql/breadth.rb                        |   2 +-
 lib/graphql/breadth/executor.rb               | 378 ++++++++++++++----
 .../breadth/executor/execution_field.rb       |  10 +
 .../breadth/executor/list_stream_field.rb     | 214 ++++++++++
 lib/graphql/breadth/field_resolvers.rb        |  19 +-
 lib/graphql/breadth/incremental.rb            |   2 +
 lib/graphql/breadth/incremental/context.rb    |  30 +-
 .../breadth/incremental/deferred_delivery.rb  |  27 +-
 lib/graphql/breadth/incremental/delivery.rb   |  38 ++
 .../breadth/incremental/list_stream_entry.rb  |  49 +++
 lib/graphql/breadth/incremental/publisher.rb  |   8 +-
 .../breadth/incremental/stream_delivery.rb    |  13 +-
 .../incremental/stream_execution_scope.rb     |  61 ++-
 lib/graphql/breadth/list_stream_chunk.rb      |  22 +
 lib/graphql/breadth/stream_source.rb          |  20 -
 .../breadth/executor/incremental_test.rb      | 309 +++++++++-----
 17 files changed, 1054 insertions(+), 247 deletions(-)
 create mode 100644 lib/graphql/breadth/executor/list_stream_field.rb
 create mode 100644 lib/graphql/breadth/incremental/delivery.rb
 create mode 100644 lib/graphql/breadth/incremental/list_stream_entry.rb
 create mode 100644 lib/graphql/breadth/list_stream_chunk.rb
 delete mode 100644 lib/graphql/breadth/stream_source.rb

diff --git a/README.md b/README.md
index 735b691..7711cbd 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,7 @@ The execution algorithm is proven at scale in production. This implementation st
 
 * Uses GraphQL Ruby schemas.
 * Currently no built-in validation or analysis, do it ahead of time.
-* Currently no stream.
+* Supports incremental `@defer` and `@stream` through the `incremental_result` entry point.
 * Supports input validations, but NOT input transformations (ie: "prepare" hooks).
 
 # Usage
@@ -677,9 +677,9 @@ 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`)
+## Incremental results (`@defer` and `@stream`)
 
-Query and mutation operations that may contain `@defer` should use `incremental_result`. This always returns a `GraphQL::Breadth::Incremental::Result`, even when the operation has no active deferred work:
+Query and mutation operations that may contain `@defer` or `@stream` should use `incremental_result`. This always returns a `GraphQL::Breadth::Incremental::Result`, even when the operation has no active incremental work:
 
 ```ruby
 result = executor.incremental_result
@@ -693,10 +693,101 @@ if result.incremental?
 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.
+When no incremental work is active, `initial_result` is the normal GraphQL result hash and `incremental?` is false. When incremental 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.
 
+### Streaming lists
+
+Fields that support `@stream` must opt into the list stream API. The ordinary `resolve` method remains responsible for non-incremental execution, while `resolve_list_stream` loads items for active stream installments:
+
+```ruby
+class ProductNodesResolver < GraphQL::Breadth::FieldResolver
+  def stream?
+    true
+  end
+
+  def resolve(exec_field, context)
+    exec_field.map_objects { |connection| connection.nodes }
+  end
+
+  def resolve_list_stream(objects, context, state:, object_states:, limit:, iteration:, field:)
+    objects.map.with_index do |connection, index|
+      object_state = object_states[index]
+      cursor = object_state[:cursor]
+      page_size = limit || 25
+
+      page = ProductLoader.load_page(
+        connection,
+        first: page_size,
+        after: cursor,
+        context: context,
+      )
+
+      object_state[:cursor] = page.end_cursor
+
+      GraphQL::Breadth::ListStreamChunk.new(
+        items: page.nodes,
+        complete: !page.has_next_page?,
+      )
+    end
+  end
+end
+```
+
+`resolve_list_stream` receives only the parent objects that still have active stream deliveries. Once an object returns a complete chunk, it is dropped from future calls. This lets resolvers keep batching active streams together while avoiding repeated work for streams that have already finished.
+
+The method arguments are:
+
+* `objects`: the active parent objects for this installment.
+* `state`: a shared hash for the streamed field instance across all installments.
+* `object_states`: one persistent hash per active object, mapped by index to `objects`.
+* `limit`: the directive's `initialCount` for the initial call, then `nil` for later calls.
+* `iteration`: the call count for this streamed field batch.
+* `field`: a `ListStreamField` call object with the streamed field's `arguments`, `path`, `lazy`, `await_all`, and other field helpers.
+
+Return one result per object. A `GraphQL::Breadth::ListStreamChunk` is the explicit form: `items:` are emitted in this installment and `complete:` controls whether that object's stream remains active. Returning an array is shorthand for "emit these items and keep going" unless the array is empty, in which case the object is complete. Returning `nil` completes the object without emitting items.
+
+If the directive uses `@stream(initialCount: 0)`, the initial result includes no list items and the first `resolve_list_stream` call happens while preparing the first subsequent payload.
+
+```ruby
+GraphQL::Breadth::ListStreamEntry.new(
+  object: object,
+  state: nil,
+)
+
+class ProductNodesResolver < GraphQL::Breadth::FieldResolver
+  def stream?
+    true
+  end
+
+  def resolve(exec_field, context)
+    exec_field.map_objects { |connection| connection.nodes }
+  end
+
+  def resolve_list_stream(stream_exec_field, context, limit:)
+    # StreamExecutionField < ExecutionField
+    # owns its own pending_objects (objects filtered after each iteration)
+    limit ||= 10
+
+    stream_exec_field.lazy(
+      MyStreamingListLoader,
+      keys: stream_exec_field.pending_entries,
+      args: { limit: limit },
+    ).then do |all_results|
+      stream_exec_field.pending_entries.map.with_index do |entry, index|
+        results = all_results[index]
+        entry.set_state(results.last&.id)
+        GraphQL::Breadth::ListStreamChunk.new(
+          items: results,
+          complete: results.size <= limit,
+        )
+      end
+    end
+  end
+end
+```
+
 ## Subscriptions
 
 Query and mutation execution use `result`, which always returns a normal GraphQL result hash. Subscription operations use `subscribe`, which returns a `GraphQL::Breadth::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:
diff --git a/lib/graphql/breadth.rb b/lib/graphql/breadth.rb
index 5bb5459..e26f302 100644
--- a/lib/graphql/breadth.rb
+++ b/lib/graphql/breadth.rb
@@ -42,7 +42,7 @@ class ExecutionField; end
 require_relative "breadth/executor/execution_promise"
 require_relative "breadth/lazy_loader"
 require_relative "breadth/tracer"
-require_relative "breadth/stream_source"
+require_relative "breadth/list_stream_chunk"
 require_relative "breadth/field_resolvers"
 require_relative "breadth/directive_resolvers"
 require_relative "breadth/subscription_response_stream"
diff --git a/lib/graphql/breadth/executor.rb b/lib/graphql/breadth/executor.rb
index f7a4942..e3e7c73 100644
--- a/lib/graphql/breadth/executor.rb
+++ b/lib/graphql/breadth/executor.rb
@@ -5,6 +5,7 @@
 require_relative "./executor/lazy_element"
 require_relative "./executor/execution_scope"
 require_relative "./executor/execution_field"
+require_relative "./executor/list_stream_field"
 require_relative "./executor/execution_directive"
 require_relative "./executor/abstract_execution_scope"
 require_relative "./executor/execution_planner"
@@ -464,7 +465,7 @@ def execute_lazy(lazy_elements)
         pending_loaders.each do |loader|
           loader_elements = loader.promised.map(&:element)
           all_aborted = loader_elements.all? do |element|
-            aborted_status_cache[element.is_a?(ExecutionField) ? element.scope : element]
+            aborted_status_cache[lazy_element_scope(element)]
           end
 
           if all_aborted
@@ -490,6 +491,9 @@ def execute_lazy(lazy_elements)
                 scope_error = ExecutionError.from(handled_error, exec_field: element.parent_field)
                 element.results.each { add_error(scope_error, _1, exec_field: element.parent_field) }
                 element.abort!
+              when ListStreamField
+                stream_error = ExecutionError.from(handled_error, exec_field: element.parent_field)
+                element.result = element.resolve_all(stream_error)
               end
             end
           ensure
@@ -515,8 +519,42 @@ def execute_lazy(lazy_elements)
             next if aborted_status_cache[element]
 
             resume_lazy_scope_execute(element)
+          when ListStreamField
+            next if aborted_status_cache[element.parent_field.scope]
+
+            resume_lazy_list_stream_field(element)
+          end
+        end
+      end
+
+      #: (LazyElement) -> ExecutionScope
+      def lazy_element_scope(element)
+        case element
+        when ExecutionField
+          element.scope
+        when ListStreamField
+          element.parent_field.scope
+        when ExecutionScope
+          element
+        else
+          raise ImplementationError, "Unknown lazy element"
+        end
+      end
+
+      #: (ListStreamField) -> void
+      def resume_lazy_list_stream_field(list_stream_field)
+        if list_stream_field.lazy_result?
+          begin
+            promise = list_stream_field.result
+            if promise_resolved?(promise, element: list_stream_field)
+              list_stream_field.result = promise.value
+            end
+          rescue ExecutionError => e
+            list_stream_field.result = list_stream_field.resolve_all(e)
           end
         end
+
+        list_stream_field.lazy_state_locked! unless list_stream_field.locked?
       end
 
       #: (ExecutionScope) -> void
@@ -582,6 +620,8 @@ def promise_resolved?(promise, element:)
             element
           when ExecutionScope
             element.parent_field
+          when ListStreamField
+            element.parent_field
           end
 
           raise handle_or_reraise(reason, exec_field:)
@@ -619,16 +659,15 @@ def execute_field(exec_field)
           exec_field.validate!
           pre_authorized = @authorization.authorized_field?(exec_field, @context)
           pre_authorized &&= @authorization.authorized_type?(exec_field.type.unwrap, @context, exec_field: exec_field)
-          stream_usage = pre_authorized ? active_stream_usage_for(exec_field, exec_field.type) : nil
 
           # each branch must assign `exec_field.result` to make it available in the final ensure block
           if !pre_authorized
             exec_field.result = exec_field.resolve_all(FieldAuthorizationError.new(exec_field: exec_field))
           elsif exec_field.directives.empty?
-            exec_field.result = resolve_field(exec_field, stream_usage)
+            exec_field.result = resolve_field(exec_field)
           else
             execute_with_directives(exec_field.directives, current_field: exec_field) do
-              exec_field.result = resolve_field(exec_field, stream_usage)
+              exec_field.result = resolve_field(exec_field)
             end
           end
         rescue StandardError => e
@@ -650,15 +689,104 @@ def execute_field(exec_field)
         end
       end
 
-      #: (ExecutionField[untyped], Incremental::StreamUsage?) -> (Array[untyped] | ExecutionPromise)
-      def resolve_field(exec_field, stream_usage)
-        if stream_usage
-          exec_field.resolver.resolve_stream(exec_field, @context, initial_count: stream_usage.initial_count)
+      #: (ExecutionField[untyped]) -> (Array[untyped] | ExecutionPromise)
+      def resolve_field(exec_field)
+        if (exec_field.resolver.stream? && stream_usage = exec_field.stream_usage)
+          resolve_field_stream(exec_field, stream_usage)
         else
           exec_field.resolver.resolve(exec_field, @context)
         end
       end
 
+      #: (ExecutionField[untyped], Incremental::StreamUsage) -> (Array[untyped] | ExecutionPromise)
+      def resolve_field_stream(exec_field, stream_usage)
+        parent_objects = exec_field.objects
+        state = exec_field.attributes[:list_stream_state] ||= {}
+        object_states = Array.new(parent_objects.length) { {} } #: Array[Hash[untyped, untyped]]
+        pending_entries = parent_objects.map.with_index do |object, index|
+          object_state = object_states[index] #: as !nil
+          Incremental::ListStreamEntry.new(object:, object_state:)
+        end #: Array[Incremental::ListStreamEntry]
+        stream_field = ListStreamField.new(
+          parent_field: exec_field,
+          pending_entries:,
+          state:,
+          limit: stream_usage.initial_count,
+        )
+        pending_entries.each { _1.field = stream_field }
+
+        if stream_usage.initial_count.zero?
+          chunks = Array.new(parent_objects.length) { ListStreamChunk.new(items: EMPTY_ARRAY, complete: false) }
+          return build_list_stream_sources(stream_field, chunks)
+        end
+
+        result = resolve_list_stream_chunks(stream_field, limit: stream_usage.initial_count)
+        if result.is_a?(ExecutionPromise)
+          result.then { |chunks| build_list_stream_sources(stream_field, chunks) }
+        else
+          build_list_stream_sources(stream_field, result)
+        end
+      end
+
+      #: (ListStreamField, limit: Integer?) -> (Array[untyped] | ExecutionPromise)
+      def resolve_list_stream_chunks(stream_field, limit:)
+        exec_field = stream_field.parent_field
+        stream_field.reset_for_resolve!(limit:)
+        stream_field.lazy_state_executing!
+        stream_field.result = exec_field.resolver.resolve_list_stream(
+          stream_field.objects,
+          @context,
+          state: stream_field.state,
+          object_states: stream_field.object_states,
+          limit:,
+          iteration: stream_field.iteration,
+          field: stream_field,
+        )
+        stream_field.iteration += 1
+
+        stream_field.lazy_state_locked! unless stream_field.lazy_result?
+
+        stream_field.result
+      end
+
+      #: (ListStreamField, Array[untyped]) -> Array[untyped]
+      def build_list_stream_sources(stream_field, chunks)
+        entries = stream_field.pending_entries
+        exec_field = stream_field.parent_field
+        unless chunks.length == entries.length
+          handle_or_reraise(ResultCountMismatchError.new(
+            exec_field: exec_field,
+            expected_count: entries.length,
+            actual_count: chunks.length,
+          ))
+          chunks = Array.new(entries.length)
+        end
+
+        chunks.each_with_index.map do |chunk, index|
+          next chunk if chunk.is_a?(StandardError)
+
+          items, complete = normalize_list_stream_chunk(exec_field, chunk)
+          entry = entries[index] #: as !nil
+          remaining_items = entry unless complete
+          stream_field.drop_pending_entries([entry]) if complete
+          Incremental::ListStreamSource.new(initial_items: items, remaining_items:)
+        end
+      end
+
+      #: (ExecutionField[untyped], untyped) -> [Array[untyped], bool]
+      def normalize_list_stream_chunk(exec_field, chunk)
+        case chunk
+        when ListStreamChunk
+          [chunk.items, chunk.complete?]
+        when Array
+          [chunk, chunk.empty?]
+        when nil
+          [EMPTY_ARRAY, true]
+        else
+          raise InvalidListResultError.new(exec_field:, result_type: chunk.class)
+        end
+      end
+
       #: (ExecutionField[untyped]) -> void
       def build_field_placeholder(exec_field)
         field_key = exec_field.key
@@ -686,7 +814,7 @@ def build_field_result(exec_field, resolved_objects)
         field_key = exec_field.key
         field_type = exec_field.type
         return_type = field_type.unwrap
-        stream_usage = active_stream_usage_for(exec_field, field_type)
+        stream_usage = exec_field.stream_usage
 
         if resolved_objects.length != parent_objects.length
           handle_or_reraise(ResultCountMismatchError.new(
@@ -746,14 +874,6 @@ def build_field_result(exec_field, resolved_objects)
         end
       end
 
-      #: (ExecutionField[untyped], singleton(GraphQL::Schema::Member)) -> Incremental::StreamUsage?
-      def active_stream_usage_for(exec_field, field_type)
-        return nil unless incremental?
-        return nil unless Util.unwrap_non_null(field_type).list?
-
-        @planner.stream_usage_for(exec_field)
-      end
-
       #: (
       #|   ExecutionField[untyped] exec_field,
       #|   untyped current_type,
@@ -806,6 +926,7 @@ def build_streaming_leaf_result(exec_field, current_type, val, stream_usage, obj
 
         initial_result = initial_items.map { build_leaf_result(exec_field, item_type, _1) }
         if register_stream
+          list_stream_entry = remaining_items if remaining_items.is_a?(Incremental::ListStreamEntry)
           register_stream_scope(
             exec_field,
             exec_field.scope.parent_type,
@@ -817,7 +938,12 @@ def build_streaming_leaf_result(exec_field, current_type, val, stream_usage, obj
             stream_usage,
             object_index,
             initial_items.length,
-            prepare: ->(stream_scope) { prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items) },
+            prepare: if list_stream_entry
+              ->(stream_scope, raw_items) { prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items, raw_items) }
+            else
+              ->(stream_scope) { prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items) }
+            end,
+            list_stream_entry:,
           )
         end
 
@@ -827,7 +953,7 @@ def build_streaming_leaf_result(exec_field, current_type, val, stream_usage, obj
       #: (ExecutionField[untyped], untyped, Integer) -> [Array[untyped], untyped, bool]
       def split_stream_items(exec_field, value, initial_count)
         case value
-        when StreamSource
+        when Incremental::ListStreamSource
           initial_items = value.initial_items
           remaining_items = value.remaining_items
           return [initial_items, remaining_items, !remaining_items.nil?]
@@ -882,6 +1008,7 @@ def register_composite_stream_scope(exec_field, item_type, stream_usage, object_
         stream_results = []
         object_paths = []
         stream_path = exec_field.object_path(object_index)
+        list_stream_entry = remaining_items if remaining_items.is_a?(Incremental::ListStreamEntry)
 
         register_stream_scope(
           exec_field,
@@ -894,23 +1021,42 @@ def register_composite_stream_scope(exec_field, item_type, stream_usage, object_
           stream_usage,
           object_index,
           initial_count,
-          prepare: ->(stream_scope) do
-            each_stream_item(exec_field, remaining_items) do |src, index|
-              before_results = stream_results.length
-              stream_items << build_and_flatmap_composite_result(exec_field, item_type, src, stream_objects, stream_results)
-              while before_results < stream_results.length
-                object_paths[before_results] = [*stream_path, initial_count + index]
-                before_results += 1
+          prepare: if list_stream_entry
+            ->(stream_scope, raw_items) do
+              raw_items.each_with_index do |src, index|
+                before_results = stream_scope.results.length
+                stream_scope.items << build_and_flatmap_composite_result(exec_field, item_type, src, stream_scope.objects, stream_scope.results)
+                while before_results < stream_scope.results.length
+                  stream_scope.object_paths[before_results] = [*stream_path, stream_scope.initial_index + index]
+                  before_results += 1
+                end
+              end
+            end
+          else
+            ->(stream_scope) do
+              each_stream_item(exec_field, remaining_items) do |src, index|
+                before_results = stream_scope.results.length
+                stream_scope.items << build_and_flatmap_composite_result(exec_field, item_type, src, stream_scope.objects, stream_scope.results)
+                while before_results < stream_scope.results.length
+                  stream_scope.object_paths[before_results] = [*stream_path, initial_count + index]
+                  before_results += 1
+                end
               end
             end
           end,
+          list_stream_entry:,
         )
       end
 
-      #: (Incremental::StreamExecutionScope, ExecutionField[untyped], untyped, untyped) -> void
-      def prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items)
-        each_stream_item(exec_field, remaining_items) do |val, _index|
-          stream_scope.items << build_leaf_result(exec_field, item_type, val)
+      #: (Incremental::StreamExecutionScope, ExecutionField[untyped], untyped, untyped, ?Array[untyped]?) -> void
+      def prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_items, raw_items = nil)
+        items = raw_items || remaining_items
+        if raw_items
+          items.each { |val| stream_scope.items << build_leaf_result(exec_field, item_type, val) }
+        else
+          each_stream_item(exec_field, items) do |val, _index|
+            stream_scope.items << build_leaf_result(exec_field, item_type, val)
+          end
         end
       end
 
@@ -926,11 +1072,12 @@ def prepare_leaf_stream_scope(stream_scope, exec_field, item_type, remaining_ite
       #|   Integer object_index,
       #|   Integer initial_count,
       #|   ?prepare: Proc?,
+      #|   ?list_stream_entry: Incremental::ListStreamEntry?,
       #| ) -> void
-      def register_stream_scope(exec_field, parent_type, selections, objects, results, items, object_paths, stream_usage, object_index, initial_count, prepare: nil)
+      def register_stream_scope(exec_field, parent_type, selections, objects, results, items, object_paths, stream_usage, object_index, initial_count, prepare: nil, list_stream_entry: nil)
         stream_path = exec_field.object_path(object_index)
         stream_delivery = Incremental::StreamDelivery.new(stream_path, stream_usage.label)
-        @incremental.register_stream_scope(Incremental::StreamExecutionScope.new(
+        stream_scope = Incremental::StreamExecutionScope.new(
           parent_field: exec_field,
           parent_type:,
           selections:,
@@ -941,7 +1088,10 @@ def register_stream_scope(exec_field, parent_type, selections, objects, results,
           delivery: stream_delivery,
           initial_index: initial_count,
           prepare:,
-        ))
+          list_stream_entry:,
+        )
+        list_stream_entry.scope = stream_scope if list_stream_entry
+        @incremental.register_stream_scope(stream_scope)
       end
 
       #: (
@@ -1204,54 +1354,140 @@ def execute_subscription
 
       #: (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
-          scopes_to_execute = ready_scopes.reject { _1.is_a?(Incremental::StreamExecutionScope) && _1.objects.empty? }
-          run!(@planner.plan_scopes(scopes_to_execute)) unless scopes_to_execute.empty?
-          has_errors = @invalidated_results.size > initial_error_count
-
-          ready_scopes.each do |exec_scope|
-            if exec_scope.is_a?(Incremental::StreamExecutionScope)
-              items, stream_errors = stream_items_for(exec_scope)
-              incremental_payloads << @incremental.stream_payload(exec_scope.delivery, items, errors: stream_errors) unless items.empty?
-              completed_deliveries << exec_scope.delivery
-            else
-              deliveries = @incremental.deliveries_for(exec_scope)
-              deliveries.each do |index, path, deferred_deliveries|
-                data = exec_scope.results[index]
-                deferred_errors = EMPTY_ARRAY
-                if has_errors
-                  data, deferred_errors = error_result_formatter.format_object(exec_scope.parent_type, exec_scope.selections, data, path)
-                end
+          pending_payloads = [] #: Array[graphql_result]
+          incremental_payloads = [] #: Array[graphql_result]
+          completed_deliveries = [] #: Array[Incremental::Delivery]
+          completed_errors_by_delivery = {}.compare_by_identity #: Hash[Incremental::Delivery, Array[error_hash]]
+          next_installment_streams = [] #: Array[Incremental::StreamExecutionScope]
 
-                if data.nil? && !deferred_errors.empty?
-                  deferred_deliveries.each { (completed_errors_by_delivery[_1] ||= []).concat(deferred_errors) }
+          loop do
+            ready_scopes = @incremental.ready_scopes
+            break if ready_scopes.empty?
+
+            prepare_ready_incremental_scopes(ready_scopes)
+
+            initial_error_count = @invalidated_results.size
+            scopes_to_execute = ready_scopes.reject { _1.is_a?(Incremental::StreamExecutionScope) && _1.objects.empty? }
+            run!(@planner.plan_scopes(scopes_to_execute)) unless scopes_to_execute.empty?
+            has_errors = @invalidated_results.size > initial_error_count
+
+            ready_scopes.each do |exec_scope|
+              if exec_scope.is_a?(Incremental::StreamExecutionScope)
+                items, stream_errors = stream_items_for(exec_scope)
+                incremental_payloads << @incremental.stream_payload(exec_scope.delivery, items, errors: stream_errors) unless items.empty?
+                if exec_scope.complete?
+                  completed_deliveries << exec_scope.delivery
                 else
-                  incremental_payloads << @incremental.incremental_payload(deferred_deliveries, path, data, errors: deferred_errors)
+                  next_installment_streams << exec_scope
+                end
+              else
+                deliveries = @incremental.deliveries_for(exec_scope)
+                deliveries.each do |index, path, deferred_deliveries|
+                  data = exec_scope.results[index]
+                  deferred_errors = EMPTY_ARRAY
+                  if has_errors
+                    data, deferred_errors = error_result_formatter.format_object(exec_scope.parent_type, exec_scope.selections, data, path)
+                  end
+
+                  if data.nil? && !deferred_errors.empty?
+                    deferred_deliveries.each { (completed_errors_by_delivery[_1] ||= []).concat(deferred_errors) }
+                  else
+                    incremental_payloads << @incremental.incremental_payload(deferred_deliveries, path, data, errors: deferred_errors)
+                  end
+                  completed_deliveries.concat(deferred_deliveries)
                 end
-                completed_deliveries.concat(deferred_deliveries)
               end
+
+              exec_scope.executed = true
+              pending_payloads.concat(@incremental.pending_payloads(@incremental.prepare_pending))
             end
+          end
+
+          next_installment_streams.each(&:finish_installment!)
+          completed_payloads = @incremental.completed_payloads(completed_deliveries, errors_by_delivery: completed_errors_by_delivery)
+          has_next = @incremental.deferred?
+          payload = { "hasNext" => has_next }
+          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
+          break unless has_next
+        end
+      end
+
+      #: (Array[Incremental::DeferredExecutionScope | Incremental::StreamExecutionScope]) -> void
+      def prepare_ready_incremental_scopes(ready_scopes)
+        list_stream_entries_by_field = Hash.new { |h, stream_field| h[stream_field] = [] }.compare_by_identity #: Hash[ListStreamField, Array[Incremental::ListStreamEntry]]
+
+        ready_scopes.each do |exec_scope|
+          if exec_scope.is_a?(Incremental::StreamExecutionScope) && exec_scope.stream?
+            list_stream_entry = exec_scope.list_stream_entry #: as !nil
+            entries = list_stream_entries_by_field[list_stream_entry.field] #: as !nil
+            entries << list_stream_entry
+          else
+            exec_scope.prepare!
+          end
+        end
+
+        prepare_list_stream_fields(list_stream_entries_by_field) unless list_stream_entries_by_field.empty?
+      end
+
+      #: (Hash[ListStreamField, Array[Incremental::ListStreamEntry]]) -> void
+      def prepare_list_stream_fields(entries_by_field)
+        stream_fields = [] #: Array[ListStreamField]
+
+        entries_by_field.each do |stream_field, entries|
+          stream_field.retain_pending_entries(entries)
+          next if stream_field.pending_entries.empty?
+
+          resolve_list_stream_chunks(stream_field, limit: nil)
+          stream_fields << stream_field
+        end
+
+        lazy_stream_fields = stream_fields.select(&:lazy_result?)
+        execute_lazy(lazy_stream_fields) unless lazy_stream_fields.empty?
+
+        stream_fields.each do |stream_field|
+          prepare_list_stream_field(stream_field)
+        end
+      end
+
+      #: (ListStreamField) -> void
+      def prepare_list_stream_field(stream_field)
+        chunks = stream_field.result #: as Array[untyped]
+        entries = stream_field.pending_entries
+
+        unless chunks.length == entries.length
+          exec_field = stream_field.parent_field
+          handle_or_reraise(ResultCountMismatchError.new(
+            exec_field: exec_field,
+            expected_count: entries.length,
+            actual_count: chunks.length,
+          ))
+          chunks = Array.new(entries.length)
+        end
+
+        completed_entries = [] #: Array[Incremental::ListStreamEntry]
+        entries.each_with_index do |entry, index|
+          stream_scope = entry.scope #: as !nil
+          chunk = chunks[index]
+          if chunk.is_a?(StandardError)
+            stream_scope.prepare_list_stream_items!([chunk])
+            stream_scope.complete!
+            completed_entries << entry
+            next
+          end
 
-            exec_scope.executed = true
-            pending_payloads.concat(@incremental.pending_payloads(@incremental.prepare_pending))
+          items, complete = normalize_list_stream_chunk(stream_field.parent_field, chunk)
+          stream_scope.prepare_list_stream_items!(items)
+          if complete
+            stream_scope.complete!
+            completed_entries << entry
           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
+        stream_field.drop_pending_entries(completed_entries) unless completed_entries.empty?
       end
 
       #: (Incremental::StreamExecutionScope) -> [Array[untyped], Array[error_hash]]
diff --git a/lib/graphql/breadth/executor/execution_field.rb b/lib/graphql/breadth/executor/execution_field.rb
index 5455e8b..5b521cf 100644
--- a/lib/graphql/breadth/executor/execution_field.rb
+++ b/lib/graphql/breadth/executor/execution_field.rb
@@ -71,6 +71,7 @@ def initialize(key, nodes:, scope:, definition:, resolver:, directives: EMPTY_AR
           @incremental_selections = incremental_selections
           @path = nil
           @schema_path = nil
+          @stream_usage = UNDEFINED
         end
 
         #: () -> Array[ObjectType]
@@ -203,6 +204,15 @@ def mutable_arguments
           @mutable_arguments ||= Util.deep_copy(arguments)
         end
 
+        #: () -> Incremental::StreamUsage?
+        def stream_usage
+          if @stream_usage.equal?(UNDEFINED)
+            @stream_usage = executor.incremental? && Util.unwrap_non_null(type).list? ? executor.planner.stream_usage_for(self) : nil
+          end
+
+          @stream_usage
+        end
+
         #: () -> void
         def validate!
           unless @argument_errors.empty?
diff --git a/lib/graphql/breadth/executor/list_stream_field.rb b/lib/graphql/breadth/executor/list_stream_field.rb
new file mode 100644
index 0000000..1a72453
--- /dev/null
+++ b/lib/graphql/breadth/executor/list_stream_field.rb
@@ -0,0 +1,214 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    class Executor
+      class ListStreamField
+        include LazyElement
+        include HasAttributes
+
+        LAZY_STATE_EXECUTING = :executing
+
+        #: ExecutionField[untyped]
+        attr_reader :parent_field
+
+        #: Array[Incremental::ListStreamEntry]
+        attr_reader :pending_entries
+
+        #: Hash[untyped, untyped]
+        attr_reader :state
+
+        #: Integer?
+        attr_reader :limit
+
+        #: Integer
+        attr_accessor :iteration
+
+        #: untyped
+        attr_accessor :result
+
+        #: (
+        #|   parent_field: ExecutionField[untyped],
+        #|   pending_entries: Array[Incremental::ListStreamEntry],
+        #|   state: Hash[untyped, untyped],
+        #|   limit: Integer?,
+        #| ) -> void
+        def initialize(parent_field:, pending_entries:, state:, limit:)
+          super()
+          @parent_field = parent_field
+          @pending_entries = pending_entries
+          @state = state
+          @limit = limit
+          @iteration = 0
+          @result = nil
+        end
+
+        #: -> Array[untyped]
+        def objects
+          @pending_entries.map(&:object)
+        end
+
+        #: -> Array[Hash[untyped, untyped]]
+        def object_states
+          @pending_entries.map(&:object_state)
+        end
+
+        #: -> Executor
+        def executor
+          @parent_field.executor
+        end
+
+        #: -> GraphQL::Query::Context
+        def context
+          @parent_field.context
+        end
+
+        #: -> FieldResolver
+        def resolver
+          @parent_field.resolver
+        end
+
+        #: -> graphql_arguments
+        def arguments
+          @parent_field.arguments
+        end
+
+        #: -> graphql_arguments
+        def mutable_arguments
+          @parent_field.mutable_arguments
+        end
+
+        #: -> singleton(GraphQL::Schema::Member)
+        def type
+          @parent_field.type
+        end
+
+        #: -> Array[GraphQL::Language::Nodes::Field]
+        def nodes
+          @parent_field.nodes
+        end
+
+        #: -> Array[selection_node]
+        def selections
+          @parent_field.selections
+        end
+
+        #: -> String
+        def key
+          @parent_field.key
+        end
+
+        #: -> String
+        def name
+          @parent_field.name
+        end
+
+        #: -> Array[String]
+        def path
+          @parent_field.path
+        end
+
+        #: (
+        #|   loader_class: singleton(LazyLoader),
+        #|   keys: Array[untyped],
+        #|   ?args: loader_args?,
+        #|   ?eager_values: Hash[untyped, untyped]?,
+        #|   ?load_nil_keys: bool,
+        #| ) -> ExecutionPromise
+        def lazy(loader_class:, keys:, args: nil, eager_values: nil, load_nil_keys: false)
+          unless allows_lazy?
+            raise LazySequencingError.new(lazy_element: self, method_name: "lazy")
+          end
+
+          executor.lazy_loader_for(loader_class, args).load(
+            element: self,
+            keys: keys,
+            eager_values: eager_values,
+            load_nil_keys: load_nil_keys,
+          )
+        end
+
+        #: (Array[ExecutionPromise]) -> ExecutionPromise
+        def await_all(promises)
+          super
+        end
+
+        #: -> bool
+        def allows_lazy?
+          @lazy_state == LAZY_STATE_EXECUTING
+        end
+
+        #: [T] (T) -> Array[T]
+        def resolve_all(value)
+          value = case value
+          when StandardError
+            handle_or_reraise(value)
+          else
+            value
+          end
+          Array.new(objects.length, value)
+        end
+
+        #: (limit: Integer?) -> void
+        def reset_for_resolve!(limit:)
+          @limit = limit
+          @result = nil
+          @sync_preloads = nil
+          @lazy_preloads = nil
+          @preload_promises = nil
+          @lazy_state = LAZY_STATE_PRELOADING
+        end
+
+        #: (Array[Incremental::ListStreamEntry]) -> void
+        def drop_pending_entries(entries)
+          @pending_entries -= entries
+        end
+
+        #: (Array[Incremental::ListStreamEntry]) -> void
+        def retain_pending_entries(entries)
+          @pending_entries &= entries
+        end
+
+        #: (Exception) -> ExecutionError
+        def handle_or_reraise(error)
+          executor.handle_or_reraise(error, exec_field: @parent_field)
+        end
+
+        #: () -> bool
+        def lazy_result?
+          @result.is_a?(ExecutionPromise)
+        end
+
+        #: () -> bool
+        def has_result?
+          !@result.nil?
+        end
+
+        #: () -> bool
+        def locked?
+          @lazy_state == LAZY_STATE_LOCKED
+        end
+
+        #: () -> String
+        def inspect
+          "#<ListStreamField: #{@parent_field.inspect}>"
+        end
+
+        #: () -> void
+        def lazy_state_executing!
+          raise LazyStateTransitionError.new(@lazy_state, LAZY_STATE_EXECUTING) unless @lazy_state == LAZY_STATE_PRELOADING
+
+          @lazy_state = LAZY_STATE_EXECUTING
+        end
+
+        private
+
+        #: () -> bool
+        def lazy_state_lockable?
+          @lazy_state == LAZY_STATE_PRELOADING || @lazy_state == LAZY_STATE_EXECUTING
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth/field_resolvers.rb b/lib/graphql/breadth/field_resolvers.rb
index 7d58ba1..941fb42 100644
--- a/lib/graphql/breadth/field_resolvers.rb
+++ b/lib/graphql/breadth/field_resolvers.rb
@@ -14,9 +14,22 @@ def resolve(exec_field, ctx)
         raise NotImplementedError, "FieldResolver#resolve must be implemented."
       end
 
-      #: (Executor::ExecutionField[untyped], GraphQL::Query::Context, initial_count: Integer) -> (Array[untyped] | Executor::ExecutionPromise)
-      def resolve_stream(exec_field, ctx, initial_count:)
-        resolve(exec_field, ctx)
+      #: -> bool
+      def stream?
+        false
+      end
+
+      #: (
+      #|   Array[untyped],
+      #|   GraphQL::Query::Context,
+      #|   state: Hash[untyped, untyped],
+      #|   object_states: Array[Hash[untyped, untyped]],
+      #|   limit: Integer?,
+      #|   iteration: Integer,
+      #|   field: Executor::ListStreamField,
+      #| ) -> (Array[untyped] | Executor::ExecutionPromise)
+      def resolve_list_stream(_objects, _ctx, state:, object_states:, limit:, iteration:, field:)
+        raise NotImplementedError, "FieldResolver#resolve_list_stream must be implemented."
       end
 
       #: (Array[untyped] | Executor::ExecutionPromise) { (Array[untyped]) -> Array[untyped] } -> (Array[untyped] | Executor::ExecutionPromise)
diff --git a/lib/graphql/breadth/incremental.rb b/lib/graphql/breadth/incremental.rb
index ca40e1a..b05aa0d 100644
--- a/lib/graphql/breadth/incremental.rb
+++ b/lib/graphql/breadth/incremental.rb
@@ -2,10 +2,12 @@
 # frozen_string_literal: true
 
 require_relative "incremental/defer_usage"
+require_relative "incremental/delivery"
 require_relative "incremental/deferred_delivery"
 require_relative "incremental/deferred_execution_scope"
 require_relative "incremental/stream_usage"
 require_relative "incremental/stream_delivery"
+require_relative "incremental/list_stream_entry"
 require_relative "incremental/stream_execution_scope"
 require_relative "incremental/partitioner"
 require_relative "incremental/selection"
diff --git a/lib/graphql/breadth/incremental/context.rb b/lib/graphql/breadth/incremental/context.rb
index 5f3fd9e..4aed528 100644
--- a/lib/graphql/breadth/incremental/context.rb
+++ b/lib/graphql/breadth/incremental/context.rb
@@ -19,12 +19,12 @@ def initialize(executor, data:)
           @executor = executor
           @data = data
           @publisher = Publisher.new
-          @deferred_scopes = []
-          @stream_scopes = []
-          @pending_deliveries = []
-          @announced_deliveries = {}.compare_by_identity
-          @completed_deliveries = {}.compare_by_identity
-          @deliveries_by_usage = {}.compare_by_identity
+          @deferred_scopes = [] #: Array[DeferredExecutionScope]
+          @stream_scopes = [] #: Array[StreamExecutionScope]
+          @pending_deliveries = [] #: Array[Delivery]
+          @announced_deliveries = {}.compare_by_identity #: Hash[Delivery, bool]
+          @completed_deliveries = {}.compare_by_identity #: Hash[Delivery, bool]
+          @deliveries_by_usage = {}.compare_by_identity #: Hash[DeferUsage, Array[DeferredDelivery]]
         end
 
         #: (Executor::ExecutionScope, Hash[String, Array[Selection]], Set[DeferUsage]) -> void
@@ -48,10 +48,11 @@ def active?
 
         #: -> bool
         def deferred?
-          @deferred_scopes.any? || @stream_scopes.any?
+          @deferred_scopes.any? { !_1.executed? && _1.ready? } ||
+            @stream_scopes.any? { !_1.complete? && _1.ready? && !deferred_path_nulled?(_1.delivery.path) }
         end
 
-        #: -> Array[DeferredDelivery | StreamDelivery]
+        #: -> Array[Delivery]
         def prepare_pending
           @deferred_scopes.each do |deferred_scope|
             next if deferred_scope.announced? || !deferred_scope.ready?
@@ -60,7 +61,7 @@ def prepare_pending
             deferred_scope.announced = true
           end
           @stream_scopes.each do |stream_scope|
-            next if stream_scope.announced? || !stream_scope.ready? || deferred_path_nulled?(stream_scope.delivery.path)
+            next if stream_scope.complete? || stream_scope.announced? || !stream_scope.ready? || deferred_path_nulled?(stream_scope.delivery.path)
 
             delivery = stream_scope.delivery
             unless @completed_deliveries[delivery] || @announced_deliveries[delivery]
@@ -78,10 +79,11 @@ def prepare_pending
 
         #: -> Array[DeferredExecutionScope | StreamExecutionScope]
         def ready_scopes
-          (
+          ready = (
             @deferred_scopes.select { _1.announced? && !_1.executed? && _1.ready? } +
-            @stream_scopes.select { _1.announced? && !_1.executed? && _1.ready? && !deferred_path_nulled?(_1.delivery.path) }
-          ).each(&:prepare!)
+            @stream_scopes.select { _1.announced? && !_1.executed? && !_1.complete? && _1.ready? && !deferred_path_nulled?(_1.delivery.path) }
+          )
+          ready
         end
 
         #: (DeferredExecutionScope) -> Array[[Integer, error_path, Array[DeferredDelivery]]]
@@ -99,7 +101,7 @@ def deliveries_for(deferred_scope)
           deliveries
         end
 
-        #: (Array[DeferredDelivery | StreamDelivery]) -> Array[graphql_result]
+        #: (Array[Delivery]) -> Array[graphql_result]
         def pending_payloads(deliveries)
           @publisher.pending(deliveries)
         end
@@ -114,7 +116,7 @@ def stream_payload(delivery, items, errors: EMPTY_ARRAY)
           @publisher.stream(delivery, items, errors:)
         end
 
-        #: (Array[DeferredDelivery | StreamDelivery], ?errors_by_delivery: Hash[DeferredDelivery | StreamDelivery, Array[error_hash]]) -> Array[graphql_result]
+        #: (Array[Delivery], ?errors_by_delivery: Hash[Delivery, 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]
diff --git a/lib/graphql/breadth/incremental/deferred_delivery.rb b/lib/graphql/breadth/incremental/deferred_delivery.rb
index c0cb49f..7680e87 100644
--- a/lib/graphql/breadth/incremental/deferred_delivery.rb
+++ b/lib/graphql/breadth/incremental/deferred_delivery.rb
@@ -4,38 +4,15 @@
 module GraphQL
   module Breadth
     module Incremental
-      class DeferredDelivery
-        #: error_path
-        attr_reader :path
-
-        #: String?
-        attr_reader :label
-
+      class DeferredDelivery < Delivery
         #: DeferredDelivery?
         attr_reader :parent
 
         #: (error_path, ?String?, ?parent: DeferredDelivery?) -> void
         def initialize(path, label = nil, parent: nil)
-          @path = path.freeze
-          @label = label
+          super(path, 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
diff --git a/lib/graphql/breadth/incremental/delivery.rb b/lib/graphql/breadth/incremental/delivery.rb
new file mode 100644
index 0000000..0a76953
--- /dev/null
+++ b/lib/graphql/breadth/incremental/delivery.rb
@@ -0,0 +1,38 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    module Incremental
+      class Delivery
+        #: error_path
+        attr_reader :path
+
+        #: String?
+        attr_reader :label
+
+        #: (error_path, ?String?) -> void
+        def initialize(path, label = nil)
+          @path = path.freeze
+          @label = label
+        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/incremental/list_stream_entry.rb b/lib/graphql/breadth/incremental/list_stream_entry.rb
new file mode 100644
index 0000000..0ddb7b1
--- /dev/null
+++ b/lib/graphql/breadth/incremental/list_stream_entry.rb
@@ -0,0 +1,49 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    module Incremental
+      class ListStreamEntry
+        #: untyped
+        attr_reader :object
+
+        #: Hash[untyped, untyped]
+        attr_reader :object_state
+
+        #: StreamExecutionScope?
+        attr_accessor :scope
+
+        #: Executor::ListStreamField
+        attr_writer :field
+
+        #: (object: untyped, object_state: Hash[untyped, untyped]) -> void
+        def initialize(object:, object_state:)
+          @object = object
+          @object_state = object_state
+          @scope = nil
+          @field = nil #: Executor::ListStreamField?
+        end
+
+        #: -> Executor::ListStreamField
+        def field
+          @field || raise(ImplementationError, "List stream entry has no field")
+        end
+      end
+
+      class ListStreamSource
+        #: Array[untyped]
+        attr_reader :initial_items
+
+        #: untyped
+        attr_reader :remaining_items
+
+        #: (initial_items: Array[untyped], remaining_items: untyped) -> void
+        def initialize(initial_items:, remaining_items:)
+          @initial_items = initial_items
+          @remaining_items = remaining_items
+        end
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth/incremental/publisher.rb b/lib/graphql/breadth/incremental/publisher.rb
index 39faa3b..358a1ce 100644
--- a/lib/graphql/breadth/incremental/publisher.rb
+++ b/lib/graphql/breadth/incremental/publisher.rb
@@ -6,11 +6,11 @@ module Breadth
     module Incremental
       class Publisher
         def initialize
-          @ids = {}.compare_by_identity
+          @ids = {}.compare_by_identity #: Hash[Delivery, String]
           @next_id = 0
         end
 
-        #: (Array[DeferredDelivery | StreamDelivery]) -> Array[graphql_result]
+        #: (Array[Delivery]) -> Array[graphql_result]
         def pending(deliveries)
           deliveries.map do |delivery|
             result = {
@@ -46,7 +46,7 @@ def stream(delivery, items, errors: EMPTY_ARRAY)
           result
         end
 
-        #: (DeferredDelivery | StreamDelivery, ?errors: Array[error_hash]) -> graphql_result
+        #: (Delivery, ?errors: Array[error_hash]) -> graphql_result
         def completed(delivery, errors: EMPTY_ARRAY)
           result = { "id" => id_for(delivery) }
           result["errors"] = errors unless errors.empty?
@@ -56,7 +56,7 @@ def completed(delivery, errors: EMPTY_ARRAY)
 
         private
 
-        #: (DeferredDelivery | StreamDelivery) -> String
+        #: (Delivery) -> String
         def id_for(delivery)
           @ids[delivery] ||= begin
             id = @next_id.to_s
diff --git a/lib/graphql/breadth/incremental/stream_delivery.rb b/lib/graphql/breadth/incremental/stream_delivery.rb
index 5c81581..7799f7b 100644
--- a/lib/graphql/breadth/incremental/stream_delivery.rb
+++ b/lib/graphql/breadth/incremental/stream_delivery.rb
@@ -4,18 +4,7 @@
 module GraphQL
   module Breadth
     module Incremental
-      class StreamDelivery
-        #: error_path
-        attr_reader :path
-
-        #: String?
-        attr_reader :label
-
-        #: (error_path, ?String?) -> void
-        def initialize(path, label = nil)
-          @path = path.freeze
-          @label = label
-        end
+      class StreamDelivery < Delivery
       end
     end
   end
diff --git a/lib/graphql/breadth/incremental/stream_execution_scope.rb b/lib/graphql/breadth/incremental/stream_execution_scope.rb
index d79321d..a5dd014 100644
--- a/lib/graphql/breadth/incremental/stream_execution_scope.rb
+++ b/lib/graphql/breadth/incremental/stream_execution_scope.rb
@@ -11,6 +11,9 @@ class StreamExecutionScope < Executor::ExecutionScope
         #: Array[untyped]
         attr_reader :items
 
+        #: Array[error_path]
+        attr_reader :object_paths
+
         #: Integer
         attr_reader :initial_index
 
@@ -20,6 +23,9 @@ class StreamExecutionScope < Executor::ExecutionScope
         #: bool
         attr_reader :prepared
 
+        #: ListStreamEntry?
+        attr_reader :list_stream_entry
+
         #: (
         #|   parent_field: Executor::ExecutionField[untyped],
         #|   parent_type: singleton(GraphQL::Schema::Object),
@@ -31,14 +37,18 @@ class StreamExecutionScope < Executor::ExecutionScope
         #|   delivery: StreamDelivery,
         #|   initial_index: Integer,
         #|   ?prepare: Proc?,
+        #|   ?list_stream_entry: ListStreamEntry?,
         #| ) -> void
-        def initialize(parent_field:, parent_type:, selections:, objects:, results:, items:, object_paths:, delivery:, initial_index:, prepare: nil)
+        def initialize(parent_field:, parent_type:, selections:, objects:, results:, items:, object_paths:, delivery:, initial_index:, prepare: nil, list_stream_entry: nil)
           @delivery = delivery
           @items = items
           @object_paths = object_paths
           @initial_index = initial_index
+          @next_index = initial_index
           @prepare = prepare
+          @list_stream_entry = list_stream_entry
           @prepared = false
+          @complete = false
           @announced = false
 
           super(
@@ -55,7 +65,7 @@ def initialize(parent_field:, parent_type:, selections:, objects:, results:, ite
         #: -> bool
         def ready?
           field = parent_field #: as !nil
-          field.scope.executed? && !field.scope.aborted?
+          !complete? && field.scope.executed? && !field.scope.aborted?
         end
 
         #: -> bool
@@ -63,16 +73,47 @@ def announced?
           @announced
         end
 
+        #: -> bool
+        def stream?
+          !!@list_stream_entry
+        end
+
+        #: -> bool
+        def complete?
+          @complete
+        end
+
+        #: -> void
+        def complete!
+          @complete = true
+        end
+
         #: -> StreamExecutionScope
         def prepare!
           unless @prepared
             @prepare&.call(self)
             @prepared = true
+            @complete = true unless stream?
           end
 
           self
         end
 
+        #: (Array[untyped]) -> StreamExecutionScope
+        def prepare_list_stream_items!(raw_items)
+          reset_installment!
+          @prepare&.call(self, raw_items)
+          @prepared = true
+          self
+        end
+
+        #: -> void
+        def finish_installment!
+          @next_index = @initial_index + @items.length
+          @prepared = false
+          self.executed = false unless complete?
+        end
+
         #: (Integer) -> error_path
         def item_path(index)
           [*@delivery.path, @initial_index + index]
@@ -87,6 +128,22 @@ def stream_item_path(index)
         def object_path(index)
           stream_item_path(index)
         end
+
+        private
+
+        #: -> void
+        def reset_installment!
+          @initial_index = @next_index
+          @items = []
+          @objects = []
+          @results = []
+          @object_paths = []
+          @fields = {}
+          @sync_preloads = nil
+          @lazy_preloads = nil
+          @preload_promises = nil
+          @lazy_state = LAZY_STATE_PRELOADING
+        end
       end
     end
   end
diff --git a/lib/graphql/breadth/list_stream_chunk.rb b/lib/graphql/breadth/list_stream_chunk.rb
new file mode 100644
index 0000000..e342fd8
--- /dev/null
+++ b/lib/graphql/breadth/list_stream_chunk.rb
@@ -0,0 +1,22 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    class ListStreamChunk
+      #: Array[untyped]
+      attr_reader :items
+
+      #: (items: Array[untyped], complete: bool) -> void
+      def initialize(items:, complete:)
+        @items = items
+        @complete = complete
+      end
+
+      #: -> bool
+      def complete?
+        @complete
+      end
+    end
+  end
+end
diff --git a/lib/graphql/breadth/stream_source.rb b/lib/graphql/breadth/stream_source.rb
deleted file mode 100644
index f2eeb1b..0000000
--- a/lib/graphql/breadth/stream_source.rb
+++ /dev/null
@@ -1,20 +0,0 @@
-# typed: true
-# frozen_string_literal: true
-
-module GraphQL
-  module Breadth
-    class StreamSource
-      #: Array[untyped]
-      attr_reader :initial_items
-
-      #: untyped
-      attr_reader :remaining_items
-
-      #: (initial_items: Array[untyped], remaining_items: untyped) -> void
-      def initialize(initial_items:, remaining_items:)
-        @initial_items = initial_items
-        @remaining_items = remaining_items
-      end
-    end
-  end
-end
diff --git a/test/graphql/breadth/executor/incremental_test.rb b/test/graphql/breadth/executor/incremental_test.rb
index a18081a..85661a2 100644
--- a/test/graphql/breadth/executor/incremental_test.rb
+++ b/test/graphql/breadth/executor/incremental_test.rb
@@ -30,26 +30,104 @@ def resolve(exec_field, _ctx)
     end
   end
 
-  class StreamSourceNodesResolver < GraphQL::Breadth::FieldResolver
+  class PagedProductNodesResolver < GraphQL::Breadth::FieldResolver
     attr_reader :calls
 
-    def initialize(initial_items:, remaining_items:)
-      @initial_items = initial_items
-      @remaining_items = remaining_items
+    def initialize(page_size: nil)
+      @page_size = page_size
       @calls = []
     end
 
+    def stream?
+      true
+    end
+
+    def resolve(exec_field, _ctx)
+      exec_field.map_objects { _1["nodes"] }
+    end
+
+    def resolve_list_stream(objects, _ctx, state:, object_states:, limit:, iteration:, field:)
+      state[:calls] = @calls
+      @calls << {
+        limit:,
+        iteration:,
+        object_count: objects.length,
+        arguments: field.arguments,
+      }
+
+      objects.map.with_index do |object, index|
+        nodes = object.fetch("nodes")
+        object_state = object_states[index]
+        offset = object_state[:offset] || 0
+        page_size = limit || @page_size || nodes.length
+        items = nodes.slice(offset, page_size) || []
+        object_state[:offset] = offset + items.length
+        complete = object_state[:offset] >= nodes.length
+
+        complete && items.empty? ? nil : GraphQL::Breadth::ListStreamChunk.new(items:, complete:)
+      end
+    end
+  end
+
+  class LazyPagedProductNodesResolver < GraphQL::Breadth::FieldResolver
+    def stream?
+      true
+    end
+
+    def resolve(exec_field, _ctx)
+      exec_field.map_objects { _1["nodes"] }
+    end
+
+    def resolve_list_stream(objects, _ctx, state:, object_states:, limit:, iteration:, field:)
+      field
+        .lazy(loader_class: BatchTrackingLoader, keys: objects.map { [field.key, _1["nodes"]] })
+        .then do |entries|
+          entries.map do |(_field_key, nodes)|
+            GraphQL::Breadth::ListStreamChunk.new(items: nodes, complete: true)
+          end
+        end
+    end
+  end
+
+  class PagedVariantNodesResolver < GraphQL::Breadth::FieldResolver
+    attr_reader :calls
+
+    def initialize
+      @calls = []
+    end
+
+    def stream?
+      true
+    end
+
     def resolve(exec_field, _ctx)
-      @calls << [:resolve]
-      exec_field.resolve_all(@initial_items + @remaining_items.to_a)
+      exec_field.map_objects { _1["nodes"] }
     end
 
-    def resolve_stream(exec_field, _ctx, initial_count:)
-      @calls << [:resolve_stream, initial_count]
-      exec_field.resolve_all(GraphQL::Breadth::StreamSource.new(
-        initial_items: @initial_items,
-        remaining_items: @remaining_items,
-      ))
+    def resolve_list_stream(objects, _ctx, state:, object_states:, limit:, iteration:, field:)
+      state[:calls] = @calls
+      @calls << {
+        limit:,
+        iteration:,
+        objects: objects.map { _1["name"] },
+        arguments: field.arguments,
+      }
+
+      objects.map.with_index do |object, index|
+        nodes = object.fetch("nodes")
+        object_state = object_states[index]
+        offset = object_state[:offset] || 0
+        page_size = limit || 1
+        items = nodes.slice(offset, page_size) || []
+        object_state[:offset] = offset + items.length
+
+        if iteration.zero?
+          GraphQL::Breadth::ListStreamChunk.new(items:, complete: false)
+        else
+          complete = object_state[:offset] >= nodes.length
+          complete && items.empty? ? nil : GraphQL::Breadth::ListStreamChunk.new(items:, complete:)
+        end
+      end
     end
   end
 
@@ -79,6 +157,38 @@ def resolve_stream(exec_field, _ctx, initial_count:)
     },
   }.freeze
 
+  PAGED_VARIANTS_SOURCE = {
+    "products" => {
+      "nodes" => [{
+        "id" => "gid://shopify/Product/1",
+        "title" => "Banana",
+        "variants" => {
+          "name" => "Banana variants",
+          "nodes" => [{
+            "id" => "gid://shopify/Variant/1",
+            "title" => "Small Banana",
+          }],
+        },
+      }, {
+        "id" => "gid://shopify/Product/2",
+        "title" => "Apple",
+        "variants" => {
+          "name" => "Apple variants",
+          "nodes" => [{
+            "id" => "gid://shopify/Variant/2",
+            "title" => "Small Apple",
+          }, {
+            "id" => "gid://shopify/Variant/3",
+            "title" => "Medium Apple",
+          }, {
+            "id" => "gid://shopify/Variant/4",
+            "title" => "Large Apple",
+          }],
+        },
+      }],
+    },
+  }.freeze
+
   def test_incremental_result_returns_normal_result_without_defer
     result = build_executor(%|{
       products(first: 2) {
@@ -937,6 +1047,7 @@ def test_ready_deferred_executions_batch_lazy_work
   end
 
   def test_incremental_result_streams_list_field
+    resolver = PagedProductNodesResolver.new
     result = build_executor(%|{
       products(first: 2) {
         nodes @stream(initialCount: 1) {
@@ -944,7 +1055,7 @@ def test_incremental_result_streams_list_field
           title
         }
       }
-    }|).incremental_result
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).incremental_result
 
     assert_equal(
       {
@@ -971,34 +1082,21 @@ def test_incremental_result_streams_list_field
     )
   end
 
-  def test_incremental_result_streams_enumerable_list_without_materializing_tail
-    pulled_ids = []
-    nodes = SOURCE.fetch("products").fetch("nodes")
-    source = {
-      "products" => {
-        "nodes" => Enumerator.new do |yielder|
-          nodes.each do |node|
-            pulled_ids << node.fetch("id")
-            yielder << node
-          end
-        end,
-      },
-    }
-
+  def test_incremental_result_streams_with_default_initial_count
+    resolver = PagedProductNodesResolver.new
     result = build_executor(%|{
       products(first: 2) {
-        nodes @stream(initialCount: 1) {
+        nodes @stream {
           id
         }
       }
-    }|, source:).incremental_result
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).incremental_result
 
-    assert_equal ["gid://shopify/Product/1"], pulled_ids
     assert_equal(
       {
         "data" => {
           "products" => {
-            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+            "nodes" => [],
           },
         },
         "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
@@ -1009,7 +1107,10 @@ def test_incremental_result_streams_enumerable_list_without_materializing_tail
     assert_equal(
       [{
         "incremental" => [{
-          "items" => [{ "id" => "gid://shopify/Product/2" }],
+          "items" => [
+            { "id" => "gid://shopify/Product/1" },
+            { "id" => "gid://shopify/Product/2" },
+          ],
           "id" => "0",
         }],
         "completed" => [{ "id" => "0" }],
@@ -1017,45 +1118,50 @@ def test_incremental_result_streams_enumerable_list_without_materializing_tail
       }],
       result.subsequent_results.to_a,
     )
-    assert_equal ["gid://shopify/Product/1", "gid://shopify/Product/2"], pulled_ids
   end
 
-  def test_incremental_result_resolves_stream_source_for_streamed_list_field
-    pulled_remaining_ids = []
-    nodes = SOURCE.fetch("products").fetch("nodes")
-    remaining_items = Enumerator.new do |yielder|
-      node = nodes.fetch(1)
-      pulled_remaining_ids << node.fetch("id")
-      yielder << node
-    end
-    nodes_resolver = StreamSourceNodesResolver.new(
-      initial_items: [nodes.fetch(0)],
-      remaining_items:,
-    )
+  def test_incremental_result_batches_opt_in_list_streams_and_drops_completed_objects
+    resolver = PagedVariantNodesResolver.new
     resolvers = BREADTH_RESOLVERS.merge(
-      "ProductConnection" => BREADTH_RESOLVERS.fetch("ProductConnection").merge(
-        "nodes" => nodes_resolver,
+      "VariantConnection" => BREADTH_RESOLVERS.fetch("VariantConnection").merge(
+        "nodes" => resolver,
       ),
     )
 
     result = build_executor(%|{
       products(first: 2) {
-        nodes @stream(initialCount: 1) {
+        nodes {
           id
+          variants(first: 3) {
+            nodes @stream(initialCount: 1) {
+              id
+            }
+          }
         }
       }
-    }|, resolvers:).incremental_result
+    }|, source: PAGED_VARIANTS_SOURCE, resolvers:).incremental_result
 
-    assert_equal [[:resolve_stream, 1]], nodes_resolver.calls
-    assert_equal [], pulled_remaining_ids
     assert_equal(
       {
         "data" => {
           "products" => {
-            "nodes" => [{ "id" => "gid://shopify/Product/1" }],
+            "nodes" => [{
+              "id" => "gid://shopify/Product/1",
+              "variants" => {
+                "nodes" => [{ "id" => "gid://shopify/Variant/1" }],
+              },
+            }, {
+              "id" => "gid://shopify/Product/2",
+              "variants" => {
+                "nodes" => [{ "id" => "gid://shopify/Variant/2" }],
+              },
+            }],
           },
         },
-        "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
+        "pending" => [
+          { "id" => "0", "path" => ["products", "nodes", 0, "variants", "nodes"] },
+          { "id" => "1", "path" => ["products", "nodes", 1, "variants", "nodes"] },
+        ],
         "hasNext" => true,
       },
       result.initial_result,
@@ -1063,62 +1169,51 @@ def test_incremental_result_resolves_stream_source_for_streamed_list_field
     assert_equal(
       [{
         "incremental" => [{
-          "items" => [{ "id" => "gid://shopify/Product/2" }],
-          "id" => "0",
+          "items" => [{ "id" => "gid://shopify/Variant/3" }],
+          "id" => "1",
         }],
         "completed" => [{ "id" => "0" }],
+        "hasNext" => true,
+      }, {
+        "incremental" => [{
+          "items" => [{ "id" => "gid://shopify/Variant/4" }],
+          "id" => "1",
+        }],
+        "completed" => [{ "id" => "1" }],
         "hasNext" => false,
       }],
       result.subsequent_results.to_a,
     )
-    assert_equal ["gid://shopify/Product/2"], pulled_remaining_ids
-  end
-
-  def test_incremental_result_streams_with_default_initial_count
-    result = build_executor(%|{
-      products(first: 2) {
-        nodes @stream {
-          id
-        }
-      }
-    }|).incremental_result
-
-    assert_equal(
-      {
-        "data" => {
-          "products" => {
-            "nodes" => [],
-          },
-        },
-        "pending" => [{ "id" => "0", "path" => ["products", "nodes"] }],
-        "hasNext" => true,
-      },
-      result.initial_result,
-    )
     assert_equal(
       [{
-        "incremental" => [{
-          "items" => [
-            { "id" => "gid://shopify/Product/1" },
-            { "id" => "gid://shopify/Product/2" },
-          ],
-          "id" => "0",
-        }],
-        "completed" => [{ "id" => "0" }],
-        "hasNext" => false,
+        limit: 1,
+        iteration: 0,
+        objects: ["Banana variants", "Apple variants"],
+        arguments: {},
+      }, {
+        limit: nil,
+        iteration: 1,
+        objects: ["Banana variants", "Apple variants"],
+        arguments: {},
+      }, {
+        limit: nil,
+        iteration: 2,
+        objects: ["Apple variants"],
+        arguments: {},
       }],
-      result.subsequent_results.to_a,
+      resolver.calls,
     )
   end
 
   def test_incremental_result_includes_stream_label
+    resolver = PagedProductNodesResolver.new
     result = build_executor(%|{
       products(first: 2) {
         nodes @stream(initialCount: 1, label: "ProductNodes") {
           id
         }
       }
-    }|).incremental_result
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).incremental_result
 
     assert_equal(
       [{ "id" => "0", "path" => ["products", "nodes"], "label" => "ProductNodes" }],
@@ -1152,7 +1247,7 @@ def test_incremental_result_does_not_stream_when_if_is_false
   end
 
   def test_ready_stream_executions_batch_lazy_work
-    resolvers = BREADTH_RESOLVERS.merge(
+    resolvers = stream_product_nodes_resolvers.merge(
       "Product" => BREADTH_RESOLVERS.fetch("Product").merge(
         "title" => LazyHashResolver.new("title"),
       ),
@@ -1176,8 +1271,40 @@ def test_ready_stream_executions_batch_lazy_work
     )
   end
 
+  def test_ready_list_stream_fields_batch_lazy_resolver_work_across_streams
+    resolver = LazyPagedProductNodesResolver.new
+    BatchTrackingLoader.perform_keys = []
+
+    result = build_executor(%|{
+      products(first: 2) {
+        first: nodes @stream {
+          id
+        }
+        second: nodes @stream {
+          id
+        }
+      }
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).incremental_result
+
+    result.subsequent_results.to_a
+
+    nodes = SOURCE.fetch("products").fetch("nodes")
+    assert_equal(
+      [[["first", nodes], ["second", nodes]]],
+      BatchTrackingLoader.perform_keys,
+    )
+  end
+
   private
 
+  def stream_product_nodes_resolvers(resolver = PagedProductNodesResolver.new)
+    BREADTH_RESOLVERS.merge(
+      "ProductConnection" => BREADTH_RESOLVERS.fetch("ProductConnection").merge(
+        "nodes" => resolver,
+      ),
+    )
+  end
+
   def build_executor(document, source: SOURCE, resolvers: BREADTH_RESOLVERS)
     GraphQL::Breadth::Executor.new(
       SCHEMA,

From 8ffbfee194da63b2d035e8a4d7fb4cff58c11703 Mon Sep 17 00:00:00 2001
From: Greg MacWilliam <greg.macwilliam@shopify.com>
Date: Fri, 26 Jun 2026 22:43:21 -0400
Subject: [PATCH 4/4] latest noodling.

---
 README.md                                     | 266 ++++++++++--------
 lib/graphql/breadth/executor.rb               |   7 +-
 .../breadth/executor/execution_field.rb       |   2 +-
 lib/graphql/breadth/field_resolvers.rb        |   9 +-
 .../breadth/executor/incremental_test.rb      |  40 +--
 5 files changed, 174 insertions(+), 150 deletions(-)

diff --git a/README.md b/README.md
index 7711cbd..829492a 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 # Breadth-first GraphQL execution
 
-_**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)._
+_**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 TypeScript 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.
@@ -64,51 +64,59 @@ query {
 }
 ```
 
-Gets built into an execution tree structured as the following pseudocode:
+Gets built into an execution tree structured as the following pseudocode. Object scopes own their selected fields; non-leaf fields have planned child scopes that are entered after the field resolves objects:
 
 ```ruby
-ExecutionScope.new(type: QueryRoot, fields: [
-  ExecutionField.new(key: "products", ExecutionScope.new(type: ProductConnection, fields: [
-    ExecutionField.new(key: "nodes", ExecutionScope.new(type: Product, fields: [
-      ExecutionField.new(key: "id"),
-      ExecutionField.new(key: "title"),
-    ]))
-  ]))
-])
+query_scope = ExecutionScope.new(parent_type: QueryRoot)
+products_field = ExecutionField.new(key: "products", scope: query_scope)
+
+product_connection_scope = ExecutionScope.new(parent_type: ProductConnection, parent_field: products_field)
+nodes_field = ExecutionField.new(key: "nodes", scope: product_connection_scope)
+
+product_scope = ExecutionScope.new(parent_type: Product, parent_field: nodes_field)
+id_field = ExecutionField.new(key: "id", scope: product_scope)
+title_field = ExecutionField.new(key: "title", scope: product_scope)
 ```
 
 This taxonomy provides the following API, which is useful while writing resolver behaviors:
 
 * **`ExecutionField`**: represents a field to execute within a resolved object scope.
   - `path`: the selection path leading to the field, composed of namespaces with no list indices.
+  - `schema_path`: the schema path leading to the field, using schema field names rather than response aliases.
   - `key`: the namespace assigned by the field's selection alias or definition name.
+  - `name`: the field's schema name.
   - `type`: the GraphQL return type of the field, may be abstract with non-null and list wrappers.
-  - `arguments`: a frozen hash of arguments provided to the selection. Argument keys are `:snake_case` symbols. Argument transformations are intentionally do not supported (i.e. the input "prepare" hook); argument formatting should be done holistically in the resolver.
+  - `arguments`: a frozen hash of arguments provided to the selection. Argument keys are `:snake_case` symbols. Argument transformations are intentionally not supported (i.e. the input "prepare" hook); argument formatting should be done holistically in the resolver.
   - `mutable_arguments`: a mutable clone of the arguments hash that can be modified.
   - `definition`: the associated GraphQL field definition. For schema reference only (avoid repurposing legacy implementation details).
   - `scope`: the parent execution scope that this field belongs to.
+  - `parent_type`: the GraphQL object type that owns the field.
+  - `planning_root`: the highest scope that still accepts planning actions while this field is being planned.
   - `resolve_all(<value>)`: resolves a value mapped to all field objects. Useful for early returns.
   - `preload(<LazyLoader>, keys: [...]?, args: { ... }?)`: Registers a lazy preloader to run before the field executes. May only be called by field planner methods.
-  - `lazy(<LazyLoader>?, keys: [...], args: { ... }?)`: defers to lazy execution and returns a Promise. Similar to GraphQL Batch with some fundamental changes, see [documentation](#lazy-resolvers). May only be called by field resolver methods.
+  - `lazy(loader_class: <LazyLoader>, keys: [...], args: { ... }?)`: defers to lazy execution and returns a Promise. Similar to GraphQL Batch with some fundamental changes, see [documentation](#lazy-resolvers). May only be called by field resolver methods.
+  - `await_all([...promises])`: combines several execution promises and resolves when all are fulfilled.
   - `attributes`: a hash intended for local caching and freeform planning notes.
   - `attribute(<name>)`: reads an attribute without allocating storage.
   - `attribute?(<name>)`: checks an attribute without allocating storage.
 * **`ExecutionScope`**: represents a resolved object scope with a known concrete object type.
   - `path`: selection path leading to the scope, composed of namespaces with no list indices.
+  - `schema_path`: schema path leading to the scope.
   - `parent`: the execution scope above this one.
   - `parent_field`: the execution field in the parent scope that opened this scope.
   - `parent_type`: the GraphQL object type of the scope. This is always a resolved object type, never an abstract interface or union.
   - `abstraction`: for scopes resolved through an interface or union, this details characteristics of that abstraction.
+  - `planning_root`: the highest scope that still accepts planning actions while this scope is being planned.
   - `preload(<LazyLoader>, keys: [...]?, args: { ... }?)`: Registers a lazy preloader to run before the scope executes. May only be called by planner methods.
   - `attributes`: a hash intended for local caching and freeform planning notes.
   - `attribute(<name>)`: reads an attribute without allocating storage.
   - `attribute?(<name>)`: checks an attribute without allocating storage.
 
-**An execution tree can only be traversed from the bottom-up**. This is extremely intentional, because traversing top-down can never see through unresolved abstractions.
+**Planning traverses each concrete execution tree from the bottom-up**. This is intentional because top-down planning cannot see through unresolved abstractions; once an abstract field resolves to concrete object types, its newly built subtree gets its own bottom-up planning pass.
 
 ## Field resolvers
 
-For each field implementation, set up a `GraphQL::Breadth::FieldResolver` or use a [keyword helper](#resolver-keywords):
+For each field implementation, set up a `GraphQL::Breadth::FieldResolver` or use a [resolver keyword](#resolver-keywords):
 
 ```ruby
 class MyFieldResolver < GraphQL::Breadth::FieldResolver
@@ -125,7 +133,7 @@ A field resolver receives:
    - `exec_field.arguments`: a hash of resolved arguments provided to the field.
 * `context`: the request context.
 
-A resolver **must return a mapped set of results** for the field's objects, or invoke a [lazy resolver hook](#lazy-resolvers). To attach a field resolver to a field, use the `GraphQL::Breadth::HasBreadthResolver` field mixin:
+A resolver **must return a mapped set of results** for the field's objects, or return an execution promise from [lazy loading](#lazy-resolvers). To attach a field resolver to a field, use the `GraphQL::Breadth::HasBreadthResolver` field mixin:
 
 ```ruby
 class BaseField < GraphQL::Schema::Field
@@ -143,6 +151,8 @@ class MyObject < BaseObject
 end
 ```
 
+Resolver classes may also be assigned directly; they are instantiated when assigned. If a schema field does not provide `breadth_resolver`, the executor falls back to the `resolvers:` map passed to `GraphQL::Breadth::Executor.new`, keyed by GraphQL type name and field name.
+
 ### Built-in resolvers
 
 The core library includes several basic resolvers for common needs:
@@ -152,6 +162,15 @@ The core library includes several basic resolvers for common needs:
 * `GraphQL::Breadth::ValueResolver.new(true)` (static value)
 * `GraphQL::Breadth::SelfResolver.new` (resolves original objects)
 
+### Resolver keywords
+
+When using `GraphQL::Breadth::HasBreadthResolver::Field`, `breadth_resolver` may be assigned one of the built-in keyword helpers:
+
+* `:method`: calls a method matching the field's original schema name.
+* `:hash_key_symbol`: reads a symbol key matching the field's original schema name.
+* `:hash_key_string`: reads a string key matching the field's original schema name.
+* `:itself`: resolves the original object.
+
 ### Early return
 
 Field resolvers may return early with a value for all objects using `resolve_all`. This is commonly used to resolve `nil` or an eager value across all field objects.
@@ -244,58 +263,74 @@ query {
 
 In the above, we'll want `Image.sources` to batch across all instances of the field, even at different document depths. LazyLoader solves this – which is breadth's analog to traditional dataloaders. Unlike traditional dataloaders though, a breadth LazyLoader binds entire key sets to a single promise, rather than building 1:1 promises. This dramatically reduces lazy overhead.
 
-### Resolver-specific lazy loading
+### LazyLoader classes
 
-For the simplest loading scenarios, field resolvers can delegate to their own lazy loading hook with keys that will batch across instances of the field:
+Lazy work is always fulfilled by a `GraphQL::Breadth::LazyLoader` class. A field resolver calls `exec_field.lazy(loader_class:, keys:, args: ...)`, which returns an execution promise. The executor pools all pending promises by loader class and argument set, runs each loader once per lazy wave, then resolves each field with its mapped result set.
 
 ```ruby
-class BasicLazyResolver < GraphQL::Breadth::FieldResolver
-  def resolve(exec_field, context)
-    exec_field.lazy(keys: exec_field.objects.map(&:id))
+class ThingLoader < GraphQL::Breadth::LazyLoader
+  def perform(ids, context)
+    Thing.where(parent_id: ids).find_each do |thing|
+      fulfill_key(thing.parent_id, thing)
+    end
   end
+end
 
-  def perform_lazy(keys, args, context)
-    things_by_key = Thing.where(parent_id: keys).index_by(&:parent_id)
-    keys.map { |key| things_by_key[key] }
+class ThingResolver < GraphQL::Breadth::FieldResolver
+  def resolve(exec_field, context)
+    exec_field.lazy(
+      loader_class: ThingLoader,
+      keys: exec_field.objects.map(&:id),
+    )
   end
 end
 ```
 
-Calling `exec_field.lazy` defers a set of keys for lazy fulfillment, which then get loaded by the resolver's `perform_lazy` hook that **must return a mapped set of results**. You can also scope lazy loading with arguments, and wrap it with pre- and post-processing steps:
+Within a loader class, call `fulfill_key` to deliver each loaded record. Lazy loaders do not require fulfillment of each provided key; unfulfilled keys resolve as `nil`. You can also scope a loader instance with arguments, and wrap the promised values with post-processing:
 
 ```ruby
-class FancyLazyResolver < GraphQL::Breadth::FieldResolver
-  def resolve(exec_field, context)
-    mapped_keys = exec_field.objects.map { |obj| obj.valid? ? obj.id : nil }
+class GroupedThingLoader < GraphQL::Breadth::LazyLoader
+  def initialize(group:)
+    super()
+    @group = group
+  end
+
+  def perform(ids, context)
+    Thing.where(parent_id: ids, group: @group).find_each do |thing|
+      fulfill_key(thing.parent_id, thing)
+    end
+  end
+end
+
+class GroupedThingResolver < GraphQL::Breadth::FieldResolver
+  def resolve(exec_field, _context)
+    keys = exec_field.objects.map { |obj| obj.valid? ? obj.id : nil }
 
     exec_field
-      .lazy(args: { group: "a" }, keys: mapped_keys)
+      .lazy(loader_class: GroupedThingLoader, args: { group: "a" }, keys: keys)
       .then do |loaded_records|
         loaded_records.map! { |record| record&.my_field }
       end
   end
-
-  def perform_lazy(keys, args, context)
-    things_by_key = Thing.where(parent_id: keys, group: args[:group]).index_by(&:parent_id)
-    keys.map { |key| things_by_key[key] }
-  end
 end
 ```
 
+Loader instances are cached per executor by `[loader_class, args]`, so fields using the same loader class and arguments share a batch. Fields using different arguments get separate loader instances.
+
 ### Nil keys
 
 It's extremely common for a mapped set of lazy keys to have `nil` positions that must be retained to match the resolver's breadth set. These nil keys should almost never be loaded, so they are omitted from batching and resolve as nil by default. If you specifically want to treat nil as a loadable value, specify `load_nil_keys: true`.
 
 ```ruby
 class MaybeNilKeysResolver < GraphQL::Breadth::FieldResolver
-  def resolve(exec_field, context)
+  def resolve(exec_field, _context)
     mapped_keys = exec_field.objects.map { |obj| obj.ready? ? obj.id : nil }
 
-    exec_field.lazy(keys: mapped_keys, load_nil_keys: true)
-  end
-
-  def perform_lazy(keys, args, context)
-    # ... keys may include nil!
+    exec_field.lazy(
+      loader_class: ThingLoader,
+      keys: mapped_keys,
+      load_nil_keys: true,
+    )
   end
 end
 ```
@@ -314,57 +349,47 @@ class MaskingResolver < GraphQL::Breadth::FieldResolver
       obj.key
     end
 
-    exec_field.lazy(keys: mapped_keys, eager_values: eager_values)
-  end
-
-  def perform_lazy(keys, args, context)
-    # ... keys won't include "zebra"
+    exec_field.lazy(
+      loader_class: ThingLoader,
+      keys: mapped_keys,
+      eager_values: eager_values,
+    )
   end
 end
 ```
 
-Eager values are specific to their field instance and will _not_ be shared by fields [using the same loader](#shared-lazy-loading). Eager values override the loader cache, so a specific field instance may eagerly resolve its own value for a key while other fields sharing the loader will still load the key as normal.
+Eager values are specific to their field instance and will _not_ be shared by fields using the same loader. Eager values override the loader cache for that promise, so a specific field instance may eagerly resolve its own value for a key while other fields sharing the loader still load the key as normal.
 
-### Shared lazy loading
+### Mapped loaders
 
-Queries shared across field resolvers need a common LazyLoader class.
+When it is cheaper to return a complete mapped result set than to fulfill records individually, implement `map?` and `perform_map`. A mapped loader **must return one result per pending loader key**.
 
 ```ruby
-class SharedLoader < GraphQL::Breadth::LazyLoader
-  def initialize(group:)
-    super()
-    @group = group
-  end
-
-  def perform(ids, context)
-    Thing.where(parent_id: ids, group: @group).to_a.each do |thing|
-      fulfill_key(thing.parent_id, thing)
-    end
+class MapLoader < GraphQL::Breadth::LazyLoader
+  def map?
+    true
   end
-end
 
-class SharedLazyResolver < GraphQL::Breadth::FieldResolver
-  def resolve(exec_field, context)
-    mapped_keys = exec_field.objects.map { |obj| obj.valid? ? obj.id : nil }
-
-    exec_field
-      .lazy(SharedLoader, args: { group: "a" }, keys: mapped_keys)
-      .then do |loaded_records|
-        loaded_records.map! { |record| record&.my_field }
-      end
+  def perform_map(keys, context)
+    things_by_key = Thing.where(parent_id: keys).index_by(&:parent_id)
+    keys.map { |key| things_by_key[key] }
   end
 end
 ```
 
-In this case `exec_field.lazy` is called with a LazyLoader class, which delegates loading to an instance of the provided loader class rather than the resolver's own `perform_lazy` method. Within a loader class, call `fulfill_key` to deliver each loaded record. Lazy loaders do NOT require fulfillment of each provided key; unfulfilled keys simply return as `nil`. You can also set up a lazy loader class to fulfill by mapped set, although this frequently adds a mapping layer that calling `fulfill_key` directly would avoid:
+### Single-result loaders
+
+For loaders that produce exactly one result for exactly one key, implement `resolve_one?`. Calls using that loader must provide exactly one key, and the promise resolves to a single object rather than an array.
 
 ```ruby
-class MapLoader < GraphQL::Breadth::LazyLoader
-  def map? = true
+class OneThingLoader < GraphQL::Breadth::LazyLoader
+  def resolve_one?
+    true
+  end
 
-  def perform_map(keys, context)
-    things_by_key = Thing.where(parent_id: keys).index_by(&:parent_id)
-    keys.map { |key| things_by_key[key] }
+  def perform(ids, context)
+    thing = Thing.find_by(id: ids.first)
+    fulfill_key(ids.first, thing) if thing
   end
 end
 ```
@@ -398,13 +423,13 @@ class AwaitingResolver < GraphQL::Breadth::FieldResolver
   def resolve(exec_field, _context)
     keys = exec_field.objects.map(&:key)
 
-    a = exec_field.lazy(PrefixLoader, args: { prefix: "a" }, keys: keys)
-    b = exec_field.lazy(PrefixLoader, args: { prefix: "b" }, keys: keys)
+    a = exec_field.lazy(loader_class: PrefixLoader, args: { prefix: "a" }, keys: keys)
+    b = exec_field.lazy(loader_class: PrefixLoader, args: { prefix: "b" }, keys: keys)
 
     exec_field
       .await_all([a, b])
       .then do |results_a, results_b|
-        exec_field.objects.map.with_index do |i|
+        exec_field.objects.map.with_index do |_object, i|
           "#{results_a[i]} + #{results_b[i]}"
         end
       end
@@ -418,8 +443,8 @@ Lazy sequencing can be chained:
 class ChainingResolver < GraphQL::Breadth::FieldResolver
   def resolve(exec_field, _context)
     exec_field
-      .lazy(PrefixLoader, args: { prefix: "a" }, keys: exec_field.objects.map(&:key))
-      .then { |results_a| exec_field.lazy(PrefixLoader, args: { prefix: "b" }, keys: results_a) }
+      .lazy(loader_class: PrefixLoader, args: { prefix: "a" }, keys: exec_field.objects.map(&:key))
+      .then { |results_a| exec_field.lazy(loader_class: PrefixLoader, args: { prefix: "b" }, keys: results_a) }
       .then { |results_b| results_b.map { |b| "#{b}-fin" } }
   end
 end
@@ -504,9 +529,9 @@ class WidgetResolver < GraphQL::Breadth::FieldResolver
 end
 ```
 
-### Preloads hooks
+### Preload hooks
 
-Some lazy preloads may need to be configured at the time of execution when objects are actually avaiable for a scope or field. The `on_preload` hook may be used during planning to configure preloads in a just-in-time manner.
+Some lazy preloads may need to be configured at the time of execution when objects are actually available for a scope or field. The `on_preload` hook may be used during planning to configure preloads in a just-in-time manner.
 
 ```ruby
 class WidgetResolver < GraphQL::Breadth::FieldResolver
@@ -571,7 +596,7 @@ Authorization gates access at three grains: permission to access types, permissi
 * `authorized_type?(type, context, exec_field: nil)`: checks if a type may be accessed before entering a scope of the type, and before executing a field that returns the type.
 * `authorized_field?(exec_field, context)`: checks if a field may be accessed before executing its resolver. This should _only_ check if the field itself is authorized; it should NOT consider the field's owner type and/or return type, which are both covered by direct type checks (see above).
 * `authorize_objects_in_scope?(exec_scope, context)`: checks if object-level authorization checks should run in this scope.
-* `unauthorized_object_indices(exec_scope, context)`: checks authorization on all scope objects, and returns an invalidation map formatted as `Hash[Integer, StandardError?]`. The returned hash maps object indicies to their corresponding authorization errors. An empty hash means no objects were invalidated.
+* `unauthorized_object_indices(exec_scope, context)`: checks authorization on all scope objects, and returns an invalidation map formatted as `Hash[Integer, StandardError?]`. The returned hash maps object indices to their corresponding authorization errors. An empty hash means no objects were invalidated.
 
 ## Runtime directives
 
@@ -598,14 +623,14 @@ query {
 }
 ```
 
-To implement a runtime directive, set up a `Breadth::DirectiveResolver` and assign it to the directive class:
+To implement a runtime directive, set up a `GraphQL::Breadth::DirectiveResolver` and assign it to the directive class:
 
 ```ruby
-class LanguageDirectiveResolver < DirectiveResolver
+class LanguageDirectiveResolver < GraphQL::Breadth::DirectiveResolver
   def resolve(exec_directive, context, current_field: nil)
     return if current_field.nil?
 
-    current_field.attribute[:lang] = exec_directive.arguments[:lang]
+    current_field.attributes[:lang] = exec_directive.arguments[:lang]
   end
 end
 
@@ -625,7 +650,7 @@ end
 Directive resolvers can be configured as block wrappers around all of GraphQL execution (QUERY / MUTATION), or around the execution of a field (FIELD). Wrapping is disabled by default because it adds overhead. To enable wrapping for a specific directive, enable it for the resolver and include a `yield` in its resolver, or pass the resolver `&block` forward:
 
 ```ruby
-class InContextDirectiveResolver < DirectiveResolver
+class InContextDirectiveResolver < GraphQL::Breadth::DirectiveResolver
   def initialize
     super(wraps: true)
   end
@@ -661,7 +686,7 @@ query {
 We expect `a` to assign a base language of `EN` that `b` inherits, and then `c` overrides with a more specific setting. Breadth execution achieves this by marking directives as _cascading_. A cascading directive will be passed down to all of its child fields within a stacking queue. A field execution then runs all directives that it inherited in the order they were queued, followed by any directives defined on the field itself.
 
 ```ruby
-class LanguageDirectiveResolver < DirectiveResolver
+class LanguageDirectiveResolver < GraphQL::Breadth::DirectiveResolver
   def initialize
     super(cascades: true)
   end
@@ -670,7 +695,7 @@ class LanguageDirectiveResolver < DirectiveResolver
     return if current_field.nil?
 
     # repeatedly write each cascading directive's value onto the field; last one wins...
-    current_field.attribute[:lang] = exec_directive.arguments[:lang]
+    current_field.attributes[:lang] = exec_directive.arguments[:lang]
   end
 end
 ```
@@ -699,7 +724,7 @@ The basic and incremental entry points are intentionally strict. Call either `re
 
 ### Streaming lists
 
-Fields that support `@stream` must opt into the list stream API. The ordinary `resolve` method remains responsible for non-incremental execution, while `resolve_list_stream` loads items for active stream installments:
+Fields that support `@stream` must opt into the list stream API. The ordinary `resolve` method remains responsible for non-incremental execution. When a stream directive is active, the executor creates a `ListStreamField` for the field instance and calls `resolve_list_stream` for each stream installment.
 
 ```ruby
 class ProductNodesResolver < GraphQL::Breadth::FieldResolver
@@ -711,11 +736,12 @@ class ProductNodesResolver < GraphQL::Breadth::FieldResolver
     exec_field.map_objects { |connection| connection.nodes }
   end
 
-  def resolve_list_stream(objects, context, state:, object_states:, limit:, iteration:, field:)
-    objects.map.with_index do |connection, index|
-      object_state = object_states[index]
+  def resolve_list_stream(field, context)
+    field.pending_entries.map do |entry|
+      connection = entry.object
+      object_state = entry.object_state
       cursor = object_state[:cursor]
-      page_size = limit || 25
+      page_size = field.limit || 25
 
       page = ProductLoader.load_page(
         connection,
@@ -735,27 +761,34 @@ class ProductNodesResolver < GraphQL::Breadth::FieldResolver
 end
 ```
 
-`resolve_list_stream` receives only the parent objects that still have active stream deliveries. Once an object returns a complete chunk, it is dropped from future calls. This lets resolvers keep batching active streams together while avoiding repeated work for streams that have already finished.
+`resolve_list_stream` receives the `GraphQL::Breadth::Executor::ListStreamField` for the active stream instance. Its `pending_entries` are only the parent objects that still have active stream deliveries. Once an entry returns a complete chunk, it is dropped from the field's `pending_entries` and will not be present in later calls. This lets resolvers keep batching active streams together while avoiding repeated work for streams that have already finished.
 
 The method arguments are:
 
-* `objects`: the active parent objects for this installment.
-* `state`: a shared hash for the streamed field instance across all installments.
-* `object_states`: one persistent hash per active object, mapped by index to `objects`.
-* `limit`: the directive's `initialCount` for the initial call, then `nil` for later calls.
-* `iteration`: the call count for this streamed field batch.
-* `field`: a `ListStreamField` call object with the streamed field's `arguments`, `path`, `lazy`, `await_all`, and other field helpers.
+* `field`: the `GraphQL::Breadth::Executor::ListStreamField` for this stream instance.
+* `context`: the request context.
+
+`ListStreamField` delegates the field shape from the original `ExecutionField`: `arguments`, `mutable_arguments`, `type`, `nodes`, `selections`, `key`, `name`, and `path` all refer to the streamed schema field. It also exposes stream-specific state:
 
-Return one result per object. A `GraphQL::Breadth::ListStreamChunk` is the explicit form: `items:` are emitted in this installment and `complete:` controls whether that object's stream remains active. Returning an array is shorthand for "emit these items and keep going" unless the array is empty, in which case the object is complete. Returning `nil` completes the object without emitting items.
+* `pending_entries`: active `GraphQL::Breadth::Incremental::ListStreamEntry` objects.
+* `state`: the shared field-instance state hash.
+* `limit`: the directive's `initialCount` when preparing the initial result, then `nil` for later calls.
+* `iteration`: the zero-based call count for this stream field.
 
-If the directive uses `@stream(initialCount: 0)`, the initial result includes no list items and the first `resolve_list_stream` call happens while preparing the first subsequent payload.
+Each `ListStreamEntry` exposes:
 
-```ruby
-GraphQL::Breadth::ListStreamEntry.new(
-  object: object,
-  state: nil,
-)
+* `object`: the active parent object for this stream installment.
+* `object_state`: a persistent state hash for that specific object.
+
+`ListStreamField` also supports the same execution helpers used by fields: `lazy(loader_class:, keys:, args: ...)`, `await_all`, `resolve_all`, `handle_or_reraise`, and `attributes`. Lazy stream resolvers should call `field.lazy`, not `exec_field.lazy`, because stream installments are resumed independently from the original field execution.
+
+Return one result per active object. A `GraphQL::Breadth::ListStreamChunk` is the explicit form: `items:` are emitted in this installment and `complete:` controls whether that object's stream remains active. Returning an array is shorthand for "emit these items and keep going" unless the array is empty, in which case the object is complete. Returning `nil` completes the object without emitting items.
 
+If the directive uses `@stream(initialCount: 0)`, the initial result includes no list items and does not call `resolve_list_stream` while building the initial response. The first resolver call happens while preparing the first subsequent payload, with `limit: nil`.
+
+Streaming resolvers can batch the active entries through a lazy loader:
+
+```ruby
 class ProductNodesResolver < GraphQL::Breadth::FieldResolver
   def stream?
     true
@@ -765,22 +798,21 @@ class ProductNodesResolver < GraphQL::Breadth::FieldResolver
     exec_field.map_objects { |connection| connection.nodes }
   end
 
-  def resolve_list_stream(stream_exec_field, context, limit:)
-    # StreamExecutionField < ExecutionField
-    # owns its own pending_objects (objects filtered after each iteration)
-    limit ||= 10
+  def resolve_list_stream(field, context)
+    page_size = field.limit || 10
 
-    stream_exec_field.lazy(
-      MyStreamingListLoader,
-      keys: stream_exec_field.pending_entries,
-      args: { limit: limit },
+    field.lazy(
+      loader_class: ProductPageLoader,
+      keys: field.pending_entries,
+      args: { limit: page_size },
     ).then do |all_results|
-      stream_exec_field.pending_entries.map.with_index do |entry, index|
+      field.pending_entries.map.with_index do |entry, index|
         results = all_results[index]
-        entry.set_state(results.last&.id)
+        entry.object_state[:cursor] = results.last&.id
+
         GraphQL::Breadth::ListStreamChunk.new(
           items: results,
-          complete: results.size <= limit,
+          complete: results.size < page_size,
         )
       end
     end
diff --git a/lib/graphql/breadth/executor.rb b/lib/graphql/breadth/executor.rb
index e3e7c73..986a6eb 100644
--- a/lib/graphql/breadth/executor.rb
+++ b/lib/graphql/breadth/executor.rb
@@ -734,13 +734,8 @@ def resolve_list_stream_chunks(stream_field, limit:)
         stream_field.reset_for_resolve!(limit:)
         stream_field.lazy_state_executing!
         stream_field.result = exec_field.resolver.resolve_list_stream(
-          stream_field.objects,
+          stream_field,
           @context,
-          state: stream_field.state,
-          object_states: stream_field.object_states,
-          limit:,
-          iteration: stream_field.iteration,
-          field: stream_field,
         )
         stream_field.iteration += 1
 
diff --git a/lib/graphql/breadth/executor/execution_field.rb b/lib/graphql/breadth/executor/execution_field.rb
index 5b521cf..73739b3 100644
--- a/lib/graphql/breadth/executor/execution_field.rb
+++ b/lib/graphql/breadth/executor/execution_field.rb
@@ -207,7 +207,7 @@ def mutable_arguments
         #: () -> Incremental::StreamUsage?
         def stream_usage
           if @stream_usage.equal?(UNDEFINED)
-            @stream_usage = executor.incremental? && Util.unwrap_non_null(type).list? ? executor.planner.stream_usage_for(self) : nil
+            @stream_usage = executor.incremental? && type.list? ? executor.planner.stream_usage_for(self) : nil
           end
 
           @stream_usage
diff --git a/lib/graphql/breadth/field_resolvers.rb b/lib/graphql/breadth/field_resolvers.rb
index 941fb42..32da439 100644
--- a/lib/graphql/breadth/field_resolvers.rb
+++ b/lib/graphql/breadth/field_resolvers.rb
@@ -20,15 +20,10 @@ def stream?
       end
 
       #: (
-      #|   Array[untyped],
+      #|   Executor::ListStreamField,
       #|   GraphQL::Query::Context,
-      #|   state: Hash[untyped, untyped],
-      #|   object_states: Array[Hash[untyped, untyped]],
-      #|   limit: Integer?,
-      #|   iteration: Integer,
-      #|   field: Executor::ListStreamField,
       #| ) -> (Array[untyped] | Executor::ExecutionPromise)
-      def resolve_list_stream(_objects, _ctx, state:, object_states:, limit:, iteration:, field:)
+      def resolve_list_stream(_field, _ctx)
         raise NotImplementedError, "FieldResolver#resolve_list_stream must be implemented."
       end
 
diff --git a/test/graphql/breadth/executor/incremental_test.rb b/test/graphql/breadth/executor/incremental_test.rb
index 85661a2..0d31d0c 100644
--- a/test/graphql/breadth/executor/incremental_test.rb
+++ b/test/graphql/breadth/executor/incremental_test.rb
@@ -46,20 +46,21 @@ def resolve(exec_field, _ctx)
       exec_field.map_objects { _1["nodes"] }
     end
 
-    def resolve_list_stream(objects, _ctx, state:, object_states:, limit:, iteration:, field:)
-      state[:calls] = @calls
+    def resolve_list_stream(field, _ctx)
+      field.state[:calls] = @calls
       @calls << {
-        limit:,
-        iteration:,
-        object_count: objects.length,
+        limit: field.limit,
+        iteration: field.iteration,
+        object_count: field.pending_entries.length,
         arguments: field.arguments,
       }
 
-      objects.map.with_index do |object, index|
+      field.pending_entries.map do |entry|
+        object = entry.object
         nodes = object.fetch("nodes")
-        object_state = object_states[index]
+        object_state = entry.object_state
         offset = object_state[:offset] || 0
-        page_size = limit || @page_size || nodes.length
+        page_size = field.limit || @page_size || nodes.length
         items = nodes.slice(offset, page_size) || []
         object_state[:offset] = offset + items.length
         complete = object_state[:offset] >= nodes.length
@@ -78,9 +79,9 @@ def resolve(exec_field, _ctx)
       exec_field.map_objects { _1["nodes"] }
     end
 
-    def resolve_list_stream(objects, _ctx, state:, object_states:, limit:, iteration:, field:)
+    def resolve_list_stream(field, _ctx)
       field
-        .lazy(loader_class: BatchTrackingLoader, keys: objects.map { [field.key, _1["nodes"]] })
+        .lazy(loader_class: BatchTrackingLoader, keys: field.pending_entries.map { [field.key, _1.object["nodes"]] })
         .then do |entries|
           entries.map do |(_field_key, nodes)|
             GraphQL::Breadth::ListStreamChunk.new(items: nodes, complete: true)
@@ -104,24 +105,25 @@ def resolve(exec_field, _ctx)
       exec_field.map_objects { _1["nodes"] }
     end
 
-    def resolve_list_stream(objects, _ctx, state:, object_states:, limit:, iteration:, field:)
-      state[:calls] = @calls
+    def resolve_list_stream(field, _ctx)
+      field.state[:calls] = @calls
       @calls << {
-        limit:,
-        iteration:,
-        objects: objects.map { _1["name"] },
+        limit: field.limit,
+        iteration: field.iteration,
+        objects: field.pending_entries.map { _1.object["name"] },
         arguments: field.arguments,
       }
 
-      objects.map.with_index do |object, index|
+      field.pending_entries.map do |entry|
+        object = entry.object
         nodes = object.fetch("nodes")
-        object_state = object_states[index]
+        object_state = entry.object_state
         offset = object_state[:offset] || 0
-        page_size = limit || 1
+        page_size = field.limit || 1
         items = nodes.slice(offset, page_size) || []
         object_state[:offset] = offset + items.length
 
-        if iteration.zero?
+        if field.iteration.zero?
           GraphQL::Breadth::ListStreamChunk.new(items:, complete: false)
         else
           complete = object_state[:offset] >= nodes.length