Let async source kernels write attributes through their frame claim

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

@@ -291,7 +291,7 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 	}));
 
 	let async_source = parsed.injects_async_source_fields();
-	let slot_value_type = crate::codegen::classify::substitute_lifetimes(&slot_value_type(output_type), "'static");
+	let slot_value_type = crate::codegen::classify::substitute_lifetimes(&crate::codegen::classify::slot_static_type(output_type), "'static");
 	let slot_field = async_source
 		.then(|| quote! { pub(super) slot: std::sync::Arc<std::sync::Mutex<std::collections::HashMap<u64, Option<gcore::gpoll::GPoll<#slot_value_type>>>>> })
 		.into_iter();
@@ -1206,7 +1206,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 
 	// The slot persists the plain value even on record wires, so the Clone
 	// bound targets the slot type, not the (possibly lifted) trait output.
-	let slot_ty = slot_value_type(&parsed.output_type);
+	let slot_ty = crate::codegen::classify::slot_static_type(&parsed.output_type);
 	let mut async_bounds = match (async_fn, future_kernel) {
 		(false, false) => Vec::new(),
 		(false, true) => vec![quote!(#slot_ty: Clone)],
@@ -1721,9 +1721,10 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 	let kernel_fields: Vec<&&ParsedField> = regular_fields.iter().filter(|field| !injected_name(&field.pat_ident.ident)).collect();
 	// A bare `Attr<M>` in the return type cannot elide its lifetime, so the
 	// kernel gets a fresh one; reference-valued writes name their real
-	// lifetime explicitly and pass through untouched.
-	let attr_injected = record_io.then(|| inject_attr_lifetimes(&parsed.output_type)).flatten();
-	let attr_lifetime = attr_injected.is_some().then(|| quote!('__attr,));
+	// lifetime explicitly and pass through untouched. An async source's value
+	// outlives the evaluation, so its writes are `'static` instead.
+	let attr_injected = record_io.then(|| inject_attr_lifetimes(&parsed.output_type, if async_source { "'static" } else { "'__attr" })).flatten();
+	let attr_lifetime = (attr_injected.is_some() && !async_source).then(|| quote!('__attr,));
 	let lane_injected = gather_carrier
 		.then(|| crate::codegen::classify::inject_lane_lifetime(attr_injected.as_ref().unwrap_or(&parsed.output_type)))
 		.flatten();
@@ -1767,7 +1768,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 			let params = snapshot_param.chain(data_kernel_params).chain(value_kernel_params);
 			quote! {
 				#[allow(clippy::too_many_arguments, clippy::type_complexity)]
-				#vis async fn #fn_name<#(#kernel_generics,)*>(#(#params),*) -> #output_type #fn_where #body
+				#vis async fn #fn_name<#(#kernel_generics,)*>(#(#params),*) -> #kernel_output #fn_where #body
 			}
 		}
 	};
@@ -1835,10 +1836,47 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 			#clamp
 		}
 	});
+	// A writing source's carrier is its record-io carrier, so the fields it
+	// passes through ride the record plan rather than the flip one.
+	let carried_prelude = carried_prelude.or_else(|| {
+		(record_io && async_source && !skips_carrier).then(|| {
+			let field = regular_fields[0];
+			let name = &field.pat_ident.ident;
+			let ty = carrier_read_ty.expect("a carrying record source reads a concrete element");
+			quote! {
+				let __src = match __cell.eval_input(0, &self.#name, __input, __frame.frames()) {
+					Ok(value) => value,
+					Err(interrupt) => return interrupt.into(),
+				};
+				let __src_rec = self.__carrier.rec(&__src);
+				unsafe { __frame.carry(__src_rec, &self.__plan) };
+				let #name: #ty = unsafe { #core_types::record::read_element(__src_rec) };
+			}
+		})
+	});
 	// Async slots persist plain values across evaluations; the source lifts
 	// the slot value onto its record wire at every merge point, into the
 	// carried frame when the node has a carrier.
-	let merge_lifted = |poll: TokenStream2| quote!(__cell.merge(__frame.lift_served(#poll, #core_types::context::ExtractArena::arena(__input))));
+	// A writing source stores the kernel's whole tuple as that plain value:
+	// the lift writes the attributes through the claim, then lifts the
+	// element, the shape the sync record tail closes with.
+	let source_writes = (record_io && async_source && !write_markers.is_empty()).then(|| {
+		let binders: Vec<Ident> = (0..write_markers.len()).map(|index| format_ident!("__attr_{index}")).collect();
+		let slots: Vec<Ident> = (0..write_markers.len()).map(|index| format_ident!("__write_{index}")).collect();
+		quote! {
+			.map(|(__element #(, #core_types::attribute::Attr(#binders))*)| {
+				#(unsafe { __frame.attr_at(self.#slots, #binders) };)*
+				__element
+			})
+		}
+	});
+	let merge_lifted = |poll: TokenStream2| match &source_writes {
+		None => quote!(__cell.merge(__frame.lift_served(#poll, #core_types::context::ExtractArena::arena(__input)))),
+		Some(writes) => quote! {{
+			let __lifted = (#poll) #writes;
+			__cell.merge(__frame.lift_served(__lifted, #core_types::context::ExtractArena::arena(__input)))
+		}},
+	};
 	// The claim drops with the frame still claimed, so a valueless exit needs
 	// no closing of its own.
 	let pending_return = quote!(#core_types::gpoll::GPoll::Pending);
@@ -2568,6 +2606,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 			let slot = format_ident!("__mat_cache_{index}");
 			quote!(#slot: ::core::default::Default::default(),)
 		});
+		let slot_default = async_source.then(|| quote!(slot: ::core::default::Default::default(),)).into_iter();
 		// A ranked input's element generic rides the struct as a phantom
 		// parameter, so the constructor declares and initializes it too.
 		let carried_type_params: Vec<&Ident> = struct_type_params
@@ -2610,6 +2649,7 @@ pub(crate) fn generate_node_impl(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 						#(#read_names)*
 						#(#write_defaults)*
 						#(#mat_cache_defaults)*
+						#(#slot_default)*
 					}
 				}
 			}

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

@@ -206,18 +206,19 @@ pub(crate) fn substitute_routing_record(output: &Type, generic: &Ident, core_typ
 	ty
 }
 
-pub(crate) fn inject_attr_lifetimes(output: &Type) -> Option<Type> {
-	struct Injector {
+pub(crate) fn inject_attr_lifetimes(output: &Type, lifetime: &str) -> Option<Type> {
+	struct Injector<'a> {
 		changed: bool,
+		lifetime: &'a str,
 	}
 
-	impl VisitMut for Injector {
+	impl VisitMut for Injector<'_> {
 		fn visit_path_segment_mut(&mut self, segment: &mut syn::PathSegment) {
 			if segment.ident == "Attr"
 				&& let PathArguments::AngleBracketed(args) = &mut segment.arguments
 				&& !args.args.iter().any(|arg| matches!(arg, GenericArgument::Lifetime(_)))
 			{
-				args.args.insert(0, GenericArgument::Lifetime(Lifetime::new("'__attr", proc_macro2::Span::call_site())));
+				args.args.insert(0, GenericArgument::Lifetime(Lifetime::new(self.lifetime, proc_macro2::Span::call_site())));
 				self.changed = true;
 			}
 			syn::visit_mut::visit_path_segment_mut(self, segment);
@@ -225,7 +226,7 @@ pub(crate) fn inject_attr_lifetimes(output: &Type) -> Option<Type> {
 	}
 
 	let mut ty = output.clone();
-	let mut injector = Injector { changed: false };
+	let mut injector = Injector { changed: false, lifetime };
 	injector.visit_type_mut(&mut ty);
 	injector.changed.then_some(ty)
 }
@@ -281,9 +282,11 @@ pub(crate) fn unbounded_generic(parsed: &ParsedNodeFn, ty: &Type) -> Option<Iden
 }
 
 pub(crate) fn record_shape(parsed: &ParsedNodeFn) -> Option<RecordShape> {
+	let source = is_async_source(parsed);
 	let value = match kernel_kind(&parsed.output_type) {
 		KernelKind::Plain => parsed.output_type.clone(),
 		KernelKind::Interrupt(inner) => inner,
+		_ if source => slot_value_type(&parsed.output_type),
 		_ => return None,
 	};
 	let writes = record_writes(&value);
@@ -291,7 +294,9 @@ pub(crate) fn record_shape(parsed: &ParsedNodeFn) -> Option<RecordShape> {
 	if !has_reads && writes.is_none() {
 		return None;
 	}
-	if parsed.is_async {
+	// An async source's slot stores the kernel's plain tuple; the per-eval lift
+	// writes it through the claim, and the reads have no wire to bind against.
+	if source && (has_reads || writes.is_none()) {
 		return None;
 	}
 	let carrier_field = parsed.fields.first()?;
@@ -371,6 +376,11 @@ pub(crate) fn record_shape(parsed: &ParsedNodeFn) -> Option<RecordShape> {
 	if matches!(carrier, RecordCarrier::None) && !removes.is_empty() {
 		return None;
 	}
+	// The byte-carried token never becomes a value, so it cannot cross a
+	// future boundary.
+	if source && matches!(carrier, RecordCarrier::Token) {
+		return None;
+	}
 	Some(RecordShape { carrier })
 }
 
@@ -605,6 +615,24 @@ pub(crate) fn is_source_kernel(output: &Type) -> bool {
 	matches!(kernel_kind(output), KernelKind::Future(_) | KernelKind::FutureInterrupt(_))
 }
 
+/// A kernel whose value completes off the evaluation: an `async fn` or a
+/// `SourceFuture` return. Its slot persists a plain value across evaluations,
+/// so nothing it returns may borrow the arena.
+pub(crate) fn is_async_source(parsed: &ParsedNodeFn) -> bool {
+	parsed.is_async || is_source_kernel(&parsed.output_type)
+}
+
+/// The type an async source's slot persists. A writing source's value outlives
+/// the evaluation, so every lifetime it names, including a bare `Attr<M>`'s
+/// elided one, is `'static`.
+pub(crate) fn slot_static_type(output: &Type) -> Type {
+	let value = slot_value_type(output);
+	match record_writes(&value).is_some() {
+		true => substitute_lifetimes(&inject_attr_lifetimes(&value, "'static").unwrap_or_else(|| value.clone()), "'static"),
+		false => value,
+	}
+}
+
 pub(crate) enum KernelKind {
 	Plain,
 	Interrupt(Type),

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

@@ -754,6 +754,49 @@ mod tests {
 		);
 	}
 
+	#[test]
+	fn bridge_record_write_async_source() {
+		let node = assert_bridge(
+			quote!(category("")),
+			quote!(
+				async fn set_opacity_async(_: impl Ctx, val: f64) -> (f64, Attr<Opacity>) {
+					(val, Attr(1.))
+				}
+			),
+		);
+		assert!(matches!(node_kind(&node), NodeKind::RecordIo), "a writing async source takes the record tail");
+		assert!(matches!(node.effect, Effect::AsyncSource), "the writes do not change the effect axis");
+	}
+
+	#[test]
+	fn bridge_record_fresh_async_source() {
+		assert_bridge(
+			quote!(category("")),
+			quote!(
+				async fn make_async(_: impl Ctx, _: (), fill: f64) -> (f64, Attr<Opacity>) {
+					(fill, Attr(1.))
+				}
+			),
+		);
+	}
+
+	/// An async source's element is the value its slot stores, so a byte-carried
+	/// generic token has no form here.
+	#[test]
+	fn a_generic_token_carrier_has_no_async_source_form() {
+		let mut parsed = parse_node_fn(
+			quote!(category("")),
+			quote!(
+				async fn tag<T>(_: impl Ctx, val: T) -> (T, Attr<Opacity>) {
+					(val, Attr(1.))
+				}
+			),
+		)
+		.unwrap();
+		parsed.replace_impl_trait_in_input();
+		assert!(record_shape(&parsed).is_none());
+	}
+
 	#[test]
 	fn bridge_routing() {
 		assert_bridge(

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

@@ -44,8 +44,9 @@ fn validate_record_io(parsed: &ParsedNodeFn) {
 		return;
 	}
 
-	if parsed.is_async || crate::codegen::is_source_kernel(&parsed.output_type) {
-		emit_error!(parsed.output_type.span(), "attribute io is not supported on async source kernels");
+	let async_source = crate::codegen::classify::is_async_source(parsed);
+	if async_source && has_reads {
+		emit_error!(parsed.fn_name.span(), "attribute reads are not supported on async source kernels, only writes");
 	}
 	if crate::codegen::is_poll_kernel(&parsed.output_type) {
 		emit_error!(parsed.output_type.span(), "attribute io needs a plain or `Result<_, Interrupt>` kernel, not a `GPoll` one");
@@ -125,6 +126,9 @@ fn validate_record_io(parsed: &ParsedNodeFn) {
 		(false, ParsedFieldType::Node(NodeParsedField { output_type, .. })) if lazy_carrier => crate::codegen::unbounded_generic(parsed, output_type),
 		_ => None,
 	};
+	if async_source && token.is_some() {
+		emit_error!(carrier.pat_ident.span(), "an async source's element crosses the future boundary as a value; a passthrough generic element has none");
+	}
 	let element = writes.as_ref().map(|writes| &writes.element).unwrap_or(&value);
 	match &token {
 		Some(token) => {