diff --git a/README.md b/README.md
index 735b691..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.
@@ -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
@@ -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,16 +695,16 @@ 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
 ```
 
 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 +718,108 @@ 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. 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
+  def stream?
+    true
+  end
+
+  def resolve(exec_field, context)
+    exec_field.map_objects { |connection| connection.nodes }
+  end
+
+  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 = field.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 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:
+
+* `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:
+
+* `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.
+
+Each `ListStreamEntry` exposes:
+
+* `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
+  end
+
+  def resolve(exec_field, context)
+    exec_field.map_objects { |connection| connection.nodes }
+  end
+
+  def resolve_list_stream(field, context)
+    page_size = field.limit || 10
+
+    field.lazy(
+      loader_class: ProductPageLoader,
+      keys: field.pending_entries,
+      args: { limit: page_size },
+    ).then do |all_results|
+      field.pending_entries.map.with_index do |entry, index|
+        results = all_results[index]
+        entry.object_state[:cursor] = results.last&.id
+
+        GraphQL::Breadth::ListStreamChunk.new(
+          items: results,
+          complete: results.size < page_size,
+        )
+      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 8fae326..e26f302 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/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 801529f..986a6eb 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,10 +519,44 @@ 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
       def resume_lazy_scope_execute(exec_scope)
         begin
@@ -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:)
@@ -624,10 +664,10 @@ def execute_field(exec_field)
           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)
           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)
             end
           end
         rescue StandardError => e
@@ -649,6 +689,99 @@ def execute_field(exec_field)
         end
       end
 
+      #: (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,
+          @context,
+        )
+        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
@@ -676,6 +809,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 = exec_field.stream_usage
 
         if resolved_objects.length != parent_objects.length
           handle_or_reraise(ResultCountMismatchError.new(
@@ -693,7 +827,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 +850,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 +869,226 @@ def build_field_result(exec_field, resolved_objects)
         end
       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?
+
+        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_result = initial_items.map do |src|
+          build_and_flatmap_composite_result(exec_field, item_type, src, next_objects, next_results)
+        end
+
+        if register_stream
+          register_composite_stream_scope(
+            exec_field,
+            item_type,
+            stream_usage,
+            object_index,
+            remaining_items,
+            initial_items.length,
+          )
+        end
+
+        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?
+
+        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 = 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,
+            EMPTY_ARRAY,
+            [],
+            [],
+            [],
+            [],
+            stream_usage,
+            object_index,
+            initial_items.length,
+            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
+
+        initial_result
+      end
+
+      #: (ExecutionField[untyped], untyped, Integer) -> [Array[untyped], untyped, bool]
+      def split_stream_items(exec_field, value, initial_count)
+        case value
+        when Incremental::ListStreamSource
+          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,
+      #|   untyped remaining_items,
+      #|   Integer initial_count,
+      #| ) -> void
+      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)
+        list_stream_entry = remaining_items if remaining_items.is_a?(Incremental::ListStreamEntry)
+
+        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,
+          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, ?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
+
+      #: (
+      #|   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,
+      #|   ?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, list_stream_entry: nil)
+        stream_path = exec_field.object_path(object_index)
+        stream_delivery = Incremental::StreamDelivery.new(stream_path, stream_usage.label)
+        stream_scope = Incremental::StreamExecutionScope.new(
+          parent_field: exec_field,
+          parent_type:,
+          selections:,
+          objects:,
+          results:,
+          items:,
+          object_paths:,
+          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
+
       #: (
       #|   ExecutionField[untyped] exec_field,
       #|   untyped current_type,
@@ -987,47 +1349,172 @@ 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
-          run!(@planner.plan_scopes(ready_scopes))
-          has_errors = @invalidated_results.size > initial_error_count
-
-          ready_scopes.each do |exec_scope|
-            deliveries = @incremental.deliveries_for(exec_scope)
-            deliveries.each do |index, path, deferred_deliveries|
-              data = exec_scope.results[index]
-              errors = EMPTY_ARRAY
-              if has_errors
-                data, errors = error_result_formatter.format_object(exec_scope.parent_type, exec_scope.selections, data, path)
-              end
-
-              if data.nil? && !errors.empty?
-                deferred_deliveries.each { (completed_errors_by_delivery[_1] ||= []).concat(errors) }
+          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]
+
+          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
+                  next_installment_streams << exec_scope
+                end
               else
-                incremental_payloads << @incremental.incremental_payload(deferred_deliveries, path, data, errors:)
+                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
               end
-              completed_deliveries.concat(deferred_deliveries)
+
+              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]]
+      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
diff --git a/lib/graphql/breadth/executor/execution_field.rb b/lib/graphql/breadth/executor/execution_field.rb
index 5455e8b..73739b3 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? && 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/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/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/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/field_resolvers.rb b/lib/graphql/breadth/field_resolvers.rb
index d32f88e..32da439 100644
--- a/lib/graphql/breadth/field_resolvers.rb
+++ b/lib/graphql/breadth/field_resolvers.rb
@@ -14,6 +14,19 @@ def resolve(exec_field, ctx)
         raise NotImplementedError, "FieldResolver#resolve must be implemented."
       end
 
+      #: -> bool
+      def stream?
+        false
+      end
+
+      #: (
+      #|   Executor::ListStreamField,
+      #|   GraphQL::Query::Context,
+      #| ) -> (Array[untyped] | Executor::ExecutionPromise)
+      def resolve_list_stream(_field, _ctx)
+        raise NotImplementedError, "FieldResolver#resolve_list_stream must be implemented."
+      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.rb b/lib/graphql/breadth/incremental.rb
index aefd681..b05aa0d 100644
--- a/lib/graphql/breadth/incremental.rb
+++ b/lib/graphql/breadth/incremental.rb
@@ -2,8 +2,13 @@
 # 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"
 require_relative "incremental/result"
diff --git a/lib/graphql/breadth/incremental/context.rb b/lib/graphql/breadth/incremental/context.rb
index c334b38..4aed528 100644
--- a/lib/graphql/breadth/incremental/context.rb
+++ b/lib/graphql/breadth/incremental/context.rb
@@ -19,11 +19,12 @@ def initialize(executor, data:)
           @executor = executor
           @data = data
           @publisher = Publisher.new
-          @deferred_scopes = []
-          @pending_deliveries = []
-          @announced_deliveries = {}.compare_by_identity
-          @completed_deliveries = {}.compare_by_identity
-          @deliveries_by_usage = {}.compare_by_identity
+          @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
@@ -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,11 @@ def active?
 
         #: -> bool
         def deferred?
-          @deferred_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]
+        #: -> Array[Delivery]
         def prepare_pending
           @deferred_scopes.each do |deferred_scope|
             next if deferred_scope.announced? || !deferred_scope.ready?
@@ -53,6 +60,16 @@ def prepare_pending
             pending_deliveries_for(deferred_scope)
             deferred_scope.announced = true
           end
+          @stream_scopes.each do |stream_scope|
+            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]
+              @announced_deliveries[delivery] = true
+              @pending_deliveries << delivery
+            end
+            stream_scope.announced = true
+          end
 
           @pending_deliveries.uniq!
           pending = @pending_deliveries
@@ -60,9 +77,13 @@ def prepare_pending
           pending
         end
 
-        #: -> Array[DeferredExecutionScope]
+        #: -> Array[DeferredExecutionScope | StreamExecutionScope]
         def ready_scopes
-          @deferred_scopes.select { _1.announced? && !_1.executed? && _1.ready? }.each(&:prepare!)
+          ready = (
+            @deferred_scopes.select { _1.announced? && !_1.executed? && _1.ready? } +
+            @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]]]
@@ -80,7 +101,7 @@ def deliveries_for(deferred_scope)
           deliveries
         end
 
-        #: (Array[DeferredDelivery]) -> Array[graphql_result]
+        #: (Array[Delivery]) -> Array[graphql_result]
         def pending_payloads(deliveries)
           @publisher.pending(deliveries)
         end
@@ -90,7 +111,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[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 a66254f..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]) -> Array[graphql_result]
+        #: (Array[Delivery]) -> 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
+
+        #: (Delivery, ?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
+        #: (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
new file mode 100644
index 0000000..7799f7b
--- /dev/null
+++ b/lib/graphql/breadth/incremental/stream_delivery.rb
@@ -0,0 +1,11 @@
+# typed: true
+# frozen_string_literal: true
+
+module GraphQL
+  module Breadth
+    module Incremental
+      class StreamDelivery < Delivery
+      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..a5dd014
--- /dev/null
+++ b/lib/graphql/breadth/incremental/stream_execution_scope.rb
@@ -0,0 +1,150 @@
+# 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
+
+        #: Array[error_path]
+        attr_reader :object_paths
+
+        #: Integer
+        attr_reader :initial_index
+
+        #: bool
+        attr_writer :announced
+
+        #: bool
+        attr_reader :prepared
+
+        #: ListStreamEntry?
+        attr_reader :list_stream_entry
+
+        #: (
+        #|   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,
+        #|   ?prepare: Proc?,
+        #|   ?list_stream_entry: ListStreamEntry?,
+        #| ) -> void
+        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(
+            executor: parent_field.executor,
+            parent_type:,
+            selections:,
+            objects:,
+            results:,
+            parent_field:,
+            deferred: true,
+          )
+        end
+
+        #: -> bool
+        def ready?
+          field = parent_field #: as !nil
+          !complete? && field.scope.executed? && !field.scope.aborted?
+        end
+
+        #: -> bool
+        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]
+        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
+
+        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
+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/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/test/graphql/breadth/executor/incremental_test.rb b/test/graphql/breadth/executor/incremental_test.rb
index db0497e..0d31d0c 100644
--- a/test/graphql/breadth/executor/incremental_test.rb
+++ b/test/graphql/breadth/executor/incremental_test.rb
@@ -30,6 +30,109 @@ def resolve(exec_field, _ctx)
     end
   end
 
+  class PagedProductNodesResolver < GraphQL::Breadth::FieldResolver
+    attr_reader :calls
+
+    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(field, _ctx)
+      field.state[:calls] = @calls
+      @calls << {
+        limit: field.limit,
+        iteration: field.iteration,
+        object_count: field.pending_entries.length,
+        arguments: field.arguments,
+      }
+
+      field.pending_entries.map do |entry|
+        object = entry.object
+        nodes = object.fetch("nodes")
+        object_state = entry.object_state
+        offset = object_state[:offset] || 0
+        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
+
+        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(field, _ctx)
+      field
+        .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)
+          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)
+      exec_field.map_objects { _1["nodes"] }
+    end
+
+    def resolve_list_stream(field, _ctx)
+      field.state[:calls] = @calls
+      @calls << {
+        limit: field.limit,
+        iteration: field.iteration,
+        objects: field.pending_entries.map { _1.object["name"] },
+        arguments: field.arguments,
+      }
+
+      field.pending_entries.map do |entry|
+        object = entry.object
+        nodes = object.fetch("nodes")
+        object_state = entry.object_state
+        offset = object_state[:offset] || 0
+        page_size = field.limit || 1
+        items = nodes.slice(offset, page_size) || []
+        object_state[:offset] = offset + items.length
+
+        if field.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
+
   SOURCE = {
     "products" => {
       "nodes" => [{
@@ -56,6 +159,38 @@ def resolve(exec_field, _ctx)
     },
   }.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) {
@@ -913,8 +1048,265 @@ 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) {
+          id
+          title
+        }
+      }
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).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
+    resolver = PagedProductNodesResolver.new
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes @stream {
+          id
+        }
+      }
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).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_batches_opt_in_list_streams_and_drops_completed_objects
+    resolver = PagedVariantNodesResolver.new
+    resolvers = BREADTH_RESOLVERS.merge(
+      "VariantConnection" => BREADTH_RESOLVERS.fetch("VariantConnection").merge(
+        "nodes" => resolver,
+      ),
+    )
+
+    result = build_executor(%|{
+      products(first: 2) {
+        nodes {
+          id
+          variants(first: 3) {
+            nodes @stream(initialCount: 1) {
+              id
+            }
+          }
+        }
+      }
+    }|, source: PAGED_VARIANTS_SOURCE, resolvers:).incremental_result
+
+    assert_equal(
+      {
+        "data" => {
+          "products" => {
+            "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", 0, "variants", "nodes"] },
+          { "id" => "1", "path" => ["products", "nodes", 1, "variants", "nodes"] },
+        ],
+        "hasNext" => true,
+      },
+      result.initial_result,
+    )
+    assert_equal(
+      [{
+        "incremental" => [{
+          "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(
+      [{
+        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: {},
+      }],
+      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
+        }
+      }
+    }|, resolvers: stream_product_nodes_resolvers(resolver)).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 = stream_product_nodes_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
+
+  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,