Wire the async source runtimes into the hosts

editor/src/node_graph_executor.rs

@@ -59,6 +59,9 @@ pub struct NodeGraphExecutor {
 	runtime_io: NodeRuntimeIO,
 	current_execution_id: u64,
 	futures: VecDeque<(u64, ExecutionContext)>,
+	/// The most recently consumed plain render execution, kept so a runtime-replayed response with the same id
+	/// (sent after an async source completion) finds its context again.
+	last_execution_context: Option<(u64, ExecutionContext)>,
 	node_graph_hash: u64,
 	/// Full path from the root document network to the node currently being inspected by the Data panel, or empty if nothing is selected.
 	/// The last element is the inspect target itself; preceding elements identify the nested subnetwork the node lives in,
@@ -108,6 +111,7 @@ impl NodeGraphExecutor {
 		let node_executor = Self {
 			futures: Default::default(),
 			runtime_io: NodeRuntimeIO::with_channels(request_sender, response_receiver),
+			last_execution_context: None,
 			node_graph_hash: 0,
 			current_execution_id: 0,
 			previous_node_to_inspect: Vec::new(),
@@ -377,9 +381,19 @@ impl NodeGraphExecutor {
 
 					let execution_context = if self.futures.front().is_some_and(|&(queued_execution_id, _)| queued_execution_id == execution_id) {
 						let (_, execution_context) = self.futures.pop_front().expect("front was just matched");
+						self.last_execution_context = Some((execution_id, execution_context.clone()));
 						execution_context
 					} else {
-						panic!("InvalidGenerationId")
+						// A runtime-replayed response re-uses an already consumed id; only plain renders may re-apply.
+						match &self.last_execution_context {
+							Some((last_execution_id, execution_context)) if *last_execution_id == execution_id => {
+								if execution_context.export_config.is_some() || execution_context.measure_fill.is_some() {
+									continue;
+								}
+								execution_context.clone()
+							}
+							_ => panic!("InvalidGenerationId"),
+						}
 					};
 
 					// TODO: Eventually remove this document upgrade code

editor/src/node_graph_executor/runtime.rs

@@ -17,7 +17,7 @@ use graphene_std::ops::ConvertAsync;
 use graphene_std::platform_application_io::canvas_utils::{Canvas, CanvasSurface, CanvasSurfaceHandle};
 use graphene_std::raster_types::Raster;
 use graphene_std::renderer::{Render, RenderParams, RenderSvgSegmentList, SvgRender, SvgSegment};
-use graphene_std::runtime::{DynGraphRuntime, DynSpawner, GraphRuntime, NoopSpawner, RuntimeHandle};
+use graphene_std::runtime::{DynGraphRuntime, DynNotifier, DynSpawner, GraphRuntime, RuntimeHandle, SourceFuture, Spawner};
 use graphene_std::transform::RenderQuality;
 use graphene_std::vector::Vector;
 use graphene_std::vector::style::RenderMode;
@@ -43,6 +43,8 @@ pub struct NodeRuntime {
 	update_thumbnails: bool,
 	#[expect(dead_code, reason = "read once the host wires a notifier onto the runtime")]
 	graph_runtime: Arc<DynGraphRuntime>,
+	/// The last plain render request, replayed when an async source completion marks the graph dirty.
+	last_render: Option<ExecutionRequest>,
 
 	editor_api: Arc<PlatformEditorApi>,
 	resources: ResourceRegistry,
@@ -122,9 +124,57 @@ impl NodeGraphUpdateSender for InternalNodeGraphUpdateSender {
 // TODO: Replace with `core::cell::LazyCell` (<https://doc.rust-lang.org/core/cell/struct.LazyCell.html>) or similar
 pub static NODE_RUNTIME: once_cell::sync::Lazy<Mutex<Option<NodeRuntime>>> = once_cell::sync::Lazy::new(|| Mutex::new(None));
 
+#[cfg(not(target_family = "wasm"))]
+pub struct TokioSpawner(Option<tokio::runtime::Runtime>);
+
+#[cfg(not(target_family = "wasm"))]
+impl TokioSpawner {
+	pub fn new() -> Self {
+		Self(Some(tokio::runtime::Runtime::new().expect("Failed to start the async source runtime")))
+	}
+}
+
+#[cfg(not(target_family = "wasm"))]
+impl Default for TokioSpawner {
+	fn default() -> Self {
+		Self::new()
+	}
+}
+
+#[cfg(not(target_family = "wasm"))]
+impl Spawner for TokioSpawner {
+	fn spawn(&self, task: SourceFuture) {
+		self.0.as_ref().expect("runtime lives until drop").spawn(task);
+	}
+}
+
+/// Dropping a tokio runtime blocks on its tasks, which panics inside an async context; the tests drop
+/// [`NodeRuntime`] from one, so shut down in the background instead.
+#[cfg(not(target_family = "wasm"))]
+impl Drop for TokioSpawner {
+	fn drop(&mut self) {
+		if let Some(runtime) = self.0.take() {
+			runtime.shutdown_background();
+		}
+	}
+}
+
+#[cfg(target_family = "wasm")]
+pub struct WasmSpawner;
+
+#[cfg(target_family = "wasm")]
+impl Spawner for WasmSpawner {
+	fn spawn(&self, task: SourceFuture) {
+		wasm_bindgen_futures::spawn_local(task);
+	}
+}
+
 impl NodeRuntime {
 	pub fn new(receiver: Receiver<GraphRuntimeRequest>, sender: Sender<NodeGraphUpdate>) -> Self {
-		let spawner: Box<DynSpawner> = Box::new(NoopSpawner);
+		#[cfg(not(target_family = "wasm"))]
+		let spawner: Box<DynSpawner> = Box::new(TokioSpawner::new());
+		#[cfg(target_family = "wasm")]
+		let spawner: Box<DynSpawner> = Box::new(WasmSpawner);
 		let graph_runtime: Arc<DynGraphRuntime> = Arc::new(GraphRuntime::new(spawner));
 		let mut executor = DynamicExecutor::default();
 		executor.set_runtime(Arc::clone(&graph_runtime));
@@ -138,6 +188,7 @@ impl NodeRuntime {
 			resources: ResourceRegistry::default(),
 			update_thumbnails: true,
 			graph_runtime: Arc::clone(&graph_runtime),
+			last_render: None,
 
 			editor_api: PlatformEditorApi {
 				editor_preferences: Box::new(EditorPreferences::default()),
@@ -188,6 +239,10 @@ impl NodeRuntime {
 					}
 
 					let for_export = execution_request.render_config.for_export;
+					if !for_export {
+						self.last_render = Some(execution_request.clone());
+					}
+
 					execution = Some(request);
 
 					// If we get an export request we always execute it immedeatly otherwise it could get deduplicated
@@ -207,6 +262,10 @@ impl NodeRuntime {
 			eyedropper.render_config.pointer = execution.render_config.pointer;
 		}
 
+		if self.executor.take_dirty() && execution.is_none() {
+			execution = self.last_render.clone().map(GraphRuntimeRequest::ExecutionRequest);
+		}
+
 		let requests = [preferences, graph, eyedropper, execution].into_iter().flatten();
 
 		for request in requests {
@@ -582,6 +641,13 @@ pub(crate) fn replace_application_io(application_io: PlatformApplicationIo) {
 	}
 }
 
+pub fn set_completion_notifier(notifier: Arc<DynNotifier>) {
+	let node_runtime = NODE_RUNTIME.lock();
+	if let Some(node_runtime) = &*node_runtime {
+		node_runtime.graph_runtime.set_notifier(notifier);
+	}
+}
+
 impl NodeRuntime {
 	pub(crate) fn replace_application_io(&mut self, application_io: PlatformApplicationIo) {
 		self.editor_api = PlatformEditorApi {