oak-editor/crates/oak-worker/tests/procpool_integration.rs

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

//! M15 S1 end-to-end: real oak-worker processes driven by the
//! main-process [`ProcessDispatcher`] — spawn, handshake, batched
//! renders into shared-memory slots, crash isolation with restart and
//! re-dispatch, and the zero-copy main-process guarantee.
//!
//! These tests spawn actual `oak-worker` child processes (located through
//! `CARGO_BIN_EXE_oak-worker`), so they double as the binary-resolution
//! regression gate for [`DispatcherConfig::worker_bin`].
//!
//! All tests in this file serialize on [`TEST_LOCK`]: they spawn worker
//! processes that inherit the process environment, and the crash test
//! sets crash-hook variables that must not leak into sibling runs.

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use oakcore_rs::{PixelFormat, Rational, TimeRange};
use oakrender::ipc::SLOT_FORMAT_BGRA8;
use oakrender::procpool::{
	main_heap_frame_copies, reset_main_heap_frame_copies, DispatcherConfig, ProcessDispatcher,
};
use oakrender::ticket::{
	AudioTicketParams, TicketPayload, TicketResult, VideoTicketParams,
};
use oakrender::worker::{Job, JobDispatch, JobSchedule};

/// Serialize every test in this file (shared process environment +
/// real child processes).
static TEST_LOCK: Mutex<()> = Mutex::new(());

fn lock_test() -> std::sync::MutexGuard<'static, ()> {
	TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner())
}

fn worker_bin() -> std::path::PathBuf {
	env!("CARGO_BIN_EXE_oak-worker").into()
}

fn config(workers: usize, slots: u32) -> DispatcherConfig {
	DispatcherConfig {
		worker_bin: Some(worker_bin()),
		workers,
		slots_per_worker: slots,
		width: 64,
		height: 64,
		slot_format: SLOT_FORMAT_BGRA8,
		batch_size: 4,
		graph_snapshot: None,
		handshake_timeout_ms: 30_000,
	}
}

fn params(time: Rational, footage: Option<(String, i32)>) -> Arc<VideoTicketParams> {
	Arc::new(VideoTicketParams {
		viewer: 1,
		time,
		force_size: Some((64, 64)),
		// No forced format: the dispatcher's default slot format (BGRA8)
		// applies — the preview path these tests exercise (M15 S3: an F32
		// force would now request an F32 slot instead).
		force_format: None,
		cache: None,
		cache_dir: None,
		cache_id: None,
		cache_timebase: None,
		footage,
		montage: Vec::new(),
	})
}

/// Submit `count` generated-frame tickets; returns the shared results
/// sink (completions land there from the dispatcher's poll pump).
fn submit(
	dispatcher: &ProcessDispatcher,
	results: &Arc<Mutex<Vec<TicketResult>>>,
	count: usize,
	footage: Option<(String, i32)>,
) {
	for i in 0..count {
		let results = results.clone();
		let footage = footage.clone();
		let job = Job {
			node_identity: 1,
			time: Rational::new(i as i64, 25),
			params: params(Rational::new(i as i64, 25), footage),
			audio: None,
			// Never invoked on the process backend (workers render from
			// the wire spec); must still be a valid producer.
			produce: Arc::new(|_, _| {
				Err(oakrender::error::Error::Failed(
					"process backend does not use the in-process producer".into(),
				))
			}),
			done: Box::new(move |result| {
				results.lock().unwrap_or_else(|e| e.into_inner()).push(result);
			}),
			schedule: JobSchedule::seek(),
		};
		assert!(dispatcher.post(job), "post accepted while alive");
	}
}

/// Pump the dispatcher until `expect` completions arrive or the deadline
/// passes.
fn pump_until(dispatcher: &ProcessDispatcher, results: &Mutex<Vec<TicketResult>>, expect: usize) {
	let deadline = Instant::now() + Duration::from_secs(60);
	loop {
		dispatcher.poll();
		if results.lock().unwrap_or_else(|e| e.into_inner()).len() >= expect {
			return;
		}
		if Instant::now() > deadline {
			let have = results.lock().unwrap_or_else(|e| e.into_inner()).len();
			panic!("timeout: {have}/{expect} completions");
		}
		std::thread::sleep(Duration::from_millis(5));
	}
}

/// Two real workers render two waves of generated frames into shm slots;
/// the main process never copies frame bytes. Slots are released as
/// frames arrive (the dispatcher's credit-based flow control then keeps
/// the remaining tickets flowing).
#[test]
fn two_workers_render_two_waves_zero_copy() {
	let _guard = lock_test();
	let dispatcher = ProcessDispatcher::new(config(2, 4)).expect("dispatcher config");
	dispatcher.start().expect("workers start + handshake");
	assert_eq!(dispatcher.worker_count(), 2);
	assert!(dispatcher.is_alive(0));
	assert!(dispatcher.is_alive(1));

	reset_main_heap_frame_copies();

	// Wave 1: more tickets than slots in one worker, so both workers and
	// the slot-recycling path are exercised.
	let results = Arc::new(Mutex::new(Vec::new()));
	submit(&dispatcher, &results, 12, None);

	let mut seen_worker = [false; 2];
	let mut completed = 0usize;
	let deadline = Instant::now() + Duration::from_secs(60);
	while completed < 12 {
		dispatcher.poll();
		let drained: Vec<TicketResult> =
			results.lock().unwrap_or_else(|e| e.into_inner()).drain(..).collect();
		for result in drained {
			let payload = result.expect("frame rendered");
			let TicketPayload::ShmFrame(frame) = payload else {
				panic!("process backend must deliver ShmFrame payloads");
			};
			assert!(frame.worker < 2);
			assert!(frame.slot < 4);
			assert_eq!(frame.meta.width, 64);
			assert_eq!(frame.meta.height, 64);
			assert_eq!(frame.meta.format, SLOT_FORMAT_BGRA8);
			assert_eq!(frame.meta.data_size, 64 * 64 * 4);
			assert_eq!(frame.meta.linesize, 64 * 4);
			// Generated frame (transparent black) converted to BGRA8: all
			// zero. Read through the mapping — never `slot_to_vec` (the
			// counted copy path).
			let pixels = frame.shm.slot_bytes(frame.slot);
			assert!(
				pixels[..frame.meta.data_size as usize]
					.iter()
					.all(|&b| b == 0),
				"generated frame is transparent black"
			);
			seen_worker[frame.worker as usize] = true;
			dispatcher.release_frame(&frame);
			completed += 1;
		}
		if Instant::now() > deadline {
			panic!("timeout: {completed}/12 completions");
		}
		if completed < 12 {
			std::thread::sleep(Duration::from_millis(5));
		}
	}
	// Zero copy: nothing bumped the main-process frame-copy counter.
	assert_eq!(main_heap_frame_copies(), 0);
	// Both workers participated (interleaved sharded claiming).
	assert!(seen_worker[0], "worker 0 rendered at least one frame");
	assert!(seen_worker[1], "worker 1 rendered at least one frame");

	// Wave 2 through the same recycled slots.
	submit(&dispatcher, &results, 8, None);
	pump_until(&dispatcher, &results, 8);
	for result in results.lock().unwrap().drain(..) {
		let payload = result.expect("second wave rendered");
		let TicketPayload::ShmFrame(frame) = payload else {
			panic!("ShmFrame payload");
		};
		dispatcher.release_frame(&frame);
	}
	assert_eq!(main_heap_frame_copies(), 0);

	dispatcher.shutdown();
	assert_eq!(main_heap_frame_copies(), 0);
}

/// A worker crashing mid-render (SIGSEGV hook) must not take down the
/// main process: the frame is re-queued, the worker restarted and the
/// ticket still completes with a rendered frame.
#[test]
fn crash_isolation_restarts_worker_and_frame_still_renders() {
	let _guard = lock_test();

	// One-shot crash hook: the worker dies with SIGSEGV while rendering
	// ticket 1; the marker file it leaves behind makes the restarted
	// worker render the re-queued frame for real.
	let marker = std::env::temp_dir().join(format!(
		"oak-procpool-crash-marker-{}",
		std::process::id()
	));
	let _ = std::fs::remove_file(&marker);
	std::env::set_var("OAK_WORKER_CRASH_ON_TICKET", "1");
	std::env::set_var("OAK_WORKER_CRASH_MARKER", &marker);
	struct EnvGuard;
	impl Drop for EnvGuard {
		fn drop(&mut self) {
			std::env::remove_var("OAK_WORKER_CRASH_ON_TICKET");
			std::env::remove_var("OAK_WORKER_CRASH_MARKER");
		}
	}
	let _env_guard = EnvGuard;

	let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config");
	dispatcher.start().expect("worker starts");
	assert_eq!(dispatcher.worker_count(), 1);

	let results = Arc::new(Mutex::new(Vec::new()));
	// The dispatcher's first ticket id is 1 — exactly the crash ticket.
	submit(&dispatcher, &results, 4, None);
	pump_until(&dispatcher, &results, 4);

	// The crash hit the worker (it restarted at least once)...
	assert!(
		dispatcher.restarts_of(0) >= 1,
		"crashed worker must be restarted (restarts={})",
		dispatcher.restarts_of(0)
	);
	// ...and every ticket still completed with a real frame.
	let mut crashed_ticket_seen = false;
	for result in results.lock().unwrap().drain(..) {
		let payload = result.expect("frame rendered despite the worker crash");
		let TicketPayload::ShmFrame(frame) = payload else {
			panic!("ShmFrame payload");
		};
		if frame.meta.id == 1 {
			crashed_ticket_seen = true;
			assert_eq!(frame.meta.width, 64);
			assert_eq!(frame.meta.data_size, 64 * 64 * 4);
		}
		dispatcher.release_frame(&frame);
	}
	assert!(crashed_ticket_seen, "ticket 1 delivered after the restart");
	assert!(marker.exists(), "the crash hook fired exactly once");
	let _ = std::fs::remove_file(&marker);

	dispatcher.shutdown();
}

/// Footage decode inside the worker process: a real H.264 frame from
/// `tests/demo.mp4` is decoded, scaled and converted into a BGRA8 slot.
#[test]
fn worker_decodes_real_footage_into_slot() {
	let _guard = lock_test();
	let demo = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/demo.mp4");
	assert!(demo.exists(), "repo fixture tests/demo.mp4 missing");

	let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config");
	dispatcher.start().expect("worker starts");

	let results = Arc::new(Mutex::new(Vec::new()));
	submit(
		&dispatcher,
		&results,
		2,
		Some((demo.display().to_string(), 0)),
	);
	pump_until(&dispatcher, &results, 2);

	for result in results.lock().unwrap().drain(..) {
		let payload = result.expect("footage frame rendered");
		let TicketPayload::ShmFrame(frame) = payload else {
			panic!("ShmFrame payload");
		};
		assert_eq!(frame.meta.width, 64);
		assert_eq!(frame.meta.height, 64);
		assert_eq!(frame.meta.format, SLOT_FORMAT_BGRA8);
		// Decoded video is opaque: every BGRA alpha byte is 255.
		let pixels = &frame.shm.slot_bytes(frame.slot)[..frame.meta.data_size as usize];
		let alpha_ok = pixels
			.chunks_exact(4)
			.filter(|px| px[3] == 255)
			.count();
		assert!(
			alpha_ok as f64 >= 0.99 * (64 * 64) as f64,
			"decoded frame must be opaque ({alpha_ok}/4096)"
		);
		dispatcher.release_frame(&frame);
	}

	dispatcher.shutdown();
}

/// Submit an empty-montage audio range pull through the dispatcher
/// (M15 S3): the worker mixes silence into a shm slot and the main
/// process reads it back as [`TicketPayload::ShmAudio`].
fn submit_audio(
	dispatcher: &ProcessDispatcher,
	results: &Arc<Mutex<Vec<TicketResult>>>,
	viewer: u64,
	start: Rational,
	duration: Rational,
) {
	let range = TimeRange::new(start, start + duration);
	let audio = Arc::new(AudioTicketParams {
		viewer,
		range,
		sample_rate: 48000,
		channel_layout: 0x3,
		montage: Vec::new(),
	});
	let results = results.clone();
	let job = Job {
		node_identity: viewer,
		time: start,
		params: Arc::new(VideoTicketParams {
			viewer,
			time: start,
			force_size: None,
			force_format: None,
			cache: None,
			cache_dir: None,
			cache_id: None,
			cache_timebase: None,
			footage: None,
			montage: Vec::new(),
		}),
		audio: Some(audio),
		// Never invoked on the process backend (the worker renders audio
		// from the wire spec); must still be a valid producer.
		produce: Arc::new(|_, _| {
			Err(oakrender::error::Error::Failed(
				"process backend does not use the in-process producer".into(),
			))
		}),
		done: Box::new(move |result| {
			results.lock().unwrap_or_else(|e| e.into_inner()).push(result);
		}),
		schedule: JobSchedule::seek(),
	};
	assert!(dispatcher.post(job), "post accepted while alive");
}

/// Audio tickets flow through the worker pool into shm slots (M15 S3):
/// empty-montage ranges render as interleaved f32 silence, the main
/// process reads them back with `ShmAudioRef::samples()` (zero copy —
/// the copy counter stays 0) and releases the slots.
#[test]
fn audio_tickets_roundtrip_through_shm_slots() {
	let _guard = lock_test();
	let dispatcher = ProcessDispatcher::new(config(2, 4)).expect("dispatcher config");
	dispatcher.start().expect("workers start");

	reset_main_heap_frame_copies();

	let results = Arc::new(Mutex::new(Vec::new()));
	// 1/24 s at 48 kHz stereo = 2000 sample frames x 2 ch = 16000 bytes —
	// fits the 64x64 BGRA8 slot (16384 bytes).
	for i in 0..4 {
		submit_audio(&dispatcher, &results, 1, Rational::new(i, 24), Rational::new(1, 24));
	}
	pump_until(&dispatcher, &results, 4);

	let mut seen_worker = [false; 2];
	for result in results.lock().unwrap().drain(..) {
		let payload = result.expect("audio rendered");
		let TicketPayload::ShmAudio(audio) = payload else {
			panic!("audio tickets must deliver ShmAudio payloads");
		};
		assert_eq!(audio.sample_rate, 48000);
		assert_eq!(audio.channel_count, 2);
		assert_eq!(audio.meta.format, oakrender::ipc::SLOT_FORMAT_AUDIO_F32);
		assert_eq!(audio.meta.channel_count, 2);
		assert_eq!(audio.meta.linesize, 2 * 4);
		assert_eq!(audio.meta.data_size, 2000 * 2 * 4);
		// Empty montage: total silence, parsed back as f32.
		let samples = audio.samples();
		assert_eq!(samples.len(), 2000 * 2);
		assert!(samples.iter().all(|&v| v == 0.0), "empty montage is silence");
		seen_worker[audio.worker as usize] = true;
		dispatcher.release_audio_frame(&audio);
	}
	assert!(seen_worker[0] && seen_worker[1], "both workers rendered audio");
	// Samples are read via the slot mapping and parsed into a Vec — never
	// through the counted `slot_to_vec` copy path.
	assert_eq!(main_heap_frame_copies(), 0);

	dispatcher.shutdown();
}

/// A claim batch that mixes audio and video tickets must assign slots in
/// the worker's acquisition order (the video message is processed first,
/// the audio message second). Interleaved assignment scrambled the
/// worker's free ring and flooded "slot assignment mismatch" failures
/// during playback — this is the regression guard.
#[test]
fn mixed_audio_video_batch_keeps_slot_assignment_order() {
	let _guard = lock_test();
	// One worker, four slots: the first four posts dispatch singly (post
	// pumps once itself), the remaining posts queue behind the busy
	// slots. As completions are released below, claims gather up to
	// `batch_size` queued tickets — mixed audio/video batches.
	let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config");
	dispatcher.start().expect("worker starts");

	let results = Arc::new(Mutex::new(Vec::new()));
	for i in 0..8 {
		submit(&dispatcher, &results, 1, None);
		submit_audio(
			&dispatcher,
			&results,
			1,
			Rational::new(i, 24),
			Rational::new(1, 24),
		);
	}
	// Drain completions, releasing each slot immediately so the queued
	// tickets keep flowing into new (mixed) claims.
	let deadline = Instant::now() + Duration::from_secs(60);
	let mut total = 0usize;
	loop {
		dispatcher.poll();
		let done: Vec<TicketResult> = results
			.lock()
			.unwrap_or_else(|e| e.into_inner())
			.drain(..)
			.collect();
		total += done.len();
		for result in &done {
			match result {
				Ok(TicketPayload::ShmFrame(frame)) => dispatcher.release_frame(frame),
				Ok(TicketPayload::ShmAudio(audio)) => dispatcher.release_audio_frame(audio),
				Err(e) => panic!("mixed-batch ticket failed: {e}"),
				_ => panic!("unexpected payload variant"),
			}
		}
		if total >= 16 {
			break;
		}
		if Instant::now() > deadline {
			panic!("timeout waiting for the mixed batch ({total}/16)");
		}
		std::thread::sleep(Duration::from_millis(2));
	}

	dispatcher.shutdown();
}

/// An audio render crashing mid-mix (SIGSEGV hook) must not take down the
/// main process: the audio ticket is re-queued, the worker restarted and
/// the samples still arrive.
#[test]
fn audio_crash_isolation_restarts_worker_and_audio_still_renders() {
	let _guard = lock_test();

	let marker = std::env::temp_dir().join(format!(
		"oak-procpool-audio-crash-marker-{}",
		std::process::id()
	));
	let _ = std::fs::remove_file(&marker);
	std::env::set_var("OAK_WORKER_CRASH_ON_TICKET", "1");
	std::env::set_var("OAK_WORKER_CRASH_MARKER", &marker);
	struct EnvGuard;
	impl Drop for EnvGuard {
		fn drop(&mut self) {
			std::env::remove_var("OAK_WORKER_CRASH_ON_TICKET");
			std::env::remove_var("OAK_WORKER_CRASH_MARKER");
		}
	}
	let _env_guard = EnvGuard;

	let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config");
	dispatcher.start().expect("worker starts");

	let results = Arc::new(Mutex::new(Vec::new()));
	// The dispatcher's first ticket id is 1 — exactly the crash ticket
	// (an audio ticket this time).
	for i in 0..4 {
		submit_audio(&dispatcher, &results, 1, Rational::new(i, 24), Rational::new(1, 24));
	}
	pump_until(&dispatcher, &results, 4);

	assert!(
		dispatcher.restarts_of(0) >= 1,
		"crashed audio worker must be restarted (restarts={})",
		dispatcher.restarts_of(0)
	);
	let mut crashed_ticket_seen = false;
	for result in results.lock().unwrap().drain(..) {
		let payload = result.expect("audio rendered despite the worker crash");
		let TicketPayload::ShmAudio(audio) = payload else {
			panic!("ShmAudio payload");
		};
		if audio.meta.id == 1 {
			crashed_ticket_seen = true;
			assert_eq!(audio.sample_rate, 48000);
			assert_eq!(audio.meta.data_size, 2000 * 2 * 4);
		}
		dispatcher.release_audio_frame(&audio);
	}
	assert!(crashed_ticket_seen, "ticket 1 delivered after the restart");
	assert!(marker.exists(), "the crash hook fired exactly once");
	let _ = std::fs::remove_file(&marker);

	dispatcher.shutdown();
}

/// An audio range too large for a practical shm slot (> 64 MB, i.e. a
/// multi-minute export) is refused by the dispatcher's `post`, so the
/// arena can fall back to main-process inline rendering (design §3.7)
/// instead of forcing a giant shared-memory segment.
#[test]
fn oversized_audio_ticket_is_refused_by_process_backend() {
	let _guard = lock_test();
	let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config");
	dispatcher.start().expect("worker starts");

	// ~3 min of 48 kHz stereo > 64 MB: post refuses it (false).
	let duration = Rational::new(175, 1); // 175 s
	let results = Arc::new(Mutex::new(Vec::new()));
	let results2 = results.clone();
	let audio = Arc::new(AudioTicketParams {
		viewer: 1,
		range: TimeRange::new(Rational::new(0, 1), duration),
		sample_rate: 48000,
		channel_layout: 0x3,
		montage: Vec::new(),
	});
	let job = Job {
		node_identity: 1,
		time: Rational::new(0, 1),
		params: Arc::new(VideoTicketParams {
			viewer: 1,
			time: Rational::new(0, 1),
			force_size: None,
			force_format: None,
			cache: None,
			cache_dir: None,
			cache_id: None,
			cache_timebase: None,
			footage: None,
			montage: Vec::new(),
		}),
		audio: Some(audio),
		produce: Arc::new(|_, _| {
			Err(oakrender::error::Error::Failed(
				"process backend does not use the in-process producer".into(),
			))
		}),
		done: Box::new(move |_| {
			results2.lock().unwrap_or_else(|e| e.into_inner()).push(());
		}),
		schedule: JobSchedule::seek(),
	};
	assert!(
		!dispatcher.post(job),
		"oversized audio must be refused so the arena falls back inline"
	);
	assert_eq!(results.lock().unwrap().len(), 0, "no completion fires");

	dispatcher.shutdown();
}
/// Per-ticket slot formats (M15 S3): a forced-F32 ticket gets an F32 slot
/// (the segment grows on demand) while the default BGRA8 tickets keep
/// BGRA8 slots — the export path's F32 request no longer round-trips
/// through BGRA8.
#[test]
fn f32_ticket_gets_f32_slot_and_bgra8_stays_bgra8() {
	let _guard = lock_test();
	let dispatcher = ProcessDispatcher::new(config(1, 4)).expect("dispatcher config");
	dispatcher.start().expect("worker starts");

	// 1. Default BGRA8 preview ticket first.
	let results = Arc::new(Mutex::new(Vec::new()));
	submit(&dispatcher, &results, 1, None);
	pump_until(&dispatcher, &results, 1);
	let bg = results.lock().unwrap().pop().unwrap().expect("frame rendered");
	let TicketPayload::ShmFrame(frame) = bg else {
		panic!("ShmFrame payload");
	};
	assert_eq!(frame.meta.format, SLOT_FORMAT_BGRA8);
	assert_eq!(frame.meta.data_size, 64 * 64 * 4);
	dispatcher.release_frame(&frame);

	// 2. Forced-F32 ticket: the segment grows on demand and the slot holds
	//    F32 RGBA (16 bytes per pixel).
	let results2 = Arc::new(Mutex::new(Vec::new()));
	{
		let results2 = results2.clone();
		let job = Job {
			node_identity: 1,
			time: Rational::new(0, 1),
			params: Arc::new(VideoTicketParams {
				viewer: 1,
				time: Rational::new(0, 1),
				force_size: Some((64, 64)),
				force_format: Some(PixelFormat::F32),
				cache: None,
				cache_dir: None,
				cache_id: None,
				cache_timebase: None,
				footage: None,
				montage: Vec::new(),
			}),
			audio: None,
			produce: Arc::new(|_, _| {
				Err(oakrender::error::Error::Failed(
					"process backend does not use the in-process producer".into(),
				))
			}),
			done: Box::new(move |result| {
				results2.lock().unwrap_or_else(|e| e.into_inner()).push(result);
			}),
			schedule: JobSchedule::seek(),
		};
		assert!(dispatcher.post(job), "f32 post accepted");
	}
	pump_until(&dispatcher, &results2, 1);
	let f32res = results2.lock().unwrap().pop().unwrap().expect("f32 frame rendered");
	let TicketPayload::ShmFrame(frame) = f32res else {
		panic!("ShmFrame payload");
	};
	assert_eq!(frame.meta.format, PixelFormat::F32 as i32);
	assert_eq!(frame.meta.data_size, 64 * 64 * 16);
	// The F32 bytes are zero (transparent black pipeline output).
	let pixels = frame.shm.slot_bytes(frame.slot);
	assert!(
		pixels[..frame.meta.data_size as usize].iter().all(|&b| b == 0),
		"generated F32 frame is transparent black"
	);
	dispatcher.release_frame(&frame);

	// 3. A later default BGRA8 ticket still lands as BGRA8 in the grown
	//    segment (the wire format is per ticket).
	submit(&dispatcher, &results, 1, None);
	pump_until(&dispatcher, &results, 1);
	let bg2 = results.lock().unwrap().pop().unwrap().expect("frame rendered");
	let TicketPayload::ShmFrame(frame) = bg2 else {
		panic!("ShmFrame payload");
	};
	assert_eq!(frame.meta.format, SLOT_FORMAT_BGRA8);
	assert_eq!(frame.meta.data_size, 64 * 64 * 4);
	dispatcher.release_frame(&frame);

	dispatcher.shutdown();
}