diff --git a/src/execution/__tests__/flattenAsyncIterable-test.ts b/src/execution/__tests__/flattenAsyncIterable-test.ts
deleted file mode 100644
index d508f543f9..0000000000
--- a/src/execution/__tests__/flattenAsyncIterable-test.ts
+++ /dev/null
@@ -1,149 +0,0 @@
-import { expect } from 'chai';
-import { describe, it } from 'mocha';
-
-import { flattenAsyncIterable } from '../flattenAsyncIterable.js';
-
-describe('flattenAsyncIterable', () => {
-  it('flatten nested async generators', async () => {
-    async function* source() {
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(1.1);
-          yield await Promise.resolve(1.2);
-        })(),
-      );
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(2.1);
-          yield await Promise.resolve(2.2);
-        })(),
-      );
-    }
-
-    const doubles = flattenAsyncIterable(source());
-
-    const result = [];
-    for await (const x of doubles) {
-      result.push(x);
-    }
-    expect(result).to.deep.equal([1.1, 1.2, 2.1, 2.2]);
-  });
-
-  it('allows returning early from a nested async generator', async () => {
-    async function* source() {
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(1.1);
-          yield await Promise.resolve(1.2);
-        })(),
-      );
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(2.1); /* c8 ignore start */
-          // Not reachable, early return
-          yield await Promise.resolve(2.2);
-        })(),
-      );
-      // Not reachable, early return
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(3.1);
-          yield await Promise.resolve(3.2);
-        })(),
-      );
-    }
-    /* c8 ignore stop */
-
-    const doubles = flattenAsyncIterable(source());
-
-    expect(await doubles.next()).to.deep.equal({ value: 1.1, done: false });
-    expect(await doubles.next()).to.deep.equal({ value: 1.2, done: false });
-    expect(await doubles.next()).to.deep.equal({ value: 2.1, done: false });
-
-    // Early return
-    expect(await doubles.return()).to.deep.equal({
-      value: undefined,
-      done: true,
-    });
-
-    // Subsequent next calls
-    expect(await doubles.next()).to.deep.equal({
-      value: undefined,
-      done: true,
-    });
-    expect(await doubles.next()).to.deep.equal({
-      value: undefined,
-      done: true,
-    });
-  });
-
-  it('allows throwing errors from a nested async generator', async () => {
-    async function* source() {
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(1.1);
-          yield await Promise.resolve(1.2);
-        })(),
-      );
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(2.1); /* c8 ignore start */
-          // Not reachable, early return
-          yield await Promise.resolve(2.2);
-        })(),
-      );
-      // Not reachable, early return
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(3.1);
-          yield await Promise.resolve(3.2);
-        })(),
-      );
-    }
-    /* c8 ignore stop */
-
-    const doubles = flattenAsyncIterable(source());
-
-    expect(await doubles.next()).to.deep.equal({ value: 1.1, done: false });
-    expect(await doubles.next()).to.deep.equal({ value: 1.2, done: false });
-    expect(await doubles.next()).to.deep.equal({ value: 2.1, done: false });
-
-    // Throw error
-    let caughtError;
-    try {
-      await doubles.throw('ouch'); /* c8 ignore start */
-    } catch (e) {
-      caughtError = e;
-    }
-    expect(caughtError).to.equal('ouch');
-  });
-  it('completely yields sub-iterables even when next() called in parallel', async () => {
-    async function* source() {
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(1.1);
-          yield await Promise.resolve(1.2);
-        })(),
-      );
-      yield await Promise.resolve(
-        (async function* nested(): AsyncGenerator<number, void, void> {
-          yield await Promise.resolve(2.1);
-          yield await Promise.resolve(2.2);
-        })(),
-      );
-    }
-
-    const result = flattenAsyncIterable(source());
-
-    const promise1 = result.next();
-    const promise2 = result.next();
-    expect(await promise1).to.deep.equal({ value: 1.1, done: false });
-    expect(await promise2).to.deep.equal({ value: 1.2, done: false });
-    expect(await result.next()).to.deep.equal({ value: 2.1, done: false });
-    expect(await result.next()).to.deep.equal({ value: 2.2, done: false });
-    expect(await result.next()).to.deep.equal({
-      value: undefined,
-      done: true,
-    });
-  });
-});
diff --git a/src/execution/__tests__/subscribe-test.ts b/src/execution/__tests__/subscribe-test.ts
index 0dc2e3140c..6a903450d9 100644
--- a/src/execution/__tests__/subscribe-test.ts
+++ b/src/execution/__tests__/subscribe-test.ts
@@ -21,11 +21,7 @@ import {
 import { GraphQLSchema } from '../../type/schema.js';
 
 import type { ExecutionArgs, ExecutionResult } from '../execute.js';
-import {
-  createSourceEventStream,
-  experimentalSubscribeIncrementally,
-  subscribe,
-} from '../execute.js';
+import { createSourceEventStream, subscribe } from '../execute.js';
 
 import { SimplePubSub } from './simplePubSub.js';
 
@@ -99,10 +95,14 @@ const emailSchema = new GraphQLSchema({
 function createSubscription(
   pubsub: SimplePubSub<Email>,
   variableValues?: { readonly [variable: string]: unknown },
-  originalSubscribe: boolean = false,
 ) {
   const document = parse(`
-    subscription ($priority: Int = 0, $shouldDefer: Boolean = false, $asyncResolver: Boolean = false) {
+    subscription (
+      $priority: Int = 0
+      $shouldDefer: Boolean = false
+      $shouldStream: Boolean = false
+      $asyncResolver: Boolean = false
+    ) {
       importantEmail(priority: $priority) {
         email {
           from
@@ -113,6 +113,7 @@ function createSubscription(
         }
         ... @defer(if: $shouldDefer) {
           inbox {
+            emails @include(if: $shouldStream) @stream(if: $shouldStream)
             unread
             total
           }
@@ -145,7 +146,7 @@ function createSubscription(
     }),
   };
 
-  return (originalSubscribe ? subscribe : experimentalSubscribeIncrementally)({
+  return subscribe({
     schema: emailSchema,
     document,
     rootValue: data,
@@ -703,7 +704,7 @@ describe('Subscription Publish Phase', () => {
     });
   });
 
-  it('produces additional payloads for subscriptions with @defer', async () => {
+  it('subscribe function returns errors with @defer', async () => {
     const pubsub = new SimplePubSub<Email>();
     const subscription = await createSubscription(pubsub, {
       shouldDefer: true,
@@ -722,40 +723,23 @@ describe('Subscription Publish Phase', () => {
       }),
     ).to.equal(true);
 
-    // The previously waited on payload now has a value.
-    expect(await payload).to.deep.equal({
-      done: false,
-      value: {
-        data: {
-          importantEmail: {
-            email: {
-              from: 'yuzhi@graphql.org',
-              subject: 'Alright',
-            },
-          },
-        },
-        hasNext: true,
-      },
-    });
-
-    // Wait for the next payload from @defer
-    expect(await subscription.next()).to.deep.equal({
+    const errorPayload = {
       done: false,
       value: {
-        incremental: [
+        errors: [
           {
-            data: {
-              inbox: {
-                unread: 1,
-                total: 2,
-              },
-            },
+            message:
+              '`@defer` directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+            locations: [{ line: 8, column: 7 }],
             path: ['importantEmail'],
           },
         ],
-        hasNext: false,
+        data: { importantEmail: null },
       },
-    });
+    };
+
+    // The previously waited on payload now has a value.
+    expectJSON(await payload).toDeepEqual(errorPayload);
 
     // Another new email arrives, after all incrementally delivered payloads are received.
     expect(
@@ -768,88 +752,25 @@ describe('Subscription Publish Phase', () => {
     ).to.equal(true);
 
     // The next waited on payload will have a value.
-    expect(await subscription.next()).to.deep.equal({
-      done: false,
-      value: {
-        data: {
-          importantEmail: {
-            email: {
-              from: 'hyo@graphql.org',
-              subject: 'Tools',
-            },
-          },
-        },
-        hasNext: true,
-      },
-    });
-
-    // Another new email arrives, before the incrementally delivered payloads from the last email was received.
-    expect(
-      pubsub.emit({
-        from: 'adam@graphql.org',
-        subject: 'Important',
-        message: 'Read me please',
-        unread: true,
-      }),
-    ).to.equal(true);
-
-    // Deferred payload from previous event is received.
-    expect(await subscription.next()).to.deep.equal({
-      done: false,
-      value: {
-        incremental: [
-          {
-            data: {
-              inbox: {
-                unread: 2,
-                total: 3,
-              },
-            },
-            path: ['importantEmail'],
-          },
-        ],
-        hasNext: false,
-      },
-    });
-
-    // Next payload from last event
-    expect(await subscription.next()).to.deep.equal({
-      done: false,
-      value: {
-        data: {
-          importantEmail: {
-            email: {
-              from: 'adam@graphql.org',
-              subject: 'Important',
-            },
-          },
-        },
-        hasNext: true,
-      },
-    });
+    expectJSON(await subscription.next()).toDeepEqual(errorPayload);
 
-    // The client disconnects before the deferred payload is consumed.
-    expect(await subscription.return()).to.deep.equal({
+    expectJSON(await subscription.return()).toDeepEqual({
       done: true,
       value: undefined,
     });
 
     // Awaiting a subscription after closing it results in completed results.
-    expect(await subscription.next()).to.deep.equal({
+    expectJSON(await subscription.next()).toDeepEqual({
       done: true,
       value: undefined,
     });
   });
 
-  it('original subscribe function returns errors with @defer', async () => {
+  it('subscribe function returns errors with @stream', async () => {
     const pubsub = new SimplePubSub<Email>();
-    const subscription = await createSubscription(
-      pubsub,
-      {
-        shouldDefer: true,
-      },
-      true,
-    );
+    const subscription = await createSubscription(pubsub, {
+      shouldStream: true,
+    });
     assert(isAsyncIterable(subscription));
     // Wait for the next subscription payload.
     const payload = subscription.next();
@@ -864,23 +785,26 @@ describe('Subscription Publish Phase', () => {
       }),
     ).to.equal(true);
 
-    const errorPayload = {
+    // The previously waited on payload now has a value.
+    expectJSON(await payload).toDeepEqual({
       done: false,
       value: {
         errors: [
           {
             message:
-              'Executing this GraphQL operation would unexpectedly produce multiple payloads (due to @defer or @stream directive)',
+              '`@stream` directive not supported on subscription operations. Disable `@stream` by setting the `if` argument to `false`.',
+            locations: [{ line: 18, column: 13 }],
+            path: ['importantEmail', 'inbox', 'emails'],
           },
         ],
+        data: {
+          importantEmail: {
+            email: { from: 'yuzhi@graphql.org', subject: 'Alright' },
+            inbox: { emails: null, unread: 1, total: 2 },
+          },
+        },
       },
-    };
-
-    // The previously waited on payload now has a value.
-    expectJSON(await payload).toDeepEqual(errorPayload);
-
-    // Wait for the next payload from @defer
-    expectJSON(await subscription.next()).toDeepEqual(errorPayload);
+    });
 
     // Another new email arrives, after all incrementally delivered payloads are received.
     expect(
@@ -893,11 +817,26 @@ describe('Subscription Publish Phase', () => {
     ).to.equal(true);
 
     // The next waited on payload will have a value.
-    expectJSON(await subscription.next()).toDeepEqual(errorPayload);
-    // The next waited on payload will have a value.
-    expectJSON(await subscription.next()).toDeepEqual(errorPayload);
+    expectJSON(await subscription.next()).toDeepEqual({
+      done: false,
+      value: {
+        errors: [
+          {
+            message:
+              '`@stream` directive not supported on subscription operations. Disable `@stream` by setting the `if` argument to `false`.',
+            locations: [{ line: 18, column: 13 }],
+            path: ['importantEmail', 'inbox', 'emails'],
+          },
+        ],
+        data: {
+          importantEmail: {
+            email: { from: 'hyo@graphql.org', subject: 'Tools' },
+            inbox: { emails: null, unread: 2, total: 3 },
+          },
+        },
+      },
+    });
 
-    // The client disconnects before the deferred payload is consumed.
     expectJSON(await subscription.return()).toDeepEqual({
       done: true,
       value: undefined,
diff --git a/src/execution/collectFields.ts b/src/execution/collectFields.ts
index 098e1f45b5..17468b791f 100644
--- a/src/execution/collectFields.ts
+++ b/src/execution/collectFields.ts
@@ -1,4 +1,5 @@
 import { AccumulatorMap } from '../jsutils/AccumulatorMap.js';
+import { invariant } from '../jsutils/invariant.js';
 import type { ObjMap } from '../jsutils/ObjMap.js';
 
 import type {
@@ -6,8 +7,10 @@ import type {
   FragmentDefinitionNode,
   FragmentSpreadNode,
   InlineFragmentNode,
+  OperationDefinitionNode,
   SelectionSetNode,
 } from '../language/ast.js';
+import { OperationTypeNode } from '../language/ast.js';
 import { Kind } from '../language/kinds.js';
 
 import type { GraphQLObjectType } from '../type/definition.js';
@@ -47,7 +50,7 @@ export function collectFields(
   fragments: ObjMap<FragmentDefinitionNode>,
   variableValues: { [variable: string]: unknown },
   runtimeType: GraphQLObjectType,
-  selectionSet: SelectionSetNode,
+  operation: OperationDefinitionNode,
 ): FieldsAndPatches {
   const fields = new AccumulatorMap<string, FieldNode>();
   const patches: Array<PatchFields> = [];
@@ -55,8 +58,9 @@ export function collectFields(
     schema,
     fragments,
     variableValues,
+    operation,
     runtimeType,
-    selectionSet,
+    operation.selectionSet,
     fields,
     patches,
     new Set(),
@@ -74,10 +78,12 @@ export function collectFields(
  *
  * @internal
  */
+// eslint-disable-next-line max-params
 export function collectSubfields(
   schema: GraphQLSchema,
   fragments: ObjMap<FragmentDefinitionNode>,
   variableValues: { [variable: string]: unknown },
+  operation: OperationDefinitionNode,
   returnType: GraphQLObjectType,
   fieldNodes: ReadonlyArray<FieldNode>,
 ): FieldsAndPatches {
@@ -96,6 +102,7 @@ export function collectSubfields(
         schema,
         fragments,
         variableValues,
+        operation,
         returnType,
         node.selectionSet,
         subFieldNodes,
@@ -112,6 +119,7 @@ function collectFieldsImpl(
   schema: GraphQLSchema,
   fragments: ObjMap<FragmentDefinitionNode>,
   variableValues: { [variable: string]: unknown },
+  operation: OperationDefinitionNode,
   runtimeType: GraphQLObjectType,
   selectionSet: SelectionSetNode,
   fields: AccumulatorMap<string, FieldNode>,
@@ -135,7 +143,7 @@ function collectFieldsImpl(
           continue;
         }
 
-        const defer = getDeferValues(variableValues, selection);
+        const defer = getDeferValues(operation, variableValues, selection);
 
         if (defer) {
           const patchFields = new AccumulatorMap<string, FieldNode>();
@@ -143,6 +151,7 @@ function collectFieldsImpl(
             schema,
             fragments,
             variableValues,
+            operation,
             runtimeType,
             selection.selectionSet,
             patchFields,
@@ -158,6 +167,7 @@ function collectFieldsImpl(
             schema,
             fragments,
             variableValues,
+            operation,
             runtimeType,
             selection.selectionSet,
             fields,
@@ -174,7 +184,7 @@ function collectFieldsImpl(
           continue;
         }
 
-        const defer = getDeferValues(variableValues, selection);
+        const defer = getDeferValues(operation, variableValues, selection);
         if (visitedFragmentNames.has(fragName) && !defer) {
           continue;
         }
@@ -197,6 +207,7 @@ function collectFieldsImpl(
             schema,
             fragments,
             variableValues,
+            operation,
             runtimeType,
             fragment.selectionSet,
             patchFields,
@@ -212,6 +223,7 @@ function collectFieldsImpl(
             schema,
             fragments,
             variableValues,
+            operation,
             runtimeType,
             fragment.selectionSet,
             fields,
@@ -231,6 +243,7 @@ function collectFieldsImpl(
  * not disabled by the "if" argument.
  */
 function getDeferValues(
+  operation: OperationDefinitionNode,
   variableValues: { [variable: string]: unknown },
   node: FragmentSpreadNode | InlineFragmentNode,
 ): undefined | { label: string | undefined } {
@@ -244,6 +257,11 @@ function getDeferValues(
     return;
   }
 
+  invariant(
+    operation.operation !== OperationTypeNode.SUBSCRIPTION,
+    '`@defer` directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+  );
+
   return {
     label: typeof defer.label === 'string' ? defer.label : undefined,
   };
diff --git a/src/execution/execute.ts b/src/execution/execute.ts
index a7ff5ce607..052cb8da25 100644
--- a/src/execution/execute.ts
+++ b/src/execution/execute.ts
@@ -52,7 +52,6 @@ import {
   collectFields,
   collectSubfields as _collectSubfields,
 } from './collectFields.js';
-import { flattenAsyncIterable } from './flattenAsyncIterable.js';
 import { mapAsyncIterable } from './mapAsyncIterable.js';
 import {
   getArgumentValues,
@@ -79,6 +78,7 @@ const collectSubfields = memoize3(
       exeContext.schema,
       exeContext.fragments,
       exeContext.variableValues,
+      exeContext.operation,
       returnType,
       fieldNodes,
     ),
@@ -532,7 +532,7 @@ function executeOperation(
     fragments,
     variableValues,
     rootType,
-    operation.selectionSet,
+    operation,
   );
   const path = undefined;
   let result;
@@ -993,6 +993,11 @@ function getStreamValues(
     'initialCount must be a positive integer',
   );
 
+  invariant(
+    exeContext.operation.operation !== OperationTypeNode.SUBSCRIPTION,
+    '`@stream` directive not supported on subscription operations. Disable `@stream` by setting the `if` argument to `false`.',
+  );
+
   return {
     initialCount: stream.initialCount,
     label: typeof stream.label === 'string' ? stream.label : undefined,
@@ -1551,11 +1556,8 @@ export const defaultFieldResolver: GraphQLFieldResolver<unknown, unknown> =
  *
  * This function does not support incremental delivery (`@defer` and `@stream`).
  * If an operation which would defer or stream data is executed with this
- * function, each `InitialIncrementalExecutionResult` and
- * `SubsequentIncrementalExecutionResult` in the result stream will be replaced
- * with an `ExecutionResult` with a single error stating that defer/stream is
- * not supported.  Use `experimentalSubscribeIncrementally` if you want to
- * support incremental delivery.
+ * function, a field error will be raised at the location of the `@defer` or
+ * `@stream` directive.
  *
  * Accepts an object with named arguments.
  */
@@ -1563,78 +1565,6 @@ export function subscribe(
   args: ExecutionArgs,
 ): PromiseOrValue<
   AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
-> {
-  const maybePromise = experimentalSubscribeIncrementally(args);
-  if (isPromise(maybePromise)) {
-    return maybePromise.then((resultOrIterable) =>
-      isAsyncIterable(resultOrIterable)
-        ? mapAsyncIterable(resultOrIterable, ensureSingleExecutionResult)
-        : resultOrIterable,
-    );
-  }
-  return isAsyncIterable(maybePromise)
-    ? mapAsyncIterable(maybePromise, ensureSingleExecutionResult)
-    : maybePromise;
-}
-
-function ensureSingleExecutionResult(
-  result:
-    | ExecutionResult
-    | InitialIncrementalExecutionResult
-    | SubsequentIncrementalExecutionResult,
-): ExecutionResult {
-  if ('hasNext' in result) {
-    return {
-      errors: [new GraphQLError(UNEXPECTED_MULTIPLE_PAYLOADS)],
-    };
-  }
-  return result;
-}
-
-/**
- * Implements the "Subscribe" algorithm described in the GraphQL specification,
- * including `@defer` and `@stream` as proposed in
- * https://github.com/graphql/graphql-spec/pull/742
- *
- * Returns a Promise which resolves to either an AsyncIterator (if successful)
- * or an ExecutionResult (error). The promise will be rejected if the schema or
- * other arguments to this function are invalid, or if the resolved event stream
- * is not an async iterable.
- *
- * If the client-provided arguments to this function do not result in a
- * compliant subscription, a GraphQL Response (ExecutionResult) with descriptive
- * errors and no data will be returned.
- *
- * If the source stream could not be created due to faulty subscription resolver
- * logic or underlying systems, the promise will resolve to a single
- * ExecutionResult containing `errors` and no `data`.
- *
- * If the operation succeeded, the promise resolves to an AsyncIterator, which
- * yields a stream of result representing the response stream.
- *
- * Each result may be an ExecutionResult with no `hasNext` (if executing the
- * event did not use `@defer` or `@stream`), or an
- * `InitialIncrementalExecutionResult` or `SubsequentIncrementalExecutionResult`
- * (if executing the event used `@defer` or `@stream`). In the case of
- * incremental execution results, each event produces a single
- * `InitialIncrementalExecutionResult` followed by one or more
- * `SubsequentIncrementalExecutionResult`s; all but the last have `hasNext: true`,
- * and the last has `hasNext: false`. There is no interleaving between results
- * generated from the same original event.
- *
- * Accepts an object with named arguments.
- */
-export function experimentalSubscribeIncrementally(
-  args: ExecutionArgs,
-): PromiseOrValue<
-  | AsyncGenerator<
-      | ExecutionResult
-      | InitialIncrementalExecutionResult
-      | SubsequentIncrementalExecutionResult,
-      void,
-      void
-    >
-  | ExecutionResult
 > {
   // If a valid execution context cannot be created due to incorrect arguments,
   // a "Response" with only errors is returned.
@@ -1656,37 +1586,10 @@ export function experimentalSubscribeIncrementally(
   return mapSourceToResponse(exeContext, resultOrStream);
 }
 
-async function* ensureAsyncIterable(
-  someExecutionResult:
-    | ExecutionResult
-    | ExperimentalIncrementalExecutionResults,
-): AsyncGenerator<
-  | ExecutionResult
-  | InitialIncrementalExecutionResult
-  | SubsequentIncrementalExecutionResult,
-  void,
-  void
-> {
-  if ('initialResult' in someExecutionResult) {
-    yield someExecutionResult.initialResult;
-    yield* someExecutionResult.subsequentResults;
-  } else {
-    yield someExecutionResult;
-  }
-}
-
 function mapSourceToResponse(
   exeContext: ExecutionContext,
   resultOrStream: ExecutionResult | AsyncIterable<unknown>,
-):
-  | AsyncGenerator<
-      | ExecutionResult
-      | InitialIncrementalExecutionResult
-      | SubsequentIncrementalExecutionResult,
-      void,
-      void
-    >
-  | ExecutionResult {
+): AsyncGenerator<ExecutionResult, void, void> | ExecutionResult {
   if (!isAsyncIterable(resultOrStream)) {
     return resultOrStream;
   }
@@ -1697,12 +1600,15 @@ function mapSourceToResponse(
   // the GraphQL specification. The `execute` function provides the
   // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
   // "ExecuteQuery" algorithm, for which `execute` is also used.
-  return flattenAsyncIterable(
-    mapAsyncIterable(resultOrStream, async (payload: unknown) =>
-      ensureAsyncIterable(
-        await executeImpl(buildPerEventExecutionContext(exeContext, payload)),
-      ),
-    ),
+  return mapAsyncIterable(
+    resultOrStream,
+    (payload: unknown) =>
+      executeImpl(
+        buildPerEventExecutionContext(exeContext, payload),
+        // typecast to ExecutionResult, not possible to return
+        // ExperimentalIncrementalExecutionResults when
+        // exeContext.operation is 'subscription'.
+      ) as ExecutionResult,
   );
 }
 
@@ -1783,7 +1689,7 @@ function executeSubscription(
     fragments,
     variableValues,
     rootType,
-    operation.selectionSet,
+    operation,
   );
 
   const firstRootField = rootFields.entries().next().value;
diff --git a/src/execution/flattenAsyncIterable.ts b/src/execution/flattenAsyncIterable.ts
deleted file mode 100644
index 22bdb02338..0000000000
--- a/src/execution/flattenAsyncIterable.ts
+++ /dev/null
@@ -1,105 +0,0 @@
-type AsyncIterableOrGenerator<T> =
-  | AsyncGenerator<T, void, void>
-  | AsyncIterable<T>;
-
-/**
- * Given an AsyncIterable of AsyncIterables, flatten all yielded results into a
- * single AsyncIterable.
- */
-export function flattenAsyncIterable<T>(
-  iterable: AsyncIterableOrGenerator<AsyncIterableOrGenerator<T>>,
-): AsyncGenerator<T, void, void> {
-  // You might think this whole function could be replaced with
-  //
-  //    async function* flattenAsyncIterable(iterable) {
-  //      for await (const subIterator of iterable) {
-  //        yield* subIterator;
-  //      }
-  //    }
-  //
-  // but calling `.return()` on the iterator it returns won't interrupt the `for await`.
-
-  const topIterator = iterable[Symbol.asyncIterator]();
-  let currentNestedIterator: AsyncIterator<T> | undefined;
-  let waitForCurrentNestedIterator: Promise<void> | undefined;
-  let done = false;
-
-  async function next(): Promise<IteratorResult<T, void>> {
-    if (done) {
-      return { value: undefined, done: true };
-    }
-
-    try {
-      if (!currentNestedIterator) {
-        // Somebody else is getting it already.
-        if (waitForCurrentNestedIterator) {
-          await waitForCurrentNestedIterator;
-          return await next();
-        }
-        // Nobody else is getting it. We should!
-        let resolve: () => void;
-        waitForCurrentNestedIterator = new Promise<void>((r) => {
-          resolve = r;
-        });
-        const topIteratorResult = await topIterator.next();
-        if (topIteratorResult.done) {
-          // Given that done only ever transitions from false to true,
-          // require-atomic-updates is being unnecessarily cautious.
-          // eslint-disable-next-line require-atomic-updates
-          done = true;
-          return await next();
-        }
-        // eslint is making a reasonable point here, but we've explicitly protected
-        // ourself from the race condition by ensuring that only the single call
-        // that assigns to waitForCurrentNestedIterator is allowed to assign to
-        // currentNestedIterator or waitForCurrentNestedIterator.
-        // eslint-disable-next-line require-atomic-updates
-        currentNestedIterator = topIteratorResult.value[Symbol.asyncIterator]();
-        // eslint-disable-next-line require-atomic-updates
-        waitForCurrentNestedIterator = undefined;
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        resolve!();
-        return await next();
-      }
-
-      const rememberCurrentNestedIterator = currentNestedIterator;
-      const nestedIteratorResult = await currentNestedIterator.next();
-      if (!nestedIteratorResult.done) {
-        return nestedIteratorResult;
-      }
-
-      // The nested iterator is done. If it's still the current one, make it not
-      // current. (If it's not the current one, somebody else has made us move on.)
-      if (currentNestedIterator === rememberCurrentNestedIterator) {
-        currentNestedIterator = undefined;
-      }
-      return await next();
-    } catch (err) {
-      done = true;
-      throw err;
-    }
-  }
-  return {
-    next,
-    async return() {
-      done = true;
-      await Promise.all([
-        currentNestedIterator?.return?.(),
-        topIterator.return?.(),
-      ]);
-      return { value: undefined, done: true };
-    },
-    async throw(error?: unknown): Promise<IteratorResult<T>> {
-      done = true;
-      await Promise.all([
-        currentNestedIterator?.throw?.(error),
-        topIterator.throw?.(error),
-      ]);
-      /* c8 ignore next */
-      throw error;
-    },
-    [Symbol.asyncIterator]() {
-      return this;
-    },
-  };
-}
diff --git a/src/execution/index.ts b/src/execution/index.ts
index 46a4688d59..61b170e100 100644
--- a/src/execution/index.ts
+++ b/src/execution/index.ts
@@ -8,7 +8,6 @@ export {
   defaultFieldResolver,
   defaultTypeResolver,
   subscribe,
-  experimentalSubscribeIncrementally,
 } from './execute.js';
 
 export type {
diff --git a/src/index.ts b/src/index.ts
index 5e05627fbe..d949c80d81 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -328,7 +328,6 @@ export {
   getVariableValues,
   getDirectiveValues,
   subscribe,
-  experimentalSubscribeIncrementally,
   createSourceEventStream,
 } from './execution/index.js';
 
diff --git a/src/validation/__tests__/DeferStreamDirectiveOnValidOperationsRule-test.ts b/src/validation/__tests__/DeferStreamDirectiveOnValidOperationsRule-test.ts
new file mode 100644
index 0000000000..3dee7fe5a1
--- /dev/null
+++ b/src/validation/__tests__/DeferStreamDirectiveOnValidOperationsRule-test.ts
@@ -0,0 +1,309 @@
+import { describe, it } from 'mocha';
+
+import { buildSchema } from '../../utilities/buildASTSchema.js';
+
+import { DeferStreamDirectiveOnValidOperationsRule } from '../rules/DeferStreamDirectiveOnValidOperationsRule.js';
+
+import { expectValidationErrorsWithSchema } from './harness.js';
+
+function expectErrors(queryStr: string) {
+  return expectValidationErrorsWithSchema(
+    schema,
+    DeferStreamDirectiveOnValidOperationsRule,
+    queryStr,
+  );
+}
+
+function expectValid(queryStr: string) {
+  expectErrors(queryStr).toDeepEqual([]);
+}
+
+const schema = buildSchema(`
+  type Message {
+    body: String
+    sender: String
+  }
+
+  type SubscriptionRoot {
+    subscriptionField: Message
+    subscriptionListField: [Message]
+  }
+
+  type MutationRoot {
+    mutationField: Message
+    mutationListField: [Message]
+  }
+
+  type QueryRoot {
+    message: Message
+    messages: [Message]
+  }
+
+  schema {
+    query: QueryRoot
+    mutation: MutationRoot
+    subscription: SubscriptionRoot
+  }
+`);
+
+describe('Validate: Defer/Stream directive on valid operations', () => {
+  it('Defer fragment spread nested in query operation', () => {
+    expectValid(`
+      {
+        message {
+          ...myFragment @defer
+        }
+      }
+      fragment myFragment on Message {
+        message {
+          body
+        }
+      }
+    `);
+  });
+  it('Defer inline fragment spread in query operation', () => {
+    expectValid(`
+      {
+        ... @defer {
+          message {
+            body
+          }
+        }
+      }
+    `);
+  });
+  it('Defer fragment spread on mutation field', () => {
+    expectValid(`
+      mutation {
+        mutationField {
+          ...myFragment @defer
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `);
+  });
+  it('Defer inline fragment spread on mutation field', () => {
+    expectValid(`
+      mutation {
+        mutationField {
+          ... @defer {
+            body
+          }
+        }
+      }
+    `);
+  });
+  it('Defer fragment spread on subscription field', () => {
+    expectErrors(`
+      subscription {
+        subscriptionField {
+          ...myFragment @defer
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `).toDeepEqual([
+      {
+        locations: [{ column: 25, line: 4 }],
+        message:
+          'Defer directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+      },
+    ]);
+  });
+  it('Defer fragment spread with boolean true if argument', () => {
+    expectErrors(`
+      subscription {
+        subscriptionField {
+          ...myFragment @defer(if: true)
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `).toDeepEqual([
+      {
+        locations: [{ column: 25, line: 4 }],
+        message:
+          'Defer directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+      },
+    ]);
+  });
+  it('Defer fragment spread with boolean false if argument', () => {
+    expectValid(`
+      subscription {
+        subscriptionField {
+          ...myFragment @defer(if: false)
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `);
+  });
+  it('Defer fragment spread on query in multi operation document', () => {
+    expectValid(`
+      subscription MySubscription {
+        subscriptionField {
+          ...myFragment
+        }
+      }
+      query MyQuery {
+        message {
+          ...myFragment @defer
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `);
+  });
+  it('Defer fragment spread on subscription in multi operation document', () => {
+    expectErrors(`
+      subscription MySubscription {
+        subscriptionField {
+          ...myFragment @defer
+        }
+      }
+      query MyQuery {
+        message {
+          ...myFragment @defer
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `).toDeepEqual([
+      {
+        locations: [{ column: 25, line: 4 }],
+        message:
+          'Defer directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+      },
+    ]);
+  });
+  it('Defer fragment spread with invalid if argument', () => {
+    expectErrors(`
+      subscription MySubscription {
+        subscriptionField {
+          ...myFragment @defer(if: "Oops")
+        }
+      }
+      fragment myFragment on Message {
+        body
+      }
+    `).toDeepEqual([
+      {
+        locations: [{ column: 25, line: 4 }],
+        message:
+          'Defer directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+      },
+    ]);
+  });
+  it('Stream on query field', () => {
+    expectValid(`
+      {
+        messages @stream {
+          name
+        }
+      }
+    `);
+  });
+  it('Stream on mutation field', () => {
+    expectValid(`
+      mutation {
+        mutationField {
+          messages @stream
+        }
+      }
+    `);
+  });
+  it('Stream on fragment on mutation field', () => {
+    expectValid(`
+      mutation {
+        mutationField {
+          ...myFragment
+        }
+      }
+      fragment myFragment on Message {
+        messages @stream
+      }
+    `);
+  });
+  it('Stream on subscription field', () => {
+    expectErrors(`
+      subscription {
+        subscriptionField {
+          messages @stream
+        }
+      }
+    `).toDeepEqual([
+      {
+        message:
+          'Stream directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+        locations: [{ line: 4, column: 20 }],
+      },
+    ]);
+  });
+  it('Stream on fragment on subscription field', () => {
+    expectErrors(`
+      subscription {
+        subscriptionField {
+          ...myFragment
+        }
+      }
+      fragment myFragment on Message {
+        messages @stream
+      }
+    `).toDeepEqual([
+      {
+        message:
+          'Stream directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+        locations: [{ line: 8, column: 18 }],
+      },
+    ]);
+  });
+  it('Stream on fragment on query in multi operation document', () => {
+    expectValid(`
+      subscription MySubscription {
+        subscriptionField {
+          message
+        }
+      }
+      query MyQuery {
+        message {
+          ...myFragment
+        }
+      }
+      fragment myFragment on Message {
+        messages @stream
+      }
+    `);
+  });
+  it('Stream on subscription in multi operation document', () => {
+    expectErrors(`
+      query MyQuery {
+        message {
+          ...myFragment
+        }
+      }
+      subscription MySubscription {
+        subscriptionField {
+          message {
+            ...myFragment
+          }
+        }
+      }
+      fragment myFragment on Message {
+        messages @stream
+      }
+    `).toDeepEqual([
+      {
+        message:
+          'Stream directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+        locations: [{ line: 15, column: 18 }],
+      },
+    ]);
+  });
+});
diff --git a/src/validation/index.ts b/src/validation/index.ts
index 8ddbf379ba..b0cc754490 100644
--- a/src/validation/index.ts
+++ b/src/validation/index.ts
@@ -12,6 +12,9 @@ export { DeferStreamDirectiveLabelRule } from './rules/DeferStreamDirectiveLabel
 // Spec Section: "Defer And Stream Directives Are Used On Valid Root Field"
 export { DeferStreamDirectiveOnRootFieldRule } from './rules/DeferStreamDirectiveOnRootFieldRule.js';
 
+// Spec Section: "Defer And Stream Directives Are Used On Valid Operations"
+export { DeferStreamDirectiveOnValidOperationsRule } from './rules/DeferStreamDirectiveOnValidOperationsRule.js';
+
 // Spec Section: "Executable Definitions"
 export { ExecutableDefinitionsRule } from './rules/ExecutableDefinitionsRule.js';
 
diff --git a/src/validation/rules/DeferStreamDirectiveOnValidOperationsRule.ts b/src/validation/rules/DeferStreamDirectiveOnValidOperationsRule.ts
new file mode 100644
index 0000000000..e8e6a292b6
--- /dev/null
+++ b/src/validation/rules/DeferStreamDirectiveOnValidOperationsRule.ts
@@ -0,0 +1,82 @@
+import { GraphQLError } from '../../error/GraphQLError.js';
+
+import type { DirectiveNode } from '../../language/ast.js';
+import { OperationTypeNode } from '../../language/ast.js';
+import { Kind } from '../../language/kinds.js';
+import type { ASTVisitor } from '../../language/visitor.js';
+
+import {
+  GraphQLDeferDirective,
+  GraphQLStreamDirective,
+} from '../../type/directives.js';
+
+import type { ValidationContext } from '../ValidationContext.js';
+
+function ifArgumentCanBeFalse(node: DirectiveNode): boolean {
+  const ifArgument = node.arguments?.find((arg) => arg.name.value === 'if');
+  if (!ifArgument) {
+    return false;
+  }
+  if (ifArgument.value.kind === Kind.BOOLEAN) {
+    if (ifArgument.value.value) {
+      return false;
+    }
+  } else if (ifArgument.value.kind !== Kind.VARIABLE) {
+    return false;
+  }
+  return true;
+}
+
+/**
+ * Defer And Stream Directives Are Used On Valid Operations
+ *
+ * A GraphQL document is only valid if defer directives are not used on root mutation or subscription types.
+ */
+export function DeferStreamDirectiveOnValidOperationsRule(
+  context: ValidationContext,
+): ASTVisitor {
+  const fragmentsUsedOnSubscriptions = new Set<string>();
+
+  return {
+    OperationDefinition(operation) {
+      if (operation.operation === OperationTypeNode.SUBSCRIPTION) {
+        for (const fragment of context.getRecursivelyReferencedFragments(
+          operation,
+        )) {
+          fragmentsUsedOnSubscriptions.add(fragment.name.value);
+        }
+      }
+    },
+    Directive(node, _key, _parent, _path, ancestors) {
+      const definitionNode = ancestors[2];
+
+      if (
+        'kind' in definitionNode &&
+        ((definitionNode.kind === Kind.FRAGMENT_DEFINITION &&
+          fragmentsUsedOnSubscriptions.has(definitionNode.name.value)) ||
+          (definitionNode.kind === Kind.OPERATION_DEFINITION &&
+            definitionNode.operation === OperationTypeNode.SUBSCRIPTION))
+      ) {
+        if (node.name.value === GraphQLDeferDirective.name) {
+          if (!ifArgumentCanBeFalse(node)) {
+            context.reportError(
+              new GraphQLError(
+                'Defer directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+                { nodes: node },
+              ),
+            );
+          }
+        } else if (node.name.value === GraphQLStreamDirective.name) {
+          if (!ifArgumentCanBeFalse(node)) {
+            context.reportError(
+              new GraphQLError(
+                'Stream directive not supported on subscription operations. Disable `@defer` by setting the `if` argument to `false`.',
+                { nodes: node },
+              ),
+            );
+          }
+        }
+      }
+    },
+  };
+}
diff --git a/src/validation/rules/SingleFieldSubscriptionsRule.ts b/src/validation/rules/SingleFieldSubscriptionsRule.ts
index bc9f639d9e..4a3d834124 100644
--- a/src/validation/rules/SingleFieldSubscriptionsRule.ts
+++ b/src/validation/rules/SingleFieldSubscriptionsRule.ts
@@ -46,7 +46,7 @@ export function SingleFieldSubscriptionsRule(
             fragments,
             variableValues,
             subscriptionType,
-            node.selectionSet,
+            node,
           );
           if (fields.size > 1) {
             const fieldSelectionLists = [...fields.values()];
diff --git a/src/validation/specifiedRules.ts b/src/validation/specifiedRules.ts
index ba4e70d901..60c967f8f0 100644
--- a/src/validation/specifiedRules.ts
+++ b/src/validation/specifiedRules.ts
@@ -2,6 +2,8 @@
 import { DeferStreamDirectiveLabelRule } from './rules/DeferStreamDirectiveLabelRule.js';
 // Spec Section: "Defer And Stream Directives Are Used On Valid Root Field"
 import { DeferStreamDirectiveOnRootFieldRule } from './rules/DeferStreamDirectiveOnRootFieldRule.js';
+// Spec Section: "Defer And Stream Directives Are Used On Valid Operations"
+import { DeferStreamDirectiveOnValidOperationsRule } from './rules/DeferStreamDirectiveOnValidOperationsRule.js';
 // Spec Section: "Executable Definitions"
 import { ExecutableDefinitionsRule } from './rules/ExecutableDefinitionsRule.js';
 // Spec Section: "Field Selections on Objects, Interfaces, and Unions Types"
@@ -100,6 +102,7 @@ export const specifiedRules: ReadonlyArray<ValidationRule> = Object.freeze([
   KnownDirectivesRule,
   UniqueDirectivesPerLocationRule,
   DeferStreamDirectiveOnRootFieldRule,
+  DeferStreamDirectiveOnValidOperationsRule,
   DeferStreamDirectiveLabelRule,
   StreamDirectiveOnListFieldRule,
   KnownArgumentNamesRule,