Build a gathered subject's copy plan on collapse and reject an empty carry

node-graph/libraries/core-types/src/record/frames.rs

@@ -112,6 +112,7 @@ impl<'e> Frames<'e> {
 			frame,
 			free: self.reborrow(),
 			filled_fields: false,
+			carried_empty: false,
 		}
 	}
 
@@ -182,4 +183,48 @@ mod tests {
 		}
 		assert!(addresses.windows(2).all(|pair| pair[0] == pair[1]), "each claim reuses the same region");
 	}
+
+	#[test]
+	#[should_panic(expected = "carried an empty plan and filled nothing")]
+	fn carrying_an_empty_plan_and_filling_nothing_refuses_to_close() {
+		// The shape that served uninitialized bytes before this guard existed: a
+		// layout that DECLARES a field, a carry whose plan turned out empty, and
+		// nothing else to fill it. The field would otherwise be the prior frame's
+		// bytes, which for a reference field is an uninitialized read.
+		let layout = Layout::default().with_writes(0, element_write::<f64>(), &[f64_field("opacity")]);
+		let mut frame_arena = FrameArena::new();
+		frame_arena.reserve(1 << 10);
+		let frames = frame_arena.frames();
+		let scope = frames.scope();
+		let mut claim = scope.claim(&layout);
+		let value = crate::record::access::RecordValue::zeroed();
+		let src = layout.rec(&value);
+		// SAFETY: an empty plan reads nothing from `src`; closing afterwards is
+		// what the guard refuses.
+		unsafe { claim.carry(src, &[]) };
+		// SAFETY: the assertion fires before anything reads the unfilled field.
+		let _ = unsafe { claim.finish() };
+	}
+
+	#[test]
+	fn an_empty_carry_whose_fields_are_written_closes_normally() {
+		// The control that pins what the guard actually tracks: the same empty
+		// carry, but the declared field is then written. Closing must succeed, so
+		// the guard is about a field left unfilled rather than about the carry.
+		let layout = Layout::default().with_writes(0, element_write::<f64>(), &[f64_field("opacity")]);
+		let offset = layout.fields.first().expect("the fixture declares one field").offset;
+		let mut frame_arena = FrameArena::new();
+		frame_arena.reserve(1 << 10);
+		let frames = frame_arena.frames();
+		let scope = frames.scope();
+		let mut claim = scope.claim(&layout);
+		let value = crate::record::access::RecordValue::zeroed();
+		let src = layout.rec(&value);
+		// SAFETY: an empty plan reads nothing from `src`.
+		unsafe { claim.carry(src, &[]) };
+		// SAFETY: `offset` is this layout's own resolved offset for an f64 field.
+		unsafe { claim.attr_at(offset, 0.5_f64) };
+		// SAFETY: the write above filled the declared field.
+		let _ = unsafe { claim.finish() };
+	}
 }

node-graph/libraries/core-types/src/record/layout.rs

@@ -478,6 +478,18 @@ pub struct LayoutMeta {
 	/// The materialized subject a reducer folds, as `(input, levels)`. The fold
 	/// consumes the whole subject input, so only the node's own levels remain.
 	pub folded: Option<(u8, u8)>,
+	/// Whether [`sources`](Self::sources)`[0]` is a GATHERED subject, whose
+	/// per-lane layout is this output's base.
+	///
+	/// Carried explicitly because it cannot be inferred from `sources` and
+	/// `level_delta`. A gathered subject that also collapses the level
+	/// (`level_delta < 0`) still has a well-defined copy plan: the gathered
+	/// lane's layout IS the base being copied into. A LAZY folding subject looks
+	/// identical from the outside - un-materialized, so it also sits in
+	/// `sources` with a negative delta - but copying its fields down is exactly
+	/// what the plan must not do. Keying on "sources is non-empty" would
+	/// conflate the two and reintroduce the defect in the other direction.
+	pub gathered: bool,
 }
 
 /// The attributes a node reads from one input, recorded on [`LayoutMeta`] for
@@ -513,6 +525,8 @@ impl LayoutMeta {
 			removes: Vec::new(),
 			level_delta: 0,
 			folded: None,
+			// A retype keeps input 0's level, so its plan is already unconditional.
+			gathered: false,
 		}
 	}
 
@@ -545,7 +559,9 @@ impl LayoutMeta {
 		let layout = self.fold(inputs);
 		let frame_bytes = layout.frame_bytes();
 		let plan = match self.sources.first() {
-			Some(&source) if self.level_delta >= 0 => {
+			// A gathered subject copies from the lane it gathered, whose layout is
+			// this output's base, so its plan holds however the level moves.
+			Some(&source) if self.level_delta >= 0 || self.gathered => {
 				let from = inputs[source as usize].expect("layout resolve source input has no layout");
 				let carry_element = matches!(self.element, ElementSpec::Carried);
 				let removes: Vec<(&str, u8)> = self.removes.clone();

node-graph/libraries/core-types/src/record/serve.rs

@@ -47,6 +47,7 @@ impl<'a> SlotRun<'a> {
 			frame: (self.layout.frame_bytes() != 0).then_some(frame),
 			free: frames.reborrow(),
 			filled_fields: false,
+			carried_empty: false,
 		}
 	}
 
@@ -86,6 +87,11 @@ pub struct FrameClaim<'e, 'l> {
 	/// Set by the writes that fill the declared fields, so the safe closers can
 	/// refuse a field-bearing frame that was never filled.
 	pub(in crate::record) filled_fields: bool,
+	/// Set where a carry ran against an EMPTY plan, which fills nothing. Kept
+	/// apart from `filled_fields` because "carried nothing" and "never carried"
+	/// are different mistakes: the first is a wiring defect the closers can
+	/// prove, the second is the ordinary shape of a fresh record.
+	pub(in crate::record) carried_empty: bool,
 }
 
 impl<'e, 'l> FrameClaim<'e, 'l> {
@@ -124,7 +130,16 @@ impl<'e, 'l> FrameClaim<'e, 'l> {
 	/// serving the source through [`Self::frames`] establishes it.
 	pub unsafe fn carry(&mut self, src: Rec<'_>, plan: &[(usize, usize, usize)]) {
 		unsafe { apply_plan(src, self.dst(), plan) };
-		self.filled_fields = true;
+		// An empty plan copied nothing, so it must not satisfy the closers'
+		// "carried or wrote its fields" guard: the frame still holds whatever
+		// bytes the previous claim left, and a declared field read out of them
+		// is an uninitialized read rather than a missing value. Recorded rather
+		// than ignored so the closers can tell "carried nothing" apart from
+		// "never carried", which are not the same mistake.
+		match plan.is_empty() {
+			true => self.carried_empty = true,
+			false => self.filled_fields = true,
+		}
 	}
 
 	/// Copies the source record's element bytes into the frame, for a gathered
@@ -207,6 +222,17 @@ impl<'e, 'l> FrameClaim<'e, 'l> {
 	/// The frame must hold a complete record of the layout, written through
 	/// the carry, element, and field writes.
 	pub unsafe fn finish(mut self) -> RecordValue<'e> {
+		// A frame that CARRIED, but through an empty plan, and then filled
+		// nothing else: its declared fields hold the previous claim's bytes, and
+		// a reference field read out of them is an uninitialized read. This is
+		// narrower than [`Self::lift`]'s guard on purpose - a frame that never
+		// carried at all may legitimately serve fields the census staged as
+		// declared defaults, so only an empty carry is evidence of the defect.
+		assert!(
+			!(self.carried_empty && !self.filled_fields && !self.layout.fields.is_empty()),
+			"a layout with {} fields carried an empty plan and filled nothing: its fields would be the prior frame's bytes",
+			self.layout.fields.len()
+		);
 		match self.frame {
 			Some(frame) => RecordValue::spilled(unsafe { Rec::new(frame.cast_const()) }),
 			// SAFETY: the inline record is the value's own bytes.

node-graph/node-macro/src/codegen/ir.rs

@@ -295,6 +295,8 @@ pub(crate) fn layout_meta_tokens(node: &Node, element_spec: TokenStream2, core_t
 		Some((index, levels)) => quote!(::core::option::Option::Some((#index, #levels))),
 		None => quote!(::core::option::Option::None),
 	};
+	// A gathered subject's plan holds however the level moves; see `LayoutMeta::gathered`.
+	let gathered = gathered_subject(node).is_some();
 	quote! {
 		#core_types::record::LayoutMeta {
 			sources: ::std::vec![#(#sources),*],
@@ -308,6 +310,7 @@ pub(crate) fn layout_meta_tokens(node: &Node, element_spec: TokenStream2, core_t
 			removes: ::std::vec![#(#removes),*],
 			level_delta: #level_delta,
 			folded: #folded,
+			gathered: #gathered,
 		}
 	}
 }

node-graph/nodes/gcore/src/record.rs

@@ -319,6 +319,19 @@ fn reverse_lanes_extent(content: ListIn<'_, f64>, _opacity: ValueIn<'_, f64>, le
 	}
 }
 
+/// Gather-carrier kernel that COLLAPSES its level: one value for the whole
+/// subject, gathered from a named lane. Before a gathered subject built its copy
+/// plan regardless of the level change, this shape declared the lane's columns
+/// and copied none of them, serving the previous frame's bytes instead.
+#[node_macro::node(category("Test"))]
+fn collapse_to_lane<'e>(_ctx: impl Ctx + core_types::context::ExtractArena<'e> + Copy, content: IList<f64>, opacity: f64) -> Result<(Lane<f64>, Attr<Opacity>), Interrupt> {
+	if content.is_empty() {
+		return Err(GraphError::past_end().into());
+	}
+	let total: f64 = (0..content.len()).map(|lane| content.get(lane)).sum();
+	Ok((content.lane(0).map_element(total), Attr(opacity)))
+}
+
 /// Gather-carrier kernel with a substituted subject: the lane's record carries
 /// exactly as it does for [`reverse_lanes`], but `map_element` replaces the
 /// element, so a node that rewrites what it produces still keeps every column
@@ -839,6 +852,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let node = install(
 			RepeatNode::new(RecordSource::new(bare_source(&base, 7.), &base, &base), count_edge, reverse_edge, &base, &count_layout, &reverse_layout),
@@ -879,6 +893,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let repeat = install(
 			RepeatNode::new(
@@ -927,6 +942,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let repeat = install(
 			RepeatNode::new(
@@ -981,6 +997,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let repeat = install(
 			RepeatNode::new(
@@ -1083,6 +1100,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 0,
 			folded: None,
+			gathered: false,
 		};
 		let node = install(
 			ExtendNode::new(RecordSource::new(base, &base_layout, &union), RecordSource::new(new, &new_layout, &union), &union),
@@ -1155,6 +1173,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 0,
 			folded: None,
+			gathered: false,
 		};
 
 		let left_inner = install(
@@ -1233,6 +1252,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 0,
 			folded: None,
+			gathered: false,
 		};
 		let extend = install(
 			ExtendNode::new(RecordSource::new(base, &base_layout, &union), RecordSource::new(new, &new_layout, &union), &union),
@@ -1281,6 +1301,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let extend_meta = || core_types::record::LayoutMeta {
 			named_writes: Vec::new(),
@@ -1294,6 +1315,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 0,
 			folded: None,
+			gathered: false,
 		};
 		let (base_count, base_count_layout) = lifted_value(2u32);
 		let (base_reverse, base_reverse_layout) = lifted_value(false);
@@ -1373,6 +1395,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 0,
 			folded: None,
+			gathered: false,
 		};
 		let build = |index: f64| {
 			let content = LeveledSourceNode {
@@ -1430,6 +1453,7 @@ mod tests {
 				removes: vec![],
 				level_delta: 0,
 				folded: None,
+				gathered: false,
 			};
 			install(
 				IndexElementsNode::new(RecordSource::new(content, &layout, &layout), index_edge, &layout, &index_layout),
@@ -1566,6 +1590,52 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn a_gathered_collapse_carries_its_named_lanes_columns() {
+		let arena = Arena::new(1 << 16).unwrap();
+		let generations = [];
+		let scope = scope_fixture(&generations, &arena);
+		let ctx = ContextImpl::root(&scope);
+
+		// Lanes carry DISTINCT transforms, so "carried lane 0's column" is
+		// distinguishable from "carried some other lane's" and from stale bytes.
+		let layout = Layout::default().with_writes(
+			1,
+			core_types::record::element_write::<f64>(),
+			&[core_types::record::FieldWrite::of::<Transform>(0), core_types::record::FieldWrite::of::<Opacity>(0)],
+		);
+		let frames = frames_for(&[&layout]);
+		let rows = [(1., 10., 0.1), (2., 30., 0.2), (3., 20., 0.3)];
+		let content = LeveledCarriedSource {
+			layout: layout.clone(),
+			rows: rows
+				.iter()
+				.map(|&(element, x, opacity)| (element, DAffine2::from_translation(glam::DVec2::new(x, 0.)), opacity))
+				.collect(),
+		};
+		let node = install(
+			CollapseToLaneNode::new(RecordSource::new(content, &layout, &layout), ValueSource::new(0.25)),
+			collapse_to_lane_layout_meta(),
+			&[Some(&layout)],
+		);
+
+		let out = Node::<ContextImpl>::layout(&node).clone();
+		assert_eq!(out.depth, 0, "the collapse serves one value rather than a level");
+		assert!(
+			out.offset_of(<Transform as AttributeMarker>::NAME, 0).is_some(),
+			"the gathered lane's undeclared column is declared on the output"
+		);
+
+		let GPoll::Final(served) = core_types::record::capture(&node, &ctx, &frames) else {
+			panic!("expected a final record");
+		};
+		assert_eq!(served.element::<f64>(), 6., "the collapsed element is the kernel's own value");
+		let transform: DAffine2 = served.attr::<Transform>();
+		assert_eq!(transform.translation.x, 10., "lane 0's transform is carried - not lane 1's or 2's, and not stale frame bytes");
+		let opacity: f64 = served.attr::<Opacity>();
+		assert_eq!(opacity, 0.25, "the declared write still overrides the carried opacity");
+	}
+
 	#[test]
 	fn a_substituted_subject_keeps_the_lanes_other_columns() {
 		let arena = Arena::new(1 << 16).unwrap();
@@ -1781,6 +1851,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let nested = install(
 			RepeatNode::new(
@@ -1838,6 +1909,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let nested = install(
 			RepeatNode::new(
@@ -1952,6 +2024,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let repeat = install(
 			RepeatNode::new(
@@ -2082,6 +2155,7 @@ mod tests {
 			removes: vec![],
 			level_delta: 1,
 			folded: None,
+			gathered: false,
 		};
 		let repeat = install(
 			RepeatNode::new(

node-graph/nodes/graphic/src/record.rs

@@ -437,6 +437,7 @@ mod tests {
 			removes: vec![],
 			level_delta,
 			folded: None,
+			gathered: false,
 		}
 	}