node: virtual graph endpoints and the Kahn-order BFS sweep (M0b core)

Per docs/zh/plans/render-pipeline-threads.md §3.8:

- oak-node/nodes/graphendpoints.rs: the GraphInput/GraphOutput
  virtual node pair — factory-registered but hidden from every create
  menu, duplicate refused, real value() semantics (the input forwards
  its feed_in row, the output publishes its tex_in as the frame).
  The input endpoint also declares a connectable feed_in port
  (documented deviation: footage/generator sources have no connectable
  inputs, so the walk needs a feeder anchor).
- graph.rs: ensure_endpoints/endpoints/is_endpoint — idempotent,
  identified by type id, default input->output edge only while the
  output's tex_in is free; remove_node refuses endpoints.
- project.rs + serializer.rs: every project graph carries the pair;
  a legacy file without endpoints migrates on load (roundtrip and
  legacy-migration tests, re-save is idempotent).
- traverser.rs: eval_graph_bfs — the endpoint-to-endpoint Kahn
  sweep. Live set = (input's forward cone U its feeder cone) INTERSECT
  (output's backward cone); multi-input nodes dequeue at zero
  in-degree over the live subgraph; deterministic ascending-id ready
  order (Graph::edges is a BTreeSet, so insertion order is
  unrecoverable — documented); time-shifted upstreams pull through
  the shared DFS memo (walk_dfs, factored out of evaluate);
  un-orderable remainder reports a named cycle; missing endpoints /
  unreachable output are errors. Eight BFS tests cover the plan's
  acceptance bullets.
- oak-render: bfs_endpoint_sweep_renders_footage_through_position —
  real clip through a real Position node via the sweep, shifted
  pixels asserted against a reference decode.
- Endpoint names localized in all eight i18n packs; storage/structure
  tests updated for the two extra nodes.

crates/oak-node/src/graph.rs

@@ -183,7 +183,14 @@ impl Graph {
 
 	/// Remove a node and all its edges (C++: ~Node + set_parent(null)
 	/// + disconnect_all side effects — see `// CPP-PARITY: node.cpp`).
+	///
+	/// Refused (`None`) for the virtual endpoints, which every graph keeps
+	/// for its lifetime — removing one would strand the evaluation walk's
+	/// anchor ([`Graph::ensure_endpoints`]).
 	pub fn remove_node(&mut self, id: NodeId) -> Option<Box<dyn NodeBehavior>> {
+		if self.is_endpoint(id) {
+			return None;
+		}
 		let entry = self.take_node(id)?;
 		Some(entry.behavior)
 	}
@@ -228,6 +235,100 @@ impl Graph {
 		ids
 	}
 
+	/// The graph's virtual endpoint pair `(input, output)`, or `None` when
+	/// either is missing. Endpoints are identified by their behavior's
+	/// type id (see [`crate::nodes::graphendpoints`]) — no node id is ever
+	/// stored for them.
+	pub fn endpoints(&self) -> Option<(NodeId, NodeId)> {
+		let input = self.endpoint_of_type(crate::nodes::graphendpoints::GRAPH_INPUT_TYPE_ID)?;
+		let output = self.endpoint_of_type(crate::nodes::graphendpoints::GRAPH_OUTPUT_TYPE_ID)?;
+		Some((input, output))
+	}
+
+	/// True when `id` names one of the virtual endpoints (`false` for a
+	/// stale id).
+	pub fn is_endpoint(&self, id: NodeId) -> bool {
+		match self.get(id) {
+			Some(entry) => {
+				let type_id = entry.behavior.type_id();
+				type_id == crate::nodes::graphendpoints::GRAPH_INPUT_TYPE_ID
+					|| type_id == crate::nodes::graphendpoints::GRAPH_OUTPUT_TYPE_ID
+			}
+			None => false,
+		}
+	}
+
+	/// The live node whose behavior reports `type_id`, if any.
+	fn endpoint_of_type(&self, type_id: &str) -> Option<NodeId> {
+		self.node_ids()
+			.into_iter()
+			.find(|&id| {
+				self.get(id)
+					.map(|e| e.behavior.type_id() == type_id)
+					.unwrap_or(false)
+			})
+	}
+
+	/// Create whichever virtual endpoints the graph is missing and wire the
+	/// default `input -> output` edge; returns the `(input, output)` pair.
+	///
+	/// Idempotent and safe to call on every project load: a graph that
+	/// already carries both endpoints is returned untouched, and a project
+	/// saved before the endpoints existed gets them here (the migration
+	/// path — see [`crate::serializer`]). A graph with only one of the pair
+	/// gets the missing node added next to it.
+	///
+	/// The default edge is only created when `output.tex_in` is free, so an
+	/// explicitly wired output is never overridden.
+	///
+	/// Constructed directly rather than through the factory: these are
+	/// built-in graph-internal nodes and must exist even in a process where
+	/// the factory table has not been installed yet.
+	pub fn ensure_endpoints(&mut self) -> (NodeId, NodeId) {
+		let input = match self.endpoint_of_type(crate::nodes::graphendpoints::GRAPH_INPUT_TYPE_ID) {
+			Some(id) => id,
+			None => {
+				let (core, behavior) = crate::nodes::graphendpoints::create_graph_input();
+				self.add_node(core, behavior)
+			}
+		};
+		let output =
+			match self.endpoint_of_type(crate::nodes::graphendpoints::GRAPH_OUTPUT_TYPE_ID) {
+				Some(id) => id,
+				None => {
+					let (core, behavior) = crate::nodes::graphendpoints::create_graph_output();
+					self.add_node(core, behavior)
+				}
+			};
+		if self
+			.connected_output(output, crate::nodes::graphendpoints::GRAPH_OUTPUT_INPUT, -1)
+			.is_none()
+		{
+			self.connect(
+				input,
+				output,
+				crate::nodes::graphendpoints::GRAPH_OUTPUT_INPUT,
+				-1,
+			)
+			.ok();
+		}
+		(input, output)
+	}
+
+	/// Test-only: insert an edge with none of [`Graph::connect`]'s
+	/// validation, so a test can build the cycle the production paths
+	/// reject at connect time.
+	#[cfg(test)]
+	pub(crate) fn force_connect(&mut self, from: NodeId, to: NodeId, input: &str, element: i32) {
+		let key = self.intern_input(input);
+		self.edges.insert(Edge {
+			from,
+			to,
+			input_key: key,
+			element,
+		});
+	}
+
 	/// Connect `from`'s output to `to.input[element]`
 	/// (C++ `Node::connect_edge` incl. cycle rejection).
 	///
@@ -765,3 +866,192 @@ fn hash_str(input: &str) -> u64 {
 	input.hash(&mut h);
 	h.finish()
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::input::Input;
+	use crate::nodes::graphendpoints::{
+		GRAPH_INPUT_TYPE_ID, GRAPH_OUTPUT_TYPE_ID, GRAPH_OUTPUT_INPUT,
+	};
+	use crate::value::{NodeValue, ValueType};
+
+	/// Minimal non-endpoint behavior for the model tests: one connectable
+	/// texture input, no output logic.
+	struct Plain;
+
+	impl NodeBehavior for Plain {
+		fn name(&self) -> &str {
+			"Plain"
+		}
+
+		fn type_id(&self) -> &str {
+			"org.olivevideoeditor.Olive.plain-test"
+		}
+
+		fn duplicate(&self, _core: &NodeCore) -> Option<Box<dyn NodeBehavior>> {
+			Some(Box::new(Plain))
+		}
+	}
+
+	fn add_plain(graph: &mut Graph) -> NodeId {
+		let mut core = NodeCore::new();
+		core.add_input(Input::new("x_in", ValueType::Texture, NodeValue::None));
+		graph.add_node(core, Box::new(Plain))
+	}
+
+	#[test]
+	fn ensure_endpoints_creates_a_wired_pair() {
+		let mut graph = Graph::new();
+		assert!(graph.endpoints().is_none());
+		assert_eq!(graph.node_count(), 0);
+
+		let (input, output) = graph.ensure_endpoints();
+		assert_eq!(graph.endpoints(), Some((input, output)));
+		assert!(graph.is_endpoint(input));
+		assert!(graph.is_endpoint(output));
+		assert_eq!(
+			graph
+				.get(input)
+				.map(|e| e.behavior.type_id())
+				.expect("input endpoint is live"),
+			GRAPH_INPUT_TYPE_ID
+		);
+		assert_eq!(
+			graph
+				.get(output)
+				.map(|e| e.behavior.type_id())
+				.expect("output endpoint is live"),
+			GRAPH_OUTPUT_TYPE_ID
+		);
+		// Default wiring: input -> output.tex_in.
+		assert_eq!(
+			graph.connected_output(output, GRAPH_OUTPUT_INPUT, -1),
+			Some(input)
+		);
+		assert_eq!(graph.node_count(), 2);
+
+		// Idempotent: a second call reuses the same pair and edge.
+		assert_eq!(graph.ensure_endpoints(), (input, output));
+		assert_eq!(graph.node_count(), 2);
+		assert_eq!(graph.output_connections(input).len(), 1);
+	}
+
+	#[test]
+	fn ensure_endpoints_completes_a_half_pair() {
+		let mut graph = Graph::new();
+		let (core, behavior) = crate::nodes::graphendpoints::create_graph_input();
+		let input = graph.add_node(core, behavior);
+		// One endpoint alone is not a pair.
+		assert!(graph.endpoints().is_none());
+		assert!(graph.is_endpoint(input));
+
+		let (input2, output) = graph.ensure_endpoints();
+		assert_eq!(input, input2);
+		assert_eq!(graph.endpoints(), Some((input, output)));
+		assert_eq!(
+			graph.connected_output(output, GRAPH_OUTPUT_INPUT, -1),
+			Some(input)
+		);
+	}
+
+	#[test]
+	fn ensure_endpoints_keeps_an_explicitly_wired_output() {
+		let mut graph = Graph::new();
+		let (input, output) = graph.ensure_endpoints();
+		graph.disconnect(input, output, GRAPH_OUTPUT_INPUT, -1);
+
+		let source = add_plain(&mut graph);
+		graph.connect(source, output, GRAPH_OUTPUT_INPUT, -1).unwrap();
+
+		// No default edge is forced over the explicit one.
+		assert_eq!(graph.ensure_endpoints(), (input, output));
+		assert_eq!(
+			graph.connected_output(output, GRAPH_OUTPUT_INPUT, -1),
+			Some(source)
+		);
+	}
+
+	#[test]
+	fn remove_node_refuses_endpoints() {
+		let mut graph = Graph::new();
+		let (input, output) = graph.ensure_endpoints();
+		let plain = add_plain(&mut graph);
+
+		assert!(graph.remove_node(input).is_none());
+		assert!(graph.remove_node(output).is_none());
+		assert!(graph.is_valid(input) && graph.is_valid(output));
+		assert_eq!(graph.endpoints(), Some((input, output)));
+		assert_eq!(
+			graph.connected_output(output, GRAPH_OUTPUT_INPUT, -1),
+			Some(input)
+		);
+
+		// Non-endpoint nodes still go away, and a removed id is not an
+		// endpoint (nor is a stale one).
+		assert!(graph.remove_node(plain).is_some());
+		assert!(!graph.is_valid(plain));
+		assert!(!graph.is_endpoint(plain));
+		assert!(!graph.is_endpoint(NodeId::INVALID));
+		assert!(graph.remove_node(NodeId::INVALID).is_none());
+	}
+
+	#[test]
+	fn duplicate_refuses_endpoints_and_dependency_copies_propagate() {
+		let mut graph = Graph::new();
+		let (input, output) = graph.ensure_endpoints();
+
+		assert!(graph
+			.copy_node_and_dependency_graph_minus_items(input)
+			.is_none());
+		assert!(graph
+			.copy_node_and_dependency_graph_minus_items(output)
+			.is_none());
+
+		// A node whose dependency graph contains an endpoint cannot be
+		// copied either: the recursion reaches the endpoint's
+		// `duplicate` -> None.
+		let plain = add_plain(&mut graph);
+		graph.connect(input, plain, "x_in", -1).unwrap();
+		assert!(graph
+			.copy_node_and_dependency_graph_minus_items(plain)
+			.is_none());
+	}
+
+	#[test]
+	fn dependency_copy_of_an_endpoint_free_subgraph_still_works() {
+		let mut graph = Graph::new();
+		let a = add_plain(&mut graph);
+		let b = add_plain(&mut graph);
+		graph.connect(a, b, "x_in", -1).unwrap();
+
+		let (copy, map) = graph
+			.copy_node_and_dependency_graph_minus_items(b)
+			.expect("an endpoint-free chain copies");
+		assert_ne!(copy, b);
+		assert_eq!(map.get(&b), Some(&copy));
+		let a_copy = map.get(&a).copied().expect("a is copied");
+		assert_ne!(a_copy, a);
+		assert_eq!(graph.connected_output(copy, "x_in", -1), Some(a_copy));
+	}
+
+	#[test]
+	fn force_connect_builds_the_cycle_connect_rejects() {
+		let mut graph = Graph::new();
+		let a = add_plain(&mut graph);
+		let b = add_plain(&mut graph);
+		graph.connect(a, b, "x_in", -1).unwrap();
+		assert_eq!(
+			graph.connect(b, a, "x_in", -1),
+			Err(crate::error::Error::State)
+		);
+
+		graph.force_connect(b, a, "x_in", -1);
+		assert!(graph.reaches(a, b));
+		assert!(graph.reaches(b, a));
+		assert_eq!(
+			graph.input_connections(a),
+			vec![(b, "x_in".to_string(), -1)]
+		);
+	}
+}

crates/oak-node/src/nodes/graphendpoints.rs

@@ -0,0 +1,396 @@
+// Oak Video Editor - Non-Linear Video Editor
+// Copyright (C) 2026 Oak Team
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//! The graph's virtual endpoints: `GraphInput` / `GraphOutput`.
+//!
+//! Oak-only design with no C++ counterpart: upstream Olive's traversal
+//! walks back from the viewer output node and has no endpoint pair (see
+//! `docs/zh/plans/render-pipeline-threads.md` §3.8). Every project graph
+//! carries exactly one pair, created by `Graph::ensure_endpoints`
+//! (default-wired `input -> output`, and only that pair — the graph
+//! refuses to remove or copy them). `GraphInput` is the start of the
+//! evaluation walk ([`crate::traverser::Traverser::eval_graph_bfs`]);
+//! `GraphOutput` is where every branch converges and its value is the
+//! frame.
+//!
+//! `GraphInput`'s `value()` forwards whatever its input row carries —
+//! the node itself produces no pixels. `GraphOutput`'s `value()` hands
+//! its `tex_in` row value back out as the node's own output, so a
+//! consumer reads the frame from the output node's table.
+//!
+//! Deviation from the plan's literal wording ("only an output port"): the
+//! input endpoint also declares one connectable texture input,
+//! `feed_in`. Footage and generator nodes have no connectable inputs at
+//! all (their media input is `NOT_CONNECTABLE`), so without this port a
+//! single-clip graph could not be wired into the walk: the walk's live
+//! set is anchored at `GraphInput`, and a source with no edge into that
+//! anchor would be pruned. The sequence renderer feeds its composite
+//! through the same port. The "only output ports" half of the plan
+//! sentence still holds literally: the input endpoint's only output port
+//! is `tex_out`.
+
+use crate::factory::NodeMeta;
+use crate::node::{Category, NodeBehavior, NodeCore};
+use crate::value::{NodeValue, NodeValueRow, NodeValueTable, ValueType};
+
+/// Stable type id of the virtual graph input endpoint. Graph model code
+/// identifies the endpoint by this id (never by storing a node id).
+pub const GRAPH_INPUT_TYPE_ID: &str = "org.olivevideoeditor.Olive.graphinput";
+
+/// Stable type id of the virtual graph output endpoint (`Graph::endpoints`
+/// and `Graph::is_endpoint` match on it).
+pub const GRAPH_OUTPUT_TYPE_ID: &str = "org.olivevideoeditor.Olive.graphoutput";
+
+/// The input endpoint's texture input id: the graph's raw data entrance.
+/// A footage, generator, or sequence-composite node's output feeds the
+/// graph through here. Type: texture; flags: not-keyframable (see the
+/// module doc for why this port exists at all).
+pub const GRAPH_INPUT_FEED_INPUT: &str = "feed_in";
+
+/// The input endpoint's output port id (the evaluation walk's root
+/// output; the sequence renderer and the node editor use it as the
+/// input node's only outgoing port).
+pub const GRAPH_INPUT_OUTPUT: &str = "tex_out";
+
+/// The output endpoint's texture input id: every live branch converges
+/// here and the node's own value is this input's value. Type: texture;
+/// flags: not-keyframable.
+pub const GRAPH_OUTPUT_INPUT: &str = "tex_in";
+
+/// The virtual graph input node. Has no member fields.
+pub struct GraphInputNode;
+
+/// The virtual graph output node. Has no member fields.
+pub struct GraphOutputNode;
+
+impl NodeBehavior for GraphInputNode {
+	/// Human-readable name (shown as the node's fixed title).
+	fn name(&self) -> &str {
+		"Graph Input"
+	}
+
+	/// Stable type id.
+	fn type_id(&self) -> &str {
+		GRAPH_INPUT_TYPE_ID
+	}
+
+	/// Categories. The pair is never in a create menu (the constructor
+	/// sets `DONT_SHOW_IN_CREATE_MENU`); the category only groups it in
+	/// listings that ignore that flag.
+	fn categories(&self) -> &[Category] {
+		&[Category::Input]
+	}
+
+	/// Description.
+	fn description(&self) -> &str {
+		"The graph's input endpoint: the evaluation walk starts here and its row is forwarded downstream."
+	}
+
+	/// Localized input names: `feed_in` -> "Feed".
+	fn input_name<'a>(&self, id: &'a str) -> &'a str {
+		match id {
+			GRAPH_INPUT_FEED_INPUT => "Feed",
+			_ => id,
+		}
+	}
+
+	/// Forward the input row: every declared input that carries a
+	/// non-`None` value is pushed under its declared type, so downstream
+	/// nodes see exactly what was fed in. The node generates no pixels
+	/// of its own (C++ `Node::value` counterpart absent — Oak-only node).
+	fn value(
+		&self,
+		core: &NodeCore,
+		inputs: &NodeValueRow,
+		_time: oak_core::Rational,
+		table: &mut NodeValueTable,
+	) {
+		for input in &core.inputs {
+			let Some(value) = inputs.get(&input.id) else {
+				continue;
+			};
+			if matches!(value, NodeValue::None) {
+				continue;
+			}
+			// `NodeValue::clone` addrefs texture handles so the table owns
+			// its own reference (released on drop).
+			table.push(input.value_type, value.clone(), None);
+		}
+	}
+
+	/// The endpoint pair is fixed: copying either node is refused (the
+	/// graph model also refuses `remove_node`).
+	fn duplicate(&self, _core: &NodeCore) -> Option<Box<dyn NodeBehavior>> {
+		None
+	}
+}
+
+impl NodeBehavior for GraphOutputNode {
+	/// Human-readable name (shown as the node's fixed title).
+	fn name(&self) -> &str {
+		"Graph Output"
+	}
+
+	/// Stable type id.
+	fn type_id(&self) -> &str {
+		GRAPH_OUTPUT_TYPE_ID
+	}
+
+	/// Categories. See [`GraphInputNode::categories`].
+	fn categories(&self) -> &[Category] {
+		&[Category::Output]
+	}
+
+	/// Description.
+	fn description(&self) -> &str {
+		"The graph's output endpoint: every live branch converges here; the node's value is the frame."
+	}
+
+	/// Localized input names: `tex_in` -> "Texture".
+	fn input_name<'a>(&self, id: &'a str) -> &'a str {
+		match id {
+			GRAPH_OUTPUT_INPUT => "Texture",
+			_ => id,
+		}
+	}
+
+	/// Evaluate outputs: push the incoming texture as this node's own
+	/// value, so the walk's result is read from the output endpoint's
+	/// table (nothing incoming -> nothing pushed). The sibling-branch
+	/// convergence itself happens in the walk (Kahn in-degree), not here.
+	fn value(
+		&self,
+		core: &NodeCore,
+		inputs: &NodeValueRow,
+		_time: oak_core::Rational,
+		table: &mut NodeValueTable,
+	) {
+		let Some(value) = inputs.get(GRAPH_OUTPUT_INPUT) else {
+			return;
+		};
+		if matches!(value, NodeValue::None) {
+			return;
+		}
+		let Some(data_type) = core.input_data_type(GRAPH_OUTPUT_INPUT) else {
+			return;
+		};
+		table.push(data_type, value.clone(), None);
+	}
+
+	/// The endpoint pair is fixed: see [`GraphInputNode::duplicate`].
+	fn duplicate(&self, _core: &NodeCore) -> Option<Box<dyn NodeBehavior>> {
+		None
+	}
+}
+
+/// Constructor: the standard `enabled_in`, one connectable,
+/// non-keyframable `feed_in` texture input (default `None` — nothing is
+/// fed in until the project wires a source), and the create-menu hiding
+/// flag. No effect flags: the node is not addable, not an effect input,
+/// and not an item.
+pub fn create_graph_input() -> (NodeCore, Box<dyn NodeBehavior>) {
+	let mut core = NodeCore::new();
+	let mut feed = crate::input::Input::new(
+		GRAPH_INPUT_FEED_INPUT,
+		ValueType::Texture,
+		NodeValue::None,
+	);
+	feed.flags |= crate::input::flags::NOT_KEYFRAMABLE;
+	core.add_input(feed);
+	core.flags |= crate::node::flags::DONT_SHOW_IN_CREATE_MENU;
+	(core, Box::new(GraphInputNode))
+}
+
+/// Constructor: the standard `enabled_in`, one connectable,
+/// non-keyframable `tex_in` texture input (default `None`), and the
+/// create-menu hiding flag. See [`create_graph_input`].
+pub fn create_graph_output() -> (NodeCore, Box<dyn NodeBehavior>) {
+	let mut core = NodeCore::new();
+	let mut tex = crate::input::Input::new(
+		GRAPH_OUTPUT_INPUT,
+		ValueType::Texture,
+		NodeValue::None,
+	);
+	tex.flags |= crate::input::flags::NOT_KEYFRAMABLE;
+	core.add_input(tex);
+	core.flags |= crate::node::flags::DONT_SHOW_IN_CREATE_MENU;
+	(core, Box::new(GraphOutputNode))
+}
+
+/// Register both endpoint types with the factory: the graph model and
+/// the project loader construct them by type id, so serialization can
+/// rebuild a loaded graph's missing endpoints. Not in the C++ menu
+/// order (no C++ counterpart) — the entries are appended after the
+/// built-ins, and the hiding flag keeps them out of every create menu.
+pub fn register(meta: &mut Vec<NodeMeta>) {
+	meta.push(NodeMeta {
+		type_id: GRAPH_INPUT_TYPE_ID,
+		name: "Graph Input",
+		categories: &[Category::Input],
+		create: create_graph_input,
+	});
+	meta.push(NodeMeta {
+		type_id: GRAPH_OUTPUT_TYPE_ID,
+		name: "Graph Output",
+		categories: &[Category::Output],
+		create: create_graph_output,
+	});
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::factory::Factory;
+
+	#[test]
+	fn names_and_type_ids() {
+		let (_, input) = create_graph_input();
+		assert_eq!(input.name(), "Graph Input");
+		assert_eq!(input.type_id(), GRAPH_INPUT_TYPE_ID);
+		assert_eq!(input.categories(), &[Category::Input]);
+		let (_, output) = create_graph_output();
+		assert_eq!(output.name(), "Graph Output");
+		assert_eq!(output.type_id(), GRAPH_OUTPUT_TYPE_ID);
+		assert_eq!(output.categories(), &[Category::Output]);
+	}
+
+	#[test]
+	fn create_sets_ports_and_hiding_flag() {
+		use crate::input::flags as input_flags;
+		use crate::node::flags as node_flags;
+
+		let (core, _) = create_graph_input();
+		let feed = core.get_input(GRAPH_INPUT_FEED_INPUT).expect("feed_in");
+		assert_eq!(feed.value_type, ValueType::Texture);
+		assert_eq!(feed.default, NodeValue::None);
+		assert_eq!(feed.flags & input_flags::NOT_KEYFRAMABLE, input_flags::NOT_KEYFRAMABLE);
+		assert!(feed.is_connectable(), "the data entrance accepts edges");
+		assert!(core.get_input(crate::node::ENABLED_INPUT).is_some());
+		assert_ne!(core.flags & node_flags::DONT_SHOW_IN_CREATE_MENU, 0);
+		// Not an effect, not an item, no effect input.
+		assert_eq!(core.flags & node_flags::VIDEO_EFFECT, 0);
+		assert_eq!(core.flags & node_flags::AUDIO_EFFECT, 0);
+		assert_eq!(core.flags & node_flags::IS_ITEM, 0);
+		assert!(core.effect_input.is_empty());
+
+		let (core, _) = create_graph_output();
+		let tex = core.get_input(GRAPH_OUTPUT_INPUT).expect("tex_in");
+		assert_eq!(tex.value_type, ValueType::Texture);
+		assert_eq!(tex.default, NodeValue::None);
+		assert!(tex.is_connectable());
+		assert_ne!(core.flags & node_flags::DONT_SHOW_IN_CREATE_MENU, 0);
+		assert_eq!(core.flags & node_flags::VIDEO_EFFECT, 0);
+	}
+
+	#[test]
+	fn duplicate_is_refused() {
+		let (core, input) = create_graph_input();
+		assert!(input.duplicate(&core).is_none());
+		let (core, output) = create_graph_output();
+		assert!(output.duplicate(&core).is_none());
+	}
+
+	#[test]
+	fn input_names_and_enabled_fallthrough() {
+		let (_, input) = create_graph_input();
+		assert_eq!(input.input_name(GRAPH_INPUT_FEED_INPUT), "Feed");
+		// The raw id (not the default "Enabled") so no pack entry is
+		// required for the structural input (the param view hides it).
+		assert_eq!(input.input_name(crate::node::ENABLED_INPUT), "enabled_in");
+		let (_, output) = create_graph_output();
+		assert_eq!(output.input_name(GRAPH_OUTPUT_INPUT), "Texture");
+		assert_eq!(output.input_name("other_in"), "other_in");
+	}
+
+	#[test]
+	fn factory_resolves_both_types() {
+		let (core, behavior) = Factory::global()
+			.create_any(GRAPH_INPUT_TYPE_ID)
+			.expect("the factory registers the graph input endpoint");
+		assert_eq!(
+			core.flags & crate::node::flags::DONT_SHOW_IN_CREATE_MENU,
+			crate::node::flags::DONT_SHOW_IN_CREATE_MENU
+		);
+		assert_eq!(behavior.type_id(), GRAPH_INPUT_TYPE_ID);
+		let (core, behavior) = Factory::global()
+			.create_any(GRAPH_OUTPUT_TYPE_ID)
+			.expect("the factory registers the graph output endpoint");
+		assert_eq!(
+			core.flags & crate::node::flags::DONT_SHOW_IN_CREATE_MENU,
+			crate::node::flags::DONT_SHOW_IN_CREATE_MENU
+		);
+		assert_eq!(behavior.type_id(), GRAPH_OUTPUT_TYPE_ID);
+	}
+
+	#[test]
+	fn graph_input_forwards_the_row() {
+		let (core, behavior) = create_graph_input();
+		let tex = NodeValue::Texture(crate::handle::CHandle::null());
+		let row = NodeValueRow::from([
+			(crate::node::ENABLED_INPUT.to_string(), NodeValue::Boolean(true)),
+			(GRAPH_INPUT_FEED_INPUT.to_string(), tex),
+		]);
+		let mut table = NodeValueTable::default();
+		behavior.value(&core, &row, oak_core::Rational::new(0, 1), &mut table);
+		assert_eq!(table.count(), 2);
+		assert_eq!(table.get(ValueType::Boolean), Some(&NodeValue::Boolean(true)));
+		assert!(matches!(table.get(ValueType::Texture), Some(NodeValue::Texture(_))));
+	}
+
+	#[test]
+	fn graph_input_skips_a_missing_or_none_row_value() {
+		let (core, behavior) = create_graph_input();
+		let mut table = NodeValueTable::default();
+		behavior.value(
+			&core,
+			&NodeValueRow::default(),
+			oak_core::Rational::new(0, 1),
+			&mut table,
+		);
+		assert!(table.is_empty(), "no row entries -> nothing forwarded");
+
+		let row = NodeValueRow::from([(
+			GRAPH_INPUT_FEED_INPUT.to_string(),
+			NodeValue::None,
+		)]);
+		behavior.value(&core, &row, oak_core::Rational::new(0, 1), &mut table);
+		assert!(table.is_empty(), "a None value is not forwarded");
+	}
+
+	#[test]
+	fn graph_output_pushes_its_texture() {
+		let (core, behavior) = create_graph_output();
+		let tex = NodeValue::Texture(crate::handle::CHandle::null());
+		let row = NodeValueRow::from([(GRAPH_OUTPUT_INPUT.to_string(), tex)]);
+		let mut table = NodeValueTable::default();
+		behavior.value(&core, &row, oak_core::Rational::new(0, 1), &mut table);
+		let handle = match table.get(ValueType::Texture) {
+			Some(NodeValue::Texture(h)) => *h,
+			other => panic!("texture expected, got {other:?}"),
+		};
+		assert!(handle.ctx.is_null());
+
+		// Nothing incoming -> nothing pushed.
+		let mut empty = NodeValueTable::default();
+		behavior.value(
+			&core,
+			&NodeValueRow::default(),
+			oak_core::Rational::new(0, 1),
+			&mut empty,
+		);
+		assert!(empty.is_empty());
+	}
+}

crates/oak-node/src/nodes/mod.rs

@@ -30,6 +30,7 @@ mod displaytransform;
 mod dropshadowfilter;
 mod flipdistortnode;
 mod generatorwithmerge;
+pub mod graphendpoints;
 pub mod group;
 mod mask;
 mod math;
@@ -190,6 +191,13 @@ pub fn register_all() {
 	ramp::register(&mut meta);
 	multicamnode::register(&mut meta);
 
+	// Oak-only virtual graph endpoints (see
+	// [`graphendpoints`](crate::nodes::graphendpoints)). No C++ menu
+	// entry exists; they are appended after the C++ order ends and are
+	// hidden from every create menu (`DONT_SHOW_IN_CREATE_MENU`), so the
+	// only observable effect on the factory table is two extra entries.
+	graphendpoints::register(&mut meta);
+
 	// OpenFX plugins have no static type ids (C++
 	// `factory.cpp::register_plugin_nodes`); the registration call is a
 	// no-op placeholder for the oakplugin bridge's runtime discovery.

crates/oak-node/src/nodes/textbackend.rs

@@ -163,18 +163,24 @@ static MEASURE: std::sync::Mutex<Option<TextMeasureBackend>> = std::sync::Mutex:
 /// Installed render hook (C++ global `g_text_render_backend`).
 static RENDER: std::sync::Mutex<Option<TextRenderBackend>> = std::sync::Mutex::new(None);
 
+/// Serializes every test that installs the process-global text backends:
+/// the tests below and `textv3`'s installer tests share it, so none of
+/// them observes another's install.
+#[cfg(test)]
+pub(crate) static TEST_BACKEND_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
 #[cfg(test)]
 mod tests {
 	use super::*;
 
-	// The two tests below share the process-global backend statics; a
-	// lock serializes them so `backend_hooks_default_none` cannot observe
-	// the hooks installed by `backend_hooks_install_and_query`.
-	static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+	// The two tests below share the process-global backend statics; the
+	// shared test lock also serializes them against `textv3`'s installer
+	// tests, so `backend_hooks_default_none` cannot observe the hooks
+	// installed by `backend_hooks_install_and_query`.
 
 	#[test]
 	fn backend_hooks_default_none() {
-		let _guard = LOCK.lock().unwrap();
+		let _guard = TEST_BACKEND_LOCK.lock().unwrap();
 		set_text_backends(None, None);
 		assert_eq!(text_measure_backend(), None);
 		assert_eq!(text_render_backend(), None);
@@ -182,7 +188,7 @@ mod tests {
 
 	#[test]
 	fn backend_hooks_install_and_query() {
-		let _guard = LOCK.lock().unwrap();
+		let _guard = TEST_BACKEND_LOCK.lock().unwrap();
 		fn measure(_r: &TextLayoutRequest) -> TextLayoutSize {
 			TextLayoutSize {
 				width: 12.0,

crates/oak-node/src/nodes/textv3.rs

@@ -1741,11 +1741,10 @@ mod tests {
 		);
 	}
 
-	/// Serializes the tests that install a process-global text backend.
-	/// (The `textbackend` module's own tests use a different lock, so they
-	/// are not mutually excluded — same exposure as the pre-existing
-	/// `measure_without_backend_returns_zero_size`.)
-	static BACKEND_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+	/// Serializes the tests that install a process-global text backend;
+	/// shared with the `textbackend` module's own tests, so no test
+	/// observes another module's install.
+	use crate::nodes::textbackend::TEST_BACKEND_LOCK as BACKEND_LOCK;
 
 	/// Measure hook for tests that only need "a backend is installed".
 	fn noop_measure(_req: &TextLayoutRequest) -> TextLayoutSize {
@@ -1802,6 +1801,8 @@ mod tests {
 
 	#[test]
 	fn value_pushes_job_when_text_nonempty() {
+		let _guard = BACKEND_LOCK.lock().unwrap();
+		crate::nodes::textbackend::set_text_backends(None, None);
 		let (core, behavior) = create();
 		let mut row = NodeValueRow::default();
 		row.insert(
@@ -1819,6 +1820,8 @@ mod tests {
 
 	#[test]
 	fn value_expands_args_from_row() {
+		let _guard = BACKEND_LOCK.lock().unwrap();
+		crate::nodes::textbackend::set_text_backends(None, None);
 		let (core, behavior) = create();
 		let mut row = NodeValueRow::default();
 		row.insert(
@@ -1860,8 +1863,8 @@ mod tests {
 		let (core, behavior) = create();
 		let mut row = NodeValueRow::default();
 		// Both text inputs are emptied explicitly: which one `value` reads
-		// depends on the process-global backend state, which the tests that
-		// install one (own lock, and `textbackend`'s) change concurrently.
+		// depends on the process-global backend state, which the installer
+		// tests change and restore concurrently.
 		row.insert(PLAIN_TEXT_INPUT.to_string(), NodeValue::Text(String::new()));
 		row.insert(TEXT_INPUT.to_string(), NodeValue::Text(String::new()));
 		let mut table = NodeValueTable::default();

crates/oak-node/src/project.rs

@@ -119,6 +119,10 @@ impl Project {
 		self.root = id;
 		self.settings
 			.insert(SETTING_ROOT.to_string(), id.identity().to_string());
+		// Every project graph carries the virtual endpoint pair (see
+		// docs/zh/plans/render-pipeline-threads.md §3.8); created last so
+		// the root folder keeps the first arena slot.
+		self.graph.ensure_endpoints();
 		Ok(())
 	}
 

crates/oak-node/src/serializer.rs

@@ -662,6 +662,12 @@ fn load_project_body(
 	// reattach each child to its bin folder.
 	resolve_folder_children(&mut project.graph, &id_map);
 
+	// Migration for projects saved before the endpoints existed (or whose
+	// file omitted them): create whatever is missing and default-wire the
+	// pair. A file that already carries both endpoints — and possibly a
+	// hand-wired output — is left untouched.
+	project.graph.ensure_endpoints();
+
 	Ok(id_map)
 }
 

crates/oak-node/src/traverser.rs

@@ -37,7 +37,7 @@
 //! (C++ `ProcessInputElement` → `GetValueAtTime`).
 //! `// CPP-PARITY: src/node/src/traverser.cpp`.
 
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeSet, HashMap, HashSet};
 
 use oak_core::{Rational, TimeRange};
 
@@ -133,55 +133,126 @@ impl Traverser {
 		// Per-pass memo: (node, time) -> evaluated output table. A shared
 		// upstream evaluates once per requested time (C++ value_cache_).
 		let mut cache: HashMap<(NodeId, Rational), NodeValueTable> = HashMap::new();
-		let mut queued: HashSet<(NodeId, Rational)> = HashSet::new();
-		let mut stack: Vec<Frame> = vec![Frame::Enter(request.root, request.time)];
-		queued.insert((request.root, request.time));
+		walk_dfs(graph, &mut cache, request.root, request.time, hooks)?;
+		Ok(cache
+			.remove(&(request.root, request.time))
+			.unwrap_or_default())
+	}
 
-		while let Some(frame) = stack.pop() {
+	/// Evaluate the graph as one Kahn-order sweep of the *live set* and
+	/// return the [`GraphOutput`](crate::nodes::graphendpoints) value —
+	/// the frame (C++ has no counterpart; the C++ traversal starts at the
+	/// consumer and walks backwards, Oak starts at the graph's input
+	/// endpoint and converges on its output endpoint).
+	///
+	/// The live set is `(downstream of the input ∨ feeding the input) ∧
+	/// (upstream of the output)`: the input's forward cone plus the nodes
+	/// that reach the input (a sequence feeds the graph through
+	/// `GraphInput.feed_in`, which puts it in that feeder cone),
+	/// intersected with the output's backward cone. Isolated nodes and
+	/// branches that never reach the output are never queued. A
+	/// connection coming from *outside* the live set leaves `None` in the
+	/// consuming row (same as a missing cache entry in the DFS walk).
+	///
+	/// Each live node is dequeued once, in ascending [`NodeId`] order
+	/// among the ready nodes, with the DFS walk's per-node semantics
+	/// (`GenerateRowValue` → `value()` → [`RenderHooks::resolve`]), and
+	/// the resolved table is what its downstream consumers see. The
+	/// ascending-id order is the only deterministic order available:
+	/// `Graph::edges` is a `BTreeSet` keyed by `(from, to, input,
+	/// element)`, so the order in which connections were made is not
+	/// stored and cannot be recovered.
+	///
+	/// Time semantics are single-time: `time` is the frame being
+	/// rendered, and the consuming node's `input_time_adjustment` applies
+	/// as usual — an upstream at a different time is evaluated through
+	/// the DFS memo (exactly what [`Traverser::evaluate`] would do),
+	/// while same-time upstreams are the sweep's own job. Multi-time
+	/// memoization across a range is M1+ work.
+	///
+	/// Errors: `NotFound` when either endpoint is missing; `Failed` when
+	/// the live set cannot be ordered (the message names a cycle found
+	/// in it) or when the output is not reachable from the input; `State`
+	/// on cancellation.
+	pub fn eval_graph_bfs(
+		&mut self,
+		graph: &Graph,
+		time: Rational,
+		hooks: &mut dyn RenderHooks,
+	) -> crate::error::Result<NodeValue> {
+		use crate::error::Error;
+		let (input, output) = graph.endpoints().ok_or(Error::NotFound)?;
+		let live = live_nodes(graph, input, output);
+
+		// In-degree over the live-induced subgraph only: an edge whose
+		// source lies outside the live set must not keep its target from
+		// ever becoming ready.
+		let mut indegree: HashMap<NodeId, usize> = HashMap::new();
+		for node in &live {
+			let n = graph
+				.input_connections(*node)
+				.iter()
+				.filter(|(from, _, _)| live.contains(from))
+				.count();
+			indegree.insert(*node, n);
+		}
+		let mut ready: BTreeSet<NodeId> = indegree
+			.iter()
+			.filter(|(_, n)| **n == 0)
+			.map(|(node, _)| *node)
+			.collect();
+
+		let mut cache: HashMap<(NodeId, Rational), NodeValueTable> = HashMap::new();
+		let mut done = 0usize;
+		while let Some(node) = ready.iter().next().copied() {
+			ready.remove(&node);
 			if hooks.is_cancelled() {
 				return Err(Error::State);
 			}
-			match frame {
-				Frame::Enter(node, time) => {
-					if cache.contains_key(&(node, time)) {
-						continue;
-					}
-					let Some(entry) = graph.get(node) else {
-						continue;
-					};
-					stack.push(Frame::Exit(node, time));
-					// Queue every connected upstream at its adjusted time.
-					for (from, input, element) in graph.input_connections(node) {
-						let from = entry
-							.behavior
-							.connected_render_output(&entry.core, &input, element)
-							.unwrap_or(from);
-						let adjusted = adjusted_time(entry, &input, element, time);
-						let key = (from, adjusted);
-						if !cache.contains_key(&key) && queued.insert(key) {
-							stack.push(Frame::Enter(from, adjusted));
-						}
-					}
+			let Some(entry) = graph.get(node) else {
+				continue;
+			};
+			done += 1;
+			// Time-shifted upstreams are not part of this sweep; pull them
+			// through the DFS memo. A node may already own a table at
+			// `time` (a pull deeper down evaluated it out of order) — in
+			// that case it is not evaluated (and resolved) twice.
+			pull_adjusted_upstreams(graph, &mut cache, entry, node, time, hooks)?;
+			if !cache.contains_key(&(node, time)) {
+				let row = build_row(graph, &cache, entry, node, time);
+				let mut table = NodeValueTable::default();
+				entry.behavior.value(&entry.core, &row, time, &mut table);
+				hooks.resolve(node, &row, &mut table);
+				cache.insert((node, time), table);
+			}
+			for (to, _, _) in graph.output_connections(node) {
+				if !live.contains(&to) {
+					continue;
 				}
-				Frame::Exit(node, time) => {
-					if cache.contains_key(&(node, time)) {
-						continue;
+				if let Some(n) = indegree.get_mut(&to) {
+					*n -= 1;
+					if *n == 0 {
+						ready.insert(to);
 					}
-					let Some(entry) = graph.get(node) else {
-						continue;
-					};
-					let row = build_row(graph, &cache, entry, node, time);
-					let mut table = NodeValueTable::default();
-					entry.behavior.value(&entry.core, &row, time, &mut table);
-					hooks.resolve(node, &row, &mut table);
-					cache.insert((node, time), table);
 				}
 			}
 		}
 
-		Ok(cache
-			.remove(&(request.root, request.time))
-			.unwrap_or_default())
+		if done < live.len() {
+			let cycle = find_cycle(graph, &indegree);
+			return Err(Error::Failed(format!("graph contains a cycle: {cycle:?}")));
+		}
+
+		let Some(table) = cache.remove(&(output, time)) else {
+			return Err(Error::Failed(
+				"graph output is not reachable from the graph input".to_string(),
+			));
+		};
+		Ok(table
+			.get(ValueType::Texture)
+			.cloned()
+			.or_else(|| table.rows().last().map(|(_, value, _)| value.clone()))
+			.unwrap_or(NodeValue::None))
 	}
 
 	/// Invalidate walk: mark downstream caches dirty after an input
@@ -208,6 +279,171 @@ impl Default for Traverser {
 	}
 }
 
+/// Explicit-stack DFS from `root` at `time` (the C++ recursion, minus
+/// the recursion), memoizing every evaluated `(node, time)` table into
+/// `cache`. `Enter` queues the connected upstreams at their adjusted
+/// times, `Exit` builds the row and runs `GenerateRowValue` →
+/// `value()` → [`RenderHooks::resolve`]; a node is entered at most once
+/// per time (`queued` is the DFS gray set, `cache` the black set).
+///
+/// Shared by [`Traverser::evaluate`] (whole walk) and
+/// [`Traverser::eval_graph_bfs`] (time-shifted upstream pulls).
+///
+/// Errors: `State` on cancellation.
+fn walk_dfs(
+	graph: &Graph,
+	cache: &mut HashMap<(NodeId, Rational), NodeValueTable>,
+	root: NodeId,
+	time: Rational,
+	hooks: &mut dyn RenderHooks,
+) -> crate::error::Result<()> {
+	use crate::error::Error;
+	let mut queued: HashSet<(NodeId, Rational)> = HashSet::new();
+	let mut stack: Vec<Frame> = vec![Frame::Enter(root, time)];
+	queued.insert((root, time));
+
+	while let Some(frame) = stack.pop() {
+		if hooks.is_cancelled() {
+			return Err(Error::State);
+		}
+		match frame {
+			Frame::Enter(node, time) => {
+				if cache.contains_key(&(node, time)) {
+					continue;
+				}
+				let Some(entry) = graph.get(node) else {
+					continue;
+				};
+				stack.push(Frame::Exit(node, time));
+				// Queue every connected upstream at its adjusted time.
+				for (from, input, element) in graph.input_connections(node) {
+					let from = entry
+						.behavior
+						.connected_render_output(&entry.core, &input, element)
+						.unwrap_or(from);
+					let adjusted = adjusted_time(entry, &input, element, time);
+					let key = (from, adjusted);
+					if !cache.contains_key(&key) && queued.insert(key) {
+						stack.push(Frame::Enter(from, adjusted));
+					}
+				}
+			}
+			Frame::Exit(node, time) => {
+				if cache.contains_key(&(node, time)) {
+					continue;
+				}
+				let Some(entry) = graph.get(node) else {
+					continue;
+				};
+				let row = build_row(graph, cache, entry, node, time);
+				let mut table = NodeValueTable::default();
+				entry.behavior.value(&entry.core, &row, time, &mut table);
+				hooks.resolve(node, &row, &mut table);
+				cache.insert((node, time), table);
+			}
+		}
+	}
+	Ok(())
+}
+
+/// Pull the upstreams of `node` that the sweep will not visit itself:
+/// connections whose `input_time_adjustment` maps `time` to a different
+/// time are evaluated on the spot through the DFS memo (the very walk
+/// [`Traverser::evaluate`] performs, so those values carry the same
+/// semantics), and their tables land in `cache` for [`build_row`].
+///
+/// Errors: `State` on cancellation (propagated from the pull).
+fn pull_adjusted_upstreams(
+	graph: &Graph,
+	cache: &mut HashMap<(NodeId, Rational), NodeValueTable>,
+	entry: &crate::graph::NodeEntry,
+	node: NodeId,
+	time: Rational,
+	hooks: &mut dyn RenderHooks,
+) -> crate::error::Result<()> {
+	for (from, input, element) in graph.input_connections(node) {
+		let from = entry
+			.behavior
+			.connected_render_output(&entry.core, &input, element)
+			.unwrap_or(from);
+		let adjusted = adjusted_time(entry, &input, element, time);
+		if adjusted == time || !graph.is_valid(from) {
+			continue;
+		}
+		if !cache.contains_key(&(from, adjusted)) {
+			walk_dfs(graph, cache, from, adjusted, hooks)?;
+		}
+	}
+	Ok(())
+}
+
+/// Every node reachable from `root` (including `root`) walking
+/// downstream, or — with `upstream` — every node that can reach `root`.
+fn reachable(graph: &Graph, root: NodeId, upstream: bool) -> HashSet<NodeId> {
+	let mut seen: HashSet<NodeId> = HashSet::new();
+	let mut stack: Vec<NodeId> = vec![root];
+	while let Some(node) = stack.pop() {
+		if !seen.insert(node) {
+			continue;
+		}
+		if upstream {
+			stack.extend(graph.upstream(node));
+		} else {
+			stack.extend(graph.downstream(node));
+		}
+	}
+	seen
+}
+
+/// The live set of the endpoint-to-endpoint sweep: the input's forward
+/// cone plus its feeder cone (nodes that reach the input — how a
+/// sequence feeds `GraphInput.feed_in`), intersected with the output's
+/// backward cone. See [`Traverser::eval_graph_bfs`].
+fn live_nodes(graph: &Graph, input: NodeId, output: NodeId) -> HashSet<NodeId> {
+	let mut live = reachable(graph, input, false);
+	live.extend(reachable(graph, input, true));
+	let reaches_output = reachable(graph, output, true);
+	live.retain(|node| reaches_output.contains(node));
+	live
+}
+
+/// A cycle inside the nodes the sweep could not order: residual nodes
+/// (positive in-degree left) walked upstream, always taking the
+/// smallest-id residual feeder, until a node repeats — the repeated
+/// suffix is the cycle. Deterministic; empty when there is no residual
+/// node.
+fn find_cycle(graph: &Graph, indegree: &HashMap<NodeId, usize>) -> Vec<NodeId> {
+	let residual: BTreeSet<NodeId> = indegree
+		.iter()
+		.filter(|(_, n)| **n > 0)
+		.map(|(node, _)| *node)
+		.collect();
+	let mut path: Vec<NodeId> = Vec::new();
+	let mut seen: HashMap<NodeId, usize> = HashMap::new();
+	let Some(start) = residual.iter().next().copied() else {
+		return path;
+	};
+	let mut node = start;
+	loop {
+		if let Some(&at) = seen.get(&node) {
+			return path.split_off(at);
+		}
+		seen.insert(node, path.len());
+		path.push(node);
+		// Every residual node has a residual feeder: in-degree was only
+		// counted over live edges, and an unprocessed feeder keeps its
+		// own positive in-degree.
+		match graph
+			.upstream(node)
+			.into_iter()
+			.find(|from| residual.contains(from))
+		{
+			Some(from) => node = from,
+			None => return path,
+		}
+	}
+}
+
 /// The consuming node's time adjustment for `input` (C++
 /// `Node::InputTimeAdjustment` with `traverse = true`): clips map
 /// sequence time to media time, tracks clamp to the covering block. The
@@ -323,3 +559,470 @@ pub struct ValueDatabase {
 	/// Rows keyed by node input id.
 	pub rows: Vec<(String, Vec<(ValueType, NodeValue)>)>,
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::error::Error;
+	use crate::handle::CHandle;
+	use crate::input::Input;
+	use crate::node::{NodeBehavior, NodeCore};
+	use crate::nodes::graphendpoints::{GRAPH_INPUT_FEED_INPUT, GRAPH_OUTPUT_INPUT};
+	use std::sync::atomic::{AtomicUsize, Ordering};
+	use std::sync::{Arc, Mutex};
+
+	/// What a [`TestNode`] pushes to its own output table.
+	enum Emit {
+		/// Nothing at all.
+		Nothing,
+		/// A constant float.
+		Float(f64),
+		/// The time the node was evaluated at.
+		Time,
+		/// A null texture handle.
+		Texture,
+	}
+
+	/// Counting stand-in for a real node: emits on demand so a test can
+	/// observe evaluation counts, dequeue order, and input rows.
+	struct TestNode {
+		emit: Emit,
+		calls: Arc<AtomicUsize>,
+	}
+
+	impl NodeBehavior for TestNode {
+		fn name(&self) -> &str {
+			"Test Node"
+		}
+
+		fn type_id(&self) -> &str {
+			"org.olivevideoeditor.Olive.test-bfs-node"
+		}
+
+		fn value(
+			&self,
+			_core: &NodeCore,
+			_inputs: &NodeValueRow,
+			time: Rational,
+			table: &mut NodeValueTable,
+		) {
+			self.calls.fetch_add(1, Ordering::Relaxed);
+			match self.emit {
+				Emit::Nothing => {}
+				Emit::Float(value) => table.push(ValueType::Float, NodeValue::Float(value), None),
+				Emit::Time => table.push(ValueType::Float, NodeValue::Float(time.to_f64()), None),
+				Emit::Texture => {
+					table.push(ValueType::Texture, NodeValue::Texture(CHandle::null()), None)
+				}
+			}
+		}
+
+		fn duplicate(&self, _core: &NodeCore) -> Option<Box<dyn NodeBehavior>> {
+			None
+		}
+	}
+
+	/// Add a [`TestNode`] declaring `inputs`; returns its id and call counter.
+	fn add_node(
+		graph: &mut Graph,
+		inputs: &[(&str, ValueType)],
+		emit: Emit,
+	) -> (NodeId, Arc<AtomicUsize>) {
+		let calls = Arc::new(AtomicUsize::new(0));
+		let mut core = NodeCore::new();
+		for (id, value_type) in inputs {
+			core.add_input(Input::new(id, *value_type, NodeValue::None));
+		}
+		let id = graph.add_node(
+			core,
+			Box::new(TestNode {
+				emit,
+				calls: Arc::clone(&calls),
+			}),
+		);
+		(id, calls)
+	}
+
+	/// The endpoint pair with the default `input -> output` edge removed,
+	/// so a test can wire the convergence itself.
+	fn detached_endpoints(graph: &mut Graph) -> (NodeId, NodeId) {
+		let (input, output) = graph.ensure_endpoints();
+		graph.disconnect(input, output, GRAPH_OUTPUT_INPUT, -1);
+		(input, output)
+	}
+
+	/// A node that samples `val_in` one second late: observes that the
+	/// sweep pulls a time-shifted upstream through the DFS memo.
+	struct Delay {
+		rows: Arc<Mutex<Vec<NodeValueRow>>>,
+	}
+
+	impl NodeBehavior for Delay {
+		fn name(&self) -> &str {
+			"Delay"
+		}
+
+		fn type_id(&self) -> &str {
+			"org.olivevideoeditor.Olive.test-bfs-delay"
+		}
+
+		fn input_time_adjustment(
+			&self,
+			_core: &NodeCore,
+			input: &str,
+			_element: i32,
+			time: TimeRange,
+			_traverse: bool,
+		) -> TimeRange {
+			if input == "val_in" {
+				TimeRange::new(
+					time.in_() + Rational::new(1, 1),
+					time.out() + Rational::new(1, 1),
+				)
+			} else {
+				time
+			}
+		}
+
+		fn value(
+			&self,
+			_core: &NodeCore,
+			inputs: &NodeValueRow,
+			_time: Rational,
+			table: &mut NodeValueTable,
+		) {
+			self.rows.lock().expect("delay rows").push(inputs.clone());
+			if let Some(value @ NodeValue::Float(_)) = inputs.get("val_in") {
+				table.push(ValueType::Float, value.clone(), None);
+			}
+		}
+
+		fn duplicate(&self, _core: &NodeCore) -> Option<Box<dyn NodeBehavior>> {
+			None
+		}
+	}
+
+	/// Records resolve order/rows and can replace a node's resolved table.
+	struct Probe {
+		order: Vec<NodeId>,
+		rows: HashMap<NodeId, NodeValueRow>,
+		replace: Option<(NodeId, Vec<(ValueType, NodeValue)>)>,
+	}
+
+	impl Probe {
+		fn new() -> Probe {
+			Probe {
+				order: Vec::new(),
+				rows: HashMap::new(),
+				replace: None,
+			}
+		}
+	}
+
+	impl RenderHooks for Probe {
+		fn resolve(&mut self, node: NodeId, row: &NodeValueRow, table: &mut NodeValueTable) {
+			self.order.push(node);
+			self.rows.insert(node, row.clone());
+			if let Some((target, rows)) = &self.replace {
+				if *target == node {
+					table.clear();
+					for (ty, value) in rows {
+						table.push(*ty, value.clone(), None);
+					}
+				}
+			}
+		}
+	}
+
+	#[test]
+	fn bfs_starts_at_graph_input_and_skips_unreachable_nodes() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (dead, dead_calls) =
+			add_node(&mut graph, &[("x_in", ValueType::Texture)], Emit::Nothing);
+		let (_island, island_calls) = add_node(&mut graph, &[], Emit::Float(1.0));
+		graph
+			.connect(input, dead, "x_in", -1)
+			.expect("input -> dead");
+		graph
+			.connect(input, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("input -> output");
+
+		let mut probe = Probe::new();
+		let value = Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect("the input -> output edge alone is a valid graph");
+		assert_eq!(probe.order, vec![input, output]);
+		assert_eq!(dead_calls.load(Ordering::Relaxed), 0, "dead branch skipped");
+		assert_eq!(
+			island_calls.load(Ordering::Relaxed),
+			0,
+			"island never queued"
+		);
+		assert_eq!(value, NodeValue::None);
+	}
+
+	#[test]
+	fn bfs_waits_for_every_input_of_a_converging_node() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (a, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(1.0));
+		let (b, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(2.0));
+		let (merge, _) = add_node(
+			&mut graph,
+			&[("a_in", ValueType::Float), ("b_in", ValueType::Float)],
+			Emit::Float(3.0),
+		);
+		graph.connect(input, a, "tex_in", -1).expect("input -> a");
+		graph.connect(input, b, "tex_in", -1).expect("input -> b");
+		graph.connect(a, merge, "a_in", -1).expect("a -> merge");
+		graph.connect(b, merge, "b_in", -1).expect("b -> merge");
+		graph
+			.connect(merge, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("merge -> output");
+
+		let mut probe = Probe::new();
+		Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect("the converging graph is orderable");
+		assert_eq!(probe.order, vec![input, a, b, merge, output]);
+		let row = probe.rows.get(&merge).expect("merge resolved");
+		assert_eq!(row.get("a_in"), Some(&NodeValue::Float(1.0)));
+		assert_eq!(row.get("b_in"), Some(&NodeValue::Float(2.0)));
+	}
+
+	#[test]
+	fn bfs_evaluates_a_shared_source_once() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (source, source_calls) = add_node(
+			&mut graph,
+			&[("tex_in", ValueType::Texture)],
+			Emit::Float(7.0),
+		);
+		let (m, _) = add_node(&mut graph, &[("a_in", ValueType::Float)], Emit::Float(1.0));
+		let (n, _) = add_node(&mut graph, &[("a_in", ValueType::Float)], Emit::Float(2.0));
+		let (f, _) = add_node(
+			&mut graph,
+			&[("c_in", ValueType::Float), ("d_in", ValueType::Float)],
+			Emit::Float(3.0),
+		);
+		graph
+			.connect(input, source, "tex_in", -1)
+			.expect("input -> source");
+		graph.connect(source, m, "a_in", -1).expect("source -> m");
+		graph.connect(source, n, "a_in", -1).expect("source -> n");
+		graph.connect(m, f, "c_in", -1).expect("m -> f");
+		graph.connect(n, f, "d_in", -1).expect("n -> f");
+		graph
+			.connect(f, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("f -> output");
+
+		let mut probe = Probe::new();
+		Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect("the fan-out graph is orderable");
+		assert_eq!(probe.order, vec![input, source, m, n, f, output]);
+		assert_eq!(
+			source_calls.load(Ordering::Relaxed),
+			1,
+			"the shared source runs once"
+		);
+		assert_eq!(
+			probe.rows.get(&m).and_then(|r| r.get("a_in")),
+			Some(&NodeValue::Float(7.0))
+		);
+		assert_eq!(
+			probe.rows.get(&n).and_then(|r| r.get("a_in")),
+			Some(&NodeValue::Float(7.0))
+		);
+	}
+
+	#[test]
+	fn bfs_dequeue_order_is_deterministic() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (a, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(1.0));
+		let (b, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(2.0));
+		let (c, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(3.0));
+		let (merge, _) = add_node(
+			&mut graph,
+			&[
+				("a_in", ValueType::Float),
+				("b_in", ValueType::Float),
+				("c_in", ValueType::Float),
+			],
+			Emit::Float(4.0),
+		);
+		for (from, to, input) in [
+			(input, a, "tex_in"),
+			(input, b, "tex_in"),
+			(input, c, "tex_in"),
+			(a, merge, "a_in"),
+			(b, merge, "b_in"),
+			(c, merge, "c_in"),
+			(merge, output, GRAPH_OUTPUT_INPUT),
+		] {
+			graph
+				.connect(from, to, input, -1)
+				.unwrap_or_else(|e| panic!("{from:?} -> {to:?}.{input}: {e:?}"));
+		}
+
+		let mut first = Probe::new();
+		Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut first)
+			.expect("the fan-in graph is orderable");
+		let mut second = Probe::new();
+		Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut second)
+			.expect("the fan-in graph is orderable");
+		assert_eq!(first.order, vec![input, a, b, c, merge, output]);
+		assert_eq!(first.order, second.order, "same graph -> same order");
+	}
+
+	#[test]
+	fn bfs_pulls_time_shifted_upstreams_through_the_dfs_memo() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (source, source_calls) = add_node(&mut graph, &[], Emit::Time);
+		let rows = Arc::new(Mutex::new(Vec::new()));
+		let mut core = NodeCore::new();
+		core.add_input(Input::new("tex_in", ValueType::Texture, NodeValue::None));
+		core.add_input(Input::new("val_in", ValueType::Float, NodeValue::None));
+		let delay = graph.add_node(
+			core,
+			Box::new(Delay {
+				rows: Arc::clone(&rows),
+			}),
+		);
+		graph
+			.connect(input, delay, "tex_in", -1)
+			.expect("input -> delay.tex_in");
+		graph
+			.connect(source, delay, "val_in", -1)
+			.expect("source -> delay.val_in");
+		graph
+			.connect(delay, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("delay -> output");
+
+		let mut probe = Probe::new();
+		let value = Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect("an out-of-live-set feeder does not stall the sweep");
+		assert_eq!(
+			probe.order,
+			vec![input, source, delay, output],
+			"the far end is pulled during delay's turn, not queued"
+		);
+		assert_eq!(source_calls.load(Ordering::Relaxed), 1);
+		let rows = rows.lock().expect("delay rows");
+		assert_eq!(rows.len(), 1);
+		// One second late: the sweep pulled the source at time 1, not 0.
+		assert_eq!(rows[0].get("val_in"), Some(&NodeValue::Float(1.0)));
+		assert_eq!(value, NodeValue::None);
+	}
+
+	#[test]
+	fn bfs_hands_resolved_tables_to_downstream_nodes() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (a, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(1.0));
+		let (b, _) = add_node(&mut graph, &[("tex_in", ValueType::Texture)], Emit::Float(2.0));
+		let (merge, _) = add_node(
+			&mut graph,
+			&[("a_in", ValueType::Float), ("b_in", ValueType::Float)],
+			Emit::Float(3.0),
+		);
+		graph.connect(input, a, "tex_in", -1).expect("input -> a");
+		graph.connect(input, b, "tex_in", -1).expect("input -> b");
+		graph.connect(a, merge, "a_in", -1).expect("a -> merge");
+		graph.connect(b, merge, "b_in", -1).expect("b -> merge");
+		graph
+			.connect(merge, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("merge -> output");
+
+		let mut probe = Probe::new();
+		probe.replace = Some((
+			a,
+			vec![(ValueType::Float, NodeValue::Float(999.0))],
+		));
+		Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect("the converging graph is orderable");
+		let row = probe.rows.get(&merge).expect("merge resolved");
+		assert_eq!(
+			row.get("a_in"),
+			Some(&NodeValue::Float(999.0)),
+			"the downstream row carries what resolve() left in a's table"
+		);
+		assert_eq!(row.get("b_in"), Some(&NodeValue::Float(2.0)));
+	}
+
+	#[test]
+	fn bfs_reports_a_cycle_and_names_its_nodes() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (a, a_calls) = add_node(&mut graph, &[("in", ValueType::Float)], Emit::Float(1.0));
+		let (b, b_calls) = add_node(&mut graph, &[("in", ValueType::Float)], Emit::Float(2.0));
+		graph.connect(input, a, "in", -1).expect("input -> a");
+		graph.connect(a, b, "in", -1).expect("a -> b");
+		graph
+			.connect(b, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("b -> output");
+		graph.force_connect(b, a, "in", -1);
+
+		let mut probe = Probe::new();
+		let error = Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect_err("a cycle must be reported, not ordered");
+		let message = match error {
+			Error::Failed(message) => message,
+			other => panic!("Failed expected, got {other:?}"),
+		};
+		assert!(message.contains("cycle"), "message names the failure: {message}");
+		assert!(
+			message.contains(&format!("{a:?}")),
+			"message names {a:?}: {message}"
+		);
+		assert!(
+			message.contains(&format!("{b:?}")),
+			"message names {b:?}: {message}"
+		);
+		assert_eq!(probe.order, vec![input], "nothing past the cycle resolves");
+		assert_eq!(a_calls.load(Ordering::Relaxed), 0);
+		assert_eq!(b_calls.load(Ordering::Relaxed), 0);
+	}
+
+	#[test]
+	fn bfs_returns_the_output_texture() {
+		let mut graph = Graph::new();
+		let (input, output) = detached_endpoints(&mut graph);
+		let (source, source_calls) = add_node(&mut graph, &[], Emit::Texture);
+		graph
+			.connect(source, input, GRAPH_INPUT_FEED_INPUT, -1)
+			.expect("source -> feed_in");
+		graph
+			.connect(input, output, GRAPH_OUTPUT_INPUT, -1)
+			.expect("input -> output");
+
+		let mut probe = Probe::new();
+		let value = Traverser::new()
+			.eval_graph_bfs(&graph, Rational::new(0, 1), &mut probe)
+			.expect("a fed input -> output graph is orderable");
+		match value {
+			NodeValue::Texture(handle) => assert!(handle.ctx.is_null()),
+			other => panic!("texture expected, got {other:?}"),
+		}
+		assert_eq!(probe.order, vec![source, input, output]);
+		assert_eq!(source_calls.load(Ordering::Relaxed), 1);
+	}
+
+	#[test]
+	fn bfs_requires_both_endpoints() {
+		let mut probe = Probe::new();
+		let error = Traverser::new()
+			.eval_graph_bfs(&Graph::new(), Rational::new(0, 1), &mut probe)
+			.expect_err("an endpointless graph cannot be swept");
+		assert_eq!(error, Error::NotFound);
+	}
+}