Add per-field erased-read glue, the record monitor with introspection capture, and rename record constructors to new

node-graph/interpreted-executor/src/dynamic_executor.rs

@@ -145,8 +145,14 @@ impl DynamicExecutor {
 	}
 
 	/// Calls the `Node::serialize` for that specific node, returning for example the cached value for a monitor node. The node path must match the document node path.
+	/// A record capture materializes here against the arena, inside the introspection window.
 	pub fn introspect(&self, node_path: &[NodeId]) -> Result<Arc<dyn std::any::Any + Send + Sync + 'static>, IntrospectError> {
-		self.tree.introspect(node_path)
+		let result = self.tree.introspect(node_path)?;
+		if let Some(capture) = result.downcast_ref::<core_types::record::RecordCapture>() {
+			let arena = self.arena.lock().unwrap_or_else(PoisonError::into_inner);
+			return capture.materialize(&arena).map(|fields| Arc::new(fields) as Arc<_>).ok_or(IntrospectError::NoData);
+		}
+		Ok(result)
 	}
 
 	pub fn input_type(&self) -> Option<Type> {
@@ -624,6 +630,30 @@ mod test {
 		assert_eq!((&executor).execute(()).unwrap(), GPoll::Final(TaggedValue::F64(7.)));
 	}
 
+	#[test]
+	fn a_record_monitor_row_forwards_and_introspects_through_the_executor() {
+		let mut monitor = proto_node("graphene_core::memo::MonitorNode", vec![NodeId(1)]);
+		monitor.original_location.path = Some(vec![NodeId(9)]);
+		let network = ProtoNetwork {
+			inputs: vec![],
+			output: NodeId(3),
+			nodes: vec![
+				(NodeId(0), ProtoNode::value(ConstructionArgs::Value(TaggedValue::F64(7.).into()), vec![])),
+				(NodeId(1), proto_node("core_types::record::RecordLiftNode", vec![NodeId(0)])),
+				(NodeId(2), monitor),
+				(NodeId(3), proto_node("core_types::record::RecordExtractNode", vec![NodeId(2)])),
+			],
+		};
+
+		let executor = DynamicExecutor::new(network).unwrap();
+		assert_eq!((&executor).execute(()).unwrap(), GPoll::Final(TaggedValue::F64(7.)));
+		let fields = executor.introspect(&[NodeId(9)]).unwrap();
+		let fields = fields
+			.downcast_ref::<Vec<(&'static str, Box<dyn core_types::list::AnyAttributeValue>)>>()
+			.expect("a record capture materializes to its fields");
+		assert!(fields.is_empty(), "an element-only record has no attribute fields");
+	}
+
 	#[test]
 	fn a_lift_adapter_is_spliced_between_a_plain_producer_and_a_record_consumer() {
 		let network = ProtoNetwork {

node-graph/interpreted-executor/src/node_registry.rs

@@ -376,6 +376,30 @@ fn node_registry() -> HashMap<ProtoNodeIdentifier, Vec<RegistryEntry>> {
 		lend_node!(f64),
 		record_lift_node!(f64),
 		record_extract_node!(f64),
+		(
+			ProtoNodeIdentifier::new("graphene_core::memo::MonitorNode"),
+			RegistryEntry {
+				io: NodeIOTypes::new(
+					concrete!(Context),
+					core_types::Type::Record(Box::new(core_types::Type::Generic(std::borrow::Cow::Borrowed("T")))),
+					vec![core_types::registry::generic_record_edge_type("T")],
+				),
+				constructor: |inputs| {
+					if inputs.len() != 1 {
+						return Err(ConstructionError::Arity { expected: 1, got: inputs.len() });
+					}
+					let mut inputs = inputs.into_iter();
+					let handle = inputs.next().unwrap();
+					let ty = handle.ty().clone();
+					let Some(layout) = handle.layout().cloned() else {
+						return Err(ConstructionError::MissingLayout);
+					};
+					let edge = handle.downcast_erased::<core_types::registry::ErasedRecordNode>(ty.clone())?;
+					let node = core_types::record::RecordMonitor::new(edge, &layout);
+					Ok(EdgeHandle::new_erased(std::sync::Arc::new(node) as std::sync::Arc<core_types::registry::ErasedRecordNode>, ty))
+				},
+			},
+		),
 		clone_node!(f64),
 		frame_memo_node!(f64),
 		lend_node!(f32),
@@ -782,7 +806,7 @@ mod node_registry_macros {
 							return Err(ConstructionError::Arity { expected: 1, got: inputs.len() });
 						}
 						let mut inputs = inputs.into_iter();
-						let node = core_types::record::RecordLift::<$type, _>::wire(inputs.next().unwrap().downcast::<$type>()?);
+						let node = core_types::record::RecordLift::<$type, _>::new(inputs.next().unwrap().downcast::<$type>()?);
 						Ok(EdgeHandle::new_record::<$type>(std::sync::Arc::new(node) as std::sync::Arc<core_types::registry::ErasedRecordNode>))
 					},
 				},
@@ -801,7 +825,7 @@ mod node_registry_macros {
 							return Err(ConstructionError::Arity { expected: 1, got: inputs.len() });
 						}
 						let mut inputs = inputs.into_iter();
-						let node = core_types::record::RecordExtract::<$type, _>::wire(inputs.next().unwrap().downcast_record::<$type>()?);
+						let node = core_types::record::RecordExtract::<$type, _>::new(inputs.next().unwrap().downcast_record::<$type>()?);
 						Ok(EdgeHandle::new(std::sync::Arc::new(node) as std::sync::Arc<ErasedNode<$type>>))
 					},
 				},

node-graph/libraries/core-types/src/arena.rs

@@ -187,7 +187,7 @@ impl Drop for Arena {
 
 pub struct ArenaWeak<T> {
 	word: u64,
-	_marker: PhantomData<*const T>,
+	_marker: PhantomData<fn() -> T>,
 }
 
 impl<T> Clone for ArenaWeak<T> {

node-graph/libraries/core-types/src/attribute.rs

@@ -33,6 +33,10 @@ pub trait Attribute: 'static {
 	fn default<'e>() -> Self::Value<'e> {
 		Default::default()
 	}
+
+	/// # Safety
+	/// `ptr` must point at a live field of this marker's value type.
+	unsafe fn read_erased(ptr: *const u8) -> Box<dyn AnyAttributeValue>;
 }
 
 /// A kernel-facing attribute value. A parameter `Attr<A>` is a read of `A`
@@ -154,6 +158,10 @@ macro_rules! attribute {
 					$default
 				}
 			)?
+
+			unsafe fn read_erased(ptr: *const u8) -> ::std::boxed::Box<dyn $crate::list::AnyAttributeValue> {
+				::std::boxed::Box::new(unsafe { ptr.cast::<&$value>().read() }.to_owned())
+			}
 		}
 
 		$crate::attribute!(@register $marker);
@@ -171,6 +179,10 @@ macro_rules! attribute {
 					$default
 				}
 			)?
+
+			unsafe fn read_erased(ptr: *const u8) -> ::std::boxed::Box<dyn $crate::list::AnyAttributeValue> {
+				::std::boxed::Box::new(unsafe { ptr.cast::<$value>().read() })
+			}
 		}
 
 		$crate::attribute!(@register $marker);
@@ -256,6 +268,10 @@ mod tests {
 		impl Attribute for Conflict {
 			const NAME: &'static str = "opacity";
 			type Value<'e> = bool;
+
+			unsafe fn read_erased(ptr: *const u8) -> Box<dyn AnyAttributeValue> {
+				Box::new(unsafe { ptr.cast::<bool>().read() })
+			}
 		}
 		register::<Conflict>();
 	}

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

@@ -10,17 +10,52 @@ use crate::attribute;
 use crate::gpoll::GPoll;
 use crate::node::Node;
 
+/// A field write declared at wiring, carrying the marker's erased-read glue
+/// so introspection and persistence never consult the census at runtime.
+#[derive(Clone, Copy, Debug)]
+pub struct FieldWrite {
+	pub name: &'static str,
+	pub level: u8,
+	pub size: usize,
+	pub align: usize,
+	pub read_erased: unsafe fn(*const u8) -> Box<dyn crate::list::AnyAttributeValue>,
+}
+
+impl FieldWrite {
+	pub fn of<A: crate::attribute::Attribute>(level: u8) -> Self {
+		Self {
+			name: A::NAME,
+			level,
+			size: size_of::<A::Value<'static>>(),
+			align: align_of::<A::Value<'static>>(),
+			read_erased: A::read_erased,
+		}
+	}
+}
+
 /// One field of a [`Layout`]: a (name, level) key resolved to an offset.
 /// Levels are numbered innermost-out; only level 0 exists at rank 0.
-#[derive(Clone, Debug, PartialEq, Eq)]
+/// Equality is structural: the glue pointer is excluded, since fn-pointer
+/// identity is not guaranteed across codegen units and layout equality
+/// drives identity forwarding.
+#[derive(Clone, Debug)]
 pub struct FieldDesc {
 	pub name: &'static str,
 	pub level: u8,
 	pub offset: usize,
 	pub size: usize,
 	pub align: usize,
+	pub read_erased: unsafe fn(*const u8) -> Box<dyn crate::list::AnyAttributeValue>,
 }
 
+impl PartialEq for FieldDesc {
+	fn eq(&self, other: &Self) -> bool {
+		(self.name, self.level, self.offset, self.size, self.align) == (other.name, other.level, other.offset, other.size, other.align)
+	}
+}
+
+impl Eq for FieldDesc {}
+
 /// A record layout: the element at offset 0, then the written attributes in
 /// canonical order (descending alignment, then size, then name, then level).
 /// Layouts are derived data, a pure function of the upstream write set.
@@ -43,31 +78,42 @@ impl Layout {
 	/// (size, align) at `depth`, in canonical order. A (name, level) written
 	/// at a different size is a type conflict and panics; the census keeps
 	/// declared names to one type, so this only fires on wiring bugs.
-	pub fn with_writes(&self, depth: u8, element: (usize, usize), writes: &[(&'static str, u8, usize, usize)]) -> Layout {
-		let mut merged: Vec<(&'static str, u8, usize, usize)> = self.fields.iter().map(|field| (field.name, field.level, field.size, field.align)).collect();
-		for &(name, level, size, align) in writes {
-			match merged.iter().find(|(n, l, ..)| *n == name && *l == level) {
-				Some(&(.., existing_size, _)) => assert_eq!(existing_size, size, "attribute `{name}` written at two different sizes"),
-				None => merged.push((name, level, size, align)),
+	pub fn with_writes(&self, depth: u8, element: (usize, usize), writes: &[FieldWrite]) -> Layout {
+		let mut merged: Vec<FieldWrite> = self
+			.fields
+			.iter()
+			.map(|field| FieldWrite {
+				name: field.name,
+				level: field.level,
+				size: field.size,
+				align: field.align,
+				read_erased: field.read_erased,
+			})
+			.collect();
+		for &write in writes {
+			match merged.iter().find(|field| field.name == write.name && field.level == write.level) {
+				Some(existing) => assert_eq!(existing.size, write.size, "attribute `{}` written at two different sizes", write.name),
+				None => merged.push(write),
 			}
 		}
-		merged.sort_by(|a, b| b.3.cmp(&a.3).then(b.2.cmp(&a.2)).then(a.0.cmp(b.0)).then(a.1.cmp(&b.1)));
+		merged.sort_by(|a, b| b.align.cmp(&a.align).then(b.size.cmp(&a.size)).then(a.name.cmp(b.name)).then(a.level.cmp(&b.level)));
 		let (element_size, element_align) = element;
 		let mut offset = element_size;
 		let mut align = element_align.max(1);
 		let fields = merged
 			.into_iter()
-			.map(|(name, level, size, field_align)| {
-				offset = offset.next_multiple_of(field_align.max(1));
-				align = align.max(field_align);
+			.map(|write| {
+				offset = offset.next_multiple_of(write.align.max(1));
+				align = align.max(write.align);
 				let desc = FieldDesc {
-					name,
-					level,
+					name: write.name,
+					level: write.level,
 					offset,
-					size,
-					align: field_align,
+					size: write.size,
+					align: write.align,
+					read_erased: write.read_erased,
 				};
-				offset += size;
+				offset += write.size;
 				desc
 			})
 			.collect();
@@ -89,7 +135,17 @@ impl Layout {
 			assert_eq!(union.element_size, layout.element_size, "union layouts must share the element size");
 			assert_eq!(union.element_align, layout.element_align, "union layouts must share the element alignment");
 			assert_eq!(union.depth, layout.depth, "union layouts must share the depth");
-			let writes: Vec<(&'static str, u8, usize, usize)> = layout.fields.iter().map(|field| (field.name, field.level, field.size, field.align)).collect();
+			let writes: Vec<FieldWrite> = layout
+				.fields
+				.iter()
+				.map(|field| FieldWrite {
+					name: field.name,
+					level: field.level,
+					size: field.size,
+					align: field.align,
+					read_erased: field.read_erased,
+				})
+				.collect();
 			union = union.with_writes(union.depth, (union.element_size, union.element_align), &writes);
 		}
 		union
@@ -344,7 +400,7 @@ pub struct RecordSource<N> {
 }
 
 impl<N> RecordSource<N> {
-	pub fn wire(edge: N, source: &Layout, union: &Layout) -> Self {
+	pub fn new(edge: N, source: &Layout, union: &Layout) -> Self {
 		Self {
 			edge,
 			plan: SourcePlan::new(source, union),
@@ -352,6 +408,78 @@ impl<N> RecordSource<N> {
 	}
 }
 
+/// A captured record: the layout plus a generation-checked handle to the
+/// arena copy, materialized by the introspection holder, which owns the
+/// arena. A dead generation materializes to `None`, never to a stale read.
+#[derive(Clone)]
+pub struct RecordCapture {
+	layout: Layout,
+	bytes: crate::arena::ArenaWeak<Box<[u8]>>,
+}
+
+impl RecordCapture {
+	pub fn materialize(&self, arena: &crate::arena::Arena) -> Option<Vec<(&'static str, Box<dyn crate::list::AnyAttributeValue>)>> {
+		let bytes = self.bytes.upgrade(arena)?;
+		Some(
+			self.layout
+				.fields
+				.iter()
+				.map(|field| (field.name, unsafe { (field.read_erased)(bytes.as_ptr().add(field.offset)) }))
+				.collect(),
+		)
+	}
+}
+
+// TODO: Convert to a `#[node_macro::node]` node once routing nodes forward
+// layouts and the macro grows a capture capability.
+/// The monitor over a record wire: forwards the record and captures an arena
+/// copy readable through the introspection window, like a frame memo.
+pub struct RecordMonitor<N> {
+	edge: N,
+	layout: Layout,
+	capture: std::sync::Mutex<Option<RecordCapture>>,
+}
+
+impl<N> RecordMonitor<N> {
+	pub fn new(edge: N, layout: &Layout) -> Self {
+		Self {
+			edge,
+			layout: layout.clone(),
+			capture: std::sync::Mutex::new(None),
+		}
+	}
+}
+
+impl<'e, C, N> Node<C> for RecordMonitor<N>
+where
+	C: crate::context::ExtractArena<ArenaRef = &'e crate::arena::Arena>,
+	N: Node<C, Output = RecordValue<'e>>,
+{
+	type Output = RecordValue<'e>;
+
+	fn eval(&self, input: &C) -> GPoll<RecordValue<'e>> {
+		let value = self.edge.eval(input);
+		if let GPoll::Final(record) | GPoll::Partial(record) = &value {
+			let bytes: Box<[u8]> = unsafe { std::slice::from_raw_parts(record.rec().ptr(), self.layout.size) }.into();
+			let capture = input.arena().alloc(bytes).map(|(_, weak)| RecordCapture {
+				layout: self.layout.clone(),
+				bytes: weak,
+			});
+			*self.capture.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = capture;
+		}
+		value
+	}
+
+	fn layout(&self) -> Option<&Layout> {
+		Some(&self.layout)
+	}
+
+	fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
+		let capture = self.capture.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()?;
+		Some(std::sync::Arc::new(capture))
+	}
+}
+
 /// Lifts a plain producer onto a record wire: the element lands at offset 0
 /// of a fresh element-only record. `Copy` elements only until droppable
 /// elements ride records.
@@ -363,7 +491,7 @@ pub struct RecordLift<El, N> {
 }
 
 impl<El: Copy + 'static, N> RecordLift<El, N> {
-	pub fn wire(edge: N) -> Self {
+	pub fn new(edge: N) -> Self {
 		let layout = Layout::default().with_writes(0, (size_of::<El>(), align_of::<El>()), &[]);
 		let frame_bytes = layout.size.next_multiple_of(8);
 		Self {
@@ -405,7 +533,7 @@ pub struct RecordExtract<El, N> {
 }
 
 impl<El, N> RecordExtract<El, N> {
-	pub fn wire(edge: N) -> Self {
+	pub fn new(edge: N) -> Self {
 		Self {
 			edge,
 			_marker: std::marker::PhantomData,
@@ -447,13 +575,21 @@ where
 mod tests {
 	use super::*;
 
-	fn f64_field(name: &'static str) -> (&'static str, u8, usize, usize) {
-		(name, 0, 8, 8)
+	unsafe fn unread(_: *const u8) -> Box<dyn crate::list::AnyAttributeValue> {
+		unreachable!("layout-only test field")
+	}
+
+	fn sized_field(name: &'static str, size: usize, align: usize) -> FieldWrite {
+		FieldWrite { name, level: 0, size, align, read_erased: unread }
+	}
+
+	fn f64_field(name: &'static str) -> FieldWrite {
+		sized_field(name, 8, 8)
 	}
 
 	#[test]
 	fn canonical_order_and_offsets() {
-		let layout = Layout::default().with_writes(0, (8, 8), &[("tint", 0, 4, 4), f64_field("opacity"), ("flag", 0, 1, 1)]);
+		let layout = Layout::default().with_writes(0, (8, 8), &[sized_field("tint", 4, 4), f64_field("opacity"), sized_field("flag", 1, 1)]);
 		assert_eq!(layout.offset_of("opacity", 0), Some(8));
 		assert_eq!(layout.offset_of("tint", 0), Some(16));
 		assert_eq!(layout.offset_of("flag", 0), Some(20));
@@ -465,7 +601,7 @@ mod tests {
 	#[should_panic(expected = "two different sizes")]
 	fn size_conflicts_panic() {
 		let layout = Layout::default().with_writes(0, (8, 8), &[f64_field("opacity")]);
-		layout.with_writes(0, (8, 8), &[("opacity", 0, 4, 4)]);
+		layout.with_writes(0, (8, 8), &[sized_field("opacity", 4, 4)]);
 	}
 
 	#[test]

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

@@ -1241,16 +1241,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 		let write_descs: Vec<TokenStream2> = shape
 			.write_markers
 			.iter()
-			.map(|marker| {
-				quote! {
-					(
-						<#marker as #core_types::attribute::Attribute>::NAME,
-						0u8,
-						::core::mem::size_of::<<#marker as #core_types::attribute::Attribute>::Value<'static>>(),
-						::core::mem::align_of::<<#marker as #core_types::attribute::Attribute>::Value<'static>>(),
-					)
-				}
-			})
+			.map(|marker| quote!(#core_types::record::FieldWrite::of::<#marker>(0)))
 			.collect();
 		let element_dims = match &shape.element_write {
 			Some(ty) => quote!((::core::mem::size_of::<#ty>(), ::core::mem::align_of::<#ty>())),
@@ -1306,7 +1297,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 			#[automatically_derived]
 			impl<#(#data_field_generic_idents,)* #(#node_generics,)*> #mod_name::#struct_name<#(#struct_type_params,)*> {
 				#[allow(clippy::too_many_arguments)]
-				#vis fn wire(#(#edge_args,)* #(#carrier_layout_param)*) -> Self {
+				#vis fn new(#(#edge_args,)* #(#carrier_layout_param)*) -> Self {
 					#layout_binding
 					#plan_binding
 					#(#read_inits)*
@@ -1905,7 +1896,7 @@ fn record_entries_tokens(parsed: &ParsedNodeFn, struct_name: &Ident, regular_fie
 					}
 					let mut inputs = inputs.into_iter();
 					#(#downcasts)*
-					let __node = #struct_name::wire(#(#names,)* #(#wire_layout_arg)*);
+					let __node = #struct_name::new(#(#names,)* #(#wire_layout_arg)*);
 					#construct_output
 				},
 			}]

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

@@ -135,7 +135,16 @@ mod tests {
 	}
 
 	fn f64_layout(names: &[&'static str]) -> Layout {
-		let writes: Vec<(&'static str, u8, usize, usize)> = names.iter().map(|name| (*name, 0, 8, 8)).collect();
+		let writes: Vec<core_types::record::FieldWrite> = names
+			.iter()
+			.map(|name| core_types::record::FieldWrite {
+				name,
+				level: 0,
+				size: 8,
+				align: 8,
+				read_erased: <Opacity as AttributeMarker>::read_erased,
+			})
+			.collect();
 		Layout::default().with_writes(0, (8, 8), &writes)
 	}
 
@@ -168,8 +177,8 @@ mod tests {
 		let stacked = multiply_opacity_layout(&modified);
 		reserve_for(&[&source_layout, &modified, &stacked]);
 
-		let chain = MultiplyOpacityNode::wire(
-			MultiplyOpacityNode::wire(bare_source(&source_layout, 2.), ValueNode(0.5), &source_layout),
+		let chain = MultiplyOpacityNode::new(
+			MultiplyOpacityNode::new(bare_source(&source_layout, 2.), ValueNode(0.5), &source_layout),
 			ValueNode(0.5),
 			&modified,
 		);
@@ -193,7 +202,7 @@ mod tests {
 		let measured = measure_layout(&source_layout);
 		reserve_for(&[&source_layout, &measured]);
 
-		let chain = MeasureNode::wire(bare_source(&source_layout, -2.), &source_layout);
+		let chain = MeasureNode::new(bare_source(&source_layout, -2.), &source_layout);
 		let GPoll::Final(value) = chain.eval(&ctx) else {
 			panic!("expected a final record");
 		};
@@ -214,7 +223,7 @@ mod tests {
 		let measured = measure_layout(&modified);
 		reserve_for(&[&source_layout, &modified, &measured]);
 
-		let chain = MeasureNode::wire(MultiplyOpacityNode::wire(bare_source(&source_layout, -2.), ValueNode(0.5), &source_layout), &modified);
+		let chain = MeasureNode::new(MultiplyOpacityNode::new(bare_source(&source_layout, -2.), ValueNode(0.5), &source_layout), &modified);
 		let GPoll::Final(value) = chain.eval(&ctx) else {
 			panic!("expected a final record");
 		};
@@ -235,13 +244,13 @@ mod tests {
 		let shaded = shade_layout(&modified);
 		reserve_for(&[&source_layout, &modified, &shaded]);
 
-		let bare = ShadeNode::wire(bare_source(&source_layout, 4.), &source_layout);
+		let bare = ShadeNode::new(bare_source(&source_layout, 4.), &source_layout);
 		let GPoll::Final(value) = bare.eval(&ctx) else {
 			panic!("expected a final record");
 		};
 		assert_eq!(unsafe { value.rec().element::<f64>() }, 4.);
 
-		let chain = ShadeNode::wire(MultiplyOpacityNode::wire(bare_source(&source_layout, 4.), ValueNode(0.5), &source_layout), &modified);
+		let chain = ShadeNode::new(MultiplyOpacityNode::new(bare_source(&source_layout, 4.), ValueNode(0.5), &source_layout), &modified);
 		let GPoll::Final(value) = chain.eval(&ctx) else {
 			panic!("expected a final record");
 		};
@@ -263,7 +272,7 @@ mod tests {
 		let u32_faded = fade_layout(&u32_source);
 		reserve_for(&[&f64_source, &f64_faded, &u32_source, &u32_faded]);
 
-		let wide = FadeNode::wire(bare_source(&f64_source, 8.), ValueNode(0.5), &f64_source);
+		let wide = FadeNode::new(bare_source(&f64_source, 8.), ValueNode(0.5), &f64_source);
 		let GPoll::Final(value) = wide.eval(&ctx) else {
 			panic!("expected a final record");
 		};
@@ -271,7 +280,7 @@ mod tests {
 		assert_eq!(unsafe { rec.element::<f64>() }, 8.);
 		assert_eq!(unsafe { rec.read::<f64>(f64_faded.offset_of(Opacity::NAME, 0).unwrap()) }, 0.5);
 
-		let narrow = FadeNode::wire(
+		let narrow = FadeNode::new(
 			RecordSourceNode {
 				frame_bytes: frame_bytes(&u32_source),
 				element: 7u32,
@@ -299,7 +308,7 @@ mod tests {
 		let layout = source_opacity_layout();
 		reserve_for(&[&layout]);
 
-		let node = SourceOpacityNode::wire(ValueNode(3.), ValueNode(0.25));
+		let node = SourceOpacityNode::new(ValueNode(3.), ValueNode(0.25));
 		assert_eq!(Node::<ContextImpl>::layout(&node), Some(&layout));
 		let GPoll::Final(value) = node.eval(&ctx) else {
 			panic!("expected a final record");
@@ -320,7 +329,7 @@ mod tests {
 		let modified = multiply_opacity_layout(&source_layout);
 		reserve_for(&[&source_layout, &modified]);
 
-		let chain = MultiplyOpacityNode::wire(
+		let chain = MultiplyOpacityNode::new(
 			RecordSourceNode {
 				frame_bytes: frame_bytes(&source_layout),
 				element: 1.,
@@ -347,13 +356,13 @@ mod tests {
 		let modified = checked_multiply_opacity_layout(&source_layout);
 		reserve_for(&[&source_layout, &modified]);
 
-		let ok = CheckedMultiplyOpacityNode::wire(bare_source(&source_layout, 1.), ValueNode(0.5), &source_layout);
+		let ok = CheckedMultiplyOpacityNode::new(bare_source(&source_layout, 1.), ValueNode(0.5), &source_layout);
 		let GPoll::Final(value) = ok.eval(&ctx) else {
 			panic!("expected a final record");
 		};
 		assert_eq!(unsafe { value.rec().read::<f64>(modified.offset_of(Opacity::NAME, 0).unwrap()) }, 0.5);
 
-		let failing = CheckedMultiplyOpacityNode::wire(bare_source(&source_layout, 1.), ValueNode(-1.), &source_layout);
+		let failing = CheckedMultiplyOpacityNode::new(bare_source(&source_layout, 1.), ValueNode(-1.), &source_layout);
 		let GPoll::Error(error) = failing.eval(&ctx) else {
 			panic!("expected an error");
 		};
@@ -384,8 +393,8 @@ mod tests {
 		let scaled = scale_layout(&modified);
 		reserve_for(&[&source_layout, &modified, &scaled]);
 
-		let chain = ScaleNode::wire(
-			MultiplyOpacityNode::wire(bare_source(&source_layout, 2.), ValueNode(0.5), &source_layout),
+		let chain = ScaleNode::new(
+			MultiplyOpacityNode::new(bare_source(&source_layout, 2.), ValueNode(0.5), &source_layout),
 			StaticLendNode(&FACTOR),
 			&modified,
 		);
@@ -409,8 +418,8 @@ mod tests {
 		let relabeled = label_layout(&labeled);
 		reserve_for(&[&source_layout, &labeled, &relabeled]);
 
-		let chain = LabelNode::wire(
-			LabelNode::wire(bare_source(&source_layout, 1.), ValueNode(String::from("a")), &source_layout),
+		let chain = LabelNode::new(
+			LabelNode::new(bare_source(&source_layout, 1.), ValueNode(String::from("a")), &source_layout),
 			ValueNode(String::from("b")),
 			&labeled,
 		);
@@ -425,7 +434,7 @@ mod tests {
 	#[test]
 	fn census_fills_reference_defaults_from_static_data() {
 		let source = f64_layout(&[]);
-		let labeled = Layout::default().with_writes(0, (8, 8), &[(Label::NAME, 0, 16, 8)]);
+		let labeled = Layout::default().with_writes(0, (8, 8), &[core_types::record::FieldWrite::of::<Label>(0)]);
 
 		let plan = core_types::record::SourcePlan::new(&source, &labeled).unwrap();
 		let record = [5f64];
@@ -444,6 +453,35 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn record_monitor_forwards_and_captures_for_the_introspection_window() {
+		let mut arena = Arena::new(1024).unwrap();
+		let generations = [];
+
+		let layout = f64_layout(&["opacity"]);
+		reserve_for(&[&layout, &layout]);
+
+		let monitor = core_types::record::RecordMonitor::new(f64_record_source(&layout, 4., vec![(layout.offset_of("opacity", 0).unwrap(), 0.25)]), &layout);
+		{
+			let scope = scope_fixture(&generations, &arena);
+			let ctx = ContextImpl::root(&scope);
+			let GPoll::Final(value) = monitor.eval(&ctx) else {
+				panic!("expected a final record");
+			};
+			assert_eq!(unsafe { value.rec().element::<f64>() }, 4.);
+		}
+
+		let capture = Node::<ContextImpl>::serialize(&monitor).unwrap();
+		let capture = capture.downcast_ref::<core_types::record::RecordCapture>().unwrap();
+		let fields = capture.materialize(&arena).unwrap();
+		assert_eq!(fields.len(), 1);
+		assert_eq!(fields[0].0, "opacity");
+		assert_eq!(*fields[0].1.as_any().downcast_ref::<f64>().unwrap(), 0.25);
+
+		arena.reset();
+		assert!(capture.materialize(&arena).is_none(), "a dead generation materializes to nothing");
+	}
+
 	#[test]
 	fn routing_unions_branch_layouts_and_fills_census_defaults() {
 		let arena = Arena::new(1024).unwrap();
@@ -459,8 +497,8 @@ mod tests {
 		let taken = |second: bool| {
 			PickNode::new(
 				ValueNode(second),
-				RecordSource::wire(f64_record_source(&layout_a, 1., vec![(layout_a.offset_of("opacity", 0).unwrap(), 0.5)]), &layout_a, &union),
-				RecordSource::wire(f64_record_source(&layout_b, 3., vec![(layout_b.offset_of("length", 0).unwrap(), 3.)]), &layout_b, &union),
+				RecordSource::new(f64_record_source(&layout_a, 1., vec![(layout_a.offset_of("opacity", 0).unwrap(), 0.5)]), &layout_a, &union),
+				RecordSource::new(f64_record_source(&layout_b, 3., vec![(layout_b.offset_of("length", 0).unwrap(), 3.)]), &layout_b, &union),
 			)
 		};
 
@@ -495,8 +533,8 @@ mod tests {
 
 		let chain = HoldFirstNode::new(
 			ValueNode(false),
-			RecordSource::wire(f64_record_source(&layout_a, 1., vec![(layout_a.offset_of("opacity", 0).unwrap(), 0.5)]), &layout_a, &union),
-			RecordSource::wire(f64_record_source(&layout_b, 3., vec![(layout_b.offset_of("length", 0).unwrap(), 3.)]), &layout_b, &union),
+			RecordSource::new(f64_record_source(&layout_a, 1., vec![(layout_a.offset_of("opacity", 0).unwrap(), 0.5)]), &layout_a, &union),
+			RecordSource::new(f64_record_source(&layout_b, 3., vec![(layout_b.offset_of("length", 0).unwrap(), 3.)]), &layout_b, &union),
 		);
 
 		let GPoll::Final(value) = chain.eval(&ctx) else {
@@ -520,7 +558,7 @@ mod tests {
 		let base = stack::push(0);
 		stack::pop(base);
 
-		let chain = ForwardRecordNode::new(RecordSource::wire(
+		let chain = ForwardRecordNode::new(RecordSource::new(
 			f64_record_source(&layout, 4., vec![(layout.offset_of("opacity", 0).unwrap(), 0.25)]),
 			&layout,
 			&layout.clone(),
@@ -545,7 +583,7 @@ mod tests {
 		let layout = f64_layout(&["opacity"]);
 		reserve_for(&[&layout]);
 
-		let chain = ForwardRecordNode::new(RecordSource::wire(
+		let chain = ForwardRecordNode::new(RecordSource::new(
 			RecordSourceNode {
 				frame_bytes: frame_bytes(&layout),
 				element: 4.,