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

//! `RenderTask` abstract base, mirroring `src/task/src/render/render.h`.
//!
//! The abstract parent of [`crate::export::ExportTask`] and
//! [`crate::precache::PreCacheTask`]. It renders frames/audio through the
//! direct `oakrender::ticket::TicketArena` (single-lib unification: the
//! CHandle-based ticket C ABI and `crate::stubs` are gone) and hands each
//! result to a virtual hook that the subclass implements via the
//! [`RenderTaskBehavior`] trait.
//!
//! The viewer node is now a [`crate::nodeops::NodeRef`] (project + node
//! id): footage viewers render through `VideoTicketParams::footage`
//! (single-stream decode), sequence viewers are flattened into an ordered
//! clip montage (`VideoTicketParams::montage`) resolved from the
//! sequence's track lists. Tickets run on the process-wide
//! `oakrender::manager::RenderManager` arena when the manager is
//! initialized, otherwise on a private worker pool + arena owned by this
//! render run.
//!
//! ## Concurrent render loop
//!
//! Like the C++ original (`render.cpp`), the loop keeps up to
//! `max_inflight` tickets running at once (default
//! `std::thread::available_parallelism()`). Tickets report completion
//! through the arena's boxed completion callback (fired on the ticket's
//! own finishing thread), which pushes the finished ticket into a shared
//! queue and wakes the render thread (queue + condvar, exactly the C++
//! design). Tickets may complete out of order; a **reorder buffer**
//! delivers the results to the hooks in timestamp order — the audio range
//! first (the C++ queue-audio-first order), then every frame in ascending
//! time — so the observable per-frame contract (`frame_downloaded`/
//! `audio_downloaded` in order, progress updates, cancellation between
//! frames) is unchanged.
//!
//! CPP-PARITY: src/task/src/render/render.h

use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};

use oakcommon::videoparams::VideoParams;
use oaknode::footage::FootageBehavior;
use oaknode::sequence::SequenceBehavior;
use oaknode::track::{TrackBehavior, TrackListBehavior, TrackType};
use oakrender::ticket::{
	ticket_kind, AudioTicketParams, MontageClip, TicketArena, TicketId, TicketPayload,
	TicketResult, VideoTicketParams,
};
use oakrender::worker::WorkerPool;

use crate::error::{Error, Result};
use crate::nodeops::{
	find_input_footage, pixel_format_from_code, NodeRef, ProjectRef,
};
use crate::task::Task;
use oakcore_rs::{Rational, TimeRange};

/// `oakrender::ticket::ticket_kind::VIDEO` re-export (loop-local kind).
const TICKET_VIDEO: i32 = ticket_kind::VIDEO;
/// `oakrender::ticket::ticket_kind::AUDIO` re-export (loop-local kind).
const TICKET_AUDIO: i32 = ticket_kind::AUDIO;

/// Overrides applied to a render ticket, mirroring the C++ `ForceParams`
/// struct in render.h (fields map onto `oakrender_video_ticket_params`).
///
/// CPP-PARITY: src/task/src/render/render.h (ForceParams)
///
/// The color-matrix and color-output fields of the C ABI params were
/// dropped with the C ABI: the direct ticket arena carries a forced size
/// and pixel format only (the eval producer performs no color
/// management). The matrix fields are retained for API parity but unused.
#[derive(Clone, Debug, Default)]
pub struct ForceParams {
	/// Forced output width (0 = off).
	pub force_width: i32,
	/// Forced output height (0 = off).
	pub force_height: i32,
	/// Forced color matrix (row-major 4x4); retained for C++ API parity,
	/// not applied by the direct ticket arena.
	pub force_matrix: [f64; 16],
	/// Whether `force_matrix` is in effect; retained for parity, unused.
	pub has_force_matrix: bool,
	/// Forced pixel format (`oakcore_rs::PixelFormat` as int; -1 = off).
	pub force_format: i32,
	/// Forced channel count (0 = off).
	pub force_channel_count: i32,
}

/// Subclass hooks, standing in for the C++ protected virtuals
/// `download_frame`/`frame_downloaded`/`audio_downloaded`/`encode_subtitle`.
/// Frames and audio are the direct `oakrender` value types (the deleted
/// C ABI handed `CHandle`s; single-lib unification delivers the payload
/// values themselves).
pub trait RenderTaskBehavior {
	/// Called for each rendered video frame.
	fn frame_downloaded(
		&mut self,
		task: &mut Task,
		frame: &oakrender::texture::Texture,
	) -> Result<()>;
	/// Called for each rendered audio buffer.
	fn audio_downloaded(&mut self, task: &mut Task, samples: &oakrender::ticket::AudioSamples) -> Result<()>;
	/// Called to encode a subtitle.
	fn encode_subtitle(&mut self, task: &mut Task, text: &str) -> Result<()>;
}

/// The abstract render task base. Holds the shared [`Task`], the sequence
/// output params, the viewer node, and the subclass behavior.
pub struct RenderTask {
	/// The shared task base.
	pub base: Task,
	/// Output video params (value type; `None` = invalid/unavailable).
	pub video_params: Option<VideoParams>,
	/// The node being rendered (footage or sequence, see the module
	/// docs).
	pub viewer: NodeRef,
	/// Force params applied to every ticket.
	pub force_params: ForceParams,
	/// The subclass behavior.
	pub behavior: Option<Box<dyn RenderTaskBehavior + Send>>,
	// --- private render inputs (set by the subclass before render()) ---
	/// `olive::RenderMode::Mode` as int.
	mode: i32,
	/// Whether audio ranges are rendered.
	audio_enabled: bool,
	/// The range rendered by [`RenderTask::render`].
	export_range: TimeRange,
	/// Whether the render loop emits progress itself (the export task
	/// disables it and reports progress per written frame).
	native_progress_signalling: bool,
	/// Total number of frames in `export_range` (computed by `render`).
	total_frames: i64,
	/// Maximum render tickets kept in flight at once (the C++
	/// `maximum_rendered_frames`; defaults to `available_parallelism`).
	max_inflight: usize,
}

impl RenderTask {
	/// Create a render task for `viewer` (a footage or sequence node of
	/// `viewer.0`) with default (empty) render inputs.
	pub fn new(
		base: Task,
		video_params: Option<VideoParams>,
		viewer: NodeRef,
		force_params: ForceParams,
		behavior: Option<Box<dyn RenderTaskBehavior + Send>>,
	) -> RenderTask {
		RenderTask {
			base,
			video_params,
			viewer,
			force_params,
			behavior,
			mode: 0,
			audio_enabled: false,
			export_range: TimeRange::new(Rational::new(0, 1), Rational::new(0, 1)),
			native_progress_signalling: true,
			total_frames: 0,
			max_inflight: std::thread::available_parallelism()
				.map(|n| n.get())
				.unwrap_or(1)
				.max(1),
		}
	}

	/// Configure the render inputs before [`RenderTask::render`] (called by
	/// the concrete subclasses' `run`). The color-manager / frame-cache
	/// handle arguments of the deleted C ABI path are gone: the direct
	/// ticket arena has no color management (the eval producer ignores it)
	/// and the frame cache is keyed by the viewer node identity for
	/// pre-cache runs.
	pub fn set_render_inputs(&mut self, mode: i32, audio_enabled: bool, export_range: TimeRange) {
		self.mode = mode;
		self.audio_enabled = audio_enabled;
		self.export_range = export_range;
	}

	/// Enable/disable the render loop's own progress signalling.
	pub fn set_native_progress_signalling(&mut self, enabled: bool) {
		self.native_progress_signalling = enabled;
	}

	/// Override the number of render tickets kept in flight at once.
	///
	/// Defaults to `std::thread::available_parallelism()`, mirroring the
	/// C++ `std::max(1, int(std::thread::hardware_concurrency()))`. Primarily
	/// a test hook: a smaller window makes concurrent-completion tests
	/// deterministic regardless of the host core count.
	pub fn set_max_inflight(&mut self, max: usize) {
		self.max_inflight = max.max(1);
	}

	/// The number of frames computed by the last [`RenderTask::render`] run.
	pub fn total_frames(&self) -> i64 {
		self.total_frames
	}

	/// Frame duration rational from the video params (falls back to 1/1
	/// when the params are absent/invalid).
	fn timebase(&self) -> Rational {
		let Some(params) = &self.video_params else {
			return Rational::new(1, 1);
		};
		let (num, den) = params.frame_rate_as_time_base();
		if num <= 0 || den <= 0 {
			Rational::new(1, 1)
		} else {
			Rational::new(num as i64, den as i64)
		}
	}

	/// The forced output size from [`ForceParams`], or `None`.
	fn force_size(&self) -> Option<(i32, i32)> {
		if self.force_params.force_width > 0 && self.force_params.force_height > 0 {
			Some((self.force_params.force_width, self.force_params.force_height))
		} else {
			None
		}
	}

	/// The forced pixel format from [`ForceParams`], or `None`.
	fn force_format(&self) -> Option<oakcore_rs::PixelFormat> {
		if self.force_params.force_format >= 0 {
			Some(pixel_format_from_code(self.force_params.force_format))
		} else {
			None
		}
	}

	/// Flatten the video tracks of `sequence` into an ordered montage
	/// (bottom-most track first so the topmost track composites last;
	/// `// CPP-PARITY: M12 P0 montage contract`).
	fn video_montage(project: &ProjectRef, sequence: oaknode::id::NodeId, time: Rational) -> Vec<MontageClip> {
		let guard = project
			.lock()
			.unwrap_or_else(|e| e.into_inner());
		let Some(entry) = guard.graph.get(sequence) else {
			return Vec::new();
		};
		let Some(seq) = entry
			.behavior
			.as_any()
			.and_then(|a| a.downcast_ref::<SequenceBehavior>())
		else {
			return Vec::new();
		};
		let mut montage = Vec::new();
		for list_id in seq.track_lists.iter().filter(|i| i.valid()) {
			let Some(le) = guard.graph.get(*list_id) else {
				continue;
			};
			let Some(list) = le
				.behavior
				.as_any()
				.and_then(|a| a.downcast_ref::<TrackListBehavior>())
			else {
				continue;
			};
			if list.kind != TrackType::Video {
				continue;
			}
			for track_id in list.tracks.iter().rev() {
				let Some(te) = guard.graph.get(*track_id) else {
					continue;
				};
				let Some(track) = te
					.behavior
					.as_any()
					.and_then(|a| a.downcast_ref::<TrackBehavior>())
				else {
					continue;
				};
				for block_id in &track.blocks {
					let Some(core) = crate::nodeops::block_core_of(&guard.graph, *block_id)
					else {
						continue;
					};
					if time < core.in_() || time >= core.out() {
						continue;
					}
					let Some(footage_id) = find_input_footage(&guard.graph, *block_id) else {
						continue;
					};
					let Some(fe) = guard.graph.get(footage_id) else {
						continue;
					};
					let Some(footage) = fe
						.behavior
						.as_any()
						.and_then(|a| a.downcast_ref::<FootageBehavior>())
					else {
						continue;
					};
					montage.push(MontageClip {
						filename: footage.filename.clone(),
						stream_index: 0,
						in_time: core.in_(),
						out_time: core.out(),
						media_in: core.media_in,
						gain: 1.0,
					});
				}
			}
		}
		montage
	}

	/// Flatten the audio tracks of `sequence` into an audio montage
	/// (track order is irrelevant — the mixer accumulates gains).
	fn audio_montage(project: &ProjectRef, sequence: oaknode::id::NodeId, time: Rational) -> Vec<MontageClip> {
		let guard = project
			.lock()
			.unwrap_or_else(|e| e.into_inner());
		let Some(entry) = guard.graph.get(sequence) else {
			return Vec::new();
		};
		let Some(seq) = entry
			.behavior
			.as_any()
			.and_then(|a| a.downcast_ref::<SequenceBehavior>())
		else {
			return Vec::new();
		};
		let mut montage = Vec::new();
		for list_id in seq.track_lists.iter().filter(|i| i.valid()) {
			let Some(le) = guard.graph.get(*list_id) else {
				continue;
			};
			let Some(list) = le
				.behavior
				.as_any()
				.and_then(|a| a.downcast_ref::<TrackListBehavior>())
			else {
				continue;
			};
			if list.kind != TrackType::Audio {
				continue;
			}
			for track_id in &list.tracks {
				let Some(te) = guard.graph.get(*track_id) else {
					continue;
				};
				let Some(track) = te
					.behavior
					.as_any()
					.and_then(|a| a.downcast_ref::<TrackBehavior>())
				else {
					continue;
				};
				for block_id in &track.blocks {
					let Some(core) = crate::nodeops::block_core_of(&guard.graph, *block_id)
					else {
						continue;
					};
					if time < core.in_() || time >= core.out() {
						continue;
					}
					let Some(footage_id) = find_input_footage(&guard.graph, *block_id) else {
						continue;
					};
					let Some(fe) = guard.graph.get(footage_id) else {
						continue;
					};
					let Some(footage) = fe
						.behavior
						.as_any()
						.and_then(|a| a.downcast_ref::<FootageBehavior>())
					else {
						continue;
					};
					montage.push(MontageClip {
						filename: footage.filename.clone(),
						stream_index: 0,
						in_time: core.in_(),
						out_time: core.out(),
						media_in: core.media_in,
						gain: 1.0,
					});
				}
			}
		}
		montage
	}

	/// Build the video ticket params for `time` (mirrors the C++
	/// `start_video_ticket` param marshalling).
	fn build_video_ticket(&self, time: Rational) -> Result<VideoTicketParams> {
		let (project, viewer_id) = &self.viewer;
		// Inspect the viewer node WITHOUT holding the project lock across
		// the montage build below: `video_montage` locks the same project,
		// and `std::sync::Mutex` is not reentrant — holding it here
		// self-deadlocks every sequence export on the driving thread
		// (reproduced by the facade's `it_export` suite). The borrow
		// (`footage.filename`) is cloned before the lock drops.
		let footage = {
			let guard = project.lock().unwrap_or_else(|e| e.into_inner());
			let Some(entry) = guard.graph.get(*viewer_id) else {
				return Err(Error::Failed(
					"No node connected to the viewer output".to_string(),
				));
			};
			let any = entry.behavior.as_any();
			if let Some(footage) = any.and_then(|a| a.downcast_ref::<FootageBehavior>()) {
				Some(footage.filename.clone())
			} else if any
				.and_then(|a| a.downcast_ref::<SequenceBehavior>())
				.is_some()
			{
				None
			} else {
				return Err(Error::Failed(
					"No node connected to the viewer output".to_string(),
				));
			}
		};
		match footage {
			Some(filename) => Ok(VideoTicketParams {
				viewer: viewer_id.identity(),
				time,
				force_size: self.force_size(),
				force_format: self.force_format(),
				cache: Some(viewer_id.identity()),
				cache_dir: None,
				cache_id: None,
				cache_timebase: None,
				footage: Some((filename, 0)),
				montage: Vec::new(),
			}),
			// Sequence viewer: the montage is resolved without the lock.
			None => {
				let montage = Self::video_montage(project, *viewer_id, time);
				Ok(VideoTicketParams {
					viewer: viewer_id.identity(),
					time,
					force_size: self.force_size(),
					force_format: self.force_format(),
					cache: None,
					cache_dir: None,
					cache_id: None,
					cache_timebase: None,
					footage: None,
					montage,
				})
			}
		}
	}

	/// Build the audio ticket params for `range` (the Rust API renders a
	/// single range; the C++ submits one ticket per audio range).
	fn build_audio_ticket(&self, range: TimeRange) -> Result<AudioTicketParams> {
		let (project, viewer_id) = &self.viewer;
		let (sample_rate, channel_layout) = crate::nodeops::sequence_audio_params(project, *viewer_id);
		let montage = Self::audio_montage(project, *viewer_id, range.in_());
		Ok(AudioTicketParams {
			viewer: viewer_id.identity(),
			range,
			sample_rate,
			channel_layout,
			montage,
		})
	}

	/// Submit one video frame ticket at `time` (mirrors the C++
	/// `start_video_ticket`). `dispatch` is the shared completion channel
	/// handed to the ticket's finished callback.
	fn submit_video_ticket(
		&self,
		arena: &TicketArena,
		time: Rational,
		dispatch: *mut RenderDispatch,
	) -> Result<TicketId> {
		let params = self.build_video_ticket(time)?;
		let id = arena.next_id();
		let dispatch_ptr = DispatchPtr(dispatch);
		arena.submit_video_with_id(id, params, Box::new(move |result| {
			push_finished(id, result, dispatch_ptr);
		}));
		Ok(id)
	}

	/// Submit one audio ticket for `range` (see
	/// [`RenderTask::submit_video_ticket`]).
	fn submit_audio_ticket(
		&self,
		arena: &TicketArena,
		range: TimeRange,
		dispatch: *mut RenderDispatch,
	) -> Result<TicketId> {
		let params = self.build_audio_ticket(range)?;
		let id = arena.next_id();
		let dispatch_ptr = DispatchPtr(dispatch);
		arena.submit_audio_with_id(id, params, Box::new(move |result| {
			push_finished(id, result, dispatch_ptr);
		}));
		Ok(id)
	}

	/// Submit a video ticket and account it as in-flight.
	///
	/// The in-flight count is bumped **before** the submit: a ticket can
	/// complete synchronously (its callback fires before the submit returns),
	/// so the callback's decrement must always see its increment. A failed
	/// submit never fires a callback, so the bump is rolled back.
	fn start_video_ticket(
		&self,
		arena: &TicketArena,
		time: Rational,
		dispatch: *mut RenderDispatch,
		in_flight: &mut Vec<TicketId>,
	) -> Result<()> {
		unsafe {
			(&*dispatch).running.fetch_add(1, Ordering::SeqCst);
		}
		match self.submit_video_ticket(arena, time, dispatch) {
			Ok(id) => {
				in_flight.push(id);
				Ok(())
			}
			Err(e) => {
				unsafe {
					(&*dispatch).running.fetch_sub(1, Ordering::SeqCst);
				}
				Err(e)
			}
		}
	}

	/// Submit the audio ticket and account it as in-flight (see
	/// [`RenderTask::start_video_ticket`] for the counting contract).
	fn start_audio_ticket(
		&self,
		arena: &TicketArena,
		range: TimeRange,
		dispatch: *mut RenderDispatch,
		in_flight: &mut Vec<TicketId>,
	) -> Result<()> {
		unsafe {
			(&*dispatch).running.fetch_add(1, Ordering::SeqCst);
		}
		match self.submit_audio_ticket(arena, range, dispatch) {
			Ok(id) => {
				in_flight.push(id);
				Ok(())
			}
			Err(e) => {
				unsafe {
					(&*dispatch).running.fetch_sub(1, Ordering::SeqCst);
				}
				Err(e)
			}
		}
	}

	/// Map a finished ticket back to its delivery slot: the audio slot for
	/// audio tickets, the matching frame slot for video tickets.
	fn classify_ticket(
		&self,
		arena: &TicketArena,
		id: TicketId,
		slot_by_key: &HashMap<(i32, i64, i64), usize>,
	) -> Option<usize> {
		let kind = arena.kind(id)?;
		if kind == TICKET_AUDIO {
			let range = arena.range(id)?;
			slot_by_key
				.get(&(
					TICKET_AUDIO,
					range.in_().numerator(),
					range.in_().denominator(),
				))
				.copied()
		} else if kind == TICKET_VIDEO {
			let time = arena.time(id)?;
			slot_by_key
				.get(&(TICKET_VIDEO, time.numerator(), time.denominator()))
				.copied()
		} else {
			None
		}
	}

	/// Drive the whole render: keep up to `max_inflight` frame tickets in
	/// flight (audio first, then frames in timestamp order), deliver each
	/// finished ticket's result to the behavior hooks in timestamp order via
	/// the reorder buffer, and stop on cancellation or a hook error —
	/// cancelling and waiting every still-running ticket so their
	/// completions fire exactly once.
	///
	/// `task` is the live driving task (cancellation/progress/error
	/// reporting); `behavior` is the concrete subclass receiving the
	/// `frame_downloaded`/`audio_downloaded` hooks.
	pub fn render(&mut self, task: &mut Task, behavior: &mut dyn RenderTaskBehavior) -> Result<()> {
		let timebase = self.timebase();

		// Compute the frame timestamps (progress denominator mirrors the
		// C++ `total_length <= 0 -> 1` guard).
		let mut frame_times: Vec<Rational> = Vec::new();
		let mut t = self.export_range.in_();
		while t < self.export_range.out() {
			frame_times.push(t);
			t = t + timebase;
		}
		self.total_frames = frame_times.len() as i64;
		let total_length = if frame_times.is_empty() {
			1.0
		} else {
			frame_times.len() as f64
		};

		// Delivery order: the audio range first (mirrors the C++ queue
		// order), then every frame in ascending timestamp order. The reorder
		// buffer delivers to the hooks in exactly this order no matter the
		// completion order.
		let mut slots: Vec<DeliverySlot> = Vec::new();
		if self.audio_enabled {
			slots.push(DeliverySlot {
				kind: TICKET_AUDIO,
				time: self.export_range.in_(),
			});
		}
		for &ft in &frame_times {
			slots.push(DeliverySlot {
				kind: TICKET_VIDEO,
				time: ft,
			});
		}
		let mut slot_by_key: HashMap<(i32, i64, i64), usize> = HashMap::with_capacity(slots.len());
		for (i, slot) in slots.iter().enumerate() {
			slot_by_key.insert(
				(slot.kind, slot.time.numerator(), slot.time.denominator()),
				i,
			);
		}
		let total_slots = slots.len();

		// The ticket arena: the process-wide manager arena when the
		// manager is initialized, otherwise a private worker pool + arena
		// owned by this run (keeps headless/test runs self-contained).
		let (arena, mut private_pool) = match oakrender::manager::RenderManager::global() {
			Some(manager) => (manager.tickets.clone(), None),
			None => {
				let mut pool = WorkerPool::new(0);
				pool.start();
				let producer: oakrender::ticket::Producer = Arc::new(|time, params| {
					oakrender::eval::render_produced_frame(time, params)
						.map(TicketPayload::Video)
				});
				let arena = Arc::new(TicketArena::new(Arc::new(pool.clone()), producer));
				(arena, Some(pool))
			}
		};

		// The shared completion channel. Heap-allocated: the ticket
		// completion callback reaches it through its captured pointer.
		// Freed only once every in-flight ticket has fired its callback
		// (`running == 0`).
		let dispatch = Box::into_raw(Box::new(RenderDispatch::new()));
		// Shared view of the same state for the render thread (the raw
		// pointer stays valid until the box is freed below).
		let dispatch_ref = unsafe { &*dispatch };
		let window = self.max_inflight.max(1);

		// Tickets we hold (the submitter's ids); `consumed` counts tickets
		// whose finished queue copy has been popped, so
		// `in_flight.len() - consumed` is the live window.
		let mut in_flight: Vec<TicketId> = Vec::new();
		let mut consumed = 0usize;
		// Next frame timestamp to submit and next delivery slot.
		let mut next_frame_index = 0usize;
		let mut next_slot = 0usize;
		// Reorder buffer: finished tickets not yet deliverable.
		let mut pending: HashMap<usize, (TicketId, TicketResult)> = HashMap::new();
		let mut progress_counter = 0.0;
		let mut result: Result<()> = Ok(());

		// Queue audio first (mirrors the C++ order).
		if self.audio_enabled && result.is_ok() {
			if let Err(e) =
				self.start_audio_ticket(&arena, self.export_range, dispatch, &mut in_flight)
			{
				result = Err(e);
			}
		}

		// Start the initial frame window.
		if result.is_ok() {
			while in_flight.len() < window && next_frame_index < frame_times.len() {
				if let Err(e) = self.start_video_ticket(
					&arena,
					frame_times[next_frame_index],
					dispatch,
					&mut in_flight,
				) {
					result = Err(e);
					break;
				}
				next_frame_index += 1;
			}
		}

		while result.is_ok() && !task.is_cancelled() && next_slot < total_slots {
			// Drain the completion queue into the reorder buffer.
			while let Some((id, ticket_result)) = dispatch_ref.pop_finished() {
				consumed += 1;
				match self.classify_ticket(&arena, id, &slot_by_key) {
					Some(slot_index) => {
						pending.insert(slot_index, (id, ticket_result));
					}
					None => {
						result = Err(Error::Failed(
							"Render ticket reported an unexpected timestamp".to_string(),
						));
						break;
					}
				}
			}
			if result.is_err() {
				break;
			}

			// Deliver contiguous results in the observable (timestamp) order.
			while let Some((_id, ticket_result)) = pending.remove(&next_slot) {
				let slot = &slots[next_slot];
				if slot.kind == TICKET_AUDIO {
					match ticket_result {
						Ok(TicketPayload::Audio(samples)) => {
							if let Err(e) = behavior.audio_downloaded(task, &samples) {
								result = Err(e);
								break;
							}
						}
						Ok(_) => {
							result = Err(Error::Failed(
								"Audio render ticket delivered a non-audio payload".to_string(),
							));
							break;
						}
						Err(e) => {
							result = Err(Error::Failed(format!(
								"Audio render ticket failed: {e:?}"
							)));
							break;
						}
					}
				} else {
					match ticket_result {
						Ok(TicketPayload::Video(texture)) => {
							if let Err(e) = behavior.frame_downloaded(task, &texture) {
								result = Err(e);
								break;
							}
							if self.native_progress_signalling {
								progress_counter += 1.0;
								task.emit_progress(progress_counter / total_length);
							}
						}
						Ok(_) => {
							result = Err(Error::Failed(
								"Video render ticket delivered a non-video payload".to_string(),
							));
							break;
						}
						Err(e) => {
							result = Err(Error::Failed(format!(
								"Frame render ticket failed: {e:?}"
							)));
							break;
						}
					}
				}
				next_slot += 1;
			}
			if result.is_err() || task.is_cancelled() {
				break;
			}

			// Refill the window: one new ticket per delivered frame.
			while in_flight.len().saturating_sub(consumed) < window
				&& next_frame_index < frame_times.len()
			{
				if let Err(e) = self.start_video_ticket(
					&arena,
					frame_times[next_frame_index],
					dispatch,
					&mut in_flight,
				) {
					result = Err(e);
					break;
				}
				next_frame_index += 1;
			}
			if result.is_err() {
				break;
			}
			if next_slot >= total_slots {
				break;
			}

			// Wait for the next completion (a cancellation or a hook error
			// aborts the wait; in-flight tickets then finish, waking us).
			let mut guard = dispatch_ref
				.finished
				.lock()
				.unwrap_or_else(|e| e.into_inner());
			while guard.is_empty()
				&& dispatch_ref.running.load(Ordering::SeqCst) > 0
				&& !task.is_cancelled()
			{
				guard = dispatch_ref
					.cv
					.wait(guard)
					.unwrap_or_else(|e| e.into_inner());
			}
			drop(guard);
			if dispatch_ref
				.finished
				.lock()
				.unwrap_or_else(|e| e.into_inner())
				.is_empty()
				&& dispatch_ref.running.load(Ordering::SeqCst) == 0
			{
				// Every ticket finished and its queue copy was consumed.
				break;
			}
		}

		// Cancellation that aborted the loop before every slot was delivered
		// maps to the cancellation error (the sync loop returned
		// `Error::Cancelled` from its between-frames check; a render that
		// finished every frame despite a late cancel stays successful).
		if result.is_ok() && task.is_cancelled() && next_slot < total_slots {
			result = Err(Error::Cancelled);
		}

		// Tear down. On cancellation or error, cancel and wait every ticket
		// still in flight (the C++ abort path), so their completions still
		// fire exactly once. Then wait until every callback has returned
		// and free the completion channel; a private pool is shut down.
		if result.is_err() || task.is_cancelled() {
			for &id in &in_flight {
				arena.cancel(id);
				let _ = arena.wait(id);
			}
		}
		dispatch_ref.wait_idle();
		unsafe {
			drop(Box::from_raw(dispatch));
		}
		if let Some(mut pool) = private_pool.take() {
			pool.shutdown();
		}

		result
	}

	/// A detached render state used while a concrete task temporarily moves
	/// its render out of itself to drive it with itself as the behavior
	/// (avoids a self-referential borrow). Never used for actual rendering.
	pub(crate) fn placeholder() -> RenderTask {
		RenderTask::new(
			Task::new("", None),
			None,
			(
				oaknode::project::Project::new(),
				oaknode::id::NodeId::INVALID,
			),
			ForceParams::default(),
			None,
		)
	}
}

/// One deliverable unit of the render: an audio range or a frame time, in
/// the observable delivery order (audio first, then frames in time order).
struct DeliverySlot {
	/// `TICKET_AUDIO` or `TICKET_VIDEO`.
	kind: i32,
	/// Frame time (video) or range start (audio).
	time: Rational,
}

/// Shared state between the render thread and the ticket-finished
/// callbacks (the ticket's async return channel, mirroring the C++
/// `finished_mutex_`/`finished_tickets_`/`finished_wait_cond_`). One
/// instance per [`RenderTask::render`] run, reached from the completion
/// callback through its captured pointer; heap-allocated for the run and
/// freed only after every in-flight ticket has fired (`running == 0`).
struct RenderDispatch {
	/// Finished tickets in completion order (callbacks push here).
	finished: Mutex<VecDeque<(TicketId, TicketResult)>>,
	/// Tickets whose completion callback has not fired yet.
	running: AtomicUsize,
	/// Wakes the render thread when a ticket finishes.
	cv: Condvar,
}

impl RenderDispatch {
	fn new() -> RenderDispatch {
		RenderDispatch {
			finished: Mutex::new(VecDeque::new()),
			running: AtomicUsize::new(0),
			cv: Condvar::new(),
		}
	}

	/// Pop the next finished ticket; `None` when the queue is empty.
	fn pop_finished(&self) -> Option<(TicketId, TicketResult)> {
		self.finished
			.lock()
			.unwrap_or_else(|e| e.into_inner())
			.pop_front()
	}

	/// Block until every submitted ticket has fired its callback. Called
	/// before freeing `self`, so no callback can touch the state afterwards.
	fn wait_idle(&self) {
		let mut guard = self.finished.lock().unwrap_or_else(|e| e.into_inner());
		while self.running.load(Ordering::SeqCst) > 0 {
			guard = self.cv.wait(guard).unwrap_or_else(|e| e.into_inner());
		}
	}
}

/// `Send` wrapper for the raw dispatch pointer captured by the ticket
/// completion callbacks (the arena's [`oakrender::ticket::Completion`] is
/// `Send`; the pointee outlives every ticket — see [`RenderTask::render`]'s
/// `wait_idle` before freeing it).
struct DispatchPtr(*mut RenderDispatch);

// Safety: the raw pointer is exclusively owned by the render run; the
// completion callbacks only dereference it while the run is alive
// (`wait_idle` guarantees every callback returned before the free).
unsafe impl Send for DispatchPtr {}

/// Ticket-finished callback, mirroring the C++ `on_ticket_finished`. Fires
/// on the ticket's finishing thread; pushes the ticket's result into the
/// completion queue and wakes the render thread. The push precedes the
/// in-flight decrement, so a waiter never observes `running == 0` with a
/// missing queue entry.
///
/// # Safety
///
/// The wrapped pointer must point to a live `RenderDispatch` for the
/// duration of every in-flight ticket (guaranteed by
/// [`RenderTask::render`]'s `wait_idle` before freeing it).
fn push_finished(id: TicketId, result: TicketResult, dispatch: DispatchPtr) {
	let dispatch = unsafe { &*dispatch.0 };
	let mut queue = dispatch.finished.lock().unwrap_or_else(|e| e.into_inner());
	queue.push_back((id, result));
	dispatch.running.fetch_sub(1, Ordering::SeqCst);
	dispatch.cv.notify_all();
}