otf-pixels 0.2.0

//! A streaming, demand-driven image processing engine.
//!
//! Pixels is a libvips-class pipeline engine: images are lazy operation
//! graphs, pixels are pulled through the graph on demand, and memory stays
//! bounded regardless of image size. This crate is the facade — the chainable
//! [`Image`] API and the [`Image::output`] terminal — over
//! [`otf_pixels_core`]'s engine, [`otf_pixels_ops`]' kernels and the codec
//! crates.
//!
// The example writes raw bytes, so it only runs when that codec is compiled
// in. `raw` is a default feature, so docs.rs and an ordinary build both run it.
#![cfg_attr(feature = "raw", doc = "```")]
#![cfg_attr(not(feature = "raw"), doc = "```ignore")]
//! use otf_pixels::{Format, Image, ImageDescriptor, PixelFormat};
//!
//! # fn main() -> Result<(), otf_pixels::PixelsError> {
//! let descriptor = ImageDescriptor::new(4, 4, PixelFormat::Gray8)?;
//! let pixels: Vec<u8> = (0..16).collect();
//!
//! // Construction and chaining do no pixel work.
//! let bytes = Image::from_raw(descriptor, pixels)?
//!     .crop(1, 1, 2, 2)
//!     .flip()
//!     .output(Format::Raw, Default::default())
//!     .bytes()?;
//!
//! assert_eq!(bytes, [9, 10, 5, 6]);
//! # Ok(())
//! # }
//! ```
//!
//! # Errors are deferred, not swallowed
//!
//! Chaining methods take and return `Self` rather than [`Result`], so a
//! pipeline reads as one expression. An error raised mid-chain — a crop window
//! outside the image, say — is *captured* and carried to the terminal, where
//! it surfaces from [`Output::write`] or [`Output::bytes`]. Nothing is
//! silently ignored, and no operation runs after a failed one.
//!
//! # Evaluation
//!
//! Terminals run the pipeline on the demand-driven tile scheduler: output
//! tiles are evaluated in parallel and delivered to the sink in order, with
//! peak memory bounded by tiles in flight rather than image size.
//!
//! Every output runs on [`Scheduler::global`] unless told otherwise: one pool
//! of worker threads, one per core, and one tile cache, shared by every
//! pipeline in the process. That is the right setup for a server or a
//! runtime handling many images at once, and it needs no code: call
//! `output(...).bytes()` from as many threads as you like and the runs share
//! the workers. Do not build a [`Scheduler`] per request, and do not set
//! [`Output::threads`] or [`Output::scheduler_options`] to tune a busy host —
//! both give that run a private pool, spawned and joined each time, which
//! under concurrency means a pool per request competing for the same cores.
//! [`Output::with_scheduler`] runs on a scheduler you built, for a host that
//! wants image work confined to a fixed number of threads.
//!
//! [`Output::bytes_via_reference`] runs the same pipeline through the M1
//! whole-image evaluator instead. That path is slow and holds every
//! intermediate in full, but it is obviously correct, so it is the oracle the
//! scheduler is verified against.
//!
//! # Serving images
//!
//! What an image endpoint or a runtime's image API does: bytes in, a
//! thumbnail out. Opening identifies the format from its bytes, turns the
//! image upright, converts its colours to sRGB and enforces input limits;
//! nothing is decoded until the output is pulled, and a JPEG source decodes
//! at a reduced scale when the thumbnail allows it.
//!
#![cfg_attr(all(feature = "png", feature = "webp"), doc = "```")]
#![cfg_attr(not(all(feature = "png", feature = "webp")), doc = "```ignore")]
//! use otf_pixels::{
//!     EncodeOptions, Fit, Format, Image, Limits, OpenOptions, ResizeOptions,
//! };
//!
//! # fn main() -> Result<(), otf_pixels::PixelsError> {
//! # let upload = Image::from_raw(
//! #     otf_pixels::ImageDescriptor::new(64, 48, otf_pixels::PixelFormat::Rgb8)?,
//! #     vec![90; 64 * 48 * 3],
//! # )?
//! # .output(Format::Png, EncodeOptions::default())
//! # .bytes()?;
//! // Per request: bound what an untrusted upload may allocate.
//! let options = OpenOptions::default()
//!     .with_limits(Limits::default().with_max_pixels(50_000_000));
//! let image = Image::from_stream_with(std::io::Cursor::new(upload), options)?;
//!
//! let meta = image.metadata()?; // free: no pixels decoded
//! assert_eq!((meta.width, meta.height), (64, 48));
//! if let Some(animation) = image.animation() {
//!     // v1 processes the first frame; the caller decides whether that is
//!     // acceptable for this animation.
//!     let _ = animation.frame_count;
//! }
//!
//! let webp = image
//!     .resize_with(32, 32, ResizeOptions::default().with_fit(Fit::Cover))
//!     .output(Format::WebP, EncodeOptions::with_quality(80)?)
//!     .bytes()?;
//! assert_eq!(&webp[8..12], b"WEBP");
//! # Ok(())
//! # }
//! ```
//!
//! # Scope (v1)
//!
//! - **Formats**, all implemented in this workspace and checked against
//!   their reference implementations: PNG, GIF, baseline JPEG (progressive
//!   decode is wrapped), TIFF, WebP (lossy and lossless) and AVIF, read and
//!   written, plus raw pixels.
//! - **Ops**: crop, flip/flop, quarter-turn rotation and orientation,
//!   resize with sharp's five fit modes, modulate, convolve/blur/sharpen,
//!   composite, flatten, channel extraction, pixel-format and sRGB
//!   conversion.
//! - **Metadata**: EXIF/HEIF orientation applied on open; ICC profiles
//!   converted to sRGB or carried to the output; animation reported, with
//!   the first frame processed. Other metadata (EXIF, XMP) is not written
//!   out, which also strips location data from uploads.
//!
//! Not yet: multi-frame (animated) pipelines, arbitrary-angle rotation and
//! progressive JPEG output. Each will arrive as an addition, not a change:
//! [`OpenOptions::animated`] is already reserved for the first.

use otf_pixels_core::{BufferSource, Op, Prefixed, Producer, TileBuf};
use std::sync::Arc;

pub use otf_pixels_core::{
    AccessPattern, Animation, ChannelLayout, Codec, ColorModel, Decoder, EncodeOptions, Encoder,
    ErrorCode, Format, ImageDescriptor, Limit, Limits, Metadata, Orientation, PixelFormat,
    PixelsError, PlanOptions, Region, Result, RunStats, SampleKind, Scheduler, SchedulerOptions,
    Sink, Source, TileShape, evaluate as evaluate_reference,
};
pub use otf_pixels_ops::{
    Blend, Composite, Conversion, ConvertFormat, Convolve, Crop, ExtractChannel, Filter, Fit,
    Flatten, Flip, Flop, Kernel, Modulate, Quarter, Resize, ResizeOptions, Rotate, ToSrgb,
    Unconvertible,
};

#[cfg(feature = "raw")]
pub use otf_pixels_codec_raw::{RawCodec, RawDecoder, RawEncoder, RawFormat};

#[cfg(feature = "png")]
pub use otf_pixels_codec_png::{PngCodec, PngDecoder, PngEncoder};

#[cfg(feature = "gif")]
pub use otf_pixels_codec_gif::{GifCodec, GifDecoder, GifEncoder};

#[cfg(feature = "jpeg")]
pub use otf_pixels_codec_jpeg::{JpegCodec, JpegDecoder, JpegEncoder, Scale, Subsampling};

#[cfg(feature = "tiff")]
pub use otf_pixels_codec_tiff::{TiffCodec, TiffDecoder, TiffEncoder, TiffLayout};

#[cfg(feature = "webp")]
pub use otf_pixels_codec_webp::{WebPCodec, WebPDecoder, WebPEncoder};

#[cfg(feature = "avif")]
pub use otf_pixels_codec_avif::{AvifCodec, AvifDecoder, AvifEncoder};

/// How [`Image::open_with`] and [`Image::from_stream_with`] read an image.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct OpenOptions {
    /// Turn the image upright as its metadata declares — EXIF `Orientation`
    /// in JPEG, TIFF and WebP, `irot`/`imir` in AVIF — before any op.
    ///
    /// On by default (SPEC §Safety and limits): a phone photograph is stored
    /// sideways far more often than anyone wants it processed that way. Off,
    /// pixels arrive as stored, and [`Image::orient`] applies an orientation
    /// read some other way.
    pub auto_orient: bool,
    /// Convert pixels in an embedded ICC profile's colour space to sRGB,
    /// the space every op and most consumers assume (SPEC §Pixel formats).
    ///
    /// On by default, so a Display P3 phone photo or an Adobe RGB export
    /// does not come out dull or garish. Off, pixels arrive as stored with
    /// the profile attached ([`Image::icc_profile`]) and written into the
    /// output, and [`Image::to_srgb`] converts later. A profile this cannot
    /// convert is kept either way.
    pub to_srgb: bool,
    /// Ask for every frame of an animated image rather than the first.
    ///
    /// Reserved for the multi-frame pipeline, which is not implemented yet:
    /// set, opening an animated GIF or WebP fails with
    /// [`PixelsError::Unsupported`] rather than quietly processing one
    /// frame, so code written against it today states its intent and starts
    /// working when frames do. Off by default, as in sharp: the first frame
    /// is the image, and [`Image::animation`] says what was left out.
    pub animated: bool,
    /// Bounds on what a file may make the decoder allocate. A runtime facing
    /// untrusted uploads sets these per request; the default refuses images
    /// over 268 megapixels, as sharp does.
    pub limits: Limits,
}

impl OpenOptions {
    /// The defaults with `auto_orient` replaced.
    ///
    /// [`OpenOptions`] is `#[non_exhaustive]`, so outside this crate a setter
    /// is the only way to change a field.
    #[must_use]
    pub const fn with_auto_orient(mut self, auto_orient: bool) -> Self {
        self.auto_orient = auto_orient;
        self
    }

    /// The defaults with `animated` replaced.
    #[must_use]
    pub const fn with_animated(mut self, animated: bool) -> Self {
        self.animated = animated;
        self
    }

    /// The defaults with `limits` replaced.
    #[must_use]
    pub const fn with_limits(mut self, limits: Limits) -> Self {
        self.limits = limits;
        self
    }

    /// The defaults with `to_srgb` replaced.
    #[must_use]
    pub const fn with_to_srgb(mut self, to_srgb: bool) -> Self {
        self.to_srgb = to_srgb;
        self
    }
}

impl Default for OpenOptions {
    fn default() -> Self {
        Self {
            auto_orient: true,
            to_srgb: true,
            animated: false,
            limits: Limits::default(),
        }
    }
}

/// A lazily evaluated image pipeline.
///
/// Cheap to clone: clones share graph nodes rather than pixels. Chaining
/// builds graph structure and executes nothing (SPEC §Guarantees 3).
#[derive(Debug, Clone)]
pub struct Image {
    /// The graph so far, or the first error that occurred while building it.
    inner: std::result::Result<otf_pixels_core::Image, Arc<PixelsError>>,
    /// The ICC profile the pixels are in, carried to the output; `None` is
    /// sRGB, as SPEC §Pixel formats assumes.
    icc: Option<Arc<[u8]>>,
    /// The source's animation, when it has more than one frame.
    animation: Option<Arc<Animation>>,
}

impl Image {
    /// Build an image from raw pixels already in memory.
    ///
    /// `bytes` must be exactly the packed byte length of `descriptor`.
    ///
    /// # Errors
    ///
    /// Returns [`PixelsError::InvalidArgument`] if `bytes` is not exactly the
    /// packed length `descriptor` implies.
    pub fn from_raw(descriptor: ImageDescriptor, bytes: Vec<u8>) -> Result<Self> {
        let buffer = TileBuf::from_vec(descriptor.region(), descriptor.pixel, bytes)?;
        let source = BufferSource::new(descriptor, Arc::new(buffer))?;
        Ok(Self::from_producer(Arc::new(source), Format::Raw))
    }

    /// Build an image by decoding a raw pixel stream.
    ///
    /// The header parse is trivial for raw — the layout *is* the header — so
    /// this reads no bytes from `source`. Pixels are pulled at the terminal.
    ///
    /// # Errors
    ///
    /// Returns [`PixelsError::InvalidArgument`] if the layout is not
    /// representable on this platform.
    #[cfg(feature = "raw")]
    pub fn from_raw_stream(
        layout: RawFormat,
        source: impl Source + std::fmt::Debug + 'static,
    ) -> Result<Self> {
        let decoder = RawDecoder::new(layout, source)?;
        Ok(Self::from_decoder(Box::new(decoder), Format::Raw))
    }

    /// Open an image file, identifying its format from its contents, with
    /// default [`OpenOptions`] — so it is turned upright.
    ///
    /// The path's extension is **ignored**. Detection is by magic bytes only
    /// (SPEC §Formats), because a name is an attacker-controlled hint while
    /// the bytes are a fact.
    ///
    /// # Errors
    ///
    /// Returns [`PixelsError::Io`] if the file cannot be opened,
    /// [`PixelsError::Unsupported`] if no built-in codec recognises it, and
    /// [`PixelsError::Malformed`] if the header is invalid for the format its
    /// magic bytes claim.
    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Self> {
        Self::open_with(path, OpenOptions::default())
    }

    /// Open an image file with explicit [`OpenOptions`].
    ///
    /// # Errors
    ///
    /// As [`Image::open`].
    pub fn open_with(path: impl AsRef<std::path::Path>, options: OpenOptions) -> Result<Self> {
        let path = path.as_ref();
        let file = std::fs::File::open(path).map_err(|e| {
            // `PixelsError::io` takes a static context, so the path goes into
            // the wrapped error instead. Losing which file failed would make
            // the error useless in exactly the case it fires.
            let kind = e.kind();
            let detail = std::io::Error::new(kind, format!("{}: {e}", path.display()));
            PixelsError::io("opening image file", detail)
        })?;
        Self::from_stream_with(std::io::BufReader::new(file), options)
    }

    /// Build an image from a byte stream, identifying its format from the
    /// leading bytes.
    ///
    /// Sniffing reads only the longest magic prefix any known codec needs, and
    /// replays it to the decoder rather than seeking — a [`Source`] is
    /// forward-only (ADR-0005), so a pipe or socket works here exactly as a
    /// file does. A stream shorter than that prefix is not an error at this
    /// stage: it simply matches nothing.
    ///
    /// # Errors
    ///
    /// Returns [`PixelsError::Unsupported`] if no built-in codec recognises
    /// the stream, [`PixelsError::Io`] on read failure, or
    /// [`PixelsError::Malformed`] if the header is invalid for the format its
    /// magic bytes claim.
    pub fn from_stream(source: impl Source + std::fmt::Debug + 'static) -> Result<Self> {
        Self::from_stream_with(source, OpenOptions::default())
    }

    /// Build an image from a byte stream with explicit [`OpenOptions`].
    ///
    /// # Errors
    ///
    /// As [`Image::from_stream`].
    pub fn from_stream_with(
        mut source: impl Source + std::fmt::Debug + 'static,
        options: OpenOptions,
    ) -> Result<Self> {
        let codecs = sniffing_codecs();
        let longest = codecs.iter().map(|c| c.magic_len()).max().unwrap_or(0);

        // Short reads are normal, and a stream shorter than `longest` is a
        // legitimate no-match rather than a failure, so this loop stops at end
        // of input instead of demanding a full prefix.
        let mut prefix = Vec::with_capacity(longest);
        let mut buffer = vec![0_u8; longest];
        while prefix.len() < longest {
            let Some(rest) = buffer.get_mut(prefix.len()..) else {
                break;
            };
            match source.read(rest)? {
                0 => break,
                n => {
                    let Some(read) = rest.get(..n) else { break };
                    prefix.extend_from_slice(read);
                }
            }
        }

        let Some(codec) = codecs.iter().find(|codec| codec.probe(&prefix)) else {
            return Err(PixelsError::unsupported(format!(
                "no codec recognises this stream; its first {} bytes are {:02x?}",
                prefix.len().min(8),
                prefix.get(..prefix.len().min(8)).unwrap_or(&[])
            )));
        };
        let stream = Prefixed::new(prefix, source);
        // Unused when no decoding codec is compiled in, which is a legitimate
        // if degenerate build rather than a mistake.
        let _ = (&stream, options);

        match codec.format() {
            #[cfg(feature = "png")]
            Format::Png => {
                let decoder = PngDecoder::new(stream, options.limits)?;
                Self::decoded(Box::new(decoder), Format::Png, options)
            }
            #[cfg(feature = "gif")]
            Format::Gif => {
                let decoder = GifDecoder::new(stream, options.limits)?;
                Self::decoded(Box::new(decoder), Format::Gif, options)
            }
            #[cfg(feature = "jpeg")]
            Format::Jpeg => {
                let decoder = JpegDecoder::new(stream, options.limits)?;
                Self::decoded(Box::new(decoder), Format::Jpeg, options)
            }
            #[cfg(feature = "tiff")]
            Format::Tiff => {
                let decoder = TiffDecoder::new(stream, options.limits)?;
                Self::decoded(Box::new(decoder), Format::Tiff, options)
            }
            #[cfg(feature = "webp")]
            Format::WebP => {
                let decoder = WebPDecoder::new(stream, options.limits)?;
                Self::decoded(Box::new(decoder), Format::WebP, options)
            }
            #[cfg(feature = "avif")]
            Format::Avif => {
                let decoder = AvifDecoder::new(stream, options.limits)?;
                Self::decoded(Box::new(decoder), Format::Avif, options)
            }
            other => Err(PixelsError::unsupported(format!(
                "{other} was detected but no decoder for it is compiled in"
            ))),
        }
    }

    /// Wrap a sniffed decoder, turning it upright if `options` say so.
    ///
    /// The orientation is read before the decoder disappears into a source,
    /// which is the last point it is reachable.
    #[cfg(any(
        feature = "png",
        feature = "gif",
        feature = "jpeg",
        feature = "tiff",
        feature = "webp",
        feature = "avif"
    ))]
    fn decoded(decoder: Box<dyn Decoder>, format: Format, options: OpenOptions) -> Result<Self> {
        let orientation = decoder.orientation();
        let icc = decoder.icc_profile().map(Vec::from);
        let animation = decoder.animation().map(Arc::new);
        if options.animated && animation.is_some() {
            return Err(PixelsError::unsupported(format!(
                "{format}: multi-frame (animated) pipelines are not implemented yet; \
                 open with `animated` off to process the first frame"
            )));
        }
        let mut image = Self::from_decoder(decoder, format).with_icc_profile(icc);
        image.animation = animation;
        if options.to_srgb {
            image = image.to_srgb();
        }
        Ok(if options.auto_orient {
            image.orient(orientation)
        } else {
            image
        })
    }

    /// Build an image from any decoder whose header has already been parsed.
    ///
    /// This is the extension point for codecs living outside this crate.
    /// Pixels arrive as stored: [`Decoder::orientation`] is not applied here,
    /// so pass it to [`Image::orient`] to turn the result upright.
    #[must_use]
    pub fn from_decoder(decoder: Box<dyn Decoder>, format: Format) -> Self {
        let source = otf_pixels_core::DecodedSource::new(decoder);
        Self::from_producer(Arc::new(source), format)
    }

    /// Build an image from any pixel producer.
    #[must_use]
    pub fn from_producer(producer: Arc<dyn Producer>, format: Format) -> Self {
        Self {
            icc: None,
            animation: None,
            inner: Ok(otf_pixels_core::Image::from_producer(producer, format)),
        }
    }

    /// Header-only facts about this image: dimensions, format, pixel format.
    ///
    /// Free — descriptors are resolved as the graph is built, so this decodes
    /// nothing (SPEC §Guarantees 3).
    ///
    /// # Errors
    ///
    /// Returns the first error captured while building the pipeline, if any.
    pub fn metadata(&self) -> Result<Metadata> {
        self.graph()?.metadata()
    }

    /// The shape of this image at this point in the pipeline.
    ///
    /// # Errors
    ///
    /// Returns the first error captured while building the pipeline, if any.
    pub fn descriptor(&self) -> Result<ImageDescriptor> {
        Ok(self.graph()?.descriptor())
    }

    /// Extract the rectangular window at `(x, y)` of size `width` × `height`.
    ///
    /// A window outside the image is an error, surfaced at the terminal.
    #[must_use]
    pub fn crop(self, x: u32, y: u32, width: u32, height: u32) -> Self {
        match Crop::at(x, y, width, height) {
            Ok(op) => self.apply(Arc::new(op)),
            Err(error) => Self::failed(error),
        }
    }

    /// Mirror vertically: the top row becomes the bottom row.
    #[must_use]
    pub fn flip(self) -> Self {
        self.apply(Arc::new(Flip))
    }

    /// Mirror horizontally: the left column becomes the right column.
    #[must_use]
    pub fn flop(self) -> Self {
        self.apply(Arc::new(Flop))
    }

    /// Resample to `width` by `height` with the default filter (Lanczos3).
    #[must_use]
    pub fn resize(self, width: u32, height: u32) -> Self {
        self.resize_with(width, height, ResizeOptions::default())
    }

    /// Resample to `width` by `height` with explicit options.
    #[must_use]
    pub fn resize_with(self, width: u32, height: u32, options: ResizeOptions) -> Self {
        match Resize::new(width, height, options) {
            Ok(op) => self.apply(Arc::new(op)),
            Err(error) => Self::failed(error),
        }
    }

    /// Scale to fit inside `width` by `height`, preserving aspect ratio.
    #[must_use]
    pub fn thumbnail(self, width: u32, height: u32) -> Self {
        let options = ResizeOptions::default()
            .with_fit(Fit::Inside)
            .without_enlargement(true);
        self.resize_with(width, height, options)
    }

    /// Apply `orientation`: the stored image becomes the upright one.
    ///
    /// [`Image::open`] and [`Image::from_stream`] already do this with the
    /// orientation the file declares; this is for pixels opened with
    /// `auto_orient` off or through [`Image::from_decoder`]. It is a quarter
    /// turn and a mirror at most, and both rescale, so an oriented JPEG
    /// keeps its shrink-on-load fast path.
    #[must_use]
    pub fn orient(self, orientation: Orientation) -> Self {
        let turns = orientation.clockwise_turns();
        let image = if turns == 0 {
            self
        } else {
            self.rotate(90 * i32::from(turns))
        };
        if orientation.mirrored() {
            image.flop()
        } else {
            image
        }
    }

    /// Rotate by `degrees`, which must be a multiple of 90.
    #[must_use]
    pub fn rotate(self, degrees: i32) -> Self {
        match Rotate::degrees(degrees) {
            Ok(op) => self.apply(Arc::new(op)),
            Err(error) => Self::failed(error),
        }
    }

    /// Adjust brightness, saturation and hue.
    #[must_use]
    pub fn modulate(self, options: Modulate) -> Self {
        self.apply(Arc::new(options))
    }

    /// Convolve with `kernel`.
    #[must_use]
    pub fn convolve(self, kernel: Kernel) -> Self {
        self.apply(Arc::new(Convolve::new(kernel)))
    }

    /// Blur with a Gaussian of the given sigma.
    #[must_use]
    pub fn blur(self, sigma: f32) -> Self {
        match Kernel::gaussian(sigma) {
            Ok(kernel) => self.convolve(kernel),
            Err(error) => Self::failed(error),
        }
    }

    /// Sharpen by `amount`, a 3x3 unsharp-style kernel.
    #[must_use]
    pub fn sharpen(self, amount: f32) -> Self {
        match Kernel::sharpen(amount) {
            Ok(kernel) => self.convolve(kernel),
            Err(error) => Self::failed(error),
        }
    }

    /// Extract one channel as a greyscale image.
    #[must_use]
    pub fn extract_channel(self, index: usize) -> Self {
        self.apply(Arc::new(ExtractChannel::new(index)))
    }

    /// Composite this image against an opaque background, discarding alpha.
    #[must_use]
    pub fn flatten(self, red: u8, green: u8, blue: u8) -> Self {
        self.apply(Arc::new(Flatten::onto(red, green, blue)))
    }

    /// Draw `overlay` over this image at `(x, y)`.
    ///
    /// This is the join point for two branches of a graph: both pipelines stay
    /// lazy, and neither is evaluated until a terminal pulls on the result.
    #[must_use]
    pub fn composite(self, overlay: Self, x: i64, y: i64) -> Self {
        self.composite_with(overlay, x, y, Blend::Over)
    }

    /// Draw `overlay` over this image with an explicit blend mode.
    #[must_use]
    pub fn composite_with(self, overlay: Self, x: i64, y: i64, blend: Blend) -> Self {
        // The base's profile describes the result; the overlay's pixels are
        // composited as they are.
        let (icc, animation) = (self.icc, self.animation);
        let (base, over) = match (self.inner, overlay.inner) {
            (Ok(base), Ok(over)) => (base, over),
            // The first error wins, matching how a single chain behaves.
            (Err(error), _) | (Ok(_), Err(error)) => return Self::failed_shared(error),
        };
        let op: Arc<dyn Op> = Arc::new(Composite::at(x, y, blend));
        Self {
            inner: otf_pixels_core::Image::combine(&[base, over], op).map_err(Arc::new),
            icc,
            animation,
        }
    }

    /// A pipeline carrying an error, surfaced at the terminal.
    fn failed(error: PixelsError) -> Self {
        Self::failed_shared(Arc::new(error))
    }

    /// [`Image::failed`] with an error already shared.
    const fn failed_shared(error: Arc<PixelsError>) -> Self {
        Self {
            inner: Err(error),
            icc: None,
            animation: None,
        }
    }

    /// The source file's animation, if it has more than one frame.
    ///
    /// The pipeline processes the first frame, so this describes what the
    /// file holds rather than what will be written: frame count, loop count
    /// and per-frame durations, for a caller to decide whether a still is
    /// what it wants (see [`OpenOptions::animated`]).
    #[must_use]
    pub fn animation(&self) -> Option<&Animation> {
        self.animation.as_deref()
    }

    /// The ICC profile this image's pixels are in, if it is not sRGB.
    ///
    /// A file's embedded profile, unless the pixels were converted to sRGB
    /// on open (see [`OpenOptions`]). It is written into the output where
    /// the format has a place for one, so colours survive the round trip.
    #[must_use]
    pub fn icc_profile(&self) -> Option<&[u8]> {
        self.icc.as_deref()
    }

    /// Convert the pixels from their ICC profile's colour space to sRGB, and
    /// drop the profile.
    ///
    /// [`Image::open`] already does this unless told not to
    /// ([`OpenOptions::to_srgb`]). Matrix/TRC RGB and grey profiles convert,
    /// relative colorimetric with out-of-gamut colours clipped, as lcms2
    /// does; a profile that is sRGB in all but name is just dropped. Any
    /// other profile (LUT-based, CMYK, one that does not match the pixels)
    /// is kept, unconverted, so the output still carries it.
    #[must_use]
    pub fn to_srgb(self) -> Self {
        let Some(profile) = self.icc.clone() else {
            return self;
        };
        let Ok(descriptor) = self.descriptor() else {
            return self;
        };
        match ToSrgb::from_profile(&profile) {
            Conversion::Convert(op) if op.applies_to(descriptor.pixel) => {
                self.apply(Arc::new(op)).with_icc_profile(None)
            }
            Conversion::AlreadySrgb => self.with_icc_profile(None),
            _ => self,
        }
    }

    /// Convert to `pixel`: depth (8-bit, 16-bit, float) and layout (grey,
    /// grey with alpha, RGB, RGBA). Grey widens to RGB by repetition and RGB
    /// narrows to grey by BT.601 luma; alpha is added opaque or dropped
    /// (use [`Image::flatten`] to composite against a colour instead).
    ///
    /// Outputs need not ask for this: [`Image::output`] narrows to what the
    /// format holds by itself.
    #[must_use]
    pub fn to_pixel_format(self, pixel: PixelFormat) -> Self {
        match self.descriptor() {
            Ok(descriptor) if descriptor.pixel == pixel => self,
            _ => self.apply(Arc::new(ConvertFormat::to(pixel))),
        }
    }

    /// This image in a pixel format `format`'s encoder accepts: 8 bits for
    /// JPEG, WebP, AVIF and GIF, and integers for PNG and TIFF. A 16-bit PNG
    /// written as WebP is thereby narrowed rather than refused.
    fn encodable_as(self, format: Format) -> Self {
        let Ok(descriptor) = self.descriptor() else {
            return self;
        };
        let pixel = descriptor.pixel;
        let kind = match (format, pixel.sample_kind()) {
            (
                Format::Jpeg | Format::WebP | Format::Avif | Format::Gif,
                SampleKind::U16 | SampleKind::F32,
            ) => SampleKind::U8,
            (Format::Png | Format::Tiff, SampleKind::F32) => SampleKind::U16,
            _ => return self,
        };
        match PixelFormat::from_parts(pixel.layout(), kind) {
            Some(target) => self.to_pixel_format(target),
            None => self,
        }
    }

    /// Declare the ICC profile the pixels are in, or with `None` drop it and
    /// call them sRGB. Only the label changes, never a pixel.
    #[must_use]
    pub fn with_icc_profile(mut self, profile: Option<Vec<u8>>) -> Self {
        self.icc = profile.map(Arc::from);
        self
    }

    /// Chain an arbitrary op onto this pipeline.
    ///
    /// The escape hatch for ops defined outside this crate. Errors are
    /// deferred to the terminal, like every other chaining method.
    #[must_use]
    pub fn apply(self, op: Arc<dyn Op>) -> Self {
        Self {
            icc: self.icc,
            animation: self.animation,
            inner: match self.inner {
                Ok(image) => image.apply(op).map_err(Arc::new),
                // An earlier failure short-circuits: later ops never run.
                Err(error) => Err(error),
            },
        }
    }

    /// Choose the encoder and options for this pipeline's output.
    ///
    /// This is the single encode terminal, with format as data (ADR-0006):
    /// requesting a format that is not yet implemented is a catchable
    /// [`PixelsError::Unsupported`], not a compile error.
    #[must_use]
    pub fn output(self, format: Format, options: EncodeOptions) -> Output {
        Output {
            image: self,
            format,
            options,
            scheduler: None,
            shared: None,
        }
    }

    /// The underlying graph, or the first captured error.
    fn graph(&self) -> Result<&otf_pixels_core::Image> {
        match &self.inner {
            Ok(image) => Ok(image),
            // The error is shared, so it is rebuilt rather than moved out.
            Err(error) => Err(rebuild(error)),
        }
    }
}

/// Reconstruct an owned error from a shared one.
///
/// [`PixelsError`] is not [`Clone`] — [`std::io::Error`] is not — so a captured
/// error is rebuilt preserving its code and message. The [`ErrorCode`], which
/// is the part under semver (SPEC §Guarantees 4), is exact.
fn rebuild(error: &Arc<PixelsError>) -> PixelsError {
    let detail = error.to_string();
    match error.code() {
        ErrorCode::Io => PixelsError::io("running the pipeline", std::io::Error::other(detail)),
        ErrorCode::Malformed => PixelsError::malformed("pipeline", detail),
        ErrorCode::Unsupported => PixelsError::unsupported(detail),
        ErrorCode::InvalidArgument => PixelsError::invalid_argument("pipeline", detail),
        ErrorCode::Graph => PixelsError::graph(detail),
        ErrorCode::LimitExceeded => match **error {
            PixelsError::LimitExceeded {
                limit,
                requested,
                allowed,
            } => PixelsError::limit_exceeded(limit, requested, allowed),
            _ => PixelsError::graph(detail),
        },
        // A code added in a later version still round-trips as an error.
        _ => PixelsError::graph(detail),
    }
}

/// A pipeline with its output format chosen, ready to be pulled.
///
/// Nothing has executed yet: the terminals on this type are what pull pixels
/// through the graph.
#[derive(Debug, Clone)]
pub struct Output {
    image: Image,
    format: Format,
    options: EncodeOptions,
    /// Options for a private pool for this run, if the caller asked for one.
    scheduler: Option<SchedulerOptions>,
    /// A scheduler shared with other pipelines, if the caller supplied one.
    shared: Option<Arc<Scheduler>>,
}

impl Output {
    /// The format this output will be encoded as.
    #[must_use]
    pub const fn format(&self) -> Format {
        self.format
    }

    /// The encoder options in effect.
    #[must_use]
    pub const fn options(&self) -> EncodeOptions {
        self.options
    }

    /// Run this pipeline on a private pool of `threads` worker threads.
    ///
    /// Zero means one per available core. One gives a fully deterministic
    /// serial run, which is what the differential tests against the
    /// reference evaluator use.
    ///
    /// This is for tests, benchmarks and one-off tools. The pool is spawned
    /// for this run and joined when it ends, so a host running many outputs
    /// at once should leave this unset and let them share
    /// [`Scheduler::global`] — or pass its own with
    /// [`Output::with_scheduler`].
    #[must_use]
    pub fn threads(mut self, threads: usize) -> Self {
        self.scheduler = Some(self.scheduler.unwrap_or_default().with_threads(threads));
        self
    }

    /// Run on `scheduler` instead of [`Scheduler::global`].
    ///
    /// For a host that wants image work on a pool it sized itself — say,
    /// four threads on a sixteen-core machine — build one [`Scheduler`] at
    /// startup and pass the same one to every output. Building one per
    /// request defeats the point: each brings its own threads. Many
    /// pipelines may run on one scheduler at once, from any threads.
    /// [`Output::threads`] and [`Output::scheduler_options`] are ignored when
    /// one is set: the scheduler was configured when it was built.
    #[must_use]
    pub fn with_scheduler(mut self, scheduler: Arc<Scheduler>) -> Self {
        self.shared = Some(scheduler);
        self
    }

    /// Run this pipeline on a private pool tuned by `options`.
    ///
    /// As with [`Output::threads`], the pool exists for this run only. To
    /// tune the pool every output shares, build a [`Scheduler`] with these
    /// options once and pass it to [`Output::with_scheduler`].
    #[must_use]
    pub const fn scheduler_options(mut self, options: SchedulerOptions) -> Self {
        self.scheduler = Some(options);
        self
    }

    /// Run the pipeline, streaming encoded bytes into `sink`.
    ///
    /// Rows are encoded and written in order as they are produced. A failure
    /// anywhere fails the whole call; partial output is never reported as
    /// success (ARCHITECTURE §Failure model), though bytes already handed to
    /// `sink` are of course already gone — a caller needing all-or-nothing
    /// should write to a buffer or a temporary and commit on success.
    ///
    /// # Errors
    ///
    /// Returns [`PixelsError::Unsupported`] if the format has no encoder in
    /// this build, and otherwise any error from the pipeline or the sink.
    pub fn write(self, sink: impl Sink) -> Result<()> {
        self.write_with_stats(sink).map(|_| ())
    }

    /// Stream to `sink`, reporting what the run did.
    ///
    /// The same work as [`Output::write`], with [`RunStats`] returned instead
    /// of discarded. Use it to confirm a pipeline streamed rather than
    /// materialized, or that a JPEG thumbnail took the shrink-on-load path:
    /// `stats.reduction` is `None` when the source decoded at full size, which
    /// is the difference between a fast thumbnail and a slow one.
    ///
    /// # Errors
    ///
    /// As [`Output::write`].
    pub fn write_with_stats(self, mut sink: impl Sink) -> Result<RunStats> {
        // Rewritten before an evaluator is chosen, so the scheduler and the
        // reference evaluator are handed the same graph and keep agreeing.
        let encodable = self.image.clone().encodable_as(self.format);
        let (image, reduction) = otf_pixels_core::shrink_on_load(encodable.graph()?)?;
        let descriptor = image.descriptor();
        let mut encoder = encoder_for(self.format, self.options)?;
        encoder.set_icc_profile(self.image.icc.as_deref())?;
        encoder.write_header(&descriptor, &mut sink)?;

        // The scheduler delivers tiles; an encoder wants whole rows in order.
        // A run blocks its caller, so one started from inside a worker of the
        // global pool would hold a thread that pool needs: it gets a private
        // pool instead, as does a run whose caller asked for one.
        // A private pool is dropped, and its workers joined, when this ends.
        let scheduler = match (&self.shared, self.scheduler) {
            (Some(shared), _) => Arc::clone(shared),
            (None, Some(options)) => Arc::new(Scheduler::new(options)?),
            (None, None) if otf_pixels_core::ThreadPool::on_worker_thread() => {
                Arc::new(Scheduler::with_defaults()?)
            }
            (None, None) => Scheduler::global()?,
        };
        let mut rows = RowAssembler::new(descriptor);
        let mut stats = scheduler.run(&image, |region, tile| {
            rows.accept(region, tile, &mut |row| encoder.write_row(row, &mut sink))
        })?;
        rows.finish(&mut |row| encoder.write_row(row, &mut sink))?;
        encoder.finish(&mut sink)?;
        stats.reduction = reduction;
        Ok(stats)
    }

    /// Run the pipeline through the **reference** evaluator instead of the
    /// tile scheduler, collecting encoded bytes.
    ///
    /// The reference evaluator is single-threaded and whole-image: it holds
    /// every intermediate in full, so it is slow and its memory scales with
    /// the image. It exists because it is *obviously* correct, which makes it
    /// the oracle the scheduler is verified against — the two must produce
    /// byte-identical output for every pipeline (ROADMAP M2).
    ///
    /// Use it to verify, to debug a suspected scheduler bug, or where an image
    /// is small and determinism matters more than throughput. Prefer
    /// [`Output::bytes`] otherwise.
    ///
    /// # Errors
    ///
    /// As [`Output::bytes`].
    pub fn bytes_via_reference(self) -> Result<Vec<u8>> {
        // The same rewrite the scheduled path applies. Without it the oracle
        // would evaluate a different graph and the two would disagree wherever
        // shrink-on-load fired — which would look like a scheduler bug.
        let encodable = self.image.clone().encodable_as(self.format);
        let (image, _) = otf_pixels_core::shrink_on_load(encodable.graph()?)?;
        let descriptor = image.descriptor();
        let mut encoder = encoder_for(self.format, self.options)?;
        encoder.set_icc_profile(self.image.icc.as_deref())?;
        let mut sink = Vec::with_capacity(descriptor.byte_len().unwrap_or_default());
        encoder.write_header(&descriptor, &mut sink)?;
        otf_pixels_core::evaluate_rows(&image, |_, row| encoder.write_row(row, &mut sink))?;
        encoder.finish(&mut sink)?;
        Ok(sink)
    }

    /// Run the pipeline, collecting encoded bytes into a [`Vec`].
    ///
    /// # Errors
    ///
    /// As [`Output::write`].
    pub fn bytes(self) -> Result<Vec<u8>> {
        // Sizing the buffer up front avoids repeated growth for raw output,
        // where the encoded length is exactly the packed pixel length.
        let hint = self
            .image
            .descriptor()
            .ok()
            .and_then(|d| d.byte_len())
            .unwrap_or_default();
        let mut buffer = Vec::with_capacity(hint);
        self.write(&mut buffer)?;
        Ok(buffer)
    }
}

/// Reassembles scheduler tiles into whole rows for an encoder.
///
/// Sequential pipelines deliver full-width strips, so rows pass straight
/// through untouched — the common case costs nothing. Where a spatial op puts
/// the output on square tiles (ADR-0003), tiles arrive left-to-right within a
/// band, and a row is only complete once its band is. Those are buffered one
/// band at a time, so the cost is a band rather than an image.
#[derive(Debug)]
struct RowAssembler {
    descriptor: ImageDescriptor,
    /// The band being assembled, when tiles are narrower than the image.
    band: Option<otf_pixels_core::TileBuf>,
    /// Next row not yet emitted.
    next_row: u32,
}

impl RowAssembler {
    const fn new(descriptor: ImageDescriptor) -> Self {
        Self {
            descriptor,
            band: None,
            next_row: 0,
        }
    }

    /// Take one tile, emitting whatever rows it completes.
    fn accept(
        &mut self,
        region: Region,
        tile: &otf_pixels_core::Tile<'_>,
        emit: &mut impl FnMut(&[u8]) -> Result<()>,
    ) -> Result<()> {
        // Fast path: a full-width tile completes its own rows.
        if region.width == self.descriptor.width && self.band.is_none() {
            for y in region.y..region.y.saturating_add(region.height) {
                let row = tile
                    .row(y)
                    .ok_or_else(|| PixelsError::graph(format!("output tile is missing row {y}")))?;
                emit(row)?;
                self.next_row = y.saturating_add(1);
            }
            return Ok(());
        }

        // Narrow tile: accumulate into a full-width band, flushing the
        // previous one when a new band starts.
        let band_region = Region::new(0, region.y, self.descriptor.width, region.height);
        let starts_new_band = self
            .band
            .as_ref()
            .is_none_or(|band| band.region().y != region.y);
        if starts_new_band {
            self.flush(emit)?;
            self.band = Some(otf_pixels_core::TileBuf::zeroed(
                band_region,
                self.descriptor.pixel,
            )?);
        }
        let Some(band) = self.band.as_mut() else {
            return Err(PixelsError::graph("row band vanished"));
        };
        otf_pixels_core::copy_region(tile, &mut band.as_tile_mut()?, region)?;

        // A band is complete once its rightmost column has arrived.
        if region.right() >= u64::from(self.descriptor.width) {
            self.flush(emit)?;
        }
        Ok(())
    }

    /// Emit any buffered band's rows and drop it.
    fn flush(&mut self, emit: &mut impl FnMut(&[u8]) -> Result<()>) -> Result<()> {
        let Some(band) = self.band.take() else {
            return Ok(());
        };
        let region = band.region();
        let view = band.as_tile()?;
        for y in region.y..region.y.saturating_add(region.height) {
            let row = view
                .row(y)
                .ok_or_else(|| PixelsError::graph(format!("row band is missing row {y}")))?;
            emit(row)?;
            self.next_row = y.saturating_add(1);
        }
        Ok(())
    }

    /// Emit anything still buffered at the end of a run.
    fn finish(&mut self, emit: &mut impl FnMut(&[u8]) -> Result<()>) -> Result<()> {
        self.flush(emit)
    }
}

/// Every codec that can be sniffed, in probe order.
///
/// Raw is deliberately absent: it has no magic bytes, so it can only be
/// requested, never detected. Adding it here would make it match everything.
fn sniffing_codecs() -> Vec<Box<dyn Codec>> {
    let codecs: Vec<Box<dyn Codec>> = vec![
        #[cfg(feature = "png")]
        Box::new(PngCodec),
        #[cfg(feature = "gif")]
        Box::new(GifCodec),
        #[cfg(feature = "jpeg")]
        Box::new(JpegCodec),
        #[cfg(feature = "tiff")]
        Box::new(TiffCodec),
        #[cfg(feature = "webp")]
        Box::new(WebPCodec),
        #[cfg(feature = "avif")]
        Box::new(AvifCodec),
    ];
    codecs
}

/// Build the encoder for `format`, or report that it is not available.
fn encoder_for(format: Format, options: EncodeOptions) -> Result<Box<dyn Encoder>> {
    // Only read by codecs that have something to tune; see the comment in
    // `from_stream` for why a build with none of them still has to compile.
    let _ = &options;
    match format {
        #[cfg(feature = "raw")]
        Format::Raw => Ok(Box::new(RawEncoder::new())),
        #[cfg(feature = "png")]
        Format::Png => Ok(Box::new(PngEncoder::from_options(&options))),
        #[cfg(feature = "gif")]
        Format::Gif => Ok(Box::new(GifEncoder::from_options(&options))),
        #[cfg(feature = "jpeg")]
        Format::Jpeg => Ok(Box::new(JpegEncoder::from_options(&options))),
        #[cfg(feature = "tiff")]
        Format::Tiff => Ok(Box::new(TiffEncoder::from_options(&options))),
        #[cfg(feature = "webp")]
        Format::WebP => Ok(Box::new(WebPEncoder::from_options(&options))),
        #[cfg(feature = "avif")]
        Format::Avif => Ok(Box::new(AvifEncoder::from_options(&options))),
        #[cfg(not(feature = "raw"))]
        Format::Raw => Err(PixelsError::unsupported(
            "raw encoding requires the `raw` feature of otf-pixels",
        )),
        other => Err(PixelsError::unsupported(format!(
            "encoding {other} is not implemented yet; \
             see the ROADMAP for which milestone lands it"
        ))),
    }
}

// Gated on `raw` because almost every test here reaches pixels through the
// raw encoder, which is the only format that can express "these exact bytes".
// A build without it has nothing to compare against, so the suite compiles out
// rather than asserting less.
#[cfg(all(test, feature = "raw"))]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    reason = "tests operate on known-good values and assert shapes directly"
)]
mod tests {
    use super::*;

    fn ramp(width: u32, height: u32) -> Image {
        let descriptor = ImageDescriptor::new(width, height, PixelFormat::Gray8).unwrap();
        let len = descriptor.byte_len().unwrap();
        Image::from_raw(descriptor, (0..len).map(|i| i as u8).collect()).unwrap()
    }

    #[test]
    fn from_raw_requires_an_exactly_sized_buffer() {
        let descriptor = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
        assert!(Image::from_raw(descriptor, vec![0; 4]).is_ok());
        assert_eq!(
            Image::from_raw(descriptor, vec![0; 3]).unwrap_err().code(),
            ErrorCode::InvalidArgument
        );
        assert_eq!(
            Image::from_raw(descriptor, vec![0; 5]).unwrap_err().code(),
            ErrorCode::InvalidArgument
        );
    }

    #[test]
    fn a_chain_error_surfaces_at_the_terminal() {
        // A crop window outside the image, with more ops chained after it.
        let result = ramp(4, 4)
            .crop(3, 3, 4, 4)
            .flip()
            .flop()
            .output(Format::Raw, EncodeOptions::default())
            .bytes();
        let err = result.unwrap_err();
        assert_eq!(err.code(), ErrorCode::InvalidArgument);
    }

    #[test]
    fn an_error_short_circuits_later_ops() {
        // `crop` fails; `metadata` on the resulting pipeline reports it rather
        // than describing a shape that was never built.
        let broken = ramp(4, 4).crop(0, 0, 0, 0);
        assert!(broken.metadata().is_err());
        assert!(broken.clone().flip().descriptor().is_err());
    }

    #[test]
    fn metadata_is_free_and_reports_the_pipeline_shape() {
        let meta = ramp(8, 6).metadata().unwrap();
        assert_eq!((meta.width, meta.height), (8, 6));
        assert_eq!(meta.format, Format::Raw);
        assert_eq!(meta.pixel, PixelFormat::Gray8);
        // After a crop the shape reflects the crop, still without decoding.
        let cropped = ramp(8, 6).crop(1, 1, 3, 2).metadata().unwrap();
        assert_eq!((cropped.width, cropped.height), (3, 2));
    }

    /// Every format encodes once its feature is on (each checked by its own
    /// round-trip tests); a build without a format's feature must fail
    /// cleanly rather than produce something.
    #[cfg(not(feature = "avif"))]
    #[test]
    fn a_format_built_out_is_a_catchable_error() {
        let format = Format::Avif;
        let err = ramp(2, 2)
            .output(format, EncodeOptions::default())
            .bytes()
            .unwrap_err();
        assert_eq!(err.code(), ErrorCode::Unsupported, "{format}");
        assert!(err.to_string().contains(format.as_str()), "{err}");
    }

    /// TIFF shipped an encoder in M5 but was never added to `encoder_for`, so
    /// `output(Format::Tiff, ..)` reported the format unimplemented while the
    /// encoder sat in the crate unreachable. This is the test that would have
    /// caught it.
    #[cfg(feature = "tiff")]
    #[test]
    fn tiff_round_trips_through_the_facade() {
        let bytes = ramp(12, 9)
            .output(Format::Tiff, EncodeOptions::default())
            .bytes()
            .unwrap();
        let image = Image::from_stream(std::io::Cursor::new(bytes)).unwrap();
        let metadata = image.metadata().unwrap();
        assert_eq!(metadata.format, Format::Tiff);
        assert_eq!((metadata.width, metadata.height), (12, 9));

        // TIFF is lossless, so the pixels must come back exactly.
        let decoded = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        let expected: Vec<u8> = (0..12_u32 * 9).map(|i| i as u8).collect();
        assert_eq!(decoded, expected);
    }

    /// A JPEG written through the facade must be readable back through it,
    /// found by sniffing rather than by being told what it is.
    #[cfg(feature = "jpeg")]
    #[test]
    fn jpeg_round_trips_through_the_facade() {
        let (width, height) = (32_u32, 24_u32);
        let descriptor = ImageDescriptor::new(width, height, PixelFormat::Rgb8).unwrap();
        let pixels: Vec<u8> = (0..descriptor.byte_len().unwrap())
            .map(|i| {
                // A smooth gradient, which survives quantization well enough
                // for a tolerance this tight to mean something.
                let pixel = i / 3;
                let (x, y) = (pixel as u32 % width, pixel as u32 / width);
                match i % 3 {
                    0 => (x * 255 / width) as u8,
                    1 => (y * 255 / height) as u8,
                    _ => 128,
                }
            })
            .collect();

        let bytes = Image::from_raw(descriptor, pixels.clone())
            .unwrap()
            .output(Format::Jpeg, EncodeOptions::with_quality(95).unwrap())
            .bytes()
            .unwrap();

        let image = Image::from_stream(std::io::Cursor::new(bytes.clone())).unwrap();
        let metadata = image.metadata().unwrap();
        assert_eq!(metadata.format, Format::Jpeg, "sniffing missed the JPEG");
        assert_eq!((metadata.width, metadata.height), (width, height));
        assert_eq!(metadata.pixel, PixelFormat::Rgb8);

        let decoded = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        assert_eq!(decoded.len(), pixels.len());
        let worst = decoded
            .iter()
            .zip(&pixels)
            .map(|(&a, &b)| a.abs_diff(b))
            .max()
            .unwrap_or(0);
        assert!(worst <= 12, "worst sample differs by {worst}");
    }

    /// Overlapping band demand, which a resize always produces, must give the
    /// same pixels through the scheduler as through the reference evaluator.
    ///
    /// This is the regression test for a bug that made every streaming decode
    /// resized to more than one tile column fail outright: consecutive
    /// requests to a forward-only source overlap by the resize filter's
    /// support, and those rows are already past the stream cursor. Getting an
    /// answer at all is half of it; the other half is that the rows carried
    /// over from the retained band are the right ones, which only a comparison
    /// against the oracle can show.
    #[cfg(feature = "jpeg")]
    #[test]
    fn a_streaming_resize_matches_the_reference_evaluator() {
        for &(width, height, target) in &[
            (512_u32, 512_u32, 300_u32),
            (512, 512, 256),
            (256, 256, 200),
            (320, 240, 129),
            (64, 64, 16),
        ] {
            let source = jpeg_source(width, height);
            let scheduled = Image::from_stream(std::io::Cursor::new(source.clone()))
                .unwrap()
                .resize(target, target)
                .output(Format::Raw, EncodeOptions::default())
                .bytes()
                .unwrap_or_else(|e| panic!("{width}x{height} -> {target}: {e}"));
            let reference = Image::from_stream(std::io::Cursor::new(source))
                .unwrap()
                .resize(target, target)
                .output(Format::Raw, EncodeOptions::default())
                .bytes_via_reference()
                .unwrap();

            assert_eq!(
                scheduled.len(),
                (target * target * 3) as usize,
                "{width}x{height} -> {target}: size"
            );
            assert_eq!(
                scheduled, reference,
                "{width}x{height} -> {target}: the scheduler and the oracle disagree"
            );
        }
    }

    /// WebP round-trips through the facade, found by sniffing rather than by
    /// being told the format.
    ///
    /// Exact, because our WebP encoder is lossless — the one place in the
    /// codec set where a wrapped encoder still permits an exact assertion.
    #[cfg(feature = "webp")]
    #[test]
    fn webp_round_trips_through_the_facade() {
        let bytes = ramp(20, 12)
            .output(Format::WebP, EncodeOptions::default().with_lossless(true))
            .bytes()
            .unwrap();
        let image = Image::from_stream(std::io::Cursor::new(bytes)).unwrap();
        let metadata = image.metadata().unwrap();
        assert_eq!(metadata.format, Format::WebP, "sniffing missed the WebP");
        assert_eq!((metadata.width, metadata.height), (20, 12));
        // WebP has no greyscale mode, so one channel in comes back as three.
        assert_eq!(metadata.pixel, PixelFormat::Rgb8);

        let decoded = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        let expected: Vec<u8> = (0..20_u32 * 12)
            .flat_map(|i| {
                let value = i as u8;
                [value, value, value]
            })
            .collect();
        assert_eq!(decoded, expected, "a lossless round trip lost pixels");
    }

    /// Alpha survives a WebP round trip, which is the reason to reach for the
    /// format over JPEG in the first place.
    #[cfg(feature = "webp")]
    #[test]
    fn webp_keeps_alpha_through_the_facade() {
        let descriptor = ImageDescriptor::new(9, 7, PixelFormat::Rgba8).unwrap();
        let pixels: Vec<u8> = (0..(9 * 7))
            .flat_map(|i| [(i * 3) as u8, 40, 200, (i * 7) as u8])
            .collect();
        let bytes = Image::from_raw(descriptor, pixels.clone())
            .unwrap()
            .output(Format::WebP, EncodeOptions::default().with_lossless(true))
            .bytes()
            .unwrap();

        let image = Image::from_stream(std::io::Cursor::new(bytes)).unwrap();
        assert_eq!(image.metadata().unwrap().pixel, PixelFormat::Rgba8);
        let decoded = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        assert_eq!(decoded, pixels);
    }

    /// WebP output is lossy by default, at the requested quality: close to
    /// the source, smaller as the quality drops, and with its alpha intact,
    /// since WebP codes alpha losslessly even in a lossy file.
    #[cfg(feature = "webp")]
    #[test]
    fn webp_is_lossy_by_default_and_keeps_alpha_exact() {
        let (w, h) = (48_u32, 32_u32);
        let descriptor = ImageDescriptor::new(w, h, PixelFormat::Rgba8).unwrap();
        let pixels: Vec<u8> = (0..w * h)
            .flat_map(|i| {
                let (x, y) = (i % w, i / w);
                [(x * 5) as u8, (y * 7) as u8, 120, (x * 3 + y) as u8]
            })
            .collect();
        let encode = |options: EncodeOptions| {
            Image::from_raw(descriptor, pixels.clone())
                .unwrap()
                .output(Format::WebP, options)
                .bytes()
                .unwrap()
        };
        let good = encode(EncodeOptions::default());
        let rough = encode(EncodeOptions::with_quality(10).unwrap());
        assert!(
            rough.len() < good.len(),
            "{} vs {}",
            rough.len(),
            good.len()
        );

        let decoded = Image::from_stream(std::io::Cursor::new(good))
            .unwrap()
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        for (ours, source) in decoded.chunks_exact(4).zip(pixels.chunks_exact(4)) {
            assert_eq!(ours[3], source[3], "alpha changed");
            for c in 0..3 {
                assert!(ours[c].abs_diff(source[c]) <= 12, "{ours:?} vs {source:?}");
            }
        }
    }

    /// Shrink-on-load, end to end: a thumbnail pipeline must decode the JPEG
    /// at a reduced scale rather than at full size and then throw it away.
    #[cfg(feature = "jpeg")]
    #[test]
    fn a_thumbnail_pipeline_shrinks_the_jpeg_on_load() {
        // 512x512 down to 32x32: 1/8 covers it exactly.
        let source = jpeg_source(512, 512);

        let image = Image::from_stream(std::io::Cursor::new(source)).unwrap();
        assert_eq!(
            image.metadata().unwrap().width,
            512,
            "metadata is full size"
        );

        // Evaluated once: a stream-backed graph cannot be run twice, because
        // the bytes are gone the first time.
        let mut bytes = Vec::new();
        let stats = image
            .resize(32, 32)
            .output(Format::Raw, EncodeOptions::default())
            .write_with_stats(&mut bytes)
            .unwrap();

        let reduction = stats
            .reduction
            .expect("a 512 to 32 resize should shrink the source on load");
        assert_eq!(reduction.from, (512, 512));
        assert_eq!(
            reduction.to,
            (64, 64),
            "1/8 is the coarsest scale that still covers 32"
        );
        assert!((reduction.factor() - 64.0).abs() < 0.01);
        // And the pixels are still there, at the size that was asked for.
        assert_eq!(bytes.len(), 32 * 32 * 3);
    }

    /// A crop names coordinates in source pixels, so shrinking underneath it
    /// would return a different part of the picture.
    #[cfg(feature = "jpeg")]
    #[test]
    fn a_crop_blocks_shrink_on_load() {
        let source = jpeg_source(512, 512);
        let stats = Image::from_stream(std::io::Cursor::new(source))
            .unwrap()
            .crop(256, 256, 128, 128)
            .resize(32, 32)
            .output(Format::Raw, EncodeOptions::default())
            .write_with_stats(&mut Vec::new())
            .unwrap();
        assert!(
            stats.reduction.is_none(),
            "a crop must block shrink-on-load: {:?}",
            stats.reduction
        );
    }

    /// Cropping *after* a resize is fine — the crop is then in the resized
    /// image's coordinates, which the reduction does not move.
    #[cfg(feature = "jpeg")]
    #[test]
    fn a_crop_after_the_resize_still_allows_shrinking() {
        let source = jpeg_source(512, 512);
        let stats = Image::from_stream(std::io::Cursor::new(source))
            .unwrap()
            .resize(64, 64)
            .crop(0, 0, 32, 32)
            .output(Format::Raw, EncodeOptions::default())
            .write_with_stats(&mut Vec::new())
            .unwrap();
        // The crop is still not scale-covariant, so this conservatively does
        // not fire. Recorded as the current behaviour rather than asserted as
        // desirable: a crop below a resize could be allowed, and is not.
        assert!(stats.reduction.is_none());
    }

    /// A JPEG whose pipeline only resizes a little must not be shrunk past
    /// what it needs.
    #[cfg(feature = "jpeg")]
    #[test]
    fn a_mild_resize_shrinks_by_less_or_not_at_all() {
        let source = jpeg_source(512, 512);
        // 300 needs more than half of 512, so no scale fits.
        let stats = Image::from_stream(std::io::Cursor::new(source.clone()))
            .unwrap()
            .resize(300, 300)
            .output(Format::Raw, EncodeOptions::default())
            .write_with_stats(&mut Vec::new())
            .unwrap();
        assert!(stats.reduction.is_none(), "{:?}", stats.reduction);

        // 200 fits inside 1/2 (256) but not 1/4 (128).
        let stats = Image::from_stream(std::io::Cursor::new(source))
            .unwrap()
            .resize(200, 200)
            .output(Format::Raw, EncodeOptions::default())
            .write_with_stats(&mut Vec::new())
            .unwrap();
        assert_eq!(stats.reduction.map(|r| r.to), Some((256, 256)));
    }

    /// A JPEG of `size` square, as bytes.
    #[cfg(feature = "jpeg")]
    fn jpeg_source(width: u32, height: u32) -> Vec<u8> {
        let descriptor = ImageDescriptor::new(width, height, PixelFormat::Rgb8).unwrap();
        let pixels: Vec<u8> = (0..descriptor.byte_len().unwrap())
            .map(|i| {
                let pixel = i / 3;
                let (x, y) = (pixel as u32 % width, pixel as u32 / width);
                match i % 3 {
                    0 => (x * 255 / width) as u8,
                    1 => (y * 255 / height) as u8,
                    _ => 96,
                }
            })
            .collect();
        Image::from_raw(descriptor, pixels)
            .unwrap()
            .output(Format::Jpeg, EncodeOptions::with_quality(85).unwrap())
            .bytes()
            .unwrap()
    }

    /// The pipeline a thumbnail actually is: decode, resize, encode.
    #[cfg(feature = "jpeg")]
    #[test]
    fn a_jpeg_pipeline_resizes_and_re_encodes() {
        let descriptor = ImageDescriptor::new(64, 64, PixelFormat::Rgb8).unwrap();
        let pixels: Vec<u8> = (0..descriptor.byte_len().unwrap())
            .map(|i| ((i / 3) % 251) as u8)
            .collect();
        let source = Image::from_raw(descriptor, pixels)
            .unwrap()
            .output(Format::Jpeg, EncodeOptions::default())
            .bytes()
            .unwrap();

        let thumbnail = Image::from_stream(std::io::Cursor::new(source.clone()))
            .unwrap()
            .resize(16, 16)
            .output(Format::Jpeg, EncodeOptions::with_quality(70).unwrap())
            .bytes()
            .unwrap();

        let metadata = Image::from_stream(std::io::Cursor::new(thumbnail.clone()))
            .unwrap()
            .metadata()
            .unwrap();
        assert_eq!((metadata.width, metadata.height), (16, 16));
        assert!(
            thumbnail.len() < source.len(),
            "a 16x16 thumbnail is {} bytes against a 64x64 source's {}",
            thumbnail.len(),
            source.len()
        );
    }

    /// Greyscale must stay greyscale through the facade rather than being
    /// widened to RGB on the way out.
    #[cfg(feature = "jpeg")]
    #[test]
    fn grayscale_jpeg_keeps_one_channel_through_the_facade() {
        let bytes = ramp(24, 16)
            .output(Format::Jpeg, EncodeOptions::with_quality(90).unwrap())
            .bytes()
            .unwrap();
        let metadata = Image::from_stream(std::io::Cursor::new(bytes))
            .unwrap()
            .metadata()
            .unwrap();
        assert_eq!(metadata.pixel, PixelFormat::Gray8);
        assert_eq!((metadata.width, metadata.height), (24, 16));
    }

    #[test]
    fn write_streams_into_any_sink() {
        let mut sink = Vec::new();
        ramp(2, 2)
            .output(Format::Raw, EncodeOptions::default())
            .write(&mut sink)
            .unwrap();
        assert_eq!(sink, [0, 1, 2, 3]);
    }

    #[test]
    fn output_reports_its_settings() {
        let options = EncodeOptions::with_quality(55).unwrap();
        let output = ramp(2, 2).output(Format::Raw, options);
        assert_eq!(output.format(), Format::Raw);
        assert_eq!(output.options().quality, 55);
    }

    #[cfg(feature = "raw")]
    #[test]
    fn from_raw_stream_defers_decoding_to_the_terminal() {
        let descriptor = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
        let layout = RawFormat::packed(descriptor);
        let cursor = std::io::Cursor::new(vec![1_u8, 2, 3, 4]);
        let image = Image::from_raw_stream(layout, cursor).unwrap();
        // Header facts are available without reading pixels.
        assert_eq!(image.metadata().unwrap().width, 2);
        let bytes = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        assert_eq!(bytes, [1, 2, 3, 4]);
    }

    #[cfg(feature = "raw")]
    #[test]
    fn a_truncated_stream_fails_the_terminal_without_panicking() {
        let descriptor = ImageDescriptor::new(4, 4, PixelFormat::Gray8).unwrap();
        let layout = RawFormat::packed(descriptor);
        let cursor = std::io::Cursor::new(vec![1_u8; 5]);
        let image = Image::from_raw_stream(layout, cursor).unwrap();
        let err = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap_err();
        assert_eq!(err.code(), ErrorCode::Malformed);
    }

    #[test]
    fn images_are_send_sync_and_cheap_to_clone() {
        const fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<Image>();
        assert_send_sync::<Output>();
        let image = ramp(4, 4);
        let clone = image.clone();
        assert_eq!(
            image.metadata().unwrap().width,
            clone.metadata().unwrap().width
        );
    }

    #[cfg(feature = "png")]
    #[test]
    fn a_png_round_trips_through_the_facade() {
        let png = ramp(37, 21)
            .output(Format::Png, EncodeOptions::default())
            .bytes()
            .unwrap();
        let image = Image::from_stream(std::io::Cursor::new(png)).unwrap();
        assert_eq!(image.descriptor().unwrap().width, 37);
        assert_eq!(image.metadata().unwrap().format, Format::Png);

        let back = image
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        let expected = ramp(37, 21)
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        assert_eq!(back, expected, "pixels changed across a PNG round trip");
    }

    #[cfg(feature = "png")]
    #[test]
    fn a_pipeline_survives_a_png_round_trip() {
        // The point of sniffing is that a decoded image is an ordinary graph
        // source, so ops compose over it exactly as over raw pixels.
        let png = ramp(16, 16)
            .output(Format::Png, EncodeOptions::default())
            .bytes()
            .unwrap();
        let cropped = Image::from_stream(std::io::Cursor::new(png))
            .unwrap()
            .crop(2, 3, 8, 5)
            .flip()
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        let direct = ramp(16, 16)
            .crop(2, 3, 8, 5)
            .flip()
            .output(Format::Raw, EncodeOptions::default())
            .bytes()
            .unwrap();
        assert_eq!(cropped, direct);
    }

    #[cfg(feature = "png")]
    #[test]
    fn open_ignores_the_extension_and_reads_the_bytes() {
        let dir = std::env::temp_dir().join(format!("otf-pixels-open-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("actually-a-png.jpg");
        let png = ramp(8, 8)
            .output(Format::Png, EncodeOptions::default())
            .bytes()
            .unwrap();
        std::fs::write(&path, &png).unwrap();

        let image = Image::open(&path).unwrap();
        assert_eq!(
            image.metadata().unwrap().format,
            Format::Png,
            "extension won over content"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn open_names_the_file_it_could_not_read() {
        let error = Image::open("/nonexistent/otf-pixels/missing.png").unwrap_err();
        assert_eq!(error.code(), ErrorCode::Io);
        assert!(error.to_string().contains("missing.png"), "{error}");
    }

    #[test]
    fn an_unrecognised_stream_is_unsupported_not_a_guess() {
        for bytes in [&b""[..], &b"not an image"[..], &[0_u8; 64][..]] {
            let error = Image::from_stream(std::io::Cursor::new(bytes.to_vec())).unwrap_err();
            assert_eq!(error.code(), ErrorCode::Unsupported, "{bytes:02x?}");
        }
    }

    #[cfg(feature = "png")]
    #[test]
    fn a_truncated_png_is_malformed_not_a_panic() {
        let png = ramp(8, 8)
            .output(Format::Png, EncodeOptions::default())
            .bytes()
            .unwrap();
        // Past the signature, so sniffing succeeds and the header parse is
        // what has to fail cleanly.
        for cut in [9, 16, 24, 32, png.len() - 1] {
            let truncated = png[..cut].to_vec();
            let result = Image::from_stream(std::io::Cursor::new(truncated))
                .and_then(|i| i.output(Format::Raw, EncodeOptions::default()).bytes());
            assert!(result.is_err(), "truncating to {cut} bytes should fail");
        }
    }

    #[cfg(feature = "png")]
    #[test]
    fn sniffing_reads_only_the_magic_bytes_before_deciding() {
        // A stream that is exactly the signature and nothing else must be
        // recognised as PNG and then fail on its missing header, proving the
        // sniff does not need — or read — more than the magic.
        let error = Image::from_stream(std::io::Cursor::new(
            otf_pixels_codec_png::SIGNATURE.to_vec(),
        ))
        .unwrap_err();
        assert_eq!(error.code(), ErrorCode::Malformed, "{error}");
    }
}

/// The handles an embedder shares across threads must stay shareable: a
/// runtime builds a pipeline on one thread and runs it on a worker.
const _: () = {
    const fn shareable<T: Send + Sync>() {}
    shareable::<Image>();
    shareable::<Output>();
    shareable::<OpenOptions>();
    shareable::<PixelsError>();
    shareable::<Scheduler>();
};