From cb46be0430f23f2e60b064776139631483045087 Mon Sep 17 00:00:00 2001
From: Dennis Kobert <dennis@kobert.dev>
Date: Wed, 19 Aug 2026 21:43:14 +0000
Subject: [PATCH] Emit the reducer extent shift and the flat-span
 materialization

---
 node-graph/node-macro/src/codegen.rs | 32 +++++++++++++++++++++++++++-
 1 file changed, 31 insertions(+), 1 deletion(-)

diff --git a/node-graph/node-macro/src/codegen.rs b/node-graph/node-macro/src/codegen.rs
index 5c2838a6ee..b68ad745cd 100644
--- a/node-graph/node-macro/src/codegen.rs
+++ b/node-graph/node-macro/src/codegen.rs
@@ -771,6 +771,12 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 			.iter()
 			.any(|bound| matches!(bound, TypeParamBound::Trait(trait_bound) if trait_bound.path.segments.last().is_some_and(|segment| segment.ident == "ExtractArena")))
 	});
+	let ctx_extracts_index = ctx_param.is_some_and(|ctx_param| {
+		ctx_param
+			.bounds
+			.iter()
+			.any(|bound| matches!(bound, TypeParamBound::Trait(trait_bound) if trait_bound.path.segments.last().is_some_and(|segment| segment.ident == "ExtractIndex")))
+	});
 
 	let derives = ctx_param.is_some_and(|ctx_param| {
 		ctx_param.bounds.iter().any(|bound| match bound {
@@ -779,6 +785,9 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 		})
 	});
 	let derive_routing = derives && routing_generic.is_some();
+	// A reducer's materialized subject folds the flat span of the consumer's
+	// lane when the wire is deeper than the fold; the ctx supplies the lane.
+	let derived_materialized = |index: usize| ctx_extracts_index && node.inputs[index].subject && ir::materialized_levels(&node, index) > 0 && node.output.shape.depth == 0;
 
 	let ctx_generic = match ctx_bounds.is_empty() {
 		true => quote!(#ctx_ident),
@@ -1115,6 +1124,18 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 					let non_exact = fail(quote!(#core_types::gpoll::GraphError::new("reduce over a non-exact extent")));
 					let batch_error = fail(quote!(__error));
 					let batch_failed = fail(quote!(#core_types::gpoll::GraphError::new("reduce batch failed")));
+					// A reducer's subject on a deeper wire folds the flat span
+					// of the consumer's lane: the flat convention encodes the
+					// outer coordinate in the range, not the chain.
+					let start = match derived_materialized(index) {
+						true => quote! {
+							match #core_types::node::Node::<#ctx_ident>::layout(&self.#name).depth > #levels {
+								true => #core_types::context::ExtractIndex::innermost_index(__input) * __count as u64,
+								false => 0,
+							}
+						},
+						false => quote!(0),
+					};
 					quote! {
 						let __arena = #core_types::context::ExtractArena::arena(__input);
 						let __count = match #core_types::node::Node::extent(&self.#name, __input, #core_types::gpoll::Level::Below(#levels)) {
@@ -1122,7 +1143,8 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 							#core_types::gpoll::GPoll::Pending => #pending,
 							_ => #non_exact,
 						};
-						let __batch = match #core_types::record::materialize_batch(&self.#name, __input, 0..__count as u64, __arena) {
+						let __start: u64 = #start;
+						let __batch = match #core_types::record::materialize_batch(&self.#name, __input, __start..__start + __count as u64, __arena) {
 							#core_types::node::BatchStatus::Lent(__batch, _) => __batch,
 							#core_types::node::BatchStatus::Filled(__batch, _) => __batch.into_shared(),
 							#core_types::node::BatchStatus::Pending => #pending,
@@ -1366,6 +1388,14 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 				#path(self, __input, __level)
 			}
 		}
+	} else if let Some((subject_index, folded_levels)) = ir::folded_subject(&node).filter(|_| node.output.shape.depth == 0) {
+		// A reducer's output keeps the subject's levels above the folded ones.
+		let name = &regular_fields[subject_index as usize].pat_ident.ident;
+		quote! {
+			fn extent_at(&self, __input: &#ctx_ident, __level: u8) -> #core_types::gpoll::GPoll<#core_types::gpoll::Extent> {
+				#core_types::node::Node::extent_at(&self.#name, __input, __level + #folded_levels)
+			}
+		}
 	} else {
 		quote!()
 	};