diff --git a/apollo-federation/src/query_plan/fetch_dependency_graph.rs b/apollo-federation/src/query_plan/fetch_dependency_graph.rs
index fc5b9b1608..407f61ac98 100644
--- a/apollo-federation/src/query_plan/fetch_dependency_graph.rs
+++ b/apollo-federation/src/query_plan/fetch_dependency_graph.rs
@@ -36,6 +36,7 @@ use crate::operation::ArgumentList;
 use crate::operation::ContainmentOptions;
 use crate::operation::DirectiveList;
 use crate::operation::Field;
+use crate::operation::HasSelectionKey;
 use crate::operation::InlineFragment;
 use crate::operation::InlineFragmentSelection;
 use crate::operation::Operation;
@@ -286,6 +287,12 @@ pub(crate) struct FetchDependencyGraph {
     /// Whether this fetch dependency graph has undergone optimization (e.g. transitive reduction,
     /// removing empty/useless fetches, merging fetches with the same subgraph/path).
     is_optimized: bool,
+    /// Edges that were removed during transitive reduction but crossed a defer boundary
+    /// (source.defer_ref != target.defer_ref). These are tracked so that
+    /// `extract_children_and_deferred_dependencies` can still register the correct defer
+    /// dependencies even after the direct edge has been removed from the graph.
+    #[serde(skip)]
+    reduced_defer_edges: Vec<(NodeIndex, NodeIndex)>,
 }
 
 // TODO: Write docstrings
@@ -700,6 +707,7 @@ impl FetchDependencyGraph {
             fetch_id_generation,
             is_reduced: false,
             is_optimized: false,
+            reduced_defer_edges: Vec::new(),
         }
     }
 
@@ -1166,6 +1174,19 @@ impl FetchDependencyGraph {
             self.collect_redundant_edges(node_index, &mut redundant_edges);
         }
 
+        // Before removing redundant edges, record any that cross a defer boundary.
+        // These are needed by `extract_children_and_deferred_dependencies` to register
+        // defer dependencies that would otherwise be lost after transitive reduction.
+        for &edge in &redundant_edges {
+            if let Some((source, target)) = self.graph.edge_endpoints(edge) {
+                let source_defer = &self.graph[source].defer_ref;
+                let target_defer = &self.graph[target].defer_ref;
+                if source_defer != target_defer {
+                    self.reduced_defer_edges.push((source, target));
+                }
+            }
+        }
+
         // PORT_NOTE: JS version calls `FetchGroup.removeChild`, which calls onModification.
         if !redundant_edges.is_empty() {
             self.on_modification();
@@ -1714,6 +1735,8 @@ impl FetchDependencyGraph {
             }
         }
 
+        self.collect_reduced_defer_dependencies(node_index, &mut defer_dependencies)?;
+
         for (defer_ref, dependency) in defer_dependencies {
             self.defer_tracking.add_dependency(&defer_ref, dependency);
         }
@@ -1721,6 +1744,183 @@ impl FetchDependencyGraph {
         Ok((children, deferred_nodes))
     }
 
+    /// Collect defer dependencies from edges that were removed during transitive reduction
+    /// but crossed a defer boundary.
+    ///
+    /// Without this, an ancestor fetch whose direct edge to a deferred node was reduced
+    /// away (because a transitive path exists through an intermediate fetch) would not be
+    /// registered as a dependency, causing the deferred node to miss data at runtime.
+    ///
+    /// At runtime, each registered dependency broadcasts only its own fetch result to the
+    /// deferred node. If an ancestor's edge is reduced away, its result is never broadcast,
+    /// and the deferred node loses access to fields only that ancestor provides (e.g.
+    /// `__typename`, entity keys, or other required fields at the ancestor's response path).
+    ///
+    /// To avoid adding unnecessary dependencies, we only restore the edge when the source
+    /// node's selection has a field-name intersection with the deferred target's required
+    /// inputs. If the source doesn't provide any fields the target needs, the dependency
+    /// is purely transitive and can remain reduced.
+    fn collect_reduced_defer_dependencies(
+        &self,
+        node_index: NodeIndex,
+        defer_dependencies: &mut Vec<(DeferRef, String)>,
+    ) -> Result<(), FederationError> {
+        for &(source, target) in &self.reduced_defer_edges {
+            if source != node_index {
+                continue;
+            }
+            let node = self.node_weight(source)?;
+            let child = self.node_weight(target)?;
+            if node.defer_ref == child.defer_ref {
+                continue;
+            }
+            let Some(child_defer_ref) = &child.defer_ref else {
+                continue;
+            };
+            if node.selection_set.selection_set.selections.is_empty() {
+                continue;
+            }
+
+            // Check if the source's selection provides any fields that the deferred
+            // target's inputs require (excluding __typename which is ubiquitous).
+            let Some(inputs) = &child.inputs else {
+                continue;
+            };
+
+            // Navigate the source's selection down to the target's merge_at level,
+            // so we compare fields at the same nesting depth. When both nodes have
+            // merge_at paths, skip the common prefix and navigate only the remaining
+            // suffix (e.g., child merge_at=["start","q"] and node merge_at=["start"]
+            // means we navigate the source by just ["q"]).
+            let source_selection = match (&child.merge_at, &node.merge_at) {
+                (Some(child_path), None) => {
+                    match Self::selection_set_at_path(&node.selection_set.selection_set, child_path)
+                    {
+                        Some(ss) => ss,
+                        None => continue,
+                    }
+                }
+                (Some(child_path), Some(node_path)) => {
+                    let common_len = child_path
+                        .iter()
+                        .zip(node_path.iter())
+                        .take_while(|(a, b)| a == b)
+                        .count();
+                    let suffix = &child_path[common_len..];
+                    if suffix.is_empty() {
+                        &node.selection_set.selection_set
+                    } else {
+                        match Self::selection_set_at_path(&node.selection_set.selection_set, suffix)
+                        {
+                            Some(ss) => ss,
+                            None => continue,
+                        }
+                    }
+                }
+                _ => &node.selection_set.selection_set,
+            };
+
+            let has_intersection = inputs
+                .selection_sets_per_parent_type
+                .values()
+                .any(|input_ss| Self::has_field_intersection(source_selection, input_ss));
+
+            if has_intersection {
+                let id = *node.id.get_or_init(|| self.fetch_id_generation.next_id());
+                defer_dependencies.push((child_defer_ref.clone(), format!("{id}")));
+            }
+        }
+        Ok(())
+    }
+
+    /// Navigate a selection set down a `merge_at` path, returning the sub-selection
+    /// at that path, or `None` if the path doesn't exist in the selection.
+    /// Note: This is an over-approximation, since it does not check if fragment's type condition
+    ///       is satisfied by the path's type conditions.
+    fn selection_set_at_path<'a>(
+        selection_set: &'a SelectionSet,
+        path: &[FetchDataPathElement],
+    ) -> Option<&'a SelectionSet> {
+        let mut current = selection_set;
+        for element in path {
+            match element {
+                FetchDataPathElement::Key(name, _) => {
+                    current = Self::find_field_in_selection_set(current, name)?;
+                }
+                FetchDataPathElement::AnyIndex(_) => {
+                    // Array indices don't change the selection structure.
+                    continue;
+                }
+                _ => return None,
+            }
+        }
+        Some(current)
+    }
+
+    /// Find a field by response name in a selection set, searching through any layers of
+    /// nested inline fragments. Returns the field's sub-selection set.
+    fn find_field_in_selection_set<'a>(
+        selection_set: &'a SelectionSet,
+        name: &Name,
+    ) -> Option<&'a SelectionSet> {
+        for sel in selection_set.iter() {
+            match sel {
+                Selection::Field(fs) if fs.field.response_name() == name => {
+                    return fs.selection_set.as_ref();
+                }
+                Selection::InlineFragment(inf) => {
+                    if let Some(ss) = Self::find_field_in_selection_set(&inf.selection_set, name) {
+                        return Some(ss);
+                    }
+                }
+                _ => {}
+            }
+        }
+        None
+    }
+
+    /// Check whether two selection sets share any field selections (excluding `__typename`).
+    /// When an inline fragment (type condition) is found in `b`, it is matched against
+    /// the same type condition in `a` via `SelectionMap::get()`, so that fields are only
+    /// compared under the same type context.
+    /// NOTE: This intersection check is not meant be complete. It's meant to identify matching
+    ///       defer node dependencies. We are assuming that the selection sets `a` and `b` have
+    ///       a similar structure in the first place.
+    fn has_field_intersection(a: &SelectionSet, b: &SelectionSet) -> bool {
+        for sel in b.iter() {
+            match sel {
+                Selection::Field(fs) => {
+                    if *fs.field.name() == TYPENAME_FIELD {
+                        continue;
+                    }
+                    if a.selections.get(sel.key()).is_some() {
+                        return true;
+                    }
+                    // Here, `a` may have inline fragments without a type condition that contain
+                    // `fs` theoretically. But, our query plan won't generate such inline fragments
+                    // unnecessarily.
+                }
+                Selection::InlineFragment(b_inf) => {
+                    // Try to find a matching type condition in `a` first.
+                    if let Some(Selection::InlineFragment(a_inf)) = a.selections.get(sel.key())
+                        && Self::has_field_intersection(&a_inf.selection_set, &b_inf.selection_set)
+                    {
+                        return true;
+                    }
+
+                    if b_inf.inline_fragment.casted_type() == a.type_position {
+                        // No matching inline fragment in `a`, but `a` is already
+                        // at the same type as `b`'s type condition. Compare directly.
+                        if Self::has_field_intersection(a, &b_inf.selection_set) {
+                            return true;
+                        }
+                    }
+                }
+            }
+        }
+        false
+    }
+
     fn create_state_for_children_of_processed_node(
         &self,
         processed_index: NodeIndex,
diff --git a/apollo-federation/tests/query_plan/build_query_plan_tests/defer.rs b/apollo-federation/tests/query_plan/build_query_plan_tests/defer.rs
index e11ef2b35d..881c5c9fdd 100644
--- a/apollo-federation/tests/query_plan/build_query_plan_tests/defer.rs
+++ b/apollo-federation/tests/query_plan/build_query_plan_tests/defer.rs
@@ -3528,3 +3528,555 @@ fn defer_on_renamed_root_type() {
     "###
     );
 }
+
+// Reproduces RH-1172: when a deferred node depends on a fetch that operates at a
+// deeper path than the deferred Flatten, the transitive reduction removes the direct
+// edge from an ancestor fetch (which provides __typename and entity keys) to the
+// deferred node. This causes the deferred node to only register the deeper fetch as
+// a dependency, missing the ancestor's data at runtime.
+#[test]
+fn defer_deferred_depends_on_ancestor_with_same_merge_at_intermediate() {
+    let planner = planner!(
+        config = config_with_defer(),
+        A: r#"
+        type Query { start: Q }
+        type Q @key(fields: "id") { id: ID! extra: String @shareable }
+        "#,
+        B: r#"
+        type Q @key(fields: "id") { id: ID! data: String }
+        "#,
+        C: r#"
+        type Q @key(fields: "id extra data") {
+            id: ID! @external
+            extra: String @external
+            data: String @external
+            target: R
+        }
+        type R { x: Int }
+        "#,
+    );
+
+    // Graph before reduction:
+    //   A (root, merge_at=None) → B (entity, merge_at=["start"]) → Deferred (merge_at=["start"])
+    //   A → Deferred
+    // After reduction: A→Deferred removed (transitive path A→B→Deferred).
+    // B has the same merge_at as Deferred, but B doesn't provide `extra` — only A does.
+    // So A must still be registered as a dependency.
+    assert_plan!(planner,
+        r#"
+        query {
+          start {
+            id
+            extra
+            ... @defer {
+              target {
+                x
+              }
+            }
+          }
+        }
+        "#,
+        @r###"
+    QueryPlan {
+      Defer {
+        Primary {
+          { start { id extra } }:
+          Sequence {
+            Fetch(service: "A", id: 0) {
+              {
+                start {
+                  __typename
+                  id
+                  extra
+                }
+              }
+            },
+            Flatten(path: "start") {
+              Fetch(service: "B", id: 1) {
+                {
+                  ... on Q {
+                    __typename
+                    id
+                  }
+                } =>
+                {
+                  ... on Q {
+                    data
+                  }
+                }
+              },
+            },
+          },
+        }, [
+          Deferred(depends: [0, 1], path: "start") {
+            { target { x } }:
+            Flatten(path: "start") {
+              Fetch(service: "C") {
+                {
+                  ... on Q {
+                    __typename
+                    id
+                    extra
+                    data
+                  }
+                } =>
+                {
+                  ... on Q {
+                    target {
+                      x
+                    }
+                  }
+                }
+              },
+            },
+          },
+        ]
+      },
+    }
+    "###
+    );
+}
+
+#[test]
+fn defer_deferred_depends_on_intermediate_fetch_missing_ancestor() {
+    let planner = planner!(
+        config = config_with_defer(),
+        A: r#"
+        type Query {
+            start: Q
+        }
+
+        type Q {
+            id: ID! @shareable
+            inner: Inner @shareable
+        }
+
+        type Inner @key(fields: "id") {
+            id: ID!
+        }
+        "#,
+        B: r#"
+        type Q {
+            inner: Inner @shareable
+        }
+
+        type Inner @key(fields: "id") {
+            id: ID!
+            data: ID!
+        }
+        "#,
+        C: r#"
+        type Inner {
+            data: ID! @external
+        }
+
+        type Q @key(fields: "id inner { data }") {
+            id: ID!
+            inner: Inner @external
+            target: R
+        }
+
+        type R {
+            x: Int
+        }
+        "#,
+    );
+    // The deferred block's Flatten at "start" requires `Q { __typename id inner { data } }`.
+    // Fetch A provides `start.__typename`, `start.id`, and `start.inner`.
+    // Fetch B (id: 0) provides `inner.data`.
+    // Both are needed by the deferred Flatten, so both should be in the depends list.
+    //
+    // Regression test: The transitive reduction used to remove the direct edge from A to
+    // the deferred node (since A → B → deferred is a transitive path), so only B was
+    // registered as a dependency. The deferred node was missing A's data (__typename, id)
+    // at runtime. The fix detects that B's merge_at ("start.inner") is strictly deeper
+    // than the deferred target's merge_at ("start"), so A is also registered.
+    assert_plan!(planner,
+        r#"
+        query {
+          start {
+            __typename
+            id
+            ... on Q @defer {
+              target {
+                x
+              }
+            }
+          }
+        }
+        "#,
+        @r###"
+    QueryPlan {
+      Defer {
+        Primary {
+          { start { __typename id } }:
+          Sequence {
+            Fetch(service: "A", id: 0) {
+              {
+                start {
+                  __typename
+                  id
+                  inner {
+                    __typename
+                    id
+                  }
+                }
+              }
+            },
+            Flatten(path: "start.inner") {
+              Fetch(service: "B", id: 1) {
+                {
+                  ... on Inner {
+                    __typename
+                    id
+                  }
+                } =>
+                {
+                  ... on Inner {
+                    data
+                  }
+                }
+              },
+            },
+          },
+        }, [
+          Deferred(depends: [0, 1], path: "start") {
+            { ... on Q { target { x } } }:
+            Flatten(path: "start") {
+              Fetch(service: "C") {
+                {
+                  ... on Q {
+                    __typename
+                    id
+                    inner {
+                      data
+                    }
+                  }
+                } =>
+                {
+                  ... on Q {
+                    target {
+                      x
+                    }
+                  }
+                }
+              },
+            },
+          },
+        ]
+      },
+    }
+    "###
+    );
+}
+
+// Reproduces a variant of RH-1172: when the deferred section is reached through an
+// interface type condition (e.g. `... on P { q { ... on Q @defer { ... } } }`),
+// the merge_at path contains a TypenameEquals element. The `selection_set_at_path`
+// function needs to handle this element to correctly detect that the source fetch
+// provides fields needed by the deferred node.
+#[test]
+fn defer_deferred_depends_through_type_condition_path() {
+    let planner = planner!(
+        config = config_with_defer(),
+        A: r#"
+        type Query {
+            start: I
+        }
+
+        interface I {
+            something: Int!
+        }
+
+        type P implements I {
+            something: Int!
+            q: Q
+        }
+
+        type Q {
+            id: ID! @shareable
+            inner: Inner @shareable
+        }
+
+        type Inner @key(fields: "id") {
+            id: ID!
+        }
+        "#,
+        B: r#"
+        type Q {
+            inner: Inner @shareable
+        }
+
+        type Inner @key(fields: "id") {
+            id: ID!
+            data: ID!
+        }
+        "#,
+        C: r#"
+        type Inner {
+            data: ID! @external
+        }
+
+        type Q @key(fields: "id inner { data }") {
+            id: ID!
+            inner: Inner @external
+            target: R
+        }
+
+        type R {
+            x: Int
+        }
+        "#,
+    );
+    assert_plan!(planner,
+        r#"
+        query {
+          start {
+            ... on P {
+              q {
+                __typename
+                id
+                ... on Q @defer {
+                  target {
+                    x
+                  }
+                }
+              }
+            }
+          }
+        }
+        "#,
+
+        // The deferred block depends on both fetch 0 (provides __typename, id from A)
+        // and fetch 1 (provides inner.data from B). The merge_at path for the deferred
+        // section includes a type condition, so selection_set_at_path must look inside
+        // inline fragments to find fields.
+        @r###"
+    QueryPlan {
+      Defer {
+        Primary {
+          { start { ... on P { q { __typename id } } } }:
+          Sequence {
+            Fetch(service: "A", id: 0) {
+              {
+                start {
+                  __typename
+                  ... on P {
+                    q {
+                      __typename
+                      id
+                      inner {
+                        __typename
+                        id
+                      }
+                    }
+                  }
+                }
+              }
+            },
+            Flatten(path: "start.q.inner") {
+              Fetch(service: "B", id: 1) {
+                {
+                  ... on Inner {
+                    __typename
+                    id
+                  }
+                } =>
+                {
+                  ... on Inner {
+                    data
+                  }
+                }
+              },
+            },
+          },
+        }, [
+          Deferred(depends: [0, 1], path: "start/... on P/q") {
+            { ... on Q { target { x } } }:
+            Flatten(path: "start.q") {
+              Fetch(service: "C") {
+                {
+                  ... on Q {
+                    __typename
+                    id
+                    inner {
+                      data
+                    }
+                  }
+                } =>
+                {
+                  ... on Q {
+                    target {
+                      x
+                    }
+                  }
+                }
+              },
+            },
+          },
+        ]
+      },
+    }
+    "###
+    );
+}
+
+// Reproduces a variant of RH-1172: when the source fetch and the deferred child both
+// have merge_at paths that share a common prefix, we must skip the shared prefix and
+// navigate only the remaining suffix to compare fields at the correct nesting depth.
+//
+// Here:
+//   Fetch A (root, merge_at=None): provides `start { __typename id }`
+//   Fetch B (merge_at=["start"]): provides `q { id, inner { __typename id } }`
+//   Fetch C (merge_at=["start","q","inner"]): provides `inner { data }`
+//   Deferred (merge_at=["start","q"]): needs `Q { __typename id inner { data } }`
+//
+// B (merge_at=["start"]) and Deferred (merge_at=["start","q"]) share prefix ["start"].
+// We must navigate B's selection by suffix ["q"] to find the overlapping `id` field.
+#[test]
+fn defer_deferred_depends_on_source_with_shared_merge_at_prefix() {
+    let planner = planner!(
+        config = config_with_defer(),
+        A: r#"
+        type Query {
+            start: T
+        }
+
+        type T @key(fields: "id") {
+            id: ID!
+        }
+        "#,
+        B: r#"
+        type T @key(fields: "id") {
+            id: ID!
+            q: Q
+        }
+
+        type Q {
+            id: ID! @shareable
+            inner: Inner @shareable
+        }
+
+        type Inner @key(fields: "id") {
+            id: ID!
+        }
+        "#,
+        C: r#"
+        type Inner @key(fields: "id") {
+            id: ID!
+            data: ID!
+        }
+        "#,
+        D: r#"
+        type Inner {
+            data: ID! @external
+        }
+
+        type Q @key(fields: "id inner { data }") {
+            id: ID!
+            inner: Inner @external
+            target: R
+        }
+
+        type R {
+            x: Int
+        }
+        "#,
+    );
+    assert_plan!(planner,
+        r#"
+        query {
+          start {
+            q {
+              id
+              ... on Q @defer {
+                target {
+                  x
+                }
+              }
+            }
+          }
+        }
+        "#,
+        @r###"
+    QueryPlan {
+      Defer {
+        Primary {
+          { start { q { id } } }:
+          Sequence {
+            Fetch(service: "A") {
+              {
+                start {
+                  __typename
+                  id
+                }
+              }
+            },
+            Flatten(path: "start") {
+              Fetch(service: "B", id: 0) {
+                {
+                  ... on T {
+                    __typename
+                    id
+                  }
+                } =>
+                {
+                  ... on T {
+                    q {
+                      __typename
+                      id
+                      inner {
+                        __typename
+                        id
+                      }
+                    }
+                  }
+                }
+              },
+            },
+            Flatten(path: "start.q.inner") {
+              Fetch(service: "C", id: 1) {
+                {
+                  ... on Inner {
+                    __typename
+                    id
+                  }
+                } =>
+                {
+                  ... on Inner {
+                    data
+                  }
+                }
+              },
+            },
+          },
+        }, [
+          Deferred(depends: [0, 1], path: "start/q") {
+            { ... on Q { target { x } } }:
+            Flatten(path: "start.q") {
+              Fetch(service: "D") {
+                {
+                  ... on Q {
+                    __typename
+                    id
+                    inner {
+                      data
+                    }
+                  }
+                } =>
+                {
+                  ... on Q {
+                    target {
+                      x
+                    }
+                  }
+                }
+              },
+            },
+          },
+        ]
+      },
+    }
+    "###
+    );
+}
diff --git a/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_ancestor_with_same_merge_at_intermediate.graphql b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_ancestor_with_same_merge_at_intermediate.graphql
new file mode 100644
index 0000000000..beba24e1d3
--- /dev/null
+++ b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_ancestor_with_same_merge_at_intermediate.graphql
@@ -0,0 +1,81 @@
+# Composed from subgraphs with hash: e8e4bf1b061461248ce5eb96e5d32ad453ea4ab4
+schema
+  @link(url: "https://specs.apollo.dev/link/v1.0")
+  @link(url: "https://specs.apollo.dev/join/v0.5", for: EXECUTION)
+{
+  query: Query
+}
+
+directive @join__directive(graphs: [join__Graph!], name: String!, args: join__DirectiveArguments) repeatable on SCHEMA | OBJECT | INTERFACE | FIELD_DEFINITION
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(graph: join__Graph, requires: join__FieldSet, provides: join__FieldSet, type: String, external: Boolean, override: String, usedOverridden: Boolean, overrideLabel: String, contextArguments: [join__ContextArgument!]) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(graph: join__Graph!, interface: String!) repeatable on OBJECT | INTERFACE
+
+directive @join__type(graph: join__Graph!, key: join__FieldSet, extension: Boolean! = false, resolvable: Boolean! = true, isInterfaceObject: Boolean! = false) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(graph: join__Graph!, member: String!) repeatable on UNION
+
+directive @link(url: String, as: String, for: link__Purpose, import: [link__Import]) repeatable on SCHEMA
+
+input join__ContextArgument {
+  name: String!
+  type: String!
+  context: String!
+  selection: join__FieldValue!
+}
+
+scalar join__DirectiveArguments
+
+scalar join__FieldSet
+
+scalar join__FieldValue
+
+enum join__Graph {
+  A @join__graph(name: "A", url: "none")
+  B @join__graph(name: "B", url: "none")
+  C @join__graph(name: "C", url: "none")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+  """
+  `SECURITY` features provide metadata necessary to securely resolve fields.
+  """
+  SECURITY
+
+  """
+  `EXECUTION` features provide metadata necessary for operation execution.
+  """
+  EXECUTION
+}
+
+type Q
+  @join__type(graph: A, key: "id")
+  @join__type(graph: B, key: "id")
+  @join__type(graph: C, key: "id extra data")
+{
+  id: ID! @join__field(graph: A) @join__field(graph: B) @join__field(graph: C, external: true)
+  extra: String @join__field(graph: A) @join__field(graph: C, external: true)
+  data: String @join__field(graph: B) @join__field(graph: C, external: true)
+  target: R @join__field(graph: C)
+}
+
+type Query
+  @join__type(graph: A)
+  @join__type(graph: B)
+  @join__type(graph: C)
+{
+  start: Q @join__field(graph: A)
+}
+
+type R
+  @join__type(graph: C)
+{
+  x: Int
+}
diff --git a/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_intermediate_fetch_missing_ancestor.graphql b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_intermediate_fetch_missing_ancestor.graphql
new file mode 100644
index 0000000000..f9b57d429d
--- /dev/null
+++ b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_intermediate_fetch_missing_ancestor.graphql
@@ -0,0 +1,89 @@
+# Composed from subgraphs with hash: c30e19af15626c5dcd1fd4e537aca90396054d8d
+schema
+  @link(url: "https://specs.apollo.dev/link/v1.0")
+  @link(url: "https://specs.apollo.dev/join/v0.5", for: EXECUTION)
+{
+  query: Query
+}
+
+directive @join__directive(graphs: [join__Graph!], name: String!, args: join__DirectiveArguments) repeatable on SCHEMA | OBJECT | INTERFACE | FIELD_DEFINITION
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(graph: join__Graph, requires: join__FieldSet, provides: join__FieldSet, type: String, external: Boolean, override: String, usedOverridden: Boolean, overrideLabel: String, contextArguments: [join__ContextArgument!]) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(graph: join__Graph!, interface: String!) repeatable on OBJECT | INTERFACE
+
+directive @join__type(graph: join__Graph!, key: join__FieldSet, extension: Boolean! = false, resolvable: Boolean! = true, isInterfaceObject: Boolean! = false) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(graph: join__Graph!, member: String!) repeatable on UNION
+
+directive @link(url: String, as: String, for: link__Purpose, import: [link__Import]) repeatable on SCHEMA
+
+type Inner
+  @join__type(graph: A, key: "id")
+  @join__type(graph: B, key: "id")
+  @join__type(graph: C)
+{
+  id: ID! @join__field(graph: A) @join__field(graph: B)
+  data: ID! @join__field(graph: B) @join__field(graph: C, external: true)
+}
+
+input join__ContextArgument {
+  name: String!
+  type: String!
+  context: String!
+  selection: join__FieldValue!
+}
+
+scalar join__DirectiveArguments
+
+scalar join__FieldSet
+
+scalar join__FieldValue
+
+enum join__Graph {
+  A @join__graph(name: "A", url: "none")
+  B @join__graph(name: "B", url: "none")
+  C @join__graph(name: "C", url: "none")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+  """
+  `SECURITY` features provide metadata necessary to securely resolve fields.
+  """
+  SECURITY
+
+  """
+  `EXECUTION` features provide metadata necessary for operation execution.
+  """
+  EXECUTION
+}
+
+type Q
+  @join__type(graph: A)
+  @join__type(graph: B)
+  @join__type(graph: C, key: "id inner { data }")
+{
+  id: ID! @join__field(graph: A) @join__field(graph: C)
+  inner: Inner @join__field(graph: A) @join__field(graph: B) @join__field(graph: C, external: true)
+  target: R @join__field(graph: C)
+}
+
+type Query
+  @join__type(graph: A)
+  @join__type(graph: B)
+  @join__type(graph: C)
+{
+  start: Q @join__field(graph: A)
+}
+
+type R
+  @join__type(graph: C)
+{
+  x: Int
+}
diff --git a/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_source_with_shared_merge_at_prefix.graphql b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_source_with_shared_merge_at_prefix.graphql
new file mode 100644
index 0000000000..4a24636d21
--- /dev/null
+++ b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_on_source_with_shared_merge_at_prefix.graphql
@@ -0,0 +1,98 @@
+# Composed from subgraphs with hash: 6aa847e60aab6724a547086dc69efbc505a2c015
+schema
+  @link(url: "https://specs.apollo.dev/link/v1.0")
+  @link(url: "https://specs.apollo.dev/join/v0.5", for: EXECUTION)
+{
+  query: Query
+}
+
+directive @join__directive(graphs: [join__Graph!], name: String!, args: join__DirectiveArguments) repeatable on SCHEMA | OBJECT | INTERFACE | FIELD_DEFINITION
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(graph: join__Graph, requires: join__FieldSet, provides: join__FieldSet, type: String, external: Boolean, override: String, usedOverridden: Boolean, overrideLabel: String, contextArguments: [join__ContextArgument!]) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(graph: join__Graph!, interface: String!) repeatable on OBJECT | INTERFACE
+
+directive @join__type(graph: join__Graph!, key: join__FieldSet, extension: Boolean! = false, resolvable: Boolean! = true, isInterfaceObject: Boolean! = false) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(graph: join__Graph!, member: String!) repeatable on UNION
+
+directive @link(url: String, as: String, for: link__Purpose, import: [link__Import]) repeatable on SCHEMA
+
+type Inner
+  @join__type(graph: B, key: "id")
+  @join__type(graph: C, key: "id")
+  @join__type(graph: D)
+{
+  id: ID! @join__field(graph: B) @join__field(graph: C)
+  data: ID! @join__field(graph: C) @join__field(graph: D, external: true)
+}
+
+input join__ContextArgument {
+  name: String!
+  type: String!
+  context: String!
+  selection: join__FieldValue!
+}
+
+scalar join__DirectiveArguments
+
+scalar join__FieldSet
+
+scalar join__FieldValue
+
+enum join__Graph {
+  A @join__graph(name: "A", url: "none")
+  B @join__graph(name: "B", url: "none")
+  C @join__graph(name: "C", url: "none")
+  D @join__graph(name: "D", url: "none")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+  """
+  `SECURITY` features provide metadata necessary to securely resolve fields.
+  """
+  SECURITY
+
+  """
+  `EXECUTION` features provide metadata necessary for operation execution.
+  """
+  EXECUTION
+}
+
+type Q
+  @join__type(graph: B)
+  @join__type(graph: D, key: "id inner { data }")
+{
+  id: ID!
+  inner: Inner @join__field(graph: B) @join__field(graph: D, external: true)
+  target: R @join__field(graph: D)
+}
+
+type Query
+  @join__type(graph: A)
+  @join__type(graph: B)
+  @join__type(graph: C)
+  @join__type(graph: D)
+{
+  start: T @join__field(graph: A)
+}
+
+type R
+  @join__type(graph: D)
+{
+  x: Int
+}
+
+type T
+  @join__type(graph: A, key: "id")
+  @join__type(graph: B, key: "id")
+{
+  id: ID!
+  q: Q @join__field(graph: B)
+}
diff --git a/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_through_type_condition_path.graphql b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_through_type_condition_path.graphql
new file mode 100644
index 0000000000..48196facb7
--- /dev/null
+++ b/apollo-federation/tests/query_plan/supergraphs/defer_deferred_depends_through_type_condition_path.graphql
@@ -0,0 +1,103 @@
+# Composed from subgraphs with hash: a333facb713ac38a55cef76d900765dc94ed4622
+schema
+  @link(url: "https://specs.apollo.dev/link/v1.0")
+  @link(url: "https://specs.apollo.dev/join/v0.5", for: EXECUTION)
+{
+  query: Query
+}
+
+directive @join__directive(graphs: [join__Graph!], name: String!, args: join__DirectiveArguments) repeatable on SCHEMA | OBJECT | INTERFACE | FIELD_DEFINITION
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(graph: join__Graph, requires: join__FieldSet, provides: join__FieldSet, type: String, external: Boolean, override: String, usedOverridden: Boolean, overrideLabel: String, contextArguments: [join__ContextArgument!]) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(graph: join__Graph!, interface: String!) repeatable on OBJECT | INTERFACE
+
+directive @join__type(graph: join__Graph!, key: join__FieldSet, extension: Boolean! = false, resolvable: Boolean! = true, isInterfaceObject: Boolean! = false) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(graph: join__Graph!, member: String!) repeatable on UNION
+
+directive @link(url: String, as: String, for: link__Purpose, import: [link__Import]) repeatable on SCHEMA
+
+interface I
+  @join__type(graph: A)
+{
+  something: Int!
+}
+
+type Inner
+  @join__type(graph: A, key: "id")
+  @join__type(graph: B, key: "id")
+  @join__type(graph: C)
+{
+  id: ID! @join__field(graph: A) @join__field(graph: B)
+  data: ID! @join__field(graph: B) @join__field(graph: C, external: true)
+}
+
+input join__ContextArgument {
+  name: String!
+  type: String!
+  context: String!
+  selection: join__FieldValue!
+}
+
+scalar join__DirectiveArguments
+
+scalar join__FieldSet
+
+scalar join__FieldValue
+
+enum join__Graph {
+  A @join__graph(name: "A", url: "none")
+  B @join__graph(name: "B", url: "none")
+  C @join__graph(name: "C", url: "none")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+  """
+  `SECURITY` features provide metadata necessary to securely resolve fields.
+  """
+  SECURITY
+
+  """
+  `EXECUTION` features provide metadata necessary for operation execution.
+  """
+  EXECUTION
+}
+
+type P implements I
+  @join__implements(graph: A, interface: "I")
+  @join__type(graph: A)
+{
+  something: Int!
+  q: Q
+}
+
+type Q
+  @join__type(graph: A)
+  @join__type(graph: B)
+  @join__type(graph: C, key: "id inner { data }")
+{
+  id: ID! @join__field(graph: A) @join__field(graph: C)
+  inner: Inner @join__field(graph: A) @join__field(graph: B) @join__field(graph: C, external: true)
+  target: R @join__field(graph: C)
+}
+
+type Query
+  @join__type(graph: A)
+  @join__type(graph: B)
+  @join__type(graph: C)
+{
+  start: I @join__field(graph: A)
+}
+
+type R
+  @join__type(graph: C)
+{
+  x: Int
+}
diff --git a/apollo-router/src/query_planner/testdata/defer_depends_schema.graphql b/apollo-router/src/query_planner/testdata/defer_depends_schema.graphql
new file mode 100644
index 0000000000..f848cd3780
--- /dev/null
+++ b/apollo-router/src/query_planner/testdata/defer_depends_schema.graphql
@@ -0,0 +1,87 @@
+schema
+  @link(url: "https://specs.apollo.dev/link/v1.0")
+  @link(url: "https://specs.apollo.dev/join/v0.3", for: EXECUTION) {
+  query: Query
+}
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(
+  graph: join__Graph
+  requires: join__FieldSet
+  provides: join__FieldSet
+  type: String
+  external: Boolean
+  override: String
+  usedOverridden: Boolean
+) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(
+  graph: join__Graph!
+  interface: String!
+) repeatable on OBJECT | INTERFACE
+
+directive @join__type(
+  graph: join__Graph!
+  key: join__FieldSet
+  extension: Boolean! = false
+  resolvable: Boolean! = true
+  isInterfaceObject: Boolean! = false
+) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(
+  graph: join__Graph!
+  member: String!
+) repeatable on UNION
+
+directive @link(
+  url: String
+  as: String
+  for: link__Purpose
+  import: [link__Import]
+) repeatable on SCHEMA
+
+scalar join__FieldSet
+
+enum join__Graph {
+  X @join__graph(name: "X", url: "http://X")
+  Y @join__graph(name: "Y", url: "http://Y")
+  Z @join__graph(name: "Z", url: "http://Z")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+  SECURITY
+  EXECUTION
+}
+
+type Query @join__type(graph: X) @join__type(graph: Y) @join__type(graph: Z) {
+  start: T @join__field(graph: X)
+}
+
+type T @join__type(graph: X, key: "id") {
+  id: ID @join__field(graph: X)
+  inner: Inner @join__field(graph: X)
+}
+
+type Inner
+  @join__type(graph: X, key: "id")
+  @join__type(graph: Z, key: "id") {
+  id: String! @join__field(graph: X)
+  sub: Sub @join__field(graph: X)
+  target: R @join__field(graph: Z)
+}
+
+type Sub
+  @join__type(graph: X, key: "subId")
+  @join__type(graph: Y, key: "subId") {
+  subId: String! @join__field(graph: X)
+  data: String @join__field(graph: Y)
+}
+
+type R @join__type(graph: Z) {
+  x: String @join__field(graph: Z)
+}
diff --git a/apollo-router/src/query_planner/tests.rs b/apollo-router/src/query_planner/tests.rs
index af28576267..5fe6e66448 100644
--- a/apollo-router/src/query_planner/tests.rs
+++ b/apollo-router/src/query_planner/tests.rs
@@ -1892,3 +1892,340 @@ fn broken_plan_does_not_panic() {
         r#"[1:3] Cannot query field "invalid" on type "Query"."#
     );
 }
+
+/// Reproduces a bug (RH-1172) where a deferred node's dependency list was missing a
+/// fetch that provides `__typename` and entity keys at the deferred Flatten path.
+///
+/// The root cause was in the query planner's `collect_redundant_edges` (transitive
+/// reduction): when a direct edge from a primary fetch to a deferred fetch was also
+/// reachable transitively through an intermediate fetch, the direct edge was removed.
+/// This caused `extract_children_and_deferred_dependencies` to only register the
+/// intermediate fetch as a dependency, omitting the primary fetch that provides
+/// `__typename` and entity keys.
+///
+/// The scenario:
+/// - Primary: Fetch(X, id=fetch1) gets `start.inner { __typename id sub }`,
+///   then Flatten Fetch(Y, id=fetch0) gets `sub.data`
+/// - Deferred(depends: [fetch0, fetch1]): Flatten at `start.inner`, Fetch(Z) requires
+///   `Inner { __typename id sub { subId data } }` and returns `target`
+///
+/// With the planner fix, both fetch0 AND fetch1 are registered as dependencies,
+/// ensuring the deferred node has complete data (including `__typename` and `id`).
+#[tokio::test]
+async fn defer_depends_skips_fetch_when_typename_missing() {
+    // Query plan for:
+    //   { start { id inner { __typename id ... @defer { target { x } } } } }
+    // where inner.sub.data is fetched by a dependency (Y, id=fetch0)
+    // and the deferred fetch (Z) requires inner.__typename + id + sub { subId data }
+    let query_plan: QueryPlan = QueryPlan {
+        formatted_query_plan: Default::default(),
+        root: PlanNode::Defer {
+            primary: Primary {
+                subselection: Some("{ start { id inner { __typename id } } }".to_string()),
+                node: Some(Box::new(PlanNode::Sequence {
+                    nodes: vec![
+                        // Primary fetch: gets start with inner { __typename, id, sub }
+                        // With the planner fix, this fetch gets an id so the deferred
+                        // node can depend on it for __typename and id.
+                        PlanNode::Fetch(FetchNode {
+                            service_name: "X".into(),
+                            requires: vec![],
+                            variable_usages: vec![],
+                            operation: SerializableDocument::from_string(
+                                "{ start { __typename id inner { __typename id sub { __typename subId } } } }",
+                            ),
+                            operation_name: None,
+                            operation_kind: OperationKind::Query,
+                            id: Some("fetch1".into()),
+                            input_rewrites: None,
+                            output_rewrites: None,
+                            context_rewrites: None,
+                            schema_aware_hash: Default::default(),
+                            authorization: Default::default(),
+                        }),
+                        // Dependency fetch: gets sub.data (the deferred node depends on this)
+                        PlanNode::Flatten(FlattenNode {
+                            path: Path(vec![
+                                PathElement::Key("start".to_string(), None),
+                                PathElement::Key("inner".to_string(), None),
+                                PathElement::Key("sub".to_string(), None),
+                            ]),
+                            node: Box::new(PlanNode::Fetch(FetchNode {
+                                service_name: "Y".into(),
+                                requires: vec![
+                                    requires_selection::Selection::InlineFragment(
+                                        requires_selection::InlineFragment {
+                                            type_condition: Some(name!("Sub")),
+                                            selections: vec![
+                                                requires_selection::Selection::Field(
+                                                    requires_selection::Field {
+                                                        alias: None,
+                                                        name: name!("__typename"),
+                                                        selections: Vec::new(),
+                                                    },
+                                                ),
+                                                requires_selection::Selection::Field(
+                                                    requires_selection::Field {
+                                                        alias: None,
+                                                        name: name!("subId"),
+                                                        selections: Vec::new(),
+                                                    },
+                                                ),
+                                            ],
+                                        },
+                                    ),
+                                ],
+                                variable_usages: vec![],
+                                operation: SerializableDocument::from_string(
+                                    "query($representations:[_Any!]!){_entities(representations:$representations){...on Sub{data}}}",
+                                ),
+                                operation_name: None,
+                                operation_kind: OperationKind::Query,
+                                id: Some("fetch0".into()),
+                                input_rewrites: None,
+                                output_rewrites: None,
+                                context_rewrites: None,
+                                schema_aware_hash: Default::default(),
+                                authorization: Default::default(),
+                            })),
+                        }),
+                    ],
+                })),
+            },
+            deferred: vec![DeferredNode {
+                depends: vec![
+                    Depends {
+                        id: "fetch0".into(),
+                    },
+                    Depends {
+                        id: "fetch1".into(),
+                    },
+                ],
+                label: None,
+                query_path: Path(vec![
+                    PathElement::Key("start".to_string(), None),
+                    PathElement::Key("inner".to_string(), None),
+                ]),
+                subselection: Some("{ target { x } }".to_string()),
+                node: Some(Arc::new(PlanNode::Flatten(FlattenNode {
+                    path: Path(vec![
+                        PathElement::Key("start".to_string(), None),
+                        PathElement::Key("inner".to_string(), None),
+                    ]),
+                    node: Box::new(PlanNode::Fetch(FetchNode {
+                        service_name: "Z".into(),
+                        requires: vec![
+                            requires_selection::Selection::InlineFragment(
+                                requires_selection::InlineFragment {
+                                    type_condition: Some(name!("Inner")),
+                                    selections: vec![
+                                        requires_selection::Selection::Field(
+                                            requires_selection::Field {
+                                                alias: None,
+                                                name: name!("__typename"),
+                                                selections: Vec::new(),
+                                            },
+                                        ),
+                                        requires_selection::Selection::Field(
+                                            requires_selection::Field {
+                                                alias: None,
+                                                name: name!("id"),
+                                                selections: Vec::new(),
+                                            },
+                                        ),
+                                        requires_selection::Selection::Field(
+                                            requires_selection::Field {
+                                                alias: None,
+                                                name: name!("sub"),
+                                                selections: vec![
+                                                    requires_selection::Selection::Field(
+                                                        requires_selection::Field {
+                                                            alias: None,
+                                                            name: name!("subId"),
+                                                            selections: Vec::new(),
+                                                        },
+                                                    ),
+                                                    requires_selection::Selection::Field(
+                                                        requires_selection::Field {
+                                                            alias: None,
+                                                            name: name!("data"),
+                                                            selections: Vec::new(),
+                                                        },
+                                                    ),
+                                                ],
+                                            },
+                                        ),
+                                    ],
+                                },
+                            ),
+                        ],
+                        variable_usages: vec![],
+                        operation: SerializableDocument::from_string(
+                            "query($representations:[_Any!]!){_entities(representations:$representations){...on Inner{target{x}}}}",
+                        ),
+                        operation_name: None,
+                        operation_kind: OperationKind::Query,
+                        id: None,
+                        input_rewrites: None,
+                        output_rewrites: None,
+                        context_rewrites: None,
+                        schema_aware_hash: Default::default(),
+                        authorization: Default::default(),
+                    })),
+                }))),
+            }],
+        }
+        .into(),
+        usage_reporting: UsageReporting::Error("this is a test report key".to_string()).into(),
+        query: Arc::new(Query::empty_for_tests()),
+        query_metrics: Default::default(),
+        estimated_size: Default::default(),
+    };
+
+    // Mock X service: returns the root query data for start
+    let mut mock_x_service = plugin::test::MockSubgraphService::new();
+    mock_x_service.expect_clone().return_once(|| {
+        let mut mock_x_service = plugin::test::MockSubgraphService::new();
+        mock_x_service.expect_call().times(1).returning(|_| {
+            Ok(SubgraphResponse::fake_builder()
+                .data(serde_json::json! {{
+                    "start": {
+                        "__typename": "T",
+                        "id": "1",
+                        "inner": {
+                            "__typename": "Inner",
+                            "id": "i1",
+                            "sub": {
+                                "__typename": "Sub",
+                                "subId": "s1"
+                            }
+                        }
+                    }
+                }})
+                .build())
+        });
+        mock_x_service
+            .expect_clone()
+            .returning(plugin::test::MockSubgraphService::new);
+        mock_x_service
+    });
+
+    // Mock Y service: entity fetch for Sub, returns data field
+    let mut mock_y_service = plugin::test::MockSubgraphService::new();
+    mock_y_service.expect_clone().return_once(|| {
+        let mut mock_y_service = plugin::test::MockSubgraphService::new();
+        mock_y_service.expect_call().times(1).returning(|_| {
+            Ok(SubgraphResponse::fake_builder()
+                .data(serde_json::json! {{
+                    "_entities": [{"data": "d1"}]
+                }})
+                .build())
+        });
+        mock_y_service
+            .expect_clone()
+            .returning(plugin::test::MockSubgraphService::new);
+        mock_y_service
+    });
+
+    // Mock Z service: entity fetch for Inner, returns target
+    // If the bug exists, this service will NOT be called (fetch is skipped).
+    let mut mock_z_service = plugin::test::MockSubgraphService::new();
+    mock_z_service.expect_clone().return_once(|| {
+        let mut mock_z_service = plugin::test::MockSubgraphService::new();
+        mock_z_service.expect_call().times(1).returning(|_| {
+            Ok(SubgraphResponse::fake_builder()
+                .data(serde_json::json! {{
+                    "_entities": [{"target": {"x": "42"}}]
+                }})
+                .build())
+        });
+        mock_z_service
+            .expect_clone()
+            .returning(plugin::test::MockSubgraphService::new);
+        mock_z_service
+    });
+
+    let (sender, receiver) = tokio::sync::mpsc::channel(10);
+
+    let schema = include_str!("testdata/defer_depends_schema.graphql");
+    let schema = Arc::new(Schema::parse(schema, &Default::default()).unwrap());
+    let ssf = subgraph_service_factory(vec![
+        (
+            "X".into(),
+            Arc::new(mock_x_service) as Arc<dyn MakeSubgraphService>,
+        ),
+        (
+            "Y".into(),
+            Arc::new(mock_y_service) as Arc<dyn MakeSubgraphService>,
+        ),
+        (
+            "Z".into(),
+            Arc::new(mock_z_service) as Arc<dyn MakeSubgraphService>,
+        ),
+    ]);
+    let sf = Arc::new(FetchServiceFactory::new(
+        schema.clone(),
+        Default::default(),
+        Arc::new(ssf),
+        None,
+        Arc::new(ConnectorServiceFactory::empty(schema.clone())),
+        Arc::new(SubgraphConfiguration::<HoistOrphanErrors>::default()),
+    ));
+
+    let response = query_plan
+        .execute(
+            &Context::new(),
+            &sf,
+            &Default::default(),
+            &schema,
+            &Default::default(),
+            sender,
+            None,
+            &None,
+            None,
+        )
+        .await;
+
+    // Primary response should have start.inner with __typename and id
+    let primary_data = serde_json::to_value(&response).unwrap();
+    assert_eq!(
+        primary_data,
+        serde_json::json! {{
+            "data": {
+                "start": {
+                    "__typename": "T",
+                    "id": "1",
+                    "inner": {
+                        "__typename": "Inner",
+                        "id": "i1",
+                        "sub": {
+                            "__typename": "Sub",
+                            "subId": "s1",
+                            "data": "d1"
+                        }
+                    }
+                }
+            }
+        }}
+    );
+
+    // Deferred response should include target from Z fetch
+    let deferred = ReceiverStream::new(receiver).next().await.unwrap();
+    let deferred_data = serde_json::to_value(&deferred).unwrap();
+
+    // The deferred response data should contain target with x.
+    // If the bug exists (Z fetch skipped), target will be missing/null.
+    let inner_data = &deferred_data["data"]["start"]["inner"];
+    assert!(
+        inner_data.get("target").is_some() && !inner_data["target"].is_null(),
+        "target should not be null in the deferred response. \
+         This indicates the deferred fetch to Z was skipped because \
+         __typename was missing from the deferred node's context at the Flatten path. \
+         Got deferred response: {}",
+        serde_json::to_string_pretty(&deferred_data).unwrap()
+    );
+    assert_eq!(
+        inner_data["target"]["x"], "42",
+        "target should have x from Z fetch"
+    );
+}