moq_video/encode/

encoder.rs

//! Video encoder front end.
//!
//! Accepts raw [`Frame`]s and delegates the actual encode to a
//! [`Backend`](super::backend::Backend). The resulting frames carry Annex-B in
//! the framing the catalog importer for [`Config::codec`] expects: H.264
//! (`moq_mux::codec::h264`) or H.265 (`moq_mux::codec::h265`).

use std::marker::PhantomData;
use std::rc::Rc;

use super::Encoded;
use super::backend::{self, Backend};
use crate::{Color, Error, Frame, Rate, Size};

/// Output video codec. `#[non_exhaustive]` so new codecs can be added without
/// breaking external `match`es.
///
/// Not every codec has a backend on every platform: H.265 is hardware-only
/// (VideoToolbox on macOS today). Building an [`Encoder`] returns
/// [`Error::NoEncoder`](crate::Error::NoEncoder) when nothing can encode the
/// requested codec on this machine.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum Codec {
	/// H.264 / AVC, Annex-B with in-band SPS/PPS (the "avc3" shape). The widest
	/// support and the default.
	#[default]
	H264,
	/// H.265 / HEVC, Annex-B with in-band VPS/SPS/PPS (the "hev1" shape).
	H265,
}

/// Which encoder implementation to use. `#[non_exhaustive]` so new selection
/// strategies can be added without breaking external `match`es.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum Kind {
	/// Prefer a platform hardware encoder, falling back to OpenH264 when the
	/// `openh264` feature is enabled.
	#[default]
	Auto,
	/// Hardware only; error if none is available.
	Hardware,
	/// Software only (OpenH264 for H.264 when its feature is enabled).
	Software,
	/// A specific backend by name, e.g. `"videotoolbox"`, `"mediacodec"`,
	/// `"nvenc"`, `"vaapi"`, `"v4l2"`, or `"openh264"`.
	Named(String),
}

/// How the encoded stream is divided into groups.
///
/// Every mode opens a group at a point a subscriber can start decoding, and
/// [`Encoder::cut`] opens one on demand in any of them. `#[non_exhaustive]`: a
/// later mode adds a variant here rather than replacing the field, so match
/// with a fallback arm.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum Gop {
	/// A keyframe (an IDR with its parameter sets in band) every `interval`
	/// frames, each opening a group. A subscriber joining mid-stream waits at
	/// most that many frames for a picture it can decode.
	Keyframe {
		/// Frames per group. Must be at least 1; [`Encoder::new`] refuses 0.
		interval: u32,
	},
}

impl Gop {
	/// Keyframes every `interval` of wall time at `framerate`, rounded down to
	/// whole frames and never below one.
	pub fn keyframe_every(interval: std::time::Duration, framerate: Rate) -> Self {
		Self::Keyframe {
			interval: framerate.frames(interval).max(1),
		}
	}

	/// Refuse a configuration no backend can honor, before one is opened.
	pub(crate) fn validate(&self) -> Result<(), Error> {
		match *self {
			Self::Keyframe { interval: 0 } => Err(Error::InvalidGop(
				"a keyframe interval of 0 frames opens no group".to_owned(),
			)),
			Self::Keyframe { .. } => Ok(()),
		}
	}
}

/// Encoder configuration. `width` / `height` / `framerate` are the encoded
/// output; input frames must already be at this resolution.
///
/// `#[non_exhaustive]`: build via [`Config::new`] and set the optional fields,
/// so future knobs don't break callers.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct Config {
	pub width: u32,
	pub height: u32,
	pub framerate: Rate,
	/// Target bitrate. `None` derives a sane default
	/// from resolution and framerate (~0.07 bits per pixel per second).
	pub bitrate: Option<moq_net::bandwidth::Rate>,
	/// How the stream is divided into groups. Defaults to a keyframe every two
	/// seconds at `framerate`.
	pub gop: Gop,
	/// Output codec. Defaults to [`Codec::H264`].
	pub codec: Codec,
	pub kind: Kind,
	/// The color space of the input frames, written into the bitstream's VUI so a
	/// decoder doesn't have to guess. `None` uses [`Color::infer`], which is both
	/// what the crate's own RGB conversions produce and what a player falls back
	/// to, so leaving it unset keeps the pixels and the label in agreement.
	///
	/// Set it only when feeding frames the crate did not convert and whose space
	/// you know from elsewhere.
	pub color: Option<Color>,
}

impl Config {
	/// A config encoding `width` x `height` at `framerate`, with the default
	/// codec, GOP, and bitrate.
	pub fn new(width: u32, height: u32, framerate: Rate) -> Self {
		Self {
			width,
			height,
			framerate,
			bitrate: None,
			gop: Gop::keyframe_every(std::time::Duration::from_secs(2), framerate),
			codec: Codec::default(),
			kind: Kind::Auto,
			color: None,
		}
	}

	/// The encoded resolution.
	pub fn size(&self) -> Size {
		Size::new(self.width, self.height)
	}

	/// The catalog rendition an encoder for this config produces.
	///
	/// Resolved by opening a throwaway encoder and encoding one frame, so the profile, level, and
	/// constraints are the ones this machine writes into the bitstream rather than a guess. The
	/// backend picks all three itself (VideoToolbox asks for High, openh264 leaves it at Baseline,
	/// VAAPI writes Main) and its choice depends on the config it opened with, so reading them back
	/// is the only way to know.
	///
	/// This is what lets a track be advertised before it carries anything: a subscriber can only
	/// discover a track the catalog names, and an on-demand encoder only encodes once a subscriber
	/// arrives. Publishing the probed rendition breaks that cycle without publishing a claim that
	/// has to be corrected later, since it already says what the first keyframe will say.
	///
	/// Costs an encoder open and a single frame, so call it once per track. Call it *before*
	/// opening the encoder you will actually use, so the two never hold a codec session at once.
	///
	/// # Errors
	///
	/// Fails when this machine cannot encode the config at all, which makes it a fail-fast check:
	/// better here than on the first frame of a track that is already advertised.
	pub async fn probe(&self) -> Result<hang::catalog::VideoConfig, Error> {
		// A `Sink` rather than an `Encoder`: this runs on whatever executor thread the caller is on,
		// and the Windows backend's COM apartment has to be opened and closed on one thread.
		let mut sink = super::Sink::open(self).await?;

		// Mid-gray, since the picture only has to make the encoder emit its parameter sets.
		let size = self.size();
		let i420 = crate::I420::new(size, vec![0x80u8; crate::I420::len(size)?])?;
		let frame = Frame::new(crate::Surface::I420(i420), moq_net::Timestamp::from_micros(0)?);

		// No cut: the first picture out of a fresh encoder is a keyframe on every backend, which
		// is what carries the parameter sets, and a backend that cannot cut still probes.
		let mut encoded = sink.encode(frame).await?;
		// A backend that pipelines holds the first frame, so drain it rather than reading nothing.
		if encoded.is_empty() {
			encoded = sink.flush().await?;
		}

		let annexb: Vec<u8> = encoded.iter().flat_map(|frame| frame.payload.iter().copied()).collect();
		let parsed = match self.codec {
			Codec::H264 => moq_mux::codec::h264::config(&annexb),
			Codec::H265 => moq_mux::codec::h265::config(&annexb),
		};
		let mut rendition = parsed.map_err(|err| {
			Error::Codec(anyhow::anyhow!(
				"{} emitted no usable parameter sets: {err}",
				sink.name()
			))
		})?;

		// Neither is in the bitstream: the target bitrate is nowhere in it, and the framerate only
		// rides in an optional VUI. Fill them from the config that produced the rest.
		rendition.bitrate.get_or_insert(self.resolved_bitrate().as_bps());
		rendition.framerate.get_or_insert(self.framerate.as_f64());
		Ok(rendition)
	}

	/// Resolved input color space: explicit override, or the size-based guess
	/// every player makes for an untagged stream. Backends write this into the
	/// VUI; the crate's RGB conversions pick the same answer for the same size,
	/// so the samples match what the bitstream claims.
	pub(crate) fn resolved_color(&self) -> Color {
		self.color.unwrap_or_else(|| Color::infer(self.size()))
	}

	/// Resolved bitrate: explicit override, or a pixels-per-second estimate.
	pub(crate) fn resolved_bitrate(&self) -> moq_net::bandwidth::Rate {
		self.bitrate
			.unwrap_or_else(|| default_bitrate(self.size(), self.framerate))
	}
}

/// The bitrate an unconfigured encode resolves to: 0.07 bits per pixel per second, which matches
/// the JS publisher's default and lands ~4.4 Mbps for 1080p30.
pub(crate) fn default_bitrate(size: Size, framerate: Rate) -> moq_net::bandwidth::Rate {
	moq_net::bandwidth::Rate::from_bps((size.pixels() as f64 * framerate.as_f64() * 0.07) as u64)
}

/// Video encoder. Build one with [`Encoder::new`], feed it raw [`Frame`]s via
/// [`encode`](Self::encode), and publish the resulting [`Encoded`] access units
/// through a [`Producer`](super::Producer) built for the same [`Codec`].
///
/// An encoder is bound to the thread that opens it. Use [`Sink`](super::Sink)
/// when the owner can move between threads.
///
/// ```compile_fail
/// fn move_to_another_thread(encoder: moq_video::encode::Encoder) {
///     std::thread::spawn(move || drop(encoder));
/// }
/// ```
pub struct Encoder {
	backend: Box<dyn Backend>,
	codec: Codec,
	size: Size,
	bitrate: moq_net::bandwidth::Rate,
	/// What the backend wrote into the bitstream's VUI, kept so a frame declaring
	/// a different space is caught rather than silently mislabeled.
	color: Color,
	/// A cut asked for by [`Encoder::cut`], applied to the next frame. Held
	/// rather than applied immediately because the caller decides a group
	/// boundary before it has the frame that opens it.
	pending_cut: bool,
	/// Keeps direct use bound to the constructing thread, regardless of backend.
	_thread_bound: PhantomData<Rc<()>>,
}

impl Encoder {
	/// Open an encoder for `config`.
	pub fn new(config: &Config) -> Result<Self, Error> {
		// I420 chroma is subsampled 2x2, so the encoded resolution must be even.
		let size = config.size();
		size.validate("encoder")?;
		size.validate_encodable("encoder", config.framerate)?;
		config.gop.validate()?;

		let backend = backend::open(config)?;
		Ok(Self {
			backend,
			codec: config.codec,
			size,
			bitrate: config.resolved_bitrate(),
			color: config.resolved_color(),
			pending_cut: false,
			_thread_bound: PhantomData,
		})
	}

	/// The encoder name in use, e.g. `"videotoolbox"`.
	pub fn name(&self) -> &str {
		self.backend.name()
	}

	/// The resolution this encoder emits, which every frame fed to it must match.
	pub fn size(&self) -> Size {
		self.size
	}

	/// The current target bitrate: what [`Config::bitrate`] resolved to at open, or
	/// the last value [`set_bitrate`](Self::set_bitrate) accepted.
	pub fn bitrate(&self) -> moq_net::bandwidth::Rate {
		self.bitrate
	}

	/// Retune the live encoder to `bitrate`, taking effect from roughly the next frame. No IDR is forced, so this is cheap enough to
	/// drive from a congestion controller: pair it with
	/// [`moq_mux::rate::Control`], which decides *when* the target
	/// is worth moving.
	///
	/// Setting the rate the encoder is already at does nothing and succeeds.
	///
	/// # Errors
	///
	/// Returns [`Error::BitrateUnsupported`] if this backend can't retune while
	/// running. That's not fatal: the encoder keeps running at its current rate,
	/// so a caller driving a control loop should stop adapting rather than stop
	/// encoding.
	pub fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), Error> {
		if bitrate == self.bitrate {
			return Ok(());
		}
		// The backends speak to codec APIs that want a plain integer, so the unit is
		// unwrapped once here rather than in each of them.
		self.backend.set_bitrate(bitrate.as_bps())?;
		// Only after the backend accepts it, so a failed set doesn't leave the
		// getter reporting a rate the encoder isn't using.
		self.bitrate = bitrate;
		Ok(())
	}

	/// The codec this encoder emits. A [`Producer`](super::Producer) must be
	/// built for the same codec to publish its packets.
	pub fn codec(&self) -> Codec {
		self.codec
	}

	/// Cut a new group at the next frame, on top of the boundaries
	/// [`Config::gop`] places on its own. Today that means encoding it as a
	/// keyframe (an IDR), so a subscriber can start decoding there.
	///
	/// Rarely needed: the encoder opens groups automatically, so reach for this
	/// only when something outside the encoder needs a boundary at a specific
	/// frame. A source group boundary (transcode mirrors the source's groups) is
	/// the usual reason; a scene change or a source switch is another.
	///
	/// The request waits for the next [`encode`](Self::encode) rather than applying
	/// at once, so it is safe to call before the frame exists. Calling it repeatedly
	/// before a frame arrives cuts once, not several times.
	///
	/// # Errors
	///
	/// Returns [`Error::CutUnsupported`] when this backend cannot force a group
	/// boundary (a V4L2 driver without the force-keyframe control), in which case
	/// nothing is queued: groups keep falling where [`Config::gop`] puts them, and
	/// the caller decides whether that layout is acceptable rather than finding
	/// out from the stream.
	pub fn cut(&mut self) -> Result<(), Error> {
		if !self.backend.can_cut() {
			return Err(Error::CutUnsupported(self.backend.name()));
		}
		self.pending_cut = true;
		Ok(())
	}

	/// Encode one raw [`Frame`], whether it came from capture, a decoder (the
	/// transcode input path), or your own pixels via
	/// [`Surface::rgba`](crate::Surface::rgba).
	///
	/// Returns zero or more encoded access units, each carrying the timestamp of the
	/// raw frame it came from: a backend that buffers hands back an earlier frame's
	/// output, so the two don't always line up.
	///
	/// A GPU surface feeds a hardware encoder on the same device directly
	/// (NVDEC -> NVENC never leaves the GPU, a `CVPixelBuffer` goes straight to
	/// VideoToolbox); anything else falls back to a CPU I420 upload. The frame must
	/// already be at the encoder's resolution: scale it with
	/// [`Frame::resize`](crate::Frame::resize), which
	/// [`decode::Config::scale_hint`](crate::decode::Config::scale_hint) lets a
	/// hardware decoder make a no-op.
	pub fn encode(&mut self, frame: &Frame) -> Result<Vec<Encoded>, Error> {
		// A transposed frame is why this compares the shape rather than a byte
		// count: 240x320 and 320x240 hold the same number of bytes.
		let size = frame.size();
		if size != self.size {
			return Err(Error::Codec(anyhow::anyhow!(
				"frame {size} does not match encoder {}",
				self.size
			)));
		}
		// The VUI is fixed when the session opens, so a frame in a different space
		// gets encoded under the wrong label. Reached by resizing across the
		// standard-definition boundary, where the pixels keep their space but the
		// encoder was sized into another one; `Config::color` is the way to pin it.
		//
		// Warn rather than reject: a live gateway transcoding a source whose VUI
		// disagrees with its resolution should keep serving a mislabeled stream
		// rather than drop it, and this is no worse than the untagged stream that
		// came before. Once, because it would otherwise fire every frame.
		if let Some(color) = frame.surface.color()
			&& color != self.color
		{
			static WARN_ONCE: std::sync::Once = std::sync::Once::new();
			WARN_ONCE.call_once(|| {
				tracing::warn!(
					frame = ?color,
					encoder = ?self.color,
					"frame color space differs from the one written into the bitstream; set encode::Config::color"
				);
			});
		}
		let encoded = self.backend.encode(frame, self.pending_cut)?;
		// Cleared only once the frame is through: a failed encode produced no
		// picture, so the request still belongs to whatever comes next rather than
		// being swallowed. The size check above returns early for the same reason.
		self.pending_cut = false;
		Ok(encoded)
	}

	/// Return every access unit the codec is still holding, leaving the encoder
	/// ready for the frames that follow. Each keeps the timestamp of the raw frame
	/// it was encoded from, so a drained tail stays in step with what came before
	/// it.
	///
	/// Reach for this at a boundary the output has to respect, which for a live
	/// broadcast is a group: a hardware codec that pipelines holds the last frames
	/// of a group past its end, and they would otherwise be published into the next
	/// group ahead of its keyframe, where a consumer joining there cannot decode
	/// them. Publishing frame-by-frame with no group structure needs none of this.
	///
	/// Not free: emptying the pipeline gives up the overlap between one frame's
	/// encode and the next frame's submission, so flush at boundaries rather than
	/// per frame.
	pub fn flush(&mut self) -> Result<Vec<Encoded>, Error> {
		self.backend.flush()
	}

	/// Flush the encoder, returning any buffered frames. Each keeps the timestamp
	/// of the raw frame it was encoded from, so a drained tail stays in step with
	/// what was published before it.
	///
	/// Consumes the encoder: nothing can be encoded after a flush, so this is the
	/// last call rather than one leaving a drained encoder in your hands.
	pub fn finish(mut self) -> Result<Vec<Encoded>, Error> {
		self.backend.finish()
	}
}

#[cfg(test)]
mod tests {
	#![cfg_attr(not(feature = "openh264"), allow(dead_code, unused_imports))]

	use super::*;

	use crate::{I420, Surface};

	/// A mid-gray RGBA buffer: encodable without a camera.
	fn gray_rgba(width: u32, height: u32) -> Vec<u8> {
		vec![0x80u8; width as usize * height as usize * 4]
	}

	/// The `index`th frame of a mid-gray 30fps stream, so a round-tripped
	/// timestamp identifies the frame it came from.
	fn gray_frame(width: u32, height: u32, index: u64) -> Frame {
		let surface = Surface::rgba(&gray_rgba(width, height), Size::new(width, height)).unwrap();
		Frame::new(surface, at(index))
	}

	/// The presentation time of frame `index` at 30fps.
	fn at(index: u64) -> moq_net::Timestamp {
		moq_net::Timestamp::from_micros(index * 33_333).unwrap()
	}

	/// The payloads of some encoded frames, for the Annex-B assertions.
	fn payloads(frames: &[Encoded]) -> Vec<bytes::Bytes> {
		frames.iter().map(|f| f.payload.clone()).collect()
	}

	#[test]
	#[cfg(feature = "openh264")]
	fn software_encoder_emits_annexb() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).expect("openh264 is vendored, always available");
		assert_eq!(encoder.name(), "openh264");

		let mut frames = Vec::new();
		for i in 0..30 {
			if i == 0 {
				encoder.cut().unwrap();
			}
			frames.extend(encoder.encode(&gray_frame(320, 240, i)).unwrap());
		}
		frames.extend(encoder.finish().unwrap());

		assert!(!frames.is_empty(), "encoder produced no packets");

		// Every encoded frame carries the timestamp of the raw frame it came from,
		// so the stream stays in step even if a backend buffers.
		let micros: Vec<u128> = frames.iter().map(|f| f.timestamp.as_micros()).collect();
		assert!(
			micros.windows(2).all(|w| w[0] < w[1]),
			"encoded timestamps not strictly increasing: {micros:?}"
		);
		assert!(
			micros.iter().all(|&t| t % 33_333 == 0 && t < 30 * 33_333),
			"encoded timestamp outside the fed set: {micros:?}"
		);

		// The first packet must start with an Annex-B start code so the avc3
		// importer can find the inline SPS/PPS.
		let packets = payloads(&frames);
		let first = &packets[0];
		let has_start_code = first.starts_with(&[0, 0, 0, 1]) || first.starts_with(&[0, 0, 1]);
		assert!(
			has_start_code,
			"first packet is not Annex-B: {:02x?}",
			&first[..first.len().min(8)]
		);
	}

	/// The bring-your-own-pixels path: RGBA in through `Surface::rgba`, Annex-B out.
	#[test]
	#[cfg(feature = "openh264")]
	fn encode_rgba_surface_emits_annexb() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		let mut frames = encoder.encode(&gray_frame(320, 240, 0)).unwrap();
		frames.extend(encoder.finish().unwrap());
		assert!(!frames.is_empty());
		let packets = payloads(&frames);
		assert!(packets[0].starts_with(&[0, 0, 0, 1]) || packets[0].starts_with(&[0, 0, 1]));
	}

	/// The same path starting from planar I420 the caller already has.
	#[test]
	#[cfg(feature = "openh264")]
	fn encode_i420_surface_emits_annexb() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		// A mid-gray I420 frame: flat 0x80 across all three planes.
		let size = Size::new(320, 240);
		let i420 = I420::new(size, vec![0x80u8; I420::len(size).unwrap()]).unwrap();
		let frame = Frame::new(Surface::I420(i420), at(0));
		let mut frames = encoder.encode(&frame).unwrap();
		frames.extend(encoder.finish().unwrap());
		assert!(!frames.is_empty());
		let packets = payloads(&frames);
		assert!(packets[0].starts_with(&[0, 0, 0, 1]) || packets[0].starts_with(&[0, 0, 1]));
	}

	/// A frame that isn't the encoder's size must error rather than encode its
	/// top-left corner.
	#[test]
	fn encode_rejects_dimension_mismatch() {
		let Ok(mut encoder) = Encoder::new(&Config::new(320, 240, crate::Rate::new(30, 1).unwrap())) else {
			return;
		};
		assert!(matches!(encoder.encode(&gray_frame(640, 480, 0)), Err(Error::Codec(_))));
	}

	/// A transposed frame is why the encoder compares the shape rather than the
	/// byte count: 240x320 and 320x240 hold the same number of bytes, so a length
	/// check alone would accept this and encode garbage.
	#[test]
	fn encode_rejects_transposed_frame() {
		let Ok(mut encoder) = Encoder::new(&Config::new(320, 240, crate::Rate::new(30, 1).unwrap())) else {
			return;
		};

		let transposed = gray_frame(240, 320, 0);
		assert_eq!(
			gray_rgba(240, 320).len(),
			gray_rgba(320, 240).len(),
			"the byte counts must collide"
		);
		assert!(matches!(encoder.encode(&transposed), Err(Error::Codec(_))));
	}

	#[test]
	fn unknown_named_encoder_errors() {
		let config = Config {
			kind: Kind::Named("definitely_not_a_codec".into()),
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		// Not `NoEncoder`: that reads "no usable video encoder found (tried: )"
		// and describes a machine with no encoder rather than a name that is not
		// one. The name asked for and the names available are both here.
		let err = Encoder::new(&config).err().expect("an unknown name cannot open");
		assert!(
			matches!(&err, Error::UnknownEncoder { name, .. } if name == "definitely_not_a_codec"),
			"unexpected error: {err:?}",
		);
	}

	/// Exercises the hand-rolled VideoToolbox backend end to end on macOS:
	/// synthetic frames through the real `VTCompressionSession`, asserting the
	/// AVCC -> Annex-B conversion produces a self-contained IDR (SPS+PPS+slice).
	#[cfg(target_os = "macos")]
	#[test]
	fn videotoolbox_emits_annexb_keyframe() {
		let config = Config {
			kind: Kind::Named("videotoolbox".into()),
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).expect("videotoolbox is available on macOS");
		assert_eq!(encoder.name(), "videotoolbox");

		let mut frames = Vec::new();
		for i in 0..10 {
			if i == 0 {
				encoder.cut().unwrap();
			}
			frames.extend(encoder.encode(&gray_frame(320, 240, i)).unwrap());
		}
		frames.extend(encoder.finish().unwrap());

		assert!(!frames.is_empty(), "encoder produced no packets");
		// VideoToolbox completes each frame before returning, so the timestamps come
		// back one per input frame, in order.
		let micros: Vec<u128> = frames.iter().map(|f| f.timestamp.as_micros()).collect();
		assert!(
			micros.windows(2).all(|w| w[0] < w[1]),
			"encoded timestamps not strictly increasing: {micros:?}"
		);

		let packets = payloads(&frames);
		let first = &packets[0];
		assert!(
			first.starts_with(&[0, 0, 0, 1]) || first.starts_with(&[0, 0, 1]),
			"first packet is not Annex-B"
		);

		// The first access unit must be a self-contained IDR: SPS (7), PPS (8),
		// IDR slice (5), all spliced in-band by the AVCC -> Annex-B conversion.
		let types = nal_types(first);
		assert!(types.contains(&7), "no SPS in first packet: {types:?}");
		assert!(types.contains(&8), "no PPS in first packet: {types:?}");
		assert!(types.contains(&5), "first packet is not an IDR: {types:?}");
	}

	/// HEVC via VideoToolbox: synthetic frames through the real
	/// `VTCompressionSession` with `kCMVideoCodecType_HEVC`, asserting the
	/// HVCC -> Annex-B conversion produces a self-contained IRAP (VPS+SPS+PPS+IDR).
	#[cfg(target_os = "macos")]
	#[test]
	fn videotoolbox_emits_annexb_keyframe_h265() {
		let config = Config {
			codec: Codec::H265,
			kind: Kind::Named("videotoolbox".into()),
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).expect("videotoolbox HEVC is available on macOS");
		assert_eq!(encoder.name(), "videotoolbox");
		assert_eq!(encoder.codec(), Codec::H265);

		let mut frames = Vec::new();
		for i in 0..10 {
			if i == 0 {
				encoder.cut().unwrap();
			}
			frames.extend(encoder.encode(&gray_frame(320, 240, i)).unwrap());
		}
		frames.extend(encoder.finish().unwrap());

		assert!(!frames.is_empty(), "encoder produced no packets");
		let packets = payloads(&frames);
		let first = &packets[0];
		assert!(
			first.starts_with(&[0, 0, 0, 1]) || first.starts_with(&[0, 0, 1]),
			"first packet is not Annex-B"
		);

		// The first access unit must be a self-contained IRAP: VPS (32), SPS (33),
		// PPS (34), and an IDR slice (16..=23), spliced in-band by the conversion.
		let types = hevc_nal_types(first);
		assert!(types.contains(&32), "no VPS in first packet: {types:?}");
		assert!(types.contains(&33), "no SPS in first packet: {types:?}");
		assert!(types.contains(&34), "no PPS in first packet: {types:?}");
		assert!(
			types.iter().any(|t| (16..=23).contains(t)),
			"first packet is not an IRAP: {types:?}"
		);
	}

	/// HEVC NAL unit types in an Annex-B buffer (type = `(byte >> 1) & 0x3f`).
	#[cfg(target_os = "macos")]
	fn hevc_nal_types(annexb: &[u8]) -> Vec<u8> {
		let mut types = Vec::new();
		let mut i = 0;
		while i + 3 < annexb.len() {
			if annexb[i..i + 3] == [0, 0, 1] {
				types.push((annexb[i + 3] >> 1) & 0x3f);
				i += 3;
			} else {
				i += 1;
			}
		}
		types
	}

	/// Feed a GPU surface (NV12 `CVPixelBuffer`) straight into VideoToolbox:
	/// the zero-copy capture -> encode path, no I420 round-trip.
	#[cfg(target_os = "macos")]
	#[test]
	fn videotoolbox_encodes_surface_zero_copy() {
		let config = Config {
			kind: Kind::Named("videotoolbox".into()),
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		let mut frames = Vec::new();
		for i in 0..10 {
			if i == 0 {
				encoder.cut().unwrap();
			}
			let frame = Frame::new(Surface::PixelBuffer(nv12_surface(320, 240)), at(i));
			frames.extend(encoder.encode(&frame).unwrap());
		}
		frames.extend(encoder.finish().unwrap());

		assert!(!frames.is_empty());
		let packets = payloads(&frames);
		let types = nal_types(&packets[0]);
		assert!(
			types.contains(&7) && types.contains(&8) && types.contains(&5),
			"no IDR: {types:?}"
		);
	}

	/// A software encoder must download a GPU surface to I420 first. Exercises
	/// the NV12 -> I420 fallback path.
	#[cfg(all(target_os = "macos", feature = "openh264"))]
	#[test]
	fn openh264_downloads_surface() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		encoder.cut().unwrap();
		let frame = Frame::new(Surface::PixelBuffer(nv12_surface(320, 240)), at(0));
		let mut frames = encoder.encode(&frame).unwrap();
		frames.extend(encoder.finish().unwrap());

		assert!(!frames.is_empty());
		let packets = payloads(&frames);
		assert!(packets[0].starts_with(&[0, 0, 0, 1]) || packets[0].starts_with(&[0, 0, 1]));
	}

	/// A mid-gray NV12 `CVPixelBuffer`, the format AVFoundation/ScreenCaptureKit
	/// hand us. Y and interleaved UV planes filled with 128.
	#[cfg(target_os = "macos")]
	fn nv12_surface(width: u32, height: u32) -> crate::frame::macos::PixelBuffer {
		use std::ptr::{self, NonNull};

		use objc2_core_foundation::CFRetained;
		use objc2_core_video::{
			CVPixelBuffer, CVPixelBufferCreate, CVPixelBufferGetBaseAddressOfPlane, CVPixelBufferGetBytesPerRowOfPlane,
			CVPixelBufferLockBaseAddress, CVPixelBufferLockFlags, CVPixelBufferUnlockBaseAddress,
			kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange,
		};

		let mut raw: *mut CVPixelBuffer = ptr::null_mut();
		let status = unsafe {
			CVPixelBufferCreate(
				None,
				width as usize,
				height as usize,
				kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange,
				None,
				NonNull::new(&mut raw).unwrap(),
			)
		};
		assert_eq!(status, 0, "CVPixelBufferCreate failed");
		let buffer = unsafe { CFRetained::from_raw(NonNull::new(raw).unwrap()) };

		let flags = CVPixelBufferLockFlags(0);
		assert_eq!(unsafe { CVPixelBufferLockBaseAddress(&buffer, flags) }, 0);
		for (plane, rows) in [(0usize, height as usize), (1usize, height as usize / 2)] {
			let base = CVPixelBufferGetBaseAddressOfPlane(&buffer, plane) as *mut u8;
			let stride = CVPixelBufferGetBytesPerRowOfPlane(&buffer, plane);
			unsafe { ptr::write_bytes(base, 128, stride * rows) };
		}
		unsafe { CVPixelBufferUnlockBaseAddress(&buffer, flags) };

		crate::frame::macos::PixelBuffer::new(buffer, width, height)
	}

	/// NAL unit types in an Annex-B buffer, found via 3-byte start codes (a
	/// 4-byte `00 00 00 01` code contains `00 00 01` too, so this catches both).
	fn nal_types(annexb: &[u8]) -> Vec<u8> {
		let mut types = Vec::new();
		let mut i = 0;
		while i + 3 < annexb.len() {
			if annexb[i..i + 3] == [0, 0, 1] {
				types.push(annexb[i + 3] & 0x1f);
				i += 3;
			} else {
				i += 1;
			}
		}
		types
	}

	/// CPU path: synthetic RGBA through the Media Foundation hardware encoder
	/// (I420 -> system-memory NV12 upload). Ignored: needs a hardware encoder MFT,
	/// which GPU-less CI runners lack. Run with `--ignored`.
	#[cfg(target_os = "windows")]
	#[test]
	#[ignore]
	fn mediafoundation_cpu_rgba() {
		let config = Config {
			kind: Kind::Named("mediafoundation".into()),
			..Config::new(640, 480, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).expect("hardware H.264 encoder available");
		assert_eq!(encoder.name(), "mediafoundation");

		let mut frames = Vec::new();
		for i in 0..30 {
			if i == 0 {
				encoder.cut().unwrap();
			}
			frames.extend(encoder.encode(&gray_frame(640, 480, i)).unwrap());
		}
		frames.extend(encoder.finish().unwrap());

		assert!(!frames.is_empty(), "encoder produced no packets");
		// The MFT buffers, so packets come back stamped with the frame they were
		// encoded from rather than whichever frame was going in at the time.
		let micros: Vec<u128> = frames.iter().map(|f| f.timestamp.as_micros()).collect();
		assert!(
			micros.windows(2).all(|w| w[0] < w[1]),
			"encoded timestamps not strictly increasing: {micros:?}"
		);
		assert!(
			micros.iter().all(|&t| t % 33_333 == 0 && t < 30 * 33_333),
			"encoded timestamp outside the fed set: {micros:?}"
		);

		let packets = payloads(&frames);
		let types = nal_types(&packets[0]);
		assert!(types.contains(&7), "no SPS in first packet: {types:?}");
		assert!(types.contains(&8), "no PPS in first packet: {types:?}");
		assert!(types.contains(&5), "first packet is not an IDR: {types:?}");
	}

	/// Full zero-copy path: real camera -> D3D11 NV12 texture -> hardware encoder
	/// via the DXGI device manager, no CPU round-trip. Ignored: needs a camera and
	/// a GPU. Run with `--ignored`.
	#[cfg(all(target_os = "windows", feature = "capture"))]
	#[tokio::test]
	#[ignore]
	async fn mediafoundation_camera_texture() {
		let mut camera = crate::capture::open(&crate::capture::Config::default())
			.await
			.expect("open default camera");
		let (w, h) = (camera.width(), camera.height());

		let config = Config {
			kind: Kind::Named("mediafoundation".into()),
			..Config::new(w, h, camera.framerate().unwrap_or(30))
		};
		let mut encoder = Encoder::new(&config).expect("hardware H.264 encoder available");

		let mut frames = Vec::new();
		let mut textures = 0;
		for i in 0..30 {
			let surface = camera.read().await.expect("read camera frame");
			if matches!(surface, Some(Surface::Texture(_))) {
				textures += 1;
			}
			if i == 0 {
				encoder.cut().unwrap();
			}
			let surface = surface.expect("frame, not end of stream");
			frames.extend(encoder.encode(&Frame::new(surface, at(i))).unwrap());
		}
		frames.extend(encoder.finish().unwrap());

		// On a GPU this exercises the zero-copy texture path; the assert guards
		// against silently testing only the CPU fallback.
		assert!(textures > 0, "capture never produced a GPU texture");
		assert!(!frames.is_empty(), "encoder produced no packets");
		let packets = payloads(&frames);
		let types = nal_types(&packets[0]);
		assert!(
			types.contains(&7) && types.contains(&8) && types.contains(&5),
			"no IDR: {types:?}"
		);
	}

	/// The openh264 retune goes through the raw `set_option` FFI, so this covers
	/// both that the call is accepted and that the encoder keeps producing after
	/// it. A wrong option id or a bad `SBitrateInfo` layout would fail here.
	#[test]
	#[cfg(feature = "openh264")]
	fn set_bitrate_retunes_software_encoder() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		let opened = encoder.bitrate();
		assert_eq!(opened, config.resolved_bitrate());

		// Encode first: this is the live-retune path, once the encoder exists.
		encoder.encode(&gray_frame(320, 240, 0)).unwrap();

		let halved = opened.scaled(0.5);
		encoder.set_bitrate(halved).unwrap();
		assert_eq!(encoder.bitrate(), halved);

		// The retuned encoder must still emit a decodable keyframe, not wedge.
		let frames = encoder.encode(&gray_frame(320, 240, 1)).unwrap();
		assert!(!frames.is_empty(), "encoder produced nothing after a retune");
		let packets = payloads(&frames);
		assert!(packets[0].starts_with(&[0, 0, 0, 1]) || packets[0].starts_with(&[0, 0, 1]));
	}

	/// Regression: openh264 creates its encoder lazily on the first frame and
	/// rejects `SetOption` with `cmInitExpected` until then. A retune before any
	/// frame must be deferred to the first encode, not reported as a failure.
	#[test]
	#[cfg(feature = "openh264")]
	fn set_bitrate_before_the_first_frame_is_deferred() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		let halved = encoder.bitrate().scaled(0.5);
		encoder.set_bitrate(halved).expect("a retune before the first frame");
		assert_eq!(encoder.bitrate(), halved);

		// The deferred rate is applied during this encode, which must still work.
		let frames = encoder.encode(&gray_frame(320, 240, 0)).unwrap();
		assert!(!frames.is_empty());

		// And the encoder is live now, so a further retune takes the direct path.
		encoder.set_bitrate(halved.scaled(0.5)).unwrap();
		assert!(encoder.encode(&gray_frame(320, 240, 1)).is_ok());
	}

	/// Setting the current rate must not reach the backend at all: the control
	/// loop is allowed to be chatty, and the encoder shouldn't pay for it.
	#[test]
	#[cfg(feature = "openh264")]
	fn set_bitrate_to_current_is_a_noop() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();

		let opened = encoder.bitrate();
		encoder.set_bitrate(opened).unwrap();
		assert_eq!(encoder.bitrate(), opened);
	}

	#[test]
	fn default_bitrate_scales_with_resolution() {
		let small = Config::new(320, 240, crate::Rate::new(30, 1).unwrap()).resolved_bitrate();
		let large = Config::new(1920, 1080, crate::Rate::new(30, 1).unwrap()).resolved_bitrate();
		assert!(large > small);
		assert!(small > moq_net::bandwidth::Rate::ZERO);
	}

	/// A backend that holds each frame back by one, like the Media Foundation MFT:
	/// `encode` returns the *previous* frame's access unit and `finish` drains the
	/// last. The payload is the frame's timestamp in microseconds, so a test can
	/// tell which frame a packet came from independently of what it's stamped with.
	struct Delayed {
		pending: Option<Encoded>,
	}

	impl Backend for Delayed {
		fn encode(&mut self, frame: &Frame, _cut: bool) -> Result<Vec<Encoded>, Error> {
			let payload = bytes::Bytes::from(frame.timestamp.as_micros().to_string());
			let previous = self.pending.replace(Encoded::new(payload, frame.timestamp));
			Ok(previous.into_iter().collect())
		}

		fn flush(&mut self) -> Result<Vec<Encoded>, Error> {
			Ok(self.pending.take().into_iter().collect())
		}

		fn finish(&mut self) -> Result<Vec<Encoded>, Error> {
			self.flush()
		}

		fn set_bitrate(&mut self, _bitrate: u64) -> Result<(), Error> {
			Ok(())
		}

		fn can_cut(&self) -> bool {
			true
		}

		fn name(&self) -> &'static str {
			"delayed"
		}
	}

	/// An encoder over a hand-built backend, so a test can pick the buffering
	/// behavior rather than take whatever this machine's hardware does.
	fn encoder_with(backend: Box<dyn Backend>, config: &Config) -> Encoder {
		Encoder {
			backend,
			codec: config.codec,
			size: config.size(),
			bitrate: config.resolved_bitrate(),
			color: config.resolved_color(),
			pending_cut: false,
			_thread_bound: PhantomData,
		}
	}

	/// Records the cut flag each frame reached the codec with, so a test can
	/// check what the encoder actually asked for. `cuts` is whether it claims to
	/// honor one, the way a V4L2 driver without the force-keyframe control does
	/// not.
	struct Recorder {
		log: std::sync::Arc<std::sync::Mutex<Vec<bool>>>,
		cuts: bool,
	}

	impl Backend for Recorder {
		fn encode(&mut self, _frame: &Frame, cut: bool) -> Result<Vec<Encoded>, Error> {
			self.log.lock().unwrap().push(cut);
			Ok(Vec::new())
		}

		fn flush(&mut self) -> Result<Vec<Encoded>, Error> {
			Ok(Vec::new())
		}

		fn finish(&mut self) -> Result<Vec<Encoded>, Error> {
			Ok(Vec::new())
		}

		fn set_bitrate(&mut self, _bitrate: u64) -> Result<(), Error> {
			Ok(())
		}

		fn can_cut(&self) -> bool {
			self.cuts
		}

		fn name(&self) -> &'static str {
			"recorder"
		}
	}

	/// Group boundaries are automatic, so an untouched encoder forces none. A cut
	/// is held until a frame arrives (callers decide a group boundary before they
	/// have the frame that opens it), collapses if made twice, and clears
	/// afterwards rather than cutting every frame from then on.
	#[test]
	fn a_cut_waits_for_the_next_frame_then_clears() {
		let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
		let config = Config::new(320, 240, crate::Rate::new(30, 1).unwrap());
		let backend = Recorder {
			log: log.clone(),
			cuts: true,
		};
		let mut encoder = encoder_with(Box::new(backend), &config);

		// Nothing asked for: `Config::gop` places the boundaries on its own.
		encoder.encode(&gray_frame(320, 240, 0)).unwrap();

		// Asked twice before a frame exists: one cut, on the next frame.
		encoder.cut().unwrap();
		encoder.cut().unwrap();
		encoder.encode(&gray_frame(320, 240, 1)).unwrap();

		// And it does not carry into the frame after.
		encoder.encode(&gray_frame(320, 240, 2)).unwrap();

		assert_eq!(*log.lock().unwrap(), vec![false, true, false]);
	}

	/// Repeated boundaries each reach the codec: a cut per group is what a
	/// transcode rung asks for, and a request that only worked once would leave
	/// every later group opening on a delta frame.
	#[test]
	fn every_cut_reaches_the_codec() {
		let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
		let config = Config::new(320, 240, crate::Rate::new(30, 1).unwrap());
		let backend = Recorder {
			log: log.clone(),
			cuts: true,
		};
		let mut encoder = encoder_with(Box::new(backend), &config);

		for i in 0..6 {
			if i % 3 == 0 {
				encoder.cut().unwrap();
			}
			encoder.encode(&gray_frame(320, 240, i)).unwrap();
		}

		assert_eq!(*log.lock().unwrap(), vec![true, false, false, true, false, false]);
	}

	/// A backend that cannot force a boundary refuses the cut itself, up front,
	/// rather than queueing a request its codec would ignore. The V4L2 backend
	/// used to disable the request quietly and let groups fall on the driver's
	/// own interval, which only the subscriber could tell apart from a cut that
	/// worked.
	#[test]
	fn a_backend_that_cannot_cut_refuses_up_front() {
		let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
		let config = Config::new(320, 240, crate::Rate::new(30, 1).unwrap());
		let backend = Recorder {
			log: log.clone(),
			cuts: false,
		};
		let mut encoder = encoder_with(Box::new(backend), &config);

		let err = encoder.cut().expect_err("the backend cannot cut");
		assert!(
			matches!(err, Error::CutUnsupported("recorder")),
			"unexpected error: {err:?}"
		);
		assert!(!encoder.pending_cut, "a refused cut must not be queued");

		// The refusal names the backend, so a log line says which one to blame.
		assert!(err.to_string().contains("recorder"), "{err}");

		// Encoding carries on at the backend's own boundaries; the refusal is the
		// caller's to act on, not a broken encoder.
		encoder.encode(&gray_frame(320, 240, 0)).unwrap();
		assert_eq!(*log.lock().unwrap(), vec![false]);
	}

	/// An interval of zero opens no group, so it is refused at open rather than
	/// handed to a backend, each of which would do something different with it
	/// (openh264 keys every frame, V4L2 rejects the control, VideoToolbox reads
	/// it as unlimited).
	#[test]
	fn a_zero_keyframe_interval_is_refused() {
		let config = Config {
			gop: Gop::Keyframe { interval: 0 },
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		let Err(err) = Encoder::new(&config) else {
			panic!("an empty group cannot open");
		};
		assert!(matches!(err, Error::InvalidGop(_)), "unexpected error: {err:?}");
	}

	/// The wall-time constructor never produces the interval `Encoder::new`
	/// refuses: a window shorter than a frame rounds up to one frame, not down
	/// to none.
	#[test]
	fn a_sub_frame_keyframe_window_rounds_up_to_one_frame() {
		let framerate = crate::Rate::new(30, 1).unwrap();
		assert_eq!(
			Gop::keyframe_every(std::time::Duration::from_millis(1), framerate),
			Gop::Keyframe { interval: 1 }
		);
		assert_eq!(
			Gop::keyframe_every(std::time::Duration::from_secs(2), framerate),
			Gop::Keyframe { interval: 60 }
		);
		assert!(
			Gop::keyframe_every(std::time::Duration::ZERO, framerate)
				.validate()
				.is_ok()
		);
	}

	/// `cut()` has to reach the codec on a *warm* encoder, which is the case that
	/// matters: a fresh one emits an IDR on its first frame regardless, so only a
	/// mid-stream request proves the plumbing works. Runs on openh264, so this
	/// holds on every platform rather than only where hardware exists.
	#[test]
	#[cfg(feature = "openh264")]
	fn a_mid_stream_cut_emits_an_idr() {
		let config = Config {
			kind: Kind::Software,
			..Config::new(320, 240, crate::Rate::new(30, 1).unwrap())
		};
		// A GOP far longer than the run, so any IDR here was asked for rather than
		// inserted on schedule.
		let mut encoder = Encoder::new(&Config {
			gop: Gop::Keyframe { interval: 1000 },
			..config
		})
		.unwrap();

		let mut per_frame = Vec::new();
		for i in 0..6 {
			// Frame 0 opens the stream; ask again at frame 3, mid-stream.
			if i == 3 {
				encoder.cut().unwrap();
			}
			let encoded = encoder.encode(&gray_frame(320, 240, i)).unwrap();
			let joined: Vec<u8> = encoded.iter().flat_map(|f| f.payload.iter()).copied().collect();
			per_frame.push(nal_types(&joined));
		}

		// The requested frame is a self-contained IDR: SPS (7) and PPS (8) inline
		// ahead of an IDR slice (5), which is what avc3 promises a joining subscriber.
		let asked = &per_frame[3];
		assert!(asked.contains(&5), "the requested frame is not an IDR: {asked:?}");
		assert!(asked.contains(&7), "no SPS with the requested IDR: {asked:?}");
		assert!(asked.contains(&8), "no PPS with the requested IDR: {asked:?}");

		// And the frames around it stay delta frames, so the request keyed exactly
		// one: keying every frame would pass the assertions above while destroying
		// the bitrate.
		for i in [1, 2, 4, 5] {
			assert!(
				!per_frame[i].contains(&5),
				"frame {i} was keyed without being asked: {:?}",
				per_frame[i]
			);
		}
	}

	/// A backend whose every encode fails, to pin what survives one.
	struct Failing;

	impl Backend for Failing {
		fn encode(&mut self, _frame: &Frame, _cut: bool) -> Result<Vec<Encoded>, Error> {
			Err(Error::Codec(anyhow::anyhow!("no")))
		}

		fn flush(&mut self) -> Result<Vec<Encoded>, Error> {
			Ok(Vec::new())
		}

		fn finish(&mut self) -> Result<Vec<Encoded>, Error> {
			Ok(Vec::new())
		}

		fn set_bitrate(&mut self, _bitrate: u64) -> Result<(), Error> {
			Ok(())
		}

		fn can_cut(&self) -> bool {
			true
		}

		fn name(&self) -> &'static str {
			"failing"
		}
	}

	/// A cut outlives an encode that produced no picture, whether it was rejected
	/// up front (wrong size) or failed in the backend. Dropping it would leave the
	/// next frame unkeyed, so a subscriber waits out a whole GOP for a starting
	/// point the caller already asked for.
	#[test]
	fn a_failed_encode_keeps_the_cut() {
		let config = Config::new(320, 240, crate::Rate::new(30, 1).unwrap());

		let mut encoder = encoder_with(Box::new(Failing), &config);
		encoder.cut().unwrap();
		assert!(encoder.encode(&gray_frame(320, 240, 0)).is_err());
		assert!(encoder.pending_cut, "the backend error swallowed the request");

		// Rejected before the backend ever sees it, for the same reason.
		let mut encoder = encoder_with(Box::new(Delayed { pending: None }), &config);
		encoder.cut().unwrap();
		assert!(encoder.encode(&gray_frame(640, 480, 0)).is_err());
		assert!(encoder.pending_cut, "the size check swallowed the request");

		// ...and the next frame that does go through claims it.
		encoder.encode(&gray_frame(320, 240, 1)).unwrap();
		assert!(!encoder.pending_cut);
	}

	/// Regression: a backend that buffers hands back an earlier frame's access
	/// unit, so the timestamp has to ride through the encoder with the picture.
	/// Stamping output with whatever frame is going in at the time (or the tail
	/// with one arbitrary time) shifts every packet by the encoder delay and
	/// collapses the drained tail onto a single instant.
	#[test]
	fn a_buffering_backend_keeps_each_frames_timestamp() {
		let config = Config::new(320, 240, crate::Rate::new(30, 1).unwrap());
		let mut encoder = encoder_with(Box::new(Delayed { pending: None }), &config);

		// The packet handed back while frame `i` goes in belongs to frame `i - 1`, so
		// it has to be stamped one frame back. Stamping at the call site (the only
		// option when encode just returned bytes) would shift the whole stream.
		for i in 0..5 {
			let encoded = encoder.encode(&gray_frame(320, 240, i)).unwrap();
			if i == 0 {
				assert!(encoded.is_empty(), "the first frame is still buffered");
				continue;
			}
			assert_eq!(encoded.len(), 1);
			assert_eq!(encoded[0].timestamp, at(i - 1));
			// And the packet really is the earlier frame's, not a mis-stamped copy of
			// the one going in.
			assert_eq!(&encoded[0].payload[..], at(i - 1).as_micros().to_string().as_bytes());
		}

		// The tail: frame 4 never came back from an encode call, so `finish` drains
		// it, still carrying its own time rather than one the caller picks.
		let tail = encoder.finish().unwrap();
		assert_eq!(tail.len(), 1);
		assert_eq!(tail[0].timestamp, at(4));
		assert!(
			encoder_with(Box::new(Delayed { pending: None }), &config)
				.finish()
				.unwrap()
				.is_empty()
		);
	}

	/// A group has to contain its own frames. A codec that pipelines is still
	/// holding the last of them when the group ends, so the boundary flushes it;
	/// without that they surface in the *next* group, ahead of its keyframe, where
	/// a subscriber joining there cannot decode them.
	///
	/// The encoder stays usable afterwards, which is what separates this from
	/// `finish`: a live track flushes at every group and keeps going.
	#[test]
	fn a_flush_empties_a_pipelined_backend_and_leaves_it_running() {
		let config = Config::new(320, 240, crate::Rate::new(30, 1).unwrap());
		let mut encoder = encoder_with(Box::new(Delayed { pending: None }), &config);

		let mut group = Vec::new();
		for i in 0..3 {
			group.extend(encoder.encode(&gray_frame(320, 240, i)).unwrap());
		}
		group.extend(encoder.flush().unwrap());

		// All three frames land in the group they belong to, in order.
		let times: Vec<_> = group.iter().map(|packet| packet.timestamp).collect();
		assert_eq!(times, vec![at(0), at(1), at(2)], "the group lost or reordered frames");

		// The next group starts clean: nothing carried over, and the encoder still
		// takes frames.
		let mut next = encoder.encode(&gray_frame(320, 240, 3)).unwrap();
		assert!(next.is_empty(), "frame 3 is buffered, so nothing comes back yet");
		next.extend(encoder.finish().unwrap());
		let times: Vec<_> = next.iter().map(|packet| packet.timestamp).collect();
		assert_eq!(times, vec![at(3)], "the flush left something behind");
	}

	/// The hardware encoder states the color space too, and states the one the
	/// pixels were actually converted into. VideoToolbox takes the three
	/// properties as a request, so read the SPS back rather than trusting that it
	/// honored them.
	#[cfg(target_os = "macos")]
	#[test]
	fn videotoolbox_sps_declares_the_color_space() {
		use super::backend::test_util::{BT601_DESCRIBED, BT709_DESCRIBED, declared_color};

		for (size, described) in [
			(Size::new(640, 480), BT601_DESCRIBED),
			(Size::new(1920, 1080), BT709_DESCRIBED),
		] {
			let config = Config {
				kind: Kind::Named("videotoolbox".into()),
				..Config::new(size.width, size.height, crate::Rate::new(30, 1).unwrap())
			};
			let mut encoder = Encoder::new(&config).expect("videotoolbox is available on macOS");

			let rgba = [255u8, 0, 0, 255].repeat(size.pixels() as usize);
			let surface = crate::frame::Surface::rgba(&rgba, size).unwrap();
			encoder.cut().unwrap();
			let frames = encoder
				.encode(&Frame::new(surface, moq_net::Timestamp::from_micros(0).unwrap()))
				.unwrap();

			let keyframe = frames.first().expect("a keyframe");
			assert_eq!(declared_color(&keyframe.payload), Some(described), "{size} SPS");
		}
	}

	/// Resizing across the standard-definition boundary keeps the pixels' color
	/// space but moves the encoder into another one, which would emit BT.709
	/// pixels under a BT.601 label. `Config::color` pins the real space, and the
	/// bitstream then says so.
	#[test]
	#[cfg(feature = "openh264")]
	fn config_color_pins_the_space_a_resize_carried() {
		use crate::Color;

		let big = Size::new(1280, 720);
		let small = Size::new(640, 480);

		// Converted at 720p, so the samples are BT.709.
		let rgba = vec![0x80u8; big.pixels() as usize * 4];
		let frame = Frame::new(
			crate::frame::Surface::rgba(&rgba, big).unwrap(),
			moq_net::Timestamp::from_micros(0).unwrap(),
		);
		let scaled = frame.resize(small, &crate::resize::Config::default()).unwrap();
		assert_eq!(
			scaled.surface.color(),
			Some(Color::Bt709Limited),
			"resize keeps the space"
		);

		// Left to infer, a 480p encoder writes BT.601 over those BT.709 samples. It
		// still encodes (a live gateway keeps serving) but the label is wrong, which
		// is what `Config::color` exists to fix.
		let config = Config {
			kind: Kind::Software,
			..Config::new(small.width, small.height, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();
		encoder.cut().unwrap();
		let frames = encoder.encode(&scaled).expect("a mismatch warns rather than fails");
		use super::backend::test_util::{BT601_DESCRIBED, BT709_DESCRIBED, declared_color};
		assert_eq!(
			declared_color(&frames.first().expect("a keyframe").payload),
			Some(BT601_DESCRIBED),
			"the inferred label is the wrong one, which is the case Config::color covers"
		);

		// Declaring the real space fixes the label.
		let config = Config {
			kind: Kind::Software,
			color: Some(Color::Bt709Limited),
			..Config::new(small.width, small.height, crate::Rate::new(30, 1).unwrap())
		};
		let mut encoder = Encoder::new(&config).unwrap();
		encoder.cut().unwrap();
		let frames = encoder.encode(&scaled).expect("a declared space encodes");

		let keyframe = frames.first().expect("a keyframe");
		assert_eq!(declared_color(&keyframe.payload), Some(BT709_DESCRIBED));
	}
}