// 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/>.

//! Boundary tests for the `RenderTask` base: empty ranges, stale viewer
//! nodes, hook errors, cancellation, forced sizes/formats, footage and
//! montage delivery (video + audio, audio first) and progress reporting.
//!
//! Renders run on the process-wide manager's test-only inline (`Threads`)
//! backend, serialized through one static mutex, so no GPU, no worker
//! processes and no sleeps are involved. Media is generated by
//! `oak_codec::testmedia`.

use std::sync::{Arc, Mutex, MutexGuard};

use oak_core::texture::Texture;
use oak_core::videoparams::VideoParams;
use oak_core::{PixelFormat, Rational, TimeRange};
use oak_node::block::ClipBlockBehavior;
use oak_node::footage::FootageBehavior;
use oak_node::id::NodeId;
use oak_node::node::NodeCore;
use oak_node::project::Project;
use oak_node::sequence::SequenceBehavior;
use oak_node::track::{TrackBehavior, TrackListBehavior, TrackType};
use oak_render::manager::{RenderBackendChoice, RenderManager};
use oak_render::ticket::AudioSamples;
use oak_task::error::{Error, Result};
use oak_task::nodeops::{self, ProjectRef};
use oak_task::render::{ForceParams, RenderTask, RenderTaskBehavior};
use oak_task::task::{Task, TaskEvent};

/// The manager is process-wide and the inline backend is test-only.
static RENDER_LOCK: Mutex<()> = Mutex::new(());

/// Serialize render-manager use and make sure the inline backend is up.
fn render_serial() -> MutexGuard<'static, ()> {
	let guard = RENDER_LOCK.lock().unwrap_or_else(|e| e.into_inner());
	if RenderManager::global().is_none() {
		RenderManager::init_with_backend(RenderBackendChoice::Threads).expect("render manager");
	}
	guard
}

fn clip_path(tag: &str) -> std::path::PathBuf {
	std::env::temp_dir().join(format!("oaktask_render_{tag}_{}.mp4", std::process::id()))
}

/// Locate an `oak-worker` binary for the private-dispatcher path: an
/// explicit `OAK_WORKER_BIN`, a sibling of the test executable, or the
/// workspace `target/<profile>/oak-worker`. `None` means the path is
/// unavailable in this environment (the test then skips with a reason).
fn locate_oak_worker() -> Option<std::path::PathBuf> {
	if let Some(var) = std::env::var_os("OAK_WORKER_BIN") {
		let path = std::path::PathBuf::from(var);
		if path.is_file() {
			return Some(path);
		}
	}
	if let Some(dir) = std::env::current_exe().ok().and_then(|e| e.parent().map(|p| p.to_path_buf())) {
		let path = dir.join("oak-worker");
		if path.is_file() {
			return Some(path);
		}
	}
	let manifest = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
	for profile in ["debug", "release"] {
		let path = manifest.join("../../target").join(profile).join("oak-worker");
		if path.is_file() {
			return Some(path);
		}
	}
	None
}

// ---- fixture ----------------------------------------------------------

/// A project with one probed footage node (64x64, 1 s) and one sequence.
fn footage_project(media: &str, fps: i64) -> (ProjectRef, NodeId, VideoParams) {
	let project = Project::new();
	let (footage_id, sequence) = {
		let mut p = project.lock().unwrap();

		let mut footage = FootageBehavior::new(media);
		footage.probe().expect("probe the generated clip");
		let footage_id = p.graph.add_node(NodeCore::new(), Box::new(footage));

		let (score, sbehavior) = SequenceBehavior::create();
		let sequence = p.graph.add_node(score, sbehavior);
		if let Some(seq) = p
			.graph
			.get_mut(sequence)
			.and_then(|e| e.behavior.as_any_mut())
			.and_then(|a| a.downcast_mut::<SequenceBehavior>())
		{
			seq.video_params[0].frame_rate = Rational::new(fps, 1);
			seq.video_params[0].width = 64;
			seq.video_params[0].height = 64;
		}
		(footage_id, sequence)
	};
	let params = nodeops::sequence_video_params(&project, sequence, 0).expect("sequence params");
	(project, footage_id, params)
}

/// A sequence with one video clip (and optionally an audio clip) of the
/// same media, both covering `[0, 1s)`. With `effect = Some(enabled)`, an
/// Opacity effect (factor 0) with that enable-state sits between the
/// footage and the clip so the effect-chain walk is exercised and its
/// effect is observable in the delivered pixels: the montage evaluator
/// applies the built-in Opacity on the CPU (unknown built-ins pass
/// through without a GPU, so those could not prove the chain ran).
fn sequence_project(
	media: &str,
	with_audio: bool,
	effect: Option<bool>,
) -> (ProjectRef, NodeId, VideoParams) {
	let project = Project::new();
	let sequence = {
		let mut p = project.lock().unwrap();

		let mut footage = FootageBehavior::new(media);
		footage.probe().expect("probe the generated clip");
		let footage_id = p.graph.add_node(NodeCore::new(), Box::new(footage));

		let (score, sbehavior) = SequenceBehavior::create();
		let sequence = p.graph.add_node(score, sbehavior);
		if let Some(seq) = p
			.graph
			.get_mut(sequence)
			.and_then(|e| e.behavior.as_any_mut())
			.and_then(|a| a.downcast_mut::<SequenceBehavior>())
		{
			seq.video_params[0].frame_rate = Rational::new(1, 1);
			seq.video_params[0].width = 64;
			seq.video_params[0].height = 64;
		}

		let kinds: &[TrackType] = if with_audio {
			&[TrackType::Video, TrackType::Audio]
		} else {
			&[TrackType::Video]
		};
		for &kind in kinds {
			let (tcore, tbehavior) = TrackBehavior::create();
			let track = p.graph.add_node(tcore, tbehavior);
			p.graph
				.get_mut(track)
				.unwrap()
				.behavior
				.as_any_mut()
				.unwrap()
				.downcast_mut::<TrackBehavior>()
				.unwrap()
				.kind = kind;

			let (ccore, cbehavior) = oak_node::block::clip_create();
			let clip = p.graph.add_node(ccore, cbehavior);
			if effect.is_some() && kind == TrackType::Video {
				// The Opacity module is private; the factory is the public
				// constructor. The value input is zeroed so an enabled
				// effect zeroes every delivered channel (including alpha).
				let (ecore, ebehavior) = oak_node::factory::Factory::global()
					.create_any("org.olivevideoeditor.Olive.opacity")
					.expect("opacity node registered");
				let effect_node = p.graph.add_node(ecore, ebehavior);
				p.graph
					.get_mut(effect_node)
					.unwrap()
					.core
					.set_standard_value(
						oak_node::node::ENABLED_INPUT,
						-1,
						oak_node::value::NodeValue::Boolean(effect.unwrap_or(false)),
					);
				p.graph
					.get_mut(effect_node)
					.unwrap()
					.core
					.set_standard_value("opacity_in", -1, oak_node::value::NodeValue::Float(0.0));
				p.graph
					.connect(footage_id, effect_node, "tex_in", -1)
					.expect("connect footage to effect");
				p.graph
					.connect(
						effect_node,
						clip,
						oak_node::block::clip_input::TEXTURE_INPUT,
						-1,
					)
					.expect("connect effect to clip");
			} else {
				p.graph
					.connect(
						footage_id,
						clip,
						oak_node::block::clip_input::TEXTURE_INPUT,
						-1,
					)
					.expect("connect footage to clip");
			}
			p.graph
				.get_mut(clip)
				.unwrap()
				.behavior
				.as_any_mut()
				.unwrap()
				.downcast_mut::<ClipBlockBehavior>()
				.expect("clip block")
				.core
				.range = TimeRange::new(Rational::new(0, 1), Rational::new(1, 1));
			p.graph
				.get_mut(track)
				.unwrap()
				.behavior
				.as_any_mut()
				.unwrap()
				.downcast_mut::<TrackBehavior>()
				.unwrap()
				.append_block(clip);

			let (lcore, lbehavior) = TrackListBehavior::create();
			let list = p.graph.add_node(lcore, lbehavior);
			{
				let list_b = p
					.graph
					.get_mut(list)
					.unwrap()
					.behavior
					.as_any_mut()
					.unwrap()
					.downcast_mut::<TrackListBehavior>()
					.unwrap();
				list_b.kind = kind;
				list_b.tracks.push(track);
			}
			p.graph
				.get_mut(sequence)
				.unwrap()
				.behavior
				.as_any_mut()
				.unwrap()
				.downcast_mut::<SequenceBehavior>()
				.unwrap()
				.track_lists
				.push(list);
		}
		sequence
	};

	let params = nodeops::sequence_video_params(&project, sequence, 0).expect("sequence params");
	(project, sequence, params)
}

/// Build a `RenderTask` for `viewer` with a 64x64 forced F32 render size.
fn render_task(project: &ProjectRef, viewer: NodeId, params: Option<VideoParams>) -> RenderTask {
	let mut task = RenderTask::new(
		Task::new("render", None),
		params,
		(project.clone(), viewer),
		ForceParams {
			force_width: 64,
			force_height: 64,
			// The default leaves `force_format` off (-1), so the pipeline
			// frame stays F32.
			..ForceParams::default()
		},
		None,
	);
	task.set_max_inflight(2);
	task
}

/// The first RGBA pixel of a delivered texture as f32 (downloads GPU
/// textures; `None` when the texture cannot be read back).
fn first_pixel(frame: &Texture) -> Option<[f32; 4]> {
	let f = frame.to_frame().ok()?;
	if f.width <= 0 || f.height <= 0 || f.data.len() < 16 {
		return None;
	}
	let mut px = [0.0f32; 4];
	for (i, channel) in px.iter_mut().enumerate() {
		let start = i * 4;
		*channel = f32::from_le_bytes(f.data[start..start + 4].try_into().ok()?);
	}
	Some(px)
}

/// A `RenderTaskBehavior` recording the delivered payloads in order.
#[derive(Default)]
struct Recorder {
	order: Vec<&'static str>,
	frames: Vec<(i32, i32)>,
	/// Pixel format of each delivered frame (the forced-format contract).
	formats: Vec<PixelFormat>,
	/// First RGBA pixel of each delivered frame (effect and generated-frame
	/// observability).
	first_pixels: Vec<Option<[f32; 4]>>,
	timestamps: Vec<Rational>,
	audio_samples: Vec<usize>,
	fail_frames: bool,
	fail_audio: bool,
	/// Cancel the driving task from the first frame hook (exercises the
	/// between-frames cancellation path deterministically).
	cancel_on_first_frame: bool,
	cancelled: bool,
}

impl Recorder {
	fn new() -> Recorder {
		Recorder::default()
	}
}

impl RenderTaskBehavior for Recorder {
	fn frame_downloaded(&mut self, task: &mut Task, frame: &Texture) -> Result<()> {
		self.order.push("frame");
		self.frames.push(frame.size());
		self.formats.push(frame.format());
		self.first_pixels.push(first_pixel(frame));
		self.timestamps.push(
			frame
				.to_frame()
				.map(|f| f.timestamp)
				.unwrap_or(Rational::NULL),
		);
		if self.fail_frames {
			return Err(Error::Failed("frame hook failed".to_string()));
		}
		if self.cancel_on_first_frame && !self.cancelled {
			self.cancelled = true;
			task.cancel();
		}
		Ok(())
	}

	fn audio_downloaded(&mut self, _task: &mut Task, samples: &AudioSamples) -> Result<()> {
		self.order.push("audio");
		self.audio_samples.push(samples.samples.len());
		if self.fail_audio {
			return Err(Error::Failed("audio hook failed".to_string()));
		}
		Ok(())
	}

	fn encode_subtitle(&mut self, _task: &mut Task, _text: &str) -> Result<()> {
		Err(Error::Failed(
			"subtitles are not rendered by this test".into(),
		))
	}
}

// ---- pure / prepare paths --------------------------------------------

/// A default-constructed render task exposes zero frames and every
/// configurator tolerates edge values. The window/progress configurators
/// have no getters; their observable effects are covered by
/// [`max_inflight_edges_still_deliver`] and
/// [`native_progress_signalling_can_be_disabled`].
#[test]
fn new_defaults_and_configurators_accept_edges() {
	let project = Project::new();
	let mut task = RenderTask::new(
		Task::new("render", None),
		None,
		(project, NodeId::INVALID),
		ForceParams::default(),
		None,
	);
	assert_eq!(task.total_frames(), 0);
	assert!(!task.force_params.has_force_matrix);
	// Defaults must leave every override off (-1, not 0 = U8).
	assert_eq!(task.force_params.force_format, -1);
	task.set_render_inputs(0, true, TimeRange::default());
	task.set_native_progress_signalling(false);
	task.set_max_inflight(0);
	task.set_max_inflight(usize::MAX);
	assert_eq!(task.total_frames(), 0, "configuring does not plan frames");
	assert_eq!(
		task.force_params.force_format, -1,
		"configuring keeps defaults"
	);
}

/// The edge in-flight windows (`0`, `usize::MAX`) still render: the loop
/// clamps the window to at least one ticket. There is no `max_inflight`
/// getter, so the delivered frame is the observable effect.
#[test]
fn max_inflight_edges_still_deliver() {
	let media = clip_path("inflight");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 1);

	let _guard = render_serial();
	for max in [0usize, usize::MAX] {
		let mut task = render_task(&project, footage, Some(params.clone()));
		task.set_max_inflight(max);
		task.set_render_inputs(
			0,
			false,
			TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
		);
		let mut driver = Task::new("render", None);
		let mut recorder = Recorder::new();
		assert!(
			task.render(&mut driver, &mut recorder).is_ok(),
			"max_inflight {max} must still render"
		);
		assert_eq!(
			recorder.frames,
			vec![(64, 64)],
			"max_inflight {max} delivers the frame"
		);
	}

	let _ = std::fs::remove_file(&media);
}

/// An empty export range computes no frames and submits nothing.
#[test]
fn empty_export_range_renders_nothing() {
	let media = clip_path("empty");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 1);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(0, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(task.total_frames(), 0);
	assert!(recorder.frames.is_empty());
	assert!(recorder.audio_samples.is_empty());

	let _ = std::fs::remove_file(&media);
}

/// A viewer that is not a footage/sequence node fails the render with the
/// "no node connected to the viewer output" error.
#[test]
fn invalid_viewer_fails_the_render() {
	let project = Project::new();
	let _guard = render_serial();
	let mut task = render_task(&project, NodeId::INVALID, None);
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	let error = task.render(&mut driver, &mut recorder).unwrap_err();
	assert!(
		error.to_string().contains("viewer output"),
		"the error names the viewer: {error}"
	);
	assert_eq!(task.total_frames(), 1, "the frame plan is computed first");
	assert!(recorder.frames.is_empty());
}

/// A sequence viewer with no track lists renders a generated (transparent)
/// frame instead of a montage.
#[test]
fn sequence_viewer_without_tracks_renders_a_generated_frame() {
	let project = Project::new();
	let sequence = {
		let mut p = project.lock().unwrap();
		let (score, sbehavior) = SequenceBehavior::create();
		p.graph.add_node(score, sbehavior)
	};

	let _guard = render_serial();
	let mut task = render_task(&project, sequence, None);
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.order, vec!["frame"]);
	assert_eq!(recorder.frames, vec![(64, 64)]);
	assert_eq!(
		recorder.first_pixels,
		vec![Some([0.0, 0.0, 0.0, 0.0])],
		"the generated frame is fully transparent"
	);
}

// ---- footage path -----------------------------------------------------

/// A single-footage render delivers exactly one frame and reports native
/// progress to the driver.
#[test]
fn footage_frame_is_delivered_with_progress() {
	let media = clip_path("footage");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 1);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let progress = Arc::new(Mutex::new(Vec::new()));
	let sink = progress.clone();
	let mut driver = Task::new("render", None);
	driver.set_event_listener(Box::new(move |event| {
		if let TaskEvent::Progress(value) = event {
			sink.lock().unwrap().push(value);
		}
	}));

	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.order, vec!["frame"]);
	assert_eq!(recorder.frames, vec![(64, 64)]);
	assert_eq!(recorder.timestamps, vec![Rational::new(0, 1)]);
	assert_eq!(
		*progress.lock().unwrap(),
		vec![1.0],
		"the native progress reaches 1.0 after the single frame"
	);
	assert!(!driver.is_cancelled());

	let _ = std::fs::remove_file(&media);
}

/// Forced dimensions and an explicit F32 format reach the delivered
/// texture (size and pixel format are both asserted on the payload).
#[test]
fn forced_size_and_format_are_applied() {
	let media = clip_path("force");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, _) = footage_project(&media.to_string_lossy(), 1);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, None);
	task.force_params.force_width = 32;
	task.force_params.force_height = 32;
	task.force_params.force_format = PixelFormat::F32 as i32;
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.frames, vec![(32, 32)]);
	assert_eq!(
		recorder.formats,
		vec![PixelFormat::F32],
		"the forced F32 format reaches the delivered texture"
	);

	let _ = std::fs::remove_file(&media);
}

/// A forced **non-default** format reaches the delivered texture: the
/// generated-frame path supports U8, so if `force_format` were dropped the
/// delivered texture would be F32 and this fails. (The F32 case above pins
/// the delivery contract but cannot distinguish the forced value from the
/// pipeline default.)
#[test]
fn forced_non_default_format_reaches_the_delivered_texture() {
	let project = Project::new();
	let sequence = {
		let mut p = project.lock().unwrap();
		let (score, sbehavior) = SequenceBehavior::create();
		p.graph.add_node(score, sbehavior)
	};

	let _guard = render_serial();
	let mut task = render_task(&project, sequence, None);
	task.force_params.force_format = PixelFormat::U8 as i32;
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.frames, vec![(64, 64)]);
	assert_eq!(
		recorder.formats,
		vec![PixelFormat::U8],
		"the forced U8 format reaches the delivered texture"
	);
}

/// A frame-hook error aborts the render after the first delivered frame.
#[test]
fn hook_error_stops_the_render() {
	let media = clip_path("hookerr");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	// 2 fps over one second: two frames to schedule.
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 2);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	recorder.fail_frames = true;
	let error = task.render(&mut driver, &mut recorder).unwrap_err();
	assert!(error.to_string().contains("frame hook failed"));
	assert_eq!(task.total_frames(), 2);
	assert_eq!(
		recorder.frames,
		vec![(64, 64)],
		"delivery stopped at the error"
	);

	let _ = std::fs::remove_file(&media);
}

/// With native progress signalling disabled (the export path), the render
/// loop delivers frames without emitting progress itself.
#[test]
fn native_progress_signalling_can_be_disabled() {
	let media = clip_path("noprogress");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 1);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_native_progress_signalling(false);
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let progress = Arc::new(Mutex::new(Vec::new()));
	let sink = progress.clone();
	let mut driver = Task::new("render", None);
	driver.set_event_listener(Box::new(move |event| {
		if let TaskEvent::Progress(value) = event {
			sink.lock().unwrap().push(value);
		}
	}));
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.frames.len(), 1, "frames are still delivered");
	assert!(
		progress.lock().unwrap().is_empty(),
		"the loop emits no progress of its own"
	);
	assert!(driver.error().is_none());

	let _ = std::fs::remove_file(&media);
}

/// Cancelling before the render aborts with `Error::Cancelled` and delivers
/// nothing (the in-flight ticket is drained).
#[test]
fn cancel_before_render_reports_cancelled() {
	let media = clip_path("cancel");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 1);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	driver.cancel();
	let mut recorder = Recorder::new();
	let result = task.render(&mut driver, &mut recorder);
	assert!(matches!(result, Err(Error::Cancelled)));
	assert_eq!(task.total_frames(), 1);
	assert!(
		recorder.order.is_empty(),
		"nothing is delivered after abort"
	);

	let _ = std::fs::remove_file(&media);
}

// ---- montage / audio path --------------------------------------------

/// A one-clip sequence renders through the montage compositor.
#[test]
fn sequence_montage_renders_a_frame() {
	let media = clip_path("montage");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, sequence, params) = sequence_project(&media.to_string_lossy(), false, None);

	let _guard = render_serial();
	let mut task = render_task(&project, sequence, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.order, vec!["frame"]);
	assert_eq!(recorder.frames, vec![(64, 64)]);
	assert_eq!(recorder.timestamps, vec![Rational::new(0, 1)]);

	let _ = std::fs::remove_file(&media);
}

/// An effect in the clip's chain is walked (params collected) and its
/// enable state applied: the fixture's Opacity (factor 0) zeroes every
/// delivered channel when enabled and passes the decoded clip through when
/// disabled, so the two branches differ observably — a regression that
/// drops the chain (or ignores the enable flag) cannot pass both.
#[test]
fn clip_effect_chain_is_walked_and_bypassed() {
	let media = clip_path("effect");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let _guard = render_serial();

	for enabled in [false, true] {
		let (project, sequence, params) =
			sequence_project(&media.to_string_lossy(), false, Some(enabled));

		// Fixture sanity: the clip's effect input is fed by the Opacity
		// node, and the node carries the factor-zero input the walk needs.
		let upstream_type = {
			let guard = project.lock().unwrap_or_else(|e| e.into_inner());
			let clip = guard
				.graph
				.node_ids()
				.into_iter()
				.find(|id| {
					guard
						.graph
						.get(*id)
						.and_then(|e| e.behavior.as_any())
						.and_then(|a| a.downcast_ref::<ClipBlockBehavior>())
						.is_some()
				})
				.expect("clip node");
			let input = guard.graph.get(clip).unwrap().core.effect_input.clone();
			let upstream = guard
				.graph
				.connected_output(clip, &input, -1)
				.expect("effect upstream");
			let entry = guard.graph.get(upstream).unwrap();
			assert!(
				matches!(
					entry.core.standard_value(oak_node::node::ENABLED_INPUT, -1),
					oak_node::value::NodeValue::Boolean(v) if v == enabled
				),
				"the fixture enable state"
			);
			assert!(
				matches!(
					entry.core.standard_value("opacity_in", -1),
					oak_node::value::NodeValue::Float(v) if v == 0.0
				),
				"the fixture opacity factor"
			);
			entry.behavior.type_id().to_string()
		};
		assert_eq!(upstream_type, "org.olivevideoeditor.Olive.opacity");

		let mut task = render_task(&project, sequence, Some(params));
		task.set_render_inputs(
			0,
			false,
			TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
		);
		let mut driver = Task::new("render", None);
		let mut recorder = Recorder::new();
		assert!(task.render(&mut driver, &mut recorder).is_ok());
		assert_eq!(recorder.order, vec!["frame"]);
		assert_eq!(recorder.frames, vec![(64, 64)]);

		let pixel = recorder.first_pixels[0].expect("CPU frame with a readable pixel");
		if enabled {
			assert_eq!(
				pixel,
				[0.0, 0.0, 0.0, 0.0],
				"the enabled opacity zeroes every channel"
			);
		} else {
			assert!(
				pixel != [0.0, 0.0, 0.0, 0.0] && pixel[3] > 0.0,
				"the disabled effect must pass the decoded clip through unchanged: {pixel:?}"
			);
		}
	}

	let _ = std::fs::remove_file(&media);
}

/// Audio is enabled: the audio ticket is delivered before the video frame
/// and carries the decoded samples.
#[test]
fn audio_is_delivered_before_the_video_frame() {
	let media = clip_path("audio");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, sequence, params) = sequence_project(&media.to_string_lossy(), true, None);

	let _guard = render_serial();
	let mut task = render_task(&project, sequence, Some(params));
	task.set_render_inputs(
		0,
		true,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(
		recorder.order,
		vec!["audio", "frame"],
		"the audio range is delivered first"
	);
	assert_eq!(recorder.audio_samples.len(), 1);
	assert!(
		recorder.audio_samples[0] > 0,
		"the decoded audio range is not empty"
	);
	assert_eq!(recorder.frames, vec![(64, 64)]);
	assert_eq!(task.total_frames(), 1);

	let _ = std::fs::remove_file(&media);
}

/// The frame count covers the whole export range (not just the delivered
/// window) and the timestamps advance by the timebase.
#[test]
fn total_frames_and_timestamps_follow_the_range() {
	let media = clip_path("count");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	// 4 fps over one second: frames at 0, 1/4, 1/2, 3/4.
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 4);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(task.total_frames(), 4);
	assert_eq!(
		recorder.timestamps,
		vec![
			Rational::new(0, 1),
			Rational::new(1, 4),
			Rational::new(1, 2),
			Rational::new(3, 4),
		]
	);

	let _ = std::fs::remove_file(&media);
}

/// Native progress is emitted once per delivered frame as a fraction of the
/// whole planned range (not just the in-flight window).
#[test]
fn progress_reaches_one_across_multiple_frames() {
	let media = clip_path("multiprogress");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 4);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let progress = Arc::new(Mutex::new(Vec::new()));
	let sink = progress.clone();
	let mut driver = Task::new("render", None);
	driver.set_event_listener(Box::new(move |event| {
		if let TaskEvent::Progress(value) = event {
			sink.lock().unwrap().push(value);
		}
	}));
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(recorder.frames.len(), 4);
	assert_eq!(
		*progress.lock().unwrap(),
		vec![0.25, 0.5, 0.75, 1.0],
		"one progress report per frame, normalized by the plan"
	);

	let _ = std::fs::remove_file(&media);
}

/// A range that does not start at zero plans and delivers exactly the
/// frames inside it.
#[test]
fn range_with_non_zero_start_plans_offset_frames() {
	let media = clip_path("offset");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	// 4 fps: the half-open range [1/2, 1) holds frames at 1/2 and 3/4.
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 4);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(1, 2), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	assert!(task.render(&mut driver, &mut recorder).is_ok());
	assert_eq!(task.total_frames(), 2);
	assert_eq!(
		recorder.timestamps,
		vec![Rational::new(1, 2), Rational::new(3, 4)]
	);

	let _ = std::fs::remove_file(&media);
}

/// An audio-hook error aborts the render at the audio delivery, before any
/// video frame is handed to the hooks.
#[test]
fn audio_hook_error_stops_the_render() {
	let media = clip_path("audioerr");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, sequence, params) = sequence_project(&media.to_string_lossy(), true, None);

	let _guard = render_serial();
	let mut task = render_task(&project, sequence, Some(params));
	task.set_render_inputs(
		0,
		true,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	recorder.fail_audio = true;
	let error = task.render(&mut driver, &mut recorder).unwrap_err();
	assert!(error.to_string().contains("audio hook failed"));
	assert_eq!(
		recorder.order,
		vec!["audio"],
		"the error stops the render before the frame"
	);

	let _ = std::fs::remove_file(&media);
}

/// A cancellation raised from the first frame hook aborts the render with
/// `Error::Cancelled`, delivers no further frames, and cancels the
/// still-in-flight tickets on the way out.
#[test]
fn cancel_between_frames_reports_cancelled() {
	let media = clip_path("midcancel");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	// 4 fps over one second: four frames to schedule.
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 4);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_max_inflight(1);
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	recorder.cancel_on_first_frame = true;
	let result = task.render(&mut driver, &mut recorder);
	assert!(
		matches!(result, Err(Error::Cancelled)),
		"a cancel between frames maps to Cancelled: {result:?}"
	);
	assert_eq!(task.total_frames(), 4, "the whole plan was computed");
	assert_eq!(recorder.frames.len(), 1, "delivery stopped at the cancel");
	assert!(driver.is_cancelled());

	let _ = std::fs::remove_file(&media);
}

// ---- private process dispatcher (no global manager) -------------------

/// Environment-only reason the private dispatcher cannot render here: the
/// worker pool itself could not be created or started (an unusable
/// `oak-worker`, exec denied, process-table exhaustion, ...). Everything
/// else — ticket errors, wrong payload kinds, unexpected timestamps, shm
/// conversion problems — is a product failure and must fail the test.
fn worker_pool_environment_error(error: &Error) -> bool {
	matches!(error, Error::Failed(message) if message.starts_with("render worker pool"))
}

/// Shuts the process-wide render manager down and points `OAK_WORKER_BIN`
/// at `worker` for the private-dispatcher render, restoring both on drop —
/// so a panicking render can never leak the shut-down manager or the env
/// override into later tests. [`DispatcherEnvGuard::restore`] restores
/// eagerly, before the assertions.
struct DispatcherEnvGuard {
	previous_worker_bin: Option<std::ffi::OsString>,
	restored: bool,
}

impl DispatcherEnvGuard {
	fn install(worker: &std::path::Path) -> DispatcherEnvGuard {
		let previous_worker_bin = std::env::var_os("OAK_WORKER_BIN");
		std::env::set_var("OAK_WORKER_BIN", worker);
		let guard = DispatcherEnvGuard {
			previous_worker_bin,
			restored: false,
		};
		RenderManager::shutdown();
		assert!(
			RenderManager::global().is_none(),
			"the private path needs no global manager"
		);
		guard
	}

	/// Restore the manager and the environment exactly once.
	fn restore(&mut self) {
		if self.restored {
			return;
		}
		self.restored = true;
		RenderManager::init_with_backend(RenderBackendChoice::Threads).expect("render manager");
		match self.previous_worker_bin.take() {
			Some(value) => std::env::set_var("OAK_WORKER_BIN", value),
			None => std::env::remove_var("OAK_WORKER_BIN"),
		}
	}
}

impl Drop for DispatcherEnvGuard {
	fn drop(&mut self) {
		self.restore();
	}
}

/// With the global render manager shut down, [`RenderTask::render`] stands
/// up its own process dispatcher and renders through `oak-worker`
/// processes: the `ShmFrame`/`ShmAudio` delivery seams, the poll pump and
/// the dispatcher teardown (`shm_frame_to_texture`'s BGRA8 and F32
/// branches).
///
/// Only the worker pool being unavailable is an environment skip (the
/// missing-binary case skips earlier): a pool that cannot be created or
/// started prints a reason. Every product failure — a `"Frame render
/// ticket failed"`, a bad shm payload/format, a poll timeout surfacing as
/// a ticket error — fails the test.
#[test]
fn private_dispatcher_renders_frames_and_audio_through_workers() {
	let _guard = render_serial();
	let Some(worker) = locate_oak_worker() else {
		eprintln!("oak-worker binary not built; skipping the private dispatcher render");
		return;
	};

	let media = clip_path("private");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");

	// The dispatcher resolves the worker binary from the environment when
	// it is constructed by the render call below. The guard restores the
	// manager and the env also when a render panics.
	let mut env = DispatcherEnvGuard::install(&worker);

	// Audio + video through the workers (the default BGRA8 wire format).
	let (project, sequence, params) = sequence_project(&media.to_string_lossy(), true, None);
	let mut task = render_task(&project, sequence, Some(params));
	task.set_max_inflight(2);
	task.set_render_inputs(
		0,
		true,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	let outcome = task.render(&mut driver, &mut recorder);

	// A forced F32 ticket takes the no-conversion shm branch.
	let (fproject, footage, _) = footage_project(&media.to_string_lossy(), 1);
	let mut f32_task = render_task(&fproject, footage, None);
	f32_task.force_params.force_format = PixelFormat::F32 as i32;
	f32_task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut f32_driver = Task::new("render", None);
	let mut f32_recorder = Recorder::new();
	let f32_outcome = f32_task.render(&mut f32_driver, &mut f32_recorder);

	// Restore the manager and the environment before asserting.
	env.restore();

	let mut skipped_environment = Vec::new();
	match outcome {
		Ok(()) => {
			assert_eq!(
				recorder.order,
				vec!["audio", "frame"],
				"the worker audio and frame land in order"
			);
			assert_eq!(recorder.frames, vec![(64, 64)]);
			assert_eq!(
				recorder.formats,
				vec![PixelFormat::F32],
				"the BGRA8 wire frame is converted to the F32 delivery format"
			);
			assert!(recorder.audio_samples[0] > 0);
		}
		Err(error) if worker_pool_environment_error(&error) => {
			eprintln!("private dispatcher worker pool unavailable: {error}; skipping");
			skipped_environment.push("video+audio");
		}
		Err(error) => panic!("the private dispatcher render failed (product error): {error}"),
	}
	match f32_outcome {
		Ok(()) => {
			assert_eq!(f32_recorder.frames, vec![(64, 64)]);
			assert_eq!(
				f32_recorder.formats,
				vec![PixelFormat::F32],
				"the forced F32 shm branch keeps the delivery format"
			);
		}
		Err(error) if worker_pool_environment_error(&error) => {
			eprintln!("private dispatcher F32 render unavailable: {error}; skipping");
			skipped_environment.push("f32");
		}
		Err(error) => panic!("the private dispatcher F32 render failed (product error): {error}"),
	}
	if !skipped_environment.is_empty() {
		eprintln!(
			"private dispatcher render skipped (environment unavailable): {}",
			skipped_environment.join(", ")
		);
	}

	let _ = std::fs::remove_file(&media);
}

// ---- ticket failures --------------------------------------------------

/// A footage file removed after the probe makes the frame ticket's decode
/// fail; the render loop maps the ticket error (not a hook error).
#[test]
fn missing_media_frame_ticket_reports_failure() {
	let media = clip_path("missingticket");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, footage, params) = footage_project(&media.to_string_lossy(), 1);
	// Probe already happened; remove the payload so the decode fails.
	let _ = std::fs::remove_file(&media);

	let _guard = render_serial();
	let mut task = render_task(&project, footage, Some(params));
	task.set_render_inputs(
		0,
		false,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	let error = task.render(&mut driver, &mut recorder).unwrap_err();
	assert!(
		error.to_string().contains("Frame render ticket failed"),
		"the decode failure is a ticket error: {error}"
	);
	assert!(recorder.frames.is_empty());
}

/// The same for the audio range: the audio ticket is queued first, so its
/// failure is reported as an audio ticket error before any frame.
#[test]
fn missing_media_audio_ticket_reports_failure() {
	let media = clip_path("missingaudio");
	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
	let (project, sequence, params) = sequence_project(&media.to_string_lossy(), true, None);
	let _ = std::fs::remove_file(&media);

	let _guard = render_serial();
	let mut task = render_task(&project, sequence, Some(params));
	task.set_render_inputs(
		0,
		true,
		TimeRange::new(Rational::new(0, 1), Rational::new(1, 1)),
	);
	let mut driver = Task::new("render", None);
	let mut recorder = Recorder::new();
	let error = task.render(&mut driver, &mut recorder).unwrap_err();
	assert!(
		error.to_string().contains("Audio render ticket failed"),
		"the audio decode failure names the ticket: {error}"
	);
	assert!(recorder.order.is_empty());
}