test(oak-task, oak-storage): task managers, codec bridge, storage

Task/manager lifecycles, precache and render boundaries, OTIO/FCPXML
round trips, and the write-through/library contract tests.

crates/oak-task/tests/codecbridge_test.rs

@@ -0,0 +1,306 @@
+// 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 codec task submitter bridge: register/unregister
+//! state handling and the synchronous `submit_codec_task` dispatch for
+//! conform (real clip success, missing media, bad output name, failed
+//! rename) and proxy (missing source) requests.
+//!
+//! The bridge collapses inner task failures into one codec error per kind
+//! (the interim synchronous contract), so the failure-path tests assert
+//! that outer variant/message here and the distinguishing reason on the
+//! task base through `conform_directly`/`proxy_directly`.
+//!
+//! The oakcodec submit registry is process-global, so the tests in this
+//! file serialize on one mutex.
+
+use std::sync::{Mutex, MutexGuard};
+
+use oak_codec::task::{submit_task, TaskKind, TaskRequest};
+use oak_task::codecbridge::{
+	is_codec_task_submitter_registered, register_codec_task_submitter,
+	unregister_codec_task_submitter,
+};
+use oak_task::conform::ConformTask;
+use oak_task::error::Error as TaskError;
+use oak_task::proxy::{ProxyParams, ProxyTask};
+use oak_task::task::{Task, TaskBehavior};
+
+/// Tests that mutate oakcodec's global submit callback must not race.
+static REG_LOCK: Mutex<()> = Mutex::new(());
+
+fn serial() -> MutexGuard<'static, ()> {
+	REG_LOCK.lock().unwrap_or_else(|e| e.into_inner())
+}
+
+/// Run a conform request directly through [`ConformTask`], returning
+/// `(result, task error message)`.
+///
+/// The bridge collapses every inner failure into one codec error (the
+/// interim synchronous contract's "conform task failed"), so the
+/// distinguishing reason is asserted here, on the task base the behavior
+/// writes — a bad output name must fail the name contract, not the probe.
+fn conform_directly(input: &str, output: &str) -> (oak_task::error::Result<()>, Option<String>) {
+	let req = request(TaskKind::Conform, input, output);
+	let mut inner = ConformTask::new(&req);
+	let mut driver = Task::new("conform", None);
+	let result = inner.run(&mut driver);
+	(result, driver.error().map(str::to_string))
+}
+
+/// Run a proxy request directly through [`ProxyTask`] (missing sources fail
+/// in the ffmpeg launch/run), returning the task error message.
+fn proxy_directly(input: &str, output: &str) -> Option<String> {
+	let req = request(TaskKind::Proxy, input, output);
+	let params = ProxyParams {
+		width: 320,
+		height: 180,
+		divider: 1,
+		version: 1,
+		crf: 23,
+		include_audio: false,
+		extension: "mp4".to_string(),
+		preset: "veryfast".to_string(),
+	};
+	let mut inner = ProxyTask::new(&req, params);
+	let mut driver = Task::new("proxy", None);
+	let _ = inner.run(&mut driver);
+	driver.error().map(str::to_string)
+}
+
+/// Assert the bridge rejected a submission with the expected `Failed`
+/// message (the codec error variant is always `Failed`; the per-path
+/// reason is asserted through [`conform_directly`]/[`proxy_directly`]).
+fn assert_bridge_failed(context: &str, result: oak_codec::error::Result<bool>) -> String {
+	match result {
+		Err(oak_codec::error::Error::Failed(message)) => message,
+		other => panic!("{context}: expected a Failed submission error, got {other:?}"),
+	}
+}
+
+fn work_dir(tag: &str) -> std::path::PathBuf {
+	let path = std::env::temp_dir().join(format!("oaktask_bridge_{tag}_{}", std::process::id()));
+	let _ = std::fs::remove_dir_all(&path);
+	std::fs::create_dir_all(&path).expect("work dir");
+	path
+}
+
+fn write_clip(dir: &std::path::Path, name: &str) -> std::path::PathBuf {
+	let path = dir.join(name);
+	oak_codec::testmedia::write_test_clip(&path, 64, 64, 10, 10).expect("test clip generation");
+	path
+}
+
+fn request<'a>(kind: TaskKind, input: &'a str, output: &'a str) -> TaskRequest<'a> {
+	TaskRequest {
+		kind,
+		input_filename: input,
+		output_filename: output,
+		stream_index: 1,
+		sample_rate: 48000,
+		channel_layout: 0x3,
+		sample_format: 4,
+		proxy_width: 0,
+		proxy_height: 0,
+	}
+}
+
+// ---- registration -----------------------------------------------------
+
+/// Given no callback, `submit_task` reports "nothing submitted"; the bridge
+/// registers exactly once and unregisters idempotently.
+#[test]
+fn submitter_registration_lifecycle() {
+	let _guard = serial();
+	unregister_codec_task_submitter().expect("clean registry");
+	assert!(!is_codec_task_submitter_registered());
+
+	let req = request(TaskKind::Conform, "in.mp4", "out.0.pcm");
+	assert!(
+		!submit_task(&req).expect("no callback is not an error"),
+		"nothing is submitted without a callback"
+	);
+
+	register_codec_task_submitter().expect("first registration");
+	assert!(is_codec_task_submitter_registered());
+	assert!(matches!(
+		register_codec_task_submitter(),
+		Err(TaskError::State)
+	));
+
+	unregister_codec_task_submitter().expect("unregister");
+	assert!(!is_codec_task_submitter_registered());
+	unregister_codec_task_submitter().expect("idempotent unregister");
+	assert!(!submit_task(&req).expect("no callback"));
+}
+
+// ---- conform ----------------------------------------------------------
+
+/// A real test clip conforms into per-channel PCM files through the bridge.
+#[test]
+fn conform_submission_writes_pcm_files() {
+	let _guard = serial();
+	unregister_codec_task_submitter().expect("clean registry");
+	register_codec_task_submitter().expect("register");
+
+	let dir = work_dir("conform_ok");
+	let clip = write_clip(&dir, "clip.mp4");
+	let out = dir.join("audio.0.pcm");
+	let clip_str = clip.to_string_lossy();
+	let out_str = out.to_string_lossy();
+	let req = request(TaskKind::Conform, &clip_str, &out_str);
+	let submitted = submit_task(&req);
+	assert!(submitted.is_ok(), "conform must be accepted: {submitted:?}");
+
+	// The final per-channel files exist (renamed from the working names).
+	for channel in 0..2 {
+		let final_name = dir.join(format!("audio.{channel}.pcm"));
+		let meta = std::fs::metadata(&final_name)
+			.unwrap_or_else(|e| panic!("{} missing: {e}", final_name.display()));
+		assert!(meta.len() > 0, "conform output is empty");
+		assert!(
+			!dir.join(format!("audio.{channel}.pcm.working")).exists(),
+			"working files are renamed away"
+		);
+	}
+
+	unregister_codec_task_submitter().expect("unregister");
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// A missing source file is rejected (no decoder can probe it); the
+/// path-specific reason is distinct from a bad-output-name failure.
+#[test]
+fn conform_submission_rejects_missing_media() {
+	let _guard = serial();
+	unregister_codec_task_submitter().expect("clean registry");
+	register_codec_task_submitter().expect("register");
+
+	let dir = work_dir("conform_missing");
+	let out = dir.join("audio.0.pcm");
+	let out_str = out.to_string_lossy();
+	let req = request(TaskKind::Conform, "/definitely/not/here.mp4", &out_str);
+	assert_eq!(
+		assert_bridge_failed("missing media", submit_task(&req)),
+		"conform task failed"
+	);
+
+	// The reason is the probe: no registered decoder accepts the input.
+	let (result, error) = conform_directly("/definitely/not/here.mp4", &out_str);
+	assert!(result.is_err());
+	assert_eq!(error.as_deref(), Some("Failed to create decoder"));
+
+	unregister_codec_task_submitter().expect("unregister");
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// An output name that does not follow the `.0.pcm` contract is rejected
+/// before any media work: the failure is the name contract, not a probe
+/// failure.
+#[test]
+fn conform_submission_rejects_a_bad_output_name() {
+	let _guard = serial();
+	unregister_codec_task_submitter().expect("clean registry");
+	register_codec_task_submitter().expect("register");
+
+	let dir = work_dir("conform_bad_name");
+	let clip = write_clip(&dir, "clip.mp4");
+	let out = dir.join("audio.pcm");
+	let clip_str = clip.to_string_lossy();
+	let out_str = out.to_string_lossy();
+	let req = request(TaskKind::Conform, &clip_str, &out_str);
+	assert_eq!(
+		assert_bridge_failed("bad output name", submit_task(&req)),
+		"conform task failed"
+	);
+
+	// Directly: the rejection names the filename contract, so a probe
+	// regression could not masquerade as this failure path.
+	let (result, error) = conform_directly(&clip_str, &out_str);
+	assert!(result.is_err());
+	assert_eq!(error.as_deref(), Some("Invalid conform output filename"));
+
+	unregister_codec_task_submitter().expect("unregister");
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// A rename failure while moving the working files into place fails the
+/// task after the audio was conformed — a third, distinct reason.
+#[test]
+fn conform_submission_reports_a_failed_rename() {
+	let _guard = serial();
+	unregister_codec_task_submitter().expect("clean registry");
+	register_codec_task_submitter().expect("register");
+
+	let dir = work_dir("conform_rename");
+	let clip = write_clip(&dir, "clip.mp4");
+	// The final first-channel path is an existing directory: the file
+	// rename onto it must fail.
+	std::fs::create_dir(dir.join("audio.0.pcm")).expect("blocking directory");
+	let out = dir.join("audio.0.pcm");
+	let clip_str = clip.to_string_lossy();
+	let out_str = out.to_string_lossy();
+	let req = request(TaskKind::Conform, &clip_str, &out_str);
+	assert_eq!(
+		assert_bridge_failed("failed rename", submit_task(&req)),
+		"conform task failed"
+	);
+
+	// Directly: the conform succeeded and only the rename failed.
+	let (result, error) = conform_directly(&clip_str, &out_str);
+	assert!(result.is_err());
+	assert_eq!(
+		error.as_deref(),
+		Some("Failed to move conformed audio into place")
+	);
+
+	unregister_codec_task_submitter().expect("unregister");
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+// ---- proxy ------------------------------------------------------------
+
+/// A proxy request with a missing source is rejected: the proxy params
+/// conversion runs, then the transcode (or the ffmpeg lookup) fails with
+/// an ffmpeg-side reason.
+#[test]
+fn proxy_submission_rejects_a_missing_source() {
+	let _guard = serial();
+	unregister_codec_task_submitter().expect("clean registry");
+	register_codec_task_submitter().expect("register");
+
+	let dir = work_dir("proxy_missing");
+	let out = dir.join("proxy.mp4");
+	let out_str = out.to_string_lossy();
+	let req = request(TaskKind::Proxy, "/definitely/not/here.mp4", &out_str);
+	assert_eq!(
+		assert_bridge_failed("missing source", submit_task(&req)),
+		"proxy task failed"
+	);
+
+	// Directly: the reason names the ffmpeg failure (missing binary or a
+	// non-zero transcode), never a conform-side message.
+	let error = proxy_directly("/definitely/not/here.mp4", &out_str);
+	assert!(
+		error
+			.as_deref()
+			.is_some_and(|message| message.contains("ffmpeg")),
+		"the proxy failure reason is ffmpeg-side: {error:?}"
+	);
+
+	unregister_codec_task_submitter().expect("unregister");
+	let _ = std::fs::remove_dir_all(&dir);
+}

crates/oak-task/tests/import_test.rs

@@ -0,0 +1,241 @@
+// 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 `ProjectImportTask` (media import): valid and
+//! invalid files, directories (recursive folder creation), duplicate
+//! paths, the produced undo command, and cancellation.
+//!
+//! Media comes from `oak_codec::testmedia`, so the tests run headless.
+
+use oak_node::folder::FolderBehavior;
+use oak_node::id::NodeId;
+use oak_node::project::Project;
+use oak_task::nodeops::{self, ProjectRef};
+use oak_task::project::import::ProjectImportTask;
+use oak_task::task::{Task, TaskBehavior};
+
+fn work_dir(tag: &str) -> std::path::PathBuf {
+	let path = std::env::temp_dir().join(format!(
+		"oaktask_import_{tag}_{}",
+		std::process::id()
+	));
+	let _ = std::fs::remove_dir_all(&path);
+	std::fs::create_dir_all(&path).expect("work dir");
+	path
+}
+
+fn write_clip(dir: &std::path::Path, name: &str) -> std::path::PathBuf {
+	let path = dir.join(name);
+	oak_codec::testmedia::write_test_clip(&path, 64, 64, 10, 10).expect("test clip generation");
+	path
+}
+
+/// A fresh project with its initialized root folder.
+fn project() -> (ProjectRef, NodeId) {
+	let project = Project::new();
+	let root = {
+		let mut p = project.lock().unwrap();
+		p.initialize().expect("root folder");
+		p.root
+	};
+	(project, root)
+}
+
+fn run_import(
+	task: &mut ProjectImportTask,
+	driver: &mut Task,
+) -> oak_task::error::Result<()> {
+	task.run(driver)
+}
+
+fn folder_children(project: &ProjectRef, folder: NodeId) -> Vec<NodeId> {
+	let guard = project.lock().unwrap_or_else(|e| e.into_inner());
+	guard
+		.graph
+		.get(folder)
+		.and_then(|e| e.behavior.as_any())
+		.and_then(|a| a.downcast_ref::<FolderBehavior>())
+		.map(|f| f.children.clone())
+		.unwrap_or_default()
+}
+
+#[test]
+fn import_mixes_valid_and_invalid_files() {
+	let dir = work_dir("mixed");
+	let media = write_clip(&dir, "clip.mp4");
+	let missing = dir.join("missing.mp4");
+
+	let (project, root) = project();
+	let mut task = ProjectImportTask::new(
+		Task::new("Import", None),
+		(project.clone(), root),
+		project.clone(),
+		vec![
+			media.to_string_lossy().into_owned(),
+			missing.to_string_lossy().into_owned(),
+		],
+		None,
+		2,
+	);
+	let mut driver = Task::new("Import", None);
+	assert!(run_import(&mut task, &mut driver).is_ok());
+
+	assert_eq!(task.file_count(), 2);
+	assert_eq!(task.get_file_count(), 1, "one footage imported");
+	assert_eq!(task.get_invalid_file_count(), 1, "one file failed");
+	assert!(task.has_invalid_files());
+	assert!(!task.get_imported_footage(0).is_err());
+	assert!(task.get_imported_footage(1).is_err(), "out of range");
+
+	let (project_ref, footage) = task.get_imported_footage(0).expect("footage");
+	assert_eq!(nodeops::footage_filename(&project_ref, footage), media.to_string_lossy());
+	assert!(nodeops::footage_is_valid(&project_ref, footage));
+
+	// The task's command adds the footage to the destination folder; undo
+	// removes it again.
+	let mut cmd = task.take_command().expect("command produced");
+	assert!(folder_children(&project, root).is_empty());
+	cmd.redo_now();
+	assert_eq!(folder_children(&project, root), vec![footage]);
+	cmd.undo_now();
+	assert!(folder_children(&project, root).is_empty());
+
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn import_duplicate_paths_create_one_footage_each() {
+	let dir = work_dir("dupes");
+	let media = write_clip(&dir, "clip.mp4");
+	let path = media.to_string_lossy().into_owned();
+
+	let (project, root) = project();
+	let mut task = ProjectImportTask::new(
+		Task::new("Import", None),
+		(project.clone(), root),
+		project.clone(),
+		vec![path.clone(), path],
+		None,
+		2,
+	);
+	let mut driver = Task::new("Import", None);
+	assert!(run_import(&mut task, &mut driver).is_ok());
+	assert_eq!(task.get_file_count(), 2, "each entry imports");
+	assert_eq!(task.get_invalid_file_count(), 0);
+
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn import_directory_creates_a_labeled_folder_recursively() {
+	let dir = work_dir("directory");
+	let sub = dir.join("Shots");
+	std::fs::create_dir_all(&sub).expect("subdir");
+	write_clip(&sub, "a.mp4");
+	write_clip(&sub, "b.mp4");
+
+	let (project, root) = project();
+	let mut task = ProjectImportTask::new(
+		Task::new("Import", None),
+		(project.clone(), root),
+		project.clone(),
+		vec![sub.to_string_lossy().into_owned()],
+		None,
+		2,
+	);
+	let mut driver = Task::new("Import", None);
+	assert!(run_import(&mut task, &mut driver).is_ok());
+	assert_eq!(task.get_file_count(), 2, "directory children imported");
+	assert_eq!(task.get_invalid_file_count(), 0);
+
+	// The directory became a labeled sub-folder holding both footages.
+	let mut cmd = task.take_command().expect("command produced");
+	cmd.redo_now();
+	let children = folder_children(&project, root);
+	assert_eq!(children.len(), 1, "one folder added");
+	let sub_folder = children[0];
+	assert_eq!(nodeops::node_label(&project, sub_folder), "Shots");
+	assert_eq!(folder_children(&project, sub_folder).len(), 2);
+
+	cmd.undo_now();
+	assert!(folder_children(&project, root).is_empty());
+
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn import_empty_directory_imports_nothing() {
+	let dir = work_dir("empty");
+	let empty = dir.join("Nothing");
+	std::fs::create_dir_all(&empty).expect("empty dir");
+
+	let (project, root) = project();
+	let mut task = ProjectImportTask::new(
+		Task::new("Import", None),
+		(project.clone(), root),
+		project.clone(),
+		vec![empty.to_string_lossy().into_owned()],
+		None,
+		0,
+	);
+	let mut driver = Task::new("Import", None);
+	assert!(run_import(&mut task, &mut driver).is_ok());
+	assert_eq!(task.get_file_count(), 0);
+	assert_eq!(task.get_invalid_file_count(), 0);
+	assert!(task.take_command().is_ok(), "an (empty) command is still produced");
+
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn import_cancel_reports_cancelled_and_yields_no_command() {
+	let dir = work_dir("cancel");
+	let media = write_clip(&dir, "clip.mp4");
+
+	let (project, root) = project();
+	let mut task = ProjectImportTask::new(
+		Task::new("Import", None),
+		(project.clone(), root),
+		project.clone(),
+		vec![media.to_string_lossy().into_owned()],
+		None,
+		1,
+	);
+	let mut driver = Task::new("Import", None);
+	driver.cancel();
+	let result = run_import(&mut task, &mut driver);
+	assert!(
+		matches!(result, Err(oak_task::error::Error::Cancelled)),
+		"a cancelled import must fail with Cancelled: {result:?}"
+	);
+	assert!(task.take_command().is_err(), "no command after cancellation");
+
+	let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn take_command_before_a_run_reports_state() {
+	let (project, root) = project();
+	let mut task = ProjectImportTask::new(
+		Task::new("Import", None),
+		(project.clone(), root),
+		project.clone(),
+		Vec::new(),
+		None,
+		0,
+	);
+	assert!(task.take_command().is_err());
+}

crates/oak-task/tests/loadotio_test.rs

@@ -0,0 +1,483 @@
+// 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 project load path:
+//!
+//! - `LoadOTIOTask` (`.otio` / `.fcpxml`) round-trips a saved project
+//!   through the real interchange files.
+//! - `ProjectLoadTask` (the XML `.oakproj` path) handles valid, missing
+//!   and corrupt documents.
+//! - Error contract: missing file, corrupt document, unsupported
+//!   extension, `take_project` before a run, and the import-confirmation
+//!   callback (accept / reject / global reset).
+//!
+//! The interchange save path is the production `SaveOTIOTask`; the media
+//! is a generated test clip, so the tests run headless with no fixtures.
+
+use std::sync::{Arc, Mutex};
+
+use oak_core::{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_task::nodeops;
+use oak_task::project::load::{ProjectLoadBaseTask, ProjectLoadTask};
+use oak_task::project::loadotio::{set_import_confirm_callback, LoadOTIOTask};
+use oak_task::project::saveotio::SaveOTIOTask;
+use oak_task::task::{Task, TaskBehavior};
+
+/// Tests that install the process-wide import-confirmation callback must
+/// not race each other.
+static CALLBACK_LOCK: Mutex<()> = Mutex::new(());
+
+fn work_path(tag: &str, ext: &str) -> std::path::PathBuf {
+	std::env::temp_dir().join(format!("oaktask_load_{tag}_{}.{ext}", std::process::id()))
+}
+
+fn write_clip(tag: &str) -> std::path::PathBuf {
+	let path = work_path(tag, "mp4");
+	oak_codec::testmedia::write_test_clip(&path, 64, 64, 10, 10).expect("test clip generation");
+	path
+}
+
+/// One video track carrying one clip of `footage` covering `[in, out)`.
+fn add_track_with_clip(
+	p: &mut Project,
+	footage: NodeId,
+	kind: TrackType,
+	in_: Rational,
+	out: Rational,
+) -> NodeId {
+	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);
+	p.graph
+		.connect(footage, 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(in_, out);
+	p.graph
+		.get_mut(track)
+		.unwrap()
+		.behavior
+		.as_any_mut()
+		.unwrap()
+		.downcast_mut::<TrackBehavior>()
+		.unwrap()
+		.append_block(clip);
+	track
+}
+
+/// A project with one "Edited" sequence (one video track with one 1s clip,
+/// one audio track with the same clip) rooted in the bin folder.
+fn build_project(media: &str) -> Arc<Mutex<Project>> {
+	let project = Project::new();
+	{
+		let mut p = project.lock().unwrap();
+		p.initialize().expect("root folder");
+
+		let footage = {
+			let mut f = FootageBehavior::new(media);
+			f.probe().expect("probe the generated clip");
+			p.graph.add_node(NodeCore::new(), Box::new(f))
+		};
+
+		let (score, sbehavior) = SequenceBehavior::create();
+		let edited = p.graph.add_node(score, sbehavior);
+		{
+			let video_track = add_track_with_clip(
+				&mut p,
+				footage,
+				TrackType::Video,
+				Rational::new(0, 1),
+				Rational::new(1, 1),
+			);
+			let audio_track = add_track_with_clip(
+				&mut p,
+				footage,
+				TrackType::Audio,
+				Rational::new(0, 1),
+				Rational::new(1, 1),
+			);
+			for (kind, tracks) in [
+				(TrackType::Video, vec![video_track]),
+				(TrackType::Audio, vec![audio_track]),
+			] {
+				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 = tracks;
+				}
+				p.graph
+					.get_mut(edited)
+					.unwrap()
+					.behavior
+					.as_any_mut()
+					.unwrap()
+					.downcast_mut::<SequenceBehavior>()
+					.unwrap()
+					.track_lists
+					.push(list);
+			}
+			p.graph.get_mut(edited).unwrap().core.label = "Edited".to_string();
+		}
+
+		let root = p.root;
+		p.graph
+			.get_mut(root)
+			.unwrap()
+			.behavior
+			.as_any_mut()
+			.unwrap()
+			.downcast_mut::<oak_node::folder::FolderBehavior>()
+			.expect("root folder")
+			.children
+			.push(edited);
+	}
+	project
+}
+
+/// Save `project` through the production interchange task.
+fn save(project: &Arc<Mutex<Project>>, filename: &std::path::Path) {
+	let mut driver = Task::new("Saving project...", None);
+	driver.set_behavior(Box::new(SaveOTIOTask {
+		base: Task::new("Saving project...", None),
+		project: project.clone(),
+		filename: filename.to_string_lossy().into_owned(),
+	}));
+	if let Err(e) = driver.start() {
+		panic!(
+			"save to {} failed: {e:?} / {}",
+			filename.display(),
+			driver.error().unwrap_or("unknown error")
+		);
+	}
+}
+
+/// Run the OTIO loader on `filename`; `(task error, stored project)`.
+fn load_otio(filename: &std::path::Path) -> (Option<String>, Option<Arc<Mutex<Project>>>) {
+	let mut task = Task::new("Loading project...", None);
+	let mut loader = LoadOTIOTask::new(ProjectLoadBaseTask::new(
+		Task::new("Loading project...", None),
+		filename.to_string_lossy().into_owned(),
+	));
+	let result = loader.run(&mut task);
+	let error = if result.is_err() {
+		Some(task.error().unwrap_or("unknown error").to_string())
+	} else {
+		None
+	};
+	(error, loader.base.take_project().ok())
+}
+
+/// The labels of every sequence in `project`.
+fn sequence_labels(project: &Arc<Mutex<Project>>) -> Vec<String> {
+	let guard = project.lock().unwrap_or_else(|e| e.into_inner());
+	let mut labels = Vec::new();
+	for id in guard.graph.node_ids() {
+		if let Some(entry) = guard.graph.get(id) {
+			if entry
+				.behavior
+				.as_any()
+				.and_then(|a| a.downcast_ref::<SequenceBehavior>())
+				.is_some()
+			{
+				labels.push(entry.core.label.clone());
+			}
+		}
+	}
+	labels
+}
+
+// ---- OTIO / FCPXML round trip -----------------------------------------
+
+#[test]
+fn otio_round_trip_preserves_the_sequence() {
+	let _guard = CALLBACK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+	let media = write_clip("otio");
+	let project = build_project(&media.to_string_lossy());
+	let out = work_path("roundtrip", "otio");
+	let _ = std::fs::remove_file(&out);
+	save(&project, &out);
+
+	let (error, loaded) = load_otio(&out);
+	assert_eq!(error, None, "the saved document must load");
+	let loaded = loaded.expect("loaded project");
+	assert_eq!(sequence_labels(&loaded), vec!["Edited".to_string()]);
+
+	// The sequence carries the saved tracks and clips.
+	let (sequence, _) = {
+		let guard = loaded.lock().unwrap();
+		let mut found = None;
+		for id in guard.graph.node_ids() {
+			if guard
+				.graph
+				.get(id)
+				.and_then(|e| e.behavior.as_any())
+				.and_then(|a| a.downcast_ref::<SequenceBehavior>())
+				.is_some()
+			{
+				found = Some(id);
+				break;
+			}
+		}
+		(found.expect("sequence"), ())
+	};
+	let video_list =
+		nodeops::sequence_track_list(&loaded, sequence, TrackType::Video).expect("video list");
+	let audio_list =
+		nodeops::sequence_track_list(&loaded, sequence, TrackType::Audio).expect("audio list");
+	assert_eq!(nodeops::tracklist_track_count(&loaded, video_list), 1);
+	assert_eq!(nodeops::tracklist_track_count(&loaded, audio_list), 1);
+
+	let track = nodeops::tracklist_track_at(&loaded, video_list, 0).expect("video track");
+	assert_eq!(nodeops::track_block_count(&loaded, track), 1);
+	let clip = nodeops::track_block_at(&loaded, track, 0).expect("clip");
+	assert_eq!(nodeops::block_in(&loaded, clip), Rational::new(0, 1));
+	assert_eq!(nodeops::block_length(&loaded, clip), Rational::new(1, 1));
+	let footage = nodeops::clip_footage(&loaded, clip).expect("clip footage link");
+	assert_eq!(
+		nodeops::footage_filename(&loaded, footage),
+		media.to_string_lossy()
+	);
+
+	let _ = std::fs::remove_file(&out);
+	let _ = std::fs::remove_file(&media);
+}
+
+#[test]
+fn fcpxml_round_trip_preserves_the_sequence() {
+	let _guard = CALLBACK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+	let media = write_clip("fcpxml");
+	let project = build_project(&media.to_string_lossy());
+	let out = work_path("roundtrip", "fcpxml");
+	let _ = std::fs::remove_file(&out);
+	save(&project, &out);
+
+	let (error, loaded) = load_otio(&out);
+	assert_eq!(error, None, "the saved FCPXML must load");
+	let loaded = loaded.expect("loaded project");
+	assert_eq!(sequence_labels(&loaded), vec!["Edited".to_string()]);
+
+	// The sequence carries the saved tracks and clips (the same structure
+	// the OTIO round trip asserts, through the FCPXML mapping).
+	let sequence = {
+		let guard = loaded.lock().unwrap();
+		let mut found = None;
+		for id in guard.graph.node_ids() {
+			if guard
+				.graph
+				.get(id)
+				.and_then(|e| e.behavior.as_any())
+				.and_then(|a| a.downcast_ref::<SequenceBehavior>())
+				.is_some()
+			{
+				found = Some(id);
+				break;
+			}
+		}
+		found.expect("sequence")
+	};
+	let video_list =
+		nodeops::sequence_track_list(&loaded, sequence, TrackType::Video).expect("video list");
+	let audio_list =
+		nodeops::sequence_track_list(&loaded, sequence, TrackType::Audio).expect("audio list");
+	assert_eq!(nodeops::tracklist_track_count(&loaded, video_list), 1);
+	assert_eq!(nodeops::tracklist_track_count(&loaded, audio_list), 1);
+
+	let track = nodeops::tracklist_track_at(&loaded, video_list, 0).expect("video track");
+	assert_eq!(nodeops::track_block_count(&loaded, track), 1);
+	let clip = nodeops::track_block_at(&loaded, track, 0).expect("clip");
+	assert_eq!(nodeops::block_in(&loaded, clip), Rational::new(0, 1));
+	assert_eq!(nodeops::block_length(&loaded, clip), Rational::new(1, 1));
+	let footage = nodeops::clip_footage(&loaded, clip).expect("clip footage link");
+	assert_eq!(
+		nodeops::footage_filename(&loaded, footage),
+		media.to_string_lossy()
+	);
+
+	let _ = std::fs::remove_file(&out);
+	let _ = std::fs::remove_file(&media);
+}
+
+// ---- error contract ---------------------------------------------------
+
+#[test]
+fn load_otio_rejects_a_missing_file() {
+	let _guard = CALLBACK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+	let missing = work_path("missing", "otio");
+	let (error, loaded) = load_otio(&missing);
+	assert!(error.is_some(), "a missing document must fail");
+	assert!(loaded.is_none(), "no project is stored on failure");
+}
+
+#[test]
+fn load_otio_rejects_a_corrupt_document() {
+	let _guard = CALLBACK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+	let path = work_path("corrupt", "otio");
+	std::fs::write(&path, b"{ this is not json").expect("write corrupt file");
+	let (error, loaded) = load_otio(&path);
+	assert!(error.is_some(), "a corrupt document must fail");
+	assert!(loaded.is_none());
+	let _ = std::fs::remove_file(&path);
+}
+
+#[test]
+fn load_otio_rejects_an_unknown_extension() {
+	let _guard = CALLBACK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+	let path = work_path("unknown", "txt");
+	std::fs::write(&path, b"{}").expect("write file");
+	let (error, loaded) = load_otio(&path);
+	let error = error.expect("unknown extension must fail");
+	assert!(
+		error.contains("Unknown project file format"),
+		"error names the supported formats: {error}"
+	);
+	assert!(loaded.is_none());
+	let _ = std::fs::remove_file(&path);
+}
+
+#[test]
+fn take_project_before_a_run_reports_state() {
+	let mut base = ProjectLoadBaseTask::new(Task::new("load", None), "x.oakproj".into());
+	assert!(base.take_project().is_err());
+}
+
+#[test]
+fn xml_loader_rejects_missing_and_corrupt_documents() {
+	let mut task = Task::new("load", None);
+	let mut load = ProjectLoadTask {
+		base: ProjectLoadBaseTask::new(Task::new("load", None), "/definitely/not/here.oakproj".into()),
+	};
+	assert!(load.run(&mut task).is_err());
+	assert!(task
+		.error()
+		.is_some_and(|e| e.contains("Failed to read file")));
+
+	let path = work_path("corrupt", "oakproj");
+	std::fs::write(&path, b"<not-oak>").expect("write corrupt xml");
+	let mut task = Task::new("load", None);
+	let mut load = ProjectLoadTask {
+		base: ProjectLoadBaseTask::new(
+			Task::new("load", None),
+			path.to_string_lossy().into_owned(),
+		),
+	};
+	assert!(load.run(&mut task).is_err());
+	assert!(task
+		.error()
+		.is_some_and(|e| e.contains("Failed to read XML document")));
+	let _ = std::fs::remove_file(&path);
+}
+
+#[test]
+fn xml_loader_round_trips_a_saved_project() {
+	let media = write_clip("oakproj");
+	let project = build_project(&media.to_string_lossy());
+	let xml = {
+		let guard = project.lock().unwrap();
+		oak_node::serializer::save(&guard).expect("serialize project")
+	};
+	let path = work_path("roundtrip", "oakproj");
+	std::fs::write(&path, xml).expect("write project");
+
+	let mut task = Task::new("load", None);
+	let mut load = ProjectLoadTask {
+		base: ProjectLoadBaseTask::new(
+			Task::new("load", None),
+			path.to_string_lossy().into_owned(),
+		),
+	};
+	assert!(load.run(&mut task).is_ok(), "error: {:?}", task.error());
+	let loaded = load.base.take_project().expect("loaded project");
+	assert_eq!(sequence_labels(&loaded), vec!["Edited".to_string()]);
+	{
+		let guard = loaded.lock().unwrap();
+		assert_eq!(guard.filename(), path.to_string_lossy());
+	}
+
+	let _ = std::fs::remove_file(&path);
+	let _ = std::fs::remove_file(&media);
+}
+
+// ---- import-confirmation callback -------------------------------------
+
+#[test]
+fn import_confirm_callback_can_reject_and_accept() {
+	let _guard = CALLBACK_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+	let media = write_clip("confirm");
+	let project = build_project(&media.to_string_lossy());
+	let out = work_path("confirm", "otio");
+	let _ = std::fs::remove_file(&out);
+	save(&project, &out);
+
+	// Rejecting: the task completes but hands no project to the caller.
+	let rejected: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+	let seen = rejected.clone();
+	set_import_confirm_callback(Some(Box::new(move |names| {
+		*seen.lock().unwrap() = names.to_vec();
+		false
+	})));
+	let (error, loaded) = load_otio(&out);
+	assert_eq!(error, None, "a user rejection is not a task error");
+	assert!(loaded.is_none(), "the rejected project is disposed of");
+	assert_eq!(*rejected.lock().unwrap(), vec!["Edited".to_string()]);
+
+	// Accepting: the same document loads normally.
+	set_import_confirm_callback(Some(Box::new(|_| true)));
+	let (error, loaded) = load_otio(&out);
+	assert_eq!(error, None);
+	assert!(loaded.is_some());
+
+	// Reset the global callback so other tests see the headless default.
+	set_import_confirm_callback(None);
+
+	let _ = std::fs::remove_file(&out);
+	let _ = std::fs::remove_file(&media);
+}

crates/oak-task/tests/manager_test.rs

@@ -0,0 +1,436 @@
+// 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 `TaskManager` singleton: init/shutdown state
+//! machine, task ownership (add/cancel/drain/delete), not-found cases,
+//! pointer lookups (find/take/cancel by address) and worker joins.
+//!
+//! The manager is process-wide, so every test serializes on one static
+//! mutex and starts from a shut-down singleton. The worker behaviors use
+//! atomics and `yield_now` spins — no sleeps.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard};
+
+use oak_task::error::Error;
+use oak_task::manager::TaskManager;
+use oak_task::task::{Task, TaskBehavior};
+
+/// Tests that touch the process-wide manager must not race each other.
+static MANAGER_LOCK: Mutex<()> = Mutex::new(());
+
+fn serial() -> MutexGuard<'static, ()> {
+	MANAGER_LOCK.lock().unwrap_or_else(|e| e.into_inner())
+}
+
+/// Shut down any leftover singleton, then create a fresh one.
+fn init_fresh() {
+	TaskManager::shutdown();
+	assert!(TaskManager::instance().is_none());
+	TaskManager::init().expect("manager init");
+}
+
+/// Bounded spin (no sleeps) until `flag` is set.
+fn spin_until(flag: &AtomicBool) {
+	for _ in 0..4_000_000 {
+		if flag.load(Ordering::SeqCst) {
+			return;
+		}
+		std::thread::yield_now();
+	}
+	panic!("flag was not set in time");
+}
+
+/// Marks that the behavior entered `run`, then returns success. The flag
+/// also makes `Task::wait_finished` race-free (it no-ops on a task whose
+/// worker has not reached `start` yet).
+struct Signal {
+	entered: Arc<AtomicBool>,
+}
+
+impl TaskBehavior for Signal {
+	fn run(&mut self, _task: &mut Task) -> Result<(), oak_task::error::Error> {
+		self.entered.store(true, Ordering::SeqCst);
+		Ok(())
+	}
+}
+
+/// Parks the worker until cancellation, recording both transitions.
+struct Park {
+	started: Arc<AtomicBool>,
+	cancelled: Arc<AtomicBool>,
+}
+
+impl TaskBehavior for Park {
+	fn run(&mut self, task: &mut Task) -> Result<(), oak_task::error::Error> {
+		self.started.store(true, Ordering::SeqCst);
+		while !task.is_cancelled() {
+			std::thread::yield_now();
+		}
+		self.cancelled.store(true, Ordering::SeqCst);
+		Err(oak_task::task::cancelled())
+	}
+}
+
+/// Parks until an external release flag is set, then succeeds.
+struct Gate {
+	started: Arc<AtomicBool>,
+	release: Arc<AtomicBool>,
+}
+
+impl TaskBehavior for Gate {
+	fn run(&mut self, _task: &mut Task) -> Result<(), oak_task::error::Error> {
+		self.started.store(true, Ordering::SeqCst);
+		while !self.release.load(Ordering::SeqCst) {
+			std::thread::yield_now();
+		}
+		Ok(())
+	}
+}
+
+/// Build a `Park` behavior task plus its observation flags.
+fn park_task(tag: &str) -> (Task, Arc<AtomicBool>, Arc<AtomicBool>) {
+	let started = Arc::new(AtomicBool::new(false));
+	let cancelled = Arc::new(AtomicBool::new(false));
+	let mut task = Task::new(tag, None);
+	task.set_behavior(Box::new(Park {
+		started: started.clone(),
+		cancelled: cancelled.clone(),
+	}));
+	(task, started, cancelled)
+}
+
+/// Add `task` and return its raw pointer after waiting for the worker to
+/// start running.
+fn add_and_wait(task: Task, started: &AtomicBool) -> *mut Task {
+	TaskManager::with_manager_mut(|m| m.add_task(Box::new(task))).expect("manager present");
+	spin_until(started);
+	TaskManager::with_manager(|m| m.task_ptr_at(0))
+		.expect("manager present")
+		.expect("task 0")
+}
+
+fn task_ptr(index: usize) -> *mut Task {
+	TaskManager::with_manager(|m| m.task_ptr_at(index))
+		.expect("manager present")
+		.unwrap_or_else(|e| panic!("task {index} missing: {e:?}"))
+}
+
+// ---- singleton state --------------------------------------------------
+
+/// Given a fresh singleton, `init` rejects duplicates and the accessors see
+/// the manager; `shutdown` is idempotent and clears the instance.
+#[test]
+fn init_is_a_singleton_and_shutdown_is_idempotent() {
+	let _guard = serial();
+	init_fresh();
+
+	assert!(TaskManager::instance().is_some());
+	assert!(matches!(TaskManager::init(), Err(Error::State)));
+
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(0));
+	assert_eq!(
+		TaskManager::with_manager_mut(|m| {
+			m.set_codec_submitter_registered(true);
+			m.codec_submitter_registered()
+		}),
+		Some(true)
+	);
+
+	TaskManager::shutdown();
+	assert!(TaskManager::instance().is_none());
+	assert!(TaskManager::with_manager(|m| m.get_task_count()).is_none());
+	assert!(TaskManager::with_manager_mut(|m| m.get_task_count()).is_none());
+
+	TaskManager::shutdown(); // idempotent
+	assert!(TaskManager::instance().is_none());
+}
+
+/// `init` does not register the codec task submitter: the manager flag
+/// starts `false` while the oakcodec registry stays untouched, and the
+/// flag is pure bookkeeping — flipping it installs nothing.
+#[test]
+fn init_does_not_register_the_codec_submitter() {
+	let _guard = serial();
+	init_fresh();
+
+	assert!(
+		!oak_task::codecbridge::is_codec_task_submitter_registered(),
+		"init() must not install the codec submitter"
+	);
+	assert_eq!(
+		TaskManager::with_manager(|m| m.codec_submitter_registered()),
+		Some(false),
+		"a fresh manager has not registered anything"
+	);
+
+	TaskManager::with_manager_mut(|m| m.set_codec_submitter_registered(true)).unwrap();
+	assert!(
+		!oak_task::codecbridge::is_codec_task_submitter_registered(),
+		"the bookkeeping flag never installs the oakcodec callback"
+	);
+	assert_eq!(
+		TaskManager::with_manager(|m| m.codec_submitter_registered()),
+		Some(true)
+	);
+	TaskManager::with_manager_mut(|m| m.set_codec_submitter_registered(false)).unwrap();
+	assert_eq!(
+		TaskManager::with_manager(|m| m.codec_submitter_registered()),
+		Some(false)
+	);
+}
+
+// ---- add / finish / drain --------------------------------------------
+
+/// Given a task handed to the manager, the worker runs it; it can be found
+/// by address, waited on, drained (with its join handle) and removed.
+#[test]
+fn add_task_runs_and_drain_finished_returns_the_pair() {
+	let _guard = serial();
+	init_fresh();
+
+	let entered = Arc::new(AtomicBool::new(false));
+	let mut task = Task::new("signal", None);
+	task.set_behavior(Box::new(Signal {
+		entered: entered.clone(),
+	}));
+	TaskManager::with_manager_mut(|m| m.add_task(Box::new(task))).expect("manager present");
+	spin_until(&entered);
+	let ptr = task_ptr(0);
+
+	// Deterministic completion wait through the task itself.
+	unsafe { (&*ptr).wait_finished() };
+	assert!(unsafe { (&*ptr).is_finished() });
+	assert!(unsafe { (&*ptr).succeeded() });
+
+	// The list still owns the task until drained.
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(1));
+	assert_eq!(
+		TaskManager::with_manager(|m| m.find_index(ptr)),
+		Some(Some(0))
+	);
+
+	// An unrelated task is not in the list.
+	let unrelated = Task::new("unrelated", None);
+	let unrelated_ptr: *const Task = &unrelated;
+	assert_eq!(
+		TaskManager::with_manager(|m| m.find_index(unrelated_ptr)),
+		Some(None)
+	);
+
+	let drained = TaskManager::with_manager_mut(|m| m.drain_finished()).expect("manager present");
+	assert_eq!(drained.len(), 1, "the finished task and its thread");
+	for (_task, handle) in drained {
+		handle.join().expect("worker joined");
+	}
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(0));
+	assert!(matches!(
+		TaskManager::with_manager(|m| m.task_ptr_at(0)),
+		Some(Err(Error::NotFound))
+	));
+}
+
+/// Ending a still-running task through `cancel_task` reports NotFound for
+/// out-of-range indices; `cancel_task_and_wait` stops and joins the worker.
+#[test]
+fn cancel_task_and_wait_stops_a_running_task() {
+	let _guard = serial();
+	init_fresh();
+
+	let (task, started, cancelled) = park_task("park");
+	let ptr = add_and_wait(task, &started);
+
+	assert!(matches!(
+		TaskManager::with_manager_mut(|m| m.cancel_task(7)),
+		Some(Err(Error::NotFound))
+	));
+	assert!(matches!(
+		TaskManager::with_manager(|m| m.task_ptr_at(7)),
+		Some(Err(Error::NotFound))
+	));
+
+	TaskManager::with_manager_mut(|m| m.cancel_task_and_wait(0))
+		.expect("manager present")
+		.expect("cancel index 0");
+	assert!(cancelled.load(Ordering::SeqCst));
+	assert!(unsafe { (&*ptr).is_finished() });
+	assert!(!unsafe { (&*ptr).succeeded() });
+
+	// The handle was already joined by cancel_task_and_wait, so
+	// delete_finished silently removes the finished task.
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(1));
+	TaskManager::with_manager_mut(|m| m.delete_finished()).expect("manager present");
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(0));
+}
+
+/// `drain_finished` skips the still-running tasks and leaves them in the
+/// list.
+#[test]
+fn drain_finished_keeps_running_tasks() {
+	let _guard = serial();
+	init_fresh();
+
+	// A finished task at index 0...
+	let entered = Arc::new(AtomicBool::new(false));
+	let mut done = Task::new("done", None);
+	done.set_behavior(Box::new(Signal {
+		entered: entered.clone(),
+	}));
+	TaskManager::with_manager_mut(|m| m.add_task(Box::new(done))).unwrap();
+	spin_until(&entered);
+	unsafe { (&*task_ptr(0)).wait_finished() };
+
+	// ...and a parked task at index 1.
+	let (parked, started, cancelled) = park_task("running");
+	TaskManager::with_manager_mut(|m| m.add_task(Box::new(parked))).unwrap();
+	spin_until(&started);
+
+	let drained = TaskManager::with_manager_mut(|m| m.drain_finished()).expect("manager present");
+	assert_eq!(drained.len(), 1, "only the finished task is drained");
+	for (_task, handle) in drained {
+		handle.join().expect("worker joined");
+	}
+	assert_eq!(
+		TaskManager::with_manager(|m| m.get_task_count()),
+		Some(1),
+		"the running task stays in the list"
+	);
+
+	TaskManager::shutdown();
+	assert!(cancelled.load(Ordering::SeqCst));
+}
+
+/// `delete_finished` also joins the handles `drain_finished` would return.
+#[test]
+fn delete_finished_removes_completed_tasks() {
+	let _guard = serial();
+	init_fresh();
+
+	for (index, tag) in ["a", "b"].into_iter().enumerate() {
+		let entered = Arc::new(AtomicBool::new(false));
+		let mut task = Task::new(tag, None);
+		task.set_behavior(Box::new(Signal {
+			entered: entered.clone(),
+		}));
+		TaskManager::with_manager_mut(|m| m.add_task(Box::new(task))).unwrap();
+		spin_until(&entered);
+		unsafe { (&*task_ptr(index)).wait_finished() };
+	}
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(2));
+
+	TaskManager::with_manager_mut(|m| m.delete_finished()).unwrap();
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(0));
+}
+
+// ---- pointer lookups --------------------------------------------------
+
+/// `cancel_task_by_ptr` cancels a known task and no-ops for an absent one;
+/// `take_thread_by_ptr` hands the handle out exactly once.
+#[test]
+fn cancel_and_take_by_pointer_handle_absent_targets() {
+	let _guard = serial();
+	init_fresh();
+
+	let (task, started, cancelled) = park_task("byptr");
+	let ptr = add_and_wait(task, &started);
+
+	// Known pointer: the worker thread moves out (once).
+	let handle = TaskManager::with_manager_mut(|m| m.take_thread_by_ptr(ptr))
+		.expect("manager present")
+		.expect("worker thread");
+	assert!(TaskManager::with_manager_mut(|m| m.take_thread_by_ptr(ptr))
+		.expect("manager present")
+		.is_none());
+
+	// Cancel through the pointer, then join lock-free.
+	TaskManager::with_manager_mut(|m| m.cancel_task_by_ptr(ptr)).expect("manager present");
+	handle.join().expect("worker joined");
+	assert!(cancelled.load(Ordering::SeqCst));
+	assert!(unsafe { (&*ptr).is_cancelled() });
+
+	// Absent pointer: both operations are no-ops.
+	let absent = Task::new("absent", None);
+	let absent_ptr: *const Task = &absent;
+	TaskManager::with_manager_mut(|m| m.cancel_task_by_ptr(absent_ptr)).expect("manager present");
+	assert!(
+		TaskManager::with_manager_mut(|m| m.take_thread_by_ptr(absent_ptr))
+			.expect("manager present")
+			.is_none()
+	);
+
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(1));
+	TaskManager::with_manager_mut(|m| m.delete_finished()).unwrap();
+	assert_eq!(TaskManager::with_manager(|m| m.get_task_count()), Some(0));
+}
+
+/// `wait_finished` blocks on a live task until its behavior returns.
+#[test]
+fn wait_finished_blocks_until_the_behavior_returns() {
+	let _guard = serial();
+	init_fresh();
+
+	let started = Arc::new(AtomicBool::new(false));
+	let release = Arc::new(AtomicBool::new(false));
+	let mut task = Task::new("gate", None);
+	task.set_behavior(Box::new(Gate {
+		started: started.clone(),
+		release: release.clone(),
+	}));
+	TaskManager::with_manager_mut(|m| m.add_task(Box::new(task))).unwrap();
+	spin_until(&started);
+	let ptr = task_ptr(0);
+
+	// The waiter records whether the task was finished when
+	// `wait_finished` returned; it must not return while the gate is shut.
+	let entered = Arc::new(AtomicBool::new(false));
+	let waiter_entered = entered.clone();
+	let ptr_addr = ptr as usize;
+	let waiter = std::thread::spawn(move || {
+		let task = unsafe { &*(ptr_addr as *const Task) };
+		waiter_entered.store(true, Ordering::SeqCst);
+		task.wait_finished();
+		task.is_finished()
+	});
+	spin_until(&entered);
+	assert!(!waiter.is_finished(), "the gate is still shut");
+
+	release.store(true, Ordering::SeqCst);
+	assert!(
+		waiter.join().expect("waiter thread"),
+		"wait_finished must not return before the task finished"
+	);
+	assert!(unsafe { (&*ptr).succeeded() });
+	TaskManager::with_manager_mut(|m| m.delete_finished()).unwrap();
+}
+
+/// Dropping the manager (via `shutdown`) cancels every running task before
+/// joining its worker.
+#[test]
+fn shutdown_cancels_running_tasks_before_joining() {
+	let _guard = serial();
+	init_fresh();
+
+	let (task, started, cancelled) = park_task("drop");
+	TaskManager::with_manager_mut(|m| m.add_task(Box::new(task))).unwrap();
+	spin_until(&started);
+
+	TaskManager::shutdown();
+	assert!(
+		cancelled.load(Ordering::SeqCst),
+		"the manager destructor cancels its tasks"
+	);
+	assert!(TaskManager::instance().is_none());
+}

crates/oak-task/tests/precache_test.rs

@@ -0,0 +1,293 @@
+// 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 `PreCacheTask`: construction (probed footage, missing
+//! media, invalid ids), a real one-frame render through the render loop, an
+//! empty range, cancellation before the run, and the no-op download hooks.
+//!
+//! Renders run on the process-wide manager's test-only inline (`Threads`)
+//! backend, serialized through one static mutex; the media is generated by
+//! `oak_codec::testmedia` (no network, no fixtures). Nothing here needs a
+//! GPU.
+
+use std::sync::{Arc, Mutex, MutexGuard};
+
+use oak_core::texture::{Frame, Texture};
+use oak_core::Rational;
+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_render::manager::{RenderBackendChoice, RenderManager};
+use oak_render::ticket::AudioSamples;
+use oak_task::nodeops::{self, ProjectRef};
+use oak_task::precache::PreCacheTask;
+use oak_task::render::RenderTaskBehavior;
+use oak_task::task::{Task, TaskBehavior, 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_precache_{tag}_{}.mp4", std::process::id()))
+}
+
+/// A project with one probed footage node and one sequence whose video
+/// timebase is 1 fps (so the 1 s test clip renders as exactly one frame).
+fn project_with_footage(media: &str) -> (ProjectRef, NodeId, NodeId) {
+	let project = Project::new();
+	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;
+	}
+
+	(project.clone(), footage_id, sequence)
+}
+
+/// Build a pre-cache task with a 64x64 forced render size (the sequence
+/// default is 1080p; the forced size keeps the test frame small).
+fn precache(project: &ProjectRef, footage: NodeId, sequence: NodeId, index: i32) -> PreCacheTask {
+	let mut task = PreCacheTask::new(
+		(project.clone(), footage),
+		index,
+		(project.clone(), sequence),
+	);
+	task.render.force_params.force_width = 64;
+	task.render.force_params.force_height = 64;
+	task
+}
+
+// ---- construction -----------------------------------------------------
+
+/// The task label carries the footage filename and the frame index; the
+/// public fields keep the inputs.
+#[test]
+fn new_labels_with_the_footage_filename_and_index() {
+	let media = clip_path("new");
+	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
+	let (project, footage, sequence) = project_with_footage(&media.to_string_lossy());
+
+	let task = PreCacheTask::new((project.clone(), footage), 3, (project.clone(), sequence));
+	assert_eq!(
+		task.render.base.title(),
+		format!("Pre-caching {}:3", media.to_string_lossy())
+	);
+	assert_eq!(task.index, 3);
+	assert!(Arc::ptr_eq(&task.footage.0, &project));
+	assert_eq!(task.footage.1, footage);
+	assert!(Arc::ptr_eq(&task.sequence.0, &project));
+	assert_eq!(task.sequence.1, sequence);
+
+	let _ = std::fs::remove_file(&media);
+}
+
+/// Missing media (an unprobed footage node) still constructs a labeled task
+/// with a zero-length range.
+#[test]
+fn new_accepts_unprobed_and_missing_media() {
+	let (project, _, sequence) = project_with_footage_listless();
+
+	let footage =
+		nodeops::footage_create(&project, Some("/definitely/not/here.mp4")).expect("footage node");
+	let mut task = PreCacheTask::new((project.clone(), footage), 0, (project.clone(), sequence));
+	assert_eq!(
+		task.render.base.title(),
+		"Pre-caching /definitely/not/here.mp4:0"
+	);
+	assert_eq!(
+		nodeops::node_length(&project, footage),
+		Rational::new(0, 1),
+		"the unprobed footage has no length"
+	);
+
+	// Running over the empty range is a no-op success.
+	let _guard = render_serial();
+	let mut driver = Task::new("precache", None);
+	assert!(task.run(&mut driver).is_ok());
+	assert_eq!(task.render.total_frames(), 0);
+	assert!(driver.error().is_none());
+}
+
+/// A project with no footage; only the sequence is exercised here.
+fn project_with_footage_listless() -> (ProjectRef, NodeId, NodeId) {
+	let project = Project::new();
+	let sequence = {
+		let mut p = project.lock().unwrap();
+		let (score, sbehavior) = SequenceBehavior::create();
+		p.graph.add_node(score, sbehavior)
+	};
+	(project, NodeId::INVALID, sequence)
+}
+
+// ---- run --------------------------------------------------------------
+
+/// A probed 1 s clip renders exactly one frame and reports no error.
+#[test]
+fn run_renders_the_footage_frame() {
+	let media = clip_path("run");
+	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
+	let (project, footage, sequence) = project_with_footage(&media.to_string_lossy());
+	let mut task = precache(&project, footage, sequence, 0);
+
+	let _guard = render_serial();
+	let mut driver = Task::new("precache", None);
+	let result = task.run(&mut driver);
+	assert!(result.is_ok(), "run failed: {:?}", driver.error());
+	assert_eq!(
+		task.render.total_frames(),
+		1,
+		"the 1 s range at 1 fps holds one frame"
+	);
+	assert!(!driver.is_cancelled());
+	assert!(driver.error().is_none());
+
+	let _ = std::fs::remove_file(&media);
+}
+
+/// A stale footage id yields an empty range: the run succeeds without
+/// submitting any frame.
+#[test]
+fn run_with_a_stale_footage_id_renders_nothing() {
+	let (project, _, sequence) = project_with_footage_listless();
+	let mut task = PreCacheTask::new(
+		(project.clone(), NodeId::INVALID),
+		0,
+		(project.clone(), sequence),
+	);
+
+	let _guard = render_serial();
+	let mut driver = Task::new("precache", None);
+	assert!(task.run(&mut driver).is_ok());
+	assert_eq!(task.render.total_frames(), 0);
+	assert_eq!(
+		nodeops::node_length(&project, NodeId::INVALID),
+		Rational::new(0, 1)
+	);
+}
+
+/// Cancelling before the run aborts the render with the cancellation error
+/// and delivers no frame.
+#[test]
+fn run_cancelled_before_start_reports_cancelled() {
+	let media = clip_path("cancel");
+	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
+	let (project, footage, sequence) = project_with_footage(&media.to_string_lossy());
+	let mut task = precache(&project, footage, sequence, 0);
+
+	let _guard = render_serial();
+	let mut driver = Task::new("precache", None);
+	driver.cancel();
+	let result = task.run(&mut driver);
+	assert!(result.is_err(), "a cancelled render fails");
+	assert!(driver.is_cancelled());
+	assert_eq!(
+		task.render.total_frames(),
+		1,
+		"the frame count is computed before the cancellation check"
+	);
+
+	let _ = std::fs::remove_file(&media);
+}
+
+/// The download hooks are no-ops: each accepts its payload and reports
+/// success without touching the driver.
+#[test]
+fn download_hooks_are_no_ops() {
+	let (project, _, sequence) = project_with_footage_listless();
+	let mut task = PreCacheTask::new(
+		(project.clone(), NodeId::INVALID),
+		0,
+		(project.clone(), sequence),
+	);
+	let mut driver = Task::new("precache", None);
+
+	let frame = Texture::Cpu(Frame::dummy());
+	assert!(task.frame_downloaded(&mut driver, &frame).is_ok());
+	let audio = AudioSamples {
+		samples: vec![0.0; 8],
+		sample_rate: 48000,
+		channel_layout: 0x3,
+		channel_count: 2,
+	};
+	assert!(task.audio_downloaded(&mut driver, &audio).is_ok());
+	assert!(task.encode_subtitle(&mut driver, "a subtitle").is_ok());
+	assert!(driver.error().is_none());
+}
+
+/// `run` plans the render over the full footage length and honors the
+/// window/progress configurators: the single frame still renders with the
+/// in-flight window pinned to one ticket, and disabled native progress
+/// keeps the driver's progress silent.
+#[test]
+fn render_plan_covers_the_whole_footage() {
+	let media = clip_path("range");
+	oak_codec::testmedia::write_test_clip(&media, 64, 64, 10, 10).expect("test clip");
+	let (project, footage, sequence) = project_with_footage(&media.to_string_lossy());
+	let mut task = precache(&project, footage, sequence, 0);
+
+	let progress = Arc::new(Mutex::new(Vec::new()));
+	let sink = progress.clone();
+	task.render.set_max_inflight(1);
+	task.render.set_native_progress_signalling(false);
+	let _guard = render_serial();
+	let mut driver = Task::new("precache", None);
+	driver.set_event_listener(Box::new(move |event| {
+		if let TaskEvent::Progress(value) = event {
+			sink.lock().unwrap().push(value);
+		}
+	}));
+	let result = task.run(&mut driver);
+	assert!(result.is_ok(), "run failed: {:?}", driver.error());
+
+	// The 10-frame / 10 fps test clip is 1 s long, i.e. one frame at the
+	// sequence's 1 fps timebase; the one-ticket window still delivers it.
+	assert_eq!(
+		nodeops::node_length(&project, footage),
+		Rational::new(1, 1),
+		"the probed footage length"
+	);
+	assert_eq!(task.render.total_frames(), 1);
+	assert!(
+		progress.lock().unwrap().is_empty(),
+		"disabled native progress signalling emits nothing"
+	);
+
+	let _ = std::fs::remove_file(&media);
+}

crates/oak-task/tests/task_test.rs

@@ -0,0 +1,324 @@
+// 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 `Task` base lifecycle: `start` with/without a
+//! behavior, cancellation before/after the run, error propagation,
+//! progress/event emission (including the one-shot listener), `reset`,
+//! `wait_finished`, shared cancel atoms and subscriber state.
+//!
+//! Everything is synchronous and deterministic (no sleeps, no fixtures).
+
+use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+
+use oak_core::cancelatom::CancelAtom;
+use oak_task::error::{Error, Result};
+use oak_task::task::{cancelled, system_time_ms, SubscriberState, Task, TaskBehavior, TaskEvent};
+
+// ---- behaviors --------------------------------------------------------
+
+/// Returns success without doing anything.
+struct Noop;
+
+impl TaskBehavior for Noop {
+	fn run(&mut self, _task: &mut Task) -> Result<()> {
+		Ok(())
+	}
+}
+
+/// Records an error on the task and fails.
+struct Fail;
+
+impl TaskBehavior for Fail {
+	fn run(&mut self, task: &mut Task) -> Result<()> {
+		task.set_error("disk full");
+		Err(Error::Failed("disk full".to_string()))
+	}
+}
+
+/// Asserts cancellation was requested before the run and returns the
+/// cancellation error.
+struct CancelledBehavior;
+
+impl TaskBehavior for CancelledBehavior {
+	fn run(&mut self, task: &mut Task) -> Result<()> {
+		assert!(
+			task.is_cancelled(),
+			"cancel must be visible to the behavior"
+		);
+		Err(cancelled())
+	}
+}
+
+/// Cancels itself mid-run and succeeds.
+struct SelfCancel;
+
+impl TaskBehavior for SelfCancel {
+	fn run(&mut self, task: &mut Task) -> Result<()> {
+		assert!(!task.is_cancelled());
+		task.cancel();
+		assert!(task.is_cancelled());
+		Ok(())
+	}
+}
+
+/// Emits out-of-range progress values so the clamping is observable.
+struct Progressor;
+
+impl TaskBehavior for Progressor {
+	fn run(&mut self, task: &mut Task) -> Result<()> {
+		task.emit_progress(0.25);
+		task.emit_progress(2.0);
+		task.emit_progress(-1.0);
+		Ok(())
+	}
+}
+
+fn task_with(behavior: impl TaskBehavior + Send + 'static) -> Task {
+	let mut task = Task::new("unit", None);
+	task.set_behavior(Box::new(behavior));
+	task
+}
+
+// ---- lifecycle --------------------------------------------------------
+
+/// Given no behavior, `start` succeeds, marks the task finished/succeeded
+/// and exposes an elapsed duration; the error and cancel state stay clean.
+#[test]
+fn start_without_a_behavior_succeeds() {
+	let mut task = Task::new("unit", None);
+	assert_eq!(task.title(), "unit");
+	assert!(!task.is_finished());
+	assert!(!task.succeeded());
+	assert!(!task.is_cancelled());
+	assert!(task.error().is_none());
+	assert!(task.elapsed().is_none(), "not started yet");
+
+	assert!(task.start().is_ok());
+	assert!(task.is_finished());
+	assert!(task.succeeded());
+	assert!(task.elapsed().is_some());
+	assert!(task.error().is_none());
+}
+
+/// Given a failing behavior, `start` returns the same error, the task is
+/// finished but not succeeded, and the recorded error survives.
+#[test]
+fn failing_behavior_marks_failed_and_records_the_error() {
+	let mut task = task_with(Fail);
+	let result = task.start();
+	assert!(result.is_err());
+	assert!(task.is_finished());
+	assert!(!task.succeeded());
+	assert_eq!(task.error(), Some("disk full"));
+}
+
+/// Given a cancel before `start`, the behavior observes it and the task
+/// fails with the cancellation error.
+#[test]
+fn cancel_before_start_is_visible_to_the_behavior() {
+	let mut task = task_with(CancelledBehavior);
+	task.cancel();
+	assert!(task.is_cancelled());
+	let result = task.start();
+	assert!(matches!(result, Err(Error::Cancelled)));
+	assert!(task.is_finished());
+	assert!(!task.succeeded());
+}
+
+/// Given a cancel event hook, cancelling fires it exactly once per call
+/// (and the atom stays cancelled).
+#[test]
+fn cancel_fires_the_event_hook_and_is_idempotent() {
+	let fired = Arc::new(AtomicUsize::new(0));
+	let hook_flag = fired.clone();
+	let mut task = task_with(SelfCancel);
+	task.set_cancel_event(Box::new(move || {
+		hook_flag.fetch_add(1, Ordering::SeqCst);
+	}));
+
+	assert!(task.start().is_ok(), "the behavior itself succeeds");
+	assert_eq!(fired.load(Ordering::SeqCst), 1, "self-cancel fires once");
+	assert!(task.is_cancelled());
+
+	// A later cancel keeps firing the hook (the C++ cancel() contract).
+	task.cancel();
+	assert_eq!(fired.load(Ordering::SeqCst), 2);
+}
+
+/// Given a listener, the events arrive as Started, the clamped progress
+/// values, then Finished — and the listener is dropped after the run
+/// (one-shot subscription).
+#[test]
+fn events_are_ordered_clamped_and_one_shot() {
+	let events: Arc<Mutex<Vec<TaskEvent>>> = Arc::new(Mutex::new(Vec::new()));
+	let sink = events.clone();
+	let mut task = task_with(Progressor);
+	task.set_event_listener(Box::new(move |event| {
+		sink.lock().unwrap_or_else(|e| e.into_inner()).push(event);
+	}));
+
+	assert!(task.start().is_ok());
+	assert_eq!(
+		*events.lock().unwrap_or_else(|e| e.into_inner()),
+		vec![
+			TaskEvent::Started,
+			TaskEvent::Progress(0.25),
+			TaskEvent::Progress(1.0),
+			TaskEvent::Progress(0.0),
+			TaskEvent::Finished,
+		]
+	);
+
+	// One-shot: after `start` the listener is gone.
+	task.emit_progress(0.5);
+	assert_eq!(
+		events.lock().unwrap_or_else(|e| e.into_inner()).len(),
+		5,
+		"a finished task emits nothing to the dropped listener"
+	);
+}
+
+/// Given a failed run, `reset` clears every lifecycle flag and error so the
+/// task can run again.
+#[test]
+fn reset_clears_state_and_allows_a_second_run() {
+	let mut task = task_with(Fail);
+	assert!(task.start().is_err());
+	assert!(task.error().is_some());
+
+	task.reset();
+	assert!(!task.is_finished());
+	assert!(!task.succeeded());
+	assert!(task.error().is_none());
+	assert!(task.elapsed().is_none());
+
+	task.set_behavior(Box::new(Noop));
+	assert!(task.start().is_ok());
+	assert!(task.succeeded());
+}
+
+/// `wait_finished` returns immediately for a task that was never started or
+/// is already finished.
+#[test]
+fn wait_finished_returns_for_unstarted_and_finished_tasks() {
+	let never = Task::new("never", None);
+	never.wait_finished();
+	assert!(!never.is_finished());
+
+	let mut task = task_with(Noop);
+	assert!(task.start().is_ok());
+	task.wait_finished();
+	assert!(task.is_finished());
+}
+
+/// Given a subscriber state, `start` publishes the wall-clock start and
+/// 1.0/0.0 for success/failure.
+#[test]
+fn subscriber_state_tracks_start_time_and_result() {
+	let state = Arc::new(SubscriberState::default());
+	let mut task = task_with(Noop);
+	task.set_subscriber(state.clone());
+	let before = system_time_ms();
+	assert!(task.start().is_ok());
+	assert!(
+		state.start_ms.load(Ordering::SeqCst) >= before,
+		"start timestamp is published"
+	);
+	assert_eq!(state.finished_value.load(Ordering::SeqCst), 1);
+
+	let state = Arc::new(SubscriberState::default());
+	let mut failed = task_with(Fail);
+	failed.set_subscriber(state.clone());
+	assert!(failed.start().is_err());
+	assert_eq!(state.finished_value.load(Ordering::SeqCst), 0);
+}
+
+/// A shared cancel atom links an outer task and its inner base: cancelling
+/// either side is visible from the other.
+#[test]
+fn shared_cancel_atom_links_inner_and_outer_tasks() {
+	let atom = Arc::new(CancelAtom::new());
+	let mut outer = Task::new("outer", Some(atom.clone()));
+	assert!(Arc::ptr_eq(&outer.get_cancel_atom(), &atom));
+	outer.set_cancel_atom(atom.clone());
+
+	let mut inner = Task::new("inner", Some(atom));
+	inner.cancel();
+	assert!(outer.is_cancelled());
+}
+
+/// Titles and error messages round-trip through their setters.
+#[test]
+fn title_and_error_are_settable() {
+	let mut task = Task::new("old", None);
+	task.set_title("new");
+	assert_eq!(task.title(), "new");
+	assert!(task.error().is_none());
+	task.set_error("boom");
+	assert_eq!(task.error(), Some("boom"));
+}
+
+/// The module helpers: `cancelled()` is the cancellation error and
+/// `system_time_ms()` reports a plausible epoch timestamp.
+#[test]
+fn helpers_report_cancellation_and_wall_clock() {
+	assert!(matches!(cancelled(), Error::Cancelled));
+	assert!(system_time_ms() > 0);
+}
+
+/// `emit_progress` without a listener must not panic or resurrect state;
+/// a listener attached afterwards observes the clamped value (`2.0` is
+/// reported as `1.0`).
+#[test]
+fn progress_without_a_listener_is_a_no_op() {
+	let mut task = task_with(Progressor);
+	task.emit_progress(0.5);
+	assert!(!task.is_finished(), "emitting progress does not finish");
+	assert!(task.start().is_ok());
+	assert!(task.is_finished());
+
+	// The one-shot listener is gone after the run; a fresh one still
+	// receives the clamped value.
+	let events: Arc<Mutex<Vec<TaskEvent>>> = Arc::new(Mutex::new(Vec::new()));
+	let sink = events.clone();
+	task.set_event_listener(Box::new(move |event| {
+		sink.lock().unwrap_or_else(|e| e.into_inner()).push(event);
+	}));
+	task.emit_progress(2.0);
+	assert_eq!(
+		*events.lock().unwrap_or_else(|e| e.into_inner()),
+		vec![TaskEvent::Progress(1.0)],
+		"the final progress value is clamped"
+	);
+}
+
+/// A subscriber is not required for the lifecycle; setting one after a
+/// reset still receives the next run's values.
+#[test]
+fn subscriber_is_optional_and_reusable_after_reset() {
+	let mut task = task_with(Noop);
+	assert!(task.start().is_ok());
+	task.reset();
+	let state = Arc::new(SubscriberState {
+		start_ms: AtomicI64::new(0),
+		finished_value: AtomicI64::new(-1),
+	});
+	task.set_subscriber(state.clone());
+	assert!(task.start().is_ok());
+	assert!(state.start_ms.load(Ordering::SeqCst) > 0);
+	assert_eq!(state.finished_value.load(Ordering::SeqCst), 1);
+}