Export the async-source flag from the macro and dispatch superseded type deltas in tests

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

@@ -20,6 +20,8 @@ pub struct NodeMetadata {
 	pub context_features: Vec<ContextFeature>,
 	pub memoize: bool,
 	pub inject_scope: bool,
+	/// The macro appended its hidden `_runtime` and `_source` fields as the last two entries of `fields`.
+	pub async_source_fields: bool,
 }
 
 // Translation struct between macro and definition

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

@@ -18,7 +18,6 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 		fn_generics,
 		input,
 		output_type,
-		is_async,
 		fields,
 		description,
 		..
@@ -113,7 +112,7 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 		quote! { pub(super) #name: #r#gen }
 	});
 
-	let async_source = *is_async || is_source_kernel(output_type);
+	let async_source = parsed.injects_async_source_fields();
 	let slot_value_type = slot_value_type(output_type);
 	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>>>>> })
@@ -356,6 +355,7 @@ pub(crate) fn generate_node_code(crate_ident: &CrateIdent, parsed: &ParsedNodeFn
 					context_features: vec![#(ContextFeature::#context_features,)*],
 					memoize: #memoize_flag,
 					inject_scope: #inject_scope_flag,
+					async_source_fields: #async_source,
 					fields: vec![
 						#(
 							FieldMetadata {

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

@@ -998,7 +998,7 @@ pub fn new_node_fn(attr: TokenStream2, item: TokenStream2) -> syn::Result<TokenS
 	let crate_ident = CrateIdent::default();
 	let mut parsed_node = parse_node_fn(attr, item.clone()).map_err(|e| Error::new(e.span(), format!("Failed to parse node function:\n{e}")))?;
 	parsed_node.replace_impl_trait_in_input();
-	if parsed_node.is_async || crate::codegen::is_source_kernel(&parsed_node.output_type) {
+	if parsed_node.injects_async_source_fields() {
 		let core_types = crate_ident.gcore()?.clone();
 		parsed_node.inject_async_source_fields(&core_types);
 	}
@@ -1030,6 +1030,10 @@ impl ParsedNodeFn {
 		}
 	}
 
+	pub fn injects_async_source_fields(&self) -> bool {
+		self.is_async || crate::codegen::is_source_kernel(&self.output_type)
+	}
+
 	pub fn inject_async_source_fields(&mut self, core_types: &TokenStream2) {
 		let hidden_field = |name: &str, ty: Type, value_source: ParsedValueSource| ParsedField {
 			pat_ident: PatIdent {

node-graph/preprocessor/src/lib.rs

@@ -116,7 +116,13 @@ impl Preprocessor {
 		for (id, metadata) in core_types::registry::NODE_METADATA.lock().unwrap().iter() {
 			let id = id.clone();
 
-			let NodeMetadata { fields, memoize, inject_scope, .. } = metadata;
+			let NodeMetadata {
+				fields,
+				memoize,
+				inject_scope,
+				async_source_fields,
+				..
+			} = metadata;
 			let Some(implementations) = node_registry.get(&id) else { continue };
 			let valid_call_args: HashSet<_> = implementations.iter().map(|entry| entry.io.call_argument.clone()).collect();
 			let first_node_io = implementations.first().map(|entry| &entry.io).unwrap_or(const { &NodeIOTypes::empty() });
@@ -132,30 +138,12 @@ impl Preprocessor {
 			}
 
 			let mut inputs: Vec<_> = node_inputs(fields, first_node_io);
-			let input_count = inputs.len();
+			let wrapper_input_count = inputs.len() - if *async_source_fields { 2 } else { 0 };
 
-			// The node macro appends a `RuntimeHandle` scope field and a `SourceId` reflection field to every
-			// async or source kernel. They are resolved inside the substitution network, so the wrapper neither
-			// exposes them nor carries their context modification.
-			let injected_field_count = match &fields[..] {
-				[.., runtime, source] if matches!(runtime.value_source, RegistryValueSource::Scope(_)) && matches!(source.value_source, RegistryValueSource::SourceId) => 2,
-				_ => 0,
-			};
-			let wrapper_input_count = input_count - injected_field_count;
-
-			inputs.truncate(wrapper_input_count);
-
-			// The injected fields go straight onto the inner node, so the source it reflects is recorded on the
-			// node that consumes it rather than on a forwarding node that later gets dissolved.
-			let network_inputs = (0..input_count)
-				.map(|i| {
-					if i < wrapper_input_count {
-						NodeInput::node(NodeId(i as u64), 0)
-					} else {
-						injected_field_input(&fields[i])
-					}
-				})
-				.collect();
+			// The injected fields must not surface as wrapper inputs, and the `_source` reflection must sit on
+			// the kernel itself so the source id lands on a node that survives flattening.
+			let injected_inputs = inputs.split_off(wrapper_input_count);
+			let network_inputs = (0..wrapper_input_count).map(|i| NodeInput::node(NodeId(i as u64), 0)).chain(injected_inputs).collect();
 
 			let passthrough_node = ops::passthrough::IDENTIFIER;
 
@@ -207,7 +195,7 @@ impl Preprocessor {
 				})
 				.collect();
 
-			if generated_nodes == 0 && !memoize && !inject_scope && injected_field_count == 0 {
+			if generated_nodes == 0 && !memoize && !inject_scope && !async_source_fields {
 				continue;
 			}
 
@@ -279,14 +267,6 @@ impl Preprocessor {
 	}
 }
 
-fn injected_field_input(field: &registry::FieldMetadata) -> NodeInput {
-	match field.value_source {
-		RegistryValueSource::Scope(data) => NodeInput::scope(data),
-		RegistryValueSource::SourceId => NodeInput::Reflection(DocumentNodeMetadata::SourceId),
-		_ => NodeInput::value(TaggedValue::None, false),
-	}
-}
-
 pub fn node_inputs(fields: &[registry::FieldMetadata], first_node_io: &NodeIOTypes) -> Vec<NodeInput> {
 	fields
 		.iter()