fleischwolf-pdf 0.15.0

//! PDF backend for fleischwolf.
//!
//! A port of docling's standard PDF pipeline: pdfium extracts the text layer
//! (cells with bounding boxes) and renders page images; a discriminative ONNX
//! stack (layout detection, table structure, OCR) classifies regions; the cells
//! are assembled in reading order into a [`DoclingDocument`].
//!
//! Current stages: pdfium text-cell extraction + page rendering ([`pdfium_backend`])
//! and the deterministic text/reading-order assembly ([`assemble`]). The layout,
//! table-structure and OCR ONNX stages land behind [`Pipeline`] next.

mod assemble;
mod dp_lines;
pub mod layout;
mod mets;
mod ocr;
pub mod pdfium_backend;
pub mod resample;
pub mod tableformer;
pub mod textparse;
pub mod timing;

use std::collections::BTreeMap;
use std::fmt;
use std::sync::mpsc::{sync_channel, Receiver};
use std::sync::{Arc, Mutex};

use fleischwolf_core::{DoclingDocument, Node};

pub use mets::{convert_mets_gbs, convert_mets_gbs_with_options};
pub use pdfium_backend::{PdfDocument, PdfPage, TextCell};

/// Errors from the PDF backend. Detailed and surfaced (never silently skipped).
#[derive(Debug)]
pub enum PdfError {
    /// pdfium failed to bind, open, or read the document.
    Pdfium(String),
    /// The layout ONNX model failed to load or run.
    Layout(String),
    /// The OCR ONNX model failed to load or run.
    Ocr(String),
}

impl fmt::Display for PdfError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PdfError::Pdfium(m) => write!(f, "pdf: pdfium error: {m}"),
            PdfError::Layout(m) => write!(f, "pdf: {m}"),
            PdfError::Ocr(m) => write!(f, "pdf: {m}"),
        }
    }
}

impl std::error::Error for PdfError {}

impl From<pdfium_render::prelude::PdfiumError> for PdfError {
    fn from(e: pdfium_render::prelude::PdfiumError) -> Self {
        PdfError::Pdfium(e.to_string())
    }
}

/// Threads ONNX inference may use, capped by `FLEISCHWOLF_PDF_THREADS` if set.
/// Defaults to the available parallelism (ort otherwise picks a low number).
pub(crate) fn intra_threads() -> usize {
    if let Some(n) = std::env::var("FLEISCHWOLF_PDF_THREADS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|&n| n > 0)
    {
        return n;
    }
    std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(1)
}

/// True when `FLEISCHWOLF_FP32` (any value but `0`) forces the full-precision
/// models even where an INT8 variant sits next to the fp32 default.
pub(crate) fn fp32_forced() -> bool {
    std::env::var("FLEISCHWOLF_FP32")
        .map(|v| v != "0")
        .unwrap_or(false)
}

/// Resolve a default (CWD-relative) asset path. If it doesn't exist relative
/// to the current directory, try next to the executable and one level above
/// it (following symlinks — the layout `scripts/install.sh` produces:
/// `/usr/local/bin/fleischwolf` → `/usr/local/fleischwolf/bin/fleischwolf`
/// with `models/` and `.pdfium/` in `/usr/local/fleischwolf`). Lets an
/// installed binary run from any working directory with no env vars; explicit
/// env overrides never reach this. Returns `rel` unchanged when nothing
/// exists anywhere, so callers' error messages keep the familiar path.
pub(crate) fn resolve_asset(rel: &str) -> String {
    if std::path::Path::new(rel).exists() {
        return rel.to_string();
    }
    if let Some(dir) = std::env::current_exe()
        .ok()
        .and_then(|p| p.canonicalize().ok())
        .and_then(|p| p.parent().map(std::path::Path::to_path_buf))
    {
        for base in [Some(dir.as_path()), dir.parent()].into_iter().flatten() {
            let p = base.join(rel);
            if p.exists() {
                return p.to_string_lossy().into_owned();
            }
        }
    }
    rel.to_string()
}

/// Resolve a model path: an explicit env override always wins; otherwise the
/// INT8 variant of the default path when it exists on disk (the quantized
/// models are conformance-validated — see PDF_PERFORMANCE.md — and load/run
/// markedly faster on CPU), unless `FLEISCHWOLF_FP32` opts back into full
/// precision; else the fp32 default.
pub(crate) fn model_path(env: &str, fp32_default: &str, int8_default: &str) -> String {
    if let Ok(p) = std::env::var(env) {
        return p;
    }
    if !fp32_forced() {
        let p = resolve_asset(int8_default);
        if std::path::Path::new(&p).exists() {
            return p;
        }
    }
    resolve_asset(fp32_default)
}

/// One page's assembled output: typed nodes plus the page's hyperlinks, kept
/// separate so pages processed out of order can be stitched back in page order.
type PageOut = (Vec<Node>, Vec<(String, String)>);

/// The pool-wide TableFormer slot: one instance shared by every worker, loaded
/// lazily on the first table region any worker sees. Tables appear on a
/// minority of pages, so per-worker copies mostly multiplied ~0.4 GB of
/// weights+arenas by the pool size for nothing; a single shared instance keeps
/// the peak flat regardless of pool width, and a table's structure prediction
/// is independent of which worker runs it, so output is byte-identical. The
/// mutex serialises concurrent tables — the shared instance is loaded with the
/// full intra-op thread budget to compensate (one wide TableFormer instead of
/// several narrow ones).
enum TfSlot {
    /// Not attempted yet (no table seen so far).
    Unloaded,
    /// Load attempted, graphs absent — geometric fallback (warned once).
    Missing,
    Ready(tableformer::TableFormer),
}

type SharedTables = Arc<Mutex<TfSlot>>;

/// A self-contained set of the per-page models (layout, OCR). Each parallel
/// page-worker owns its own `Worker` so inference runs concurrently without
/// sharing an ONNX session (`ort`'s `Session::run` is `&mut self`); only the
/// rarely-hit TableFormer is shared (see [`TfSlot`]).
struct Worker {
    /// `None` when `no_ocr` skips layout entirely — no model load, no inference.
    layout: Option<layout::LayoutModel>,
    ocr: Option<ocr::OcrModel>,
    /// Shared TableFormer slot; `None` when `no_table_former`/`no_ocr` skip it.
    tables: Option<SharedTables>,
    /// Skip layout, OCR, and TableFormer; reconstruct text purely from the PDF's
    /// embedded text layer. See [`Pipeline::no_ocr`].
    no_ocr: bool,
}

impl Worker {
    fn load(intra: usize, tables: Option<SharedTables>, no_ocr: bool) -> Result<Self, PdfError> {
        Ok(Self {
            layout: if no_ocr {
                None
            } else {
                Some(layout::LayoutModel::load_with(intra).map_err(PdfError::Layout)?)
            },
            ocr: None,
            tables,
            no_ocr,
        })
    }

    /// Run layout (+ OCR for cell-less pages) + TableFormer and assemble page `n`
    /// into its nodes and links. Pure given the page (mutates only the worker's
    /// lazily-loaded OCR model), so it is safe to run concurrently across pages.
    fn process(&mut self, n: usize, page: &mut PdfPage) -> Result<PageOut, PdfError> {
        if self.no_ocr {
            // Fastest path: no layout/OCR/TableFormer inference at all. The PDF's
            // embedded text cells (if any) become flat, line-grouped paragraphs in
            // reading order via the same orphan-region machinery that normally
            // rescues text the detector missed — here it rescues *all* of it.
            // Pages with no embedded text layer (scanned/image-only) yield nothing;
            // convert those without `no_ocr`.
            let mut regions = Vec::new();
            assemble::add_orphan_regions(&mut regions, &page.cells);
            let table_rows = vec![None; regions.len()];
            return Ok(timing::timed("assemble_page", || {
                assemble::assemble_page(page, regions, &table_rows)
            }));
        }
        let regions = timing::timed("layout.predict", || {
            self.layout
                .as_mut()
                .expect("layout model loaded unless no_ocr")
                .predict(&page.image, page.width, page.height)
        })
        .map_err(|e| PdfError::Layout(format!("page {}: {e}", n + 1)))?;
        // Resolve overlapping detections once, before OCR.
        let mut regions = assemble::resolve(regions);
        // Emit text the detector missed as orphan text regions (docling parity).
        assemble::add_orphan_regions(&mut regions, &page.cells);
        // Drop phantom empty low-confidence picture boxes (docling parity).
        assemble::drop_false_pictures(&mut regions, &page.cells, page.width, page.height);
        // No text layer → recognise text from the page image via OCR.
        if page.cells.is_empty() {
            if self.ocr.is_none() {
                self.ocr = Some(ocr::OcrModel::load().map_err(PdfError::Ocr)?);
            }
            let cells = timing::timed("ocr.page", || {
                self.ocr
                    .as_mut()
                    .unwrap()
                    .ocr_page(&page.image, &regions, page.scale)
            })
            .map_err(|e| PdfError::Ocr(format!("page {}: {e}", n + 1)))?;
            page.cells = cells;
        }
        // TableFormer structure per table region (else geometric fallback). The
        // shared slot is only locked (and lazily loaded) when the page actually
        // has a table, so table-free documents never pay for TableFormer at all.
        let mut table_rows: Vec<Option<Vec<Vec<String>>>> = vec![None; regions.len()];
        if let Some(slot) = self.tables.as_ref() {
            if regions.iter().any(|r| r.label == "table") {
                timing::timed("tableformer", || {
                    let mut guard = slot.lock().unwrap();
                    if matches!(*guard, TfSlot::Unloaded) {
                        // Full intra-op width: tables serialise on this mutex, so
                        // the one instance gets the whole thread budget.
                        *guard = match tableformer::TableFormer::load_with(intra_threads()) {
                            Some(tf) => TfSlot::Ready(tf),
                            None => TfSlot::Missing,
                        };
                    }
                    if let TfSlot::Ready(tf) = &mut *guard {
                        for (i, r) in regions.iter().enumerate() {
                            if r.label == "table" {
                                table_rows[i] = tf.predict_table_rows(
                                    &page.image,
                                    page.height,
                                    [r.l, r.t, r.r, r.b],
                                    &page.word_cells,
                                );
                            }
                        }
                    }
                });
            }
        }
        Ok(timing::timed("assemble_page", || {
            assemble::assemble_page(page, regions, &table_rows)
        }))
    }
}

/// Per-worker ONNX intra-op threads. The layout model is memory-bandwidth bound,
/// so on a typical machine two threads per worker (sharing one in-cache copy of
/// the weights) extracts more throughput than one fat model or many single-thread
/// workers. `FLEISCHWOLF_PDF_INTRA` overrides for per-machine tuning.
fn pdf_intra() -> usize {
    if let Some(n) = std::env::var("FLEISCHWOLF_PDF_INTRA")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|&n| n > 0)
    {
        return n;
    }
    if intra_threads() >= 2 {
        2
    } else {
        1
    }
}

/// How many page-workers to spin up for a multi-page PDF. `FLEISCHWOLF_PDF_WORKERS`
/// overrides; otherwise size the pool so `workers × intra ≈ cores`, capped at 4 so
/// a worst-case pool holds a bounded amount of model memory (~0.4 GB per worker)
/// and does not oversaturate the memory bus with model-weight traffic.
fn pdf_worker_count() -> usize {
    if let Some(n) = std::env::var("FLEISCHWOLF_PDF_WORKERS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|&n| n > 0)
    {
        return n;
    }
    (intra_threads() / pdf_intra()).clamp(1, 4)
}

/// Minimum page count before a PDF is worth the parallel worker pool. Below this,
/// the serial primary (running its model on every core) is faster than fanning out
/// — the helper pool's one-time model-load cost only pays off once enough pages
/// share it. `FLEISCHWOLF_PDF_PARALLEL_MIN` overrides.
fn pdf_parallel_min() -> usize {
    std::env::var("FLEISCHWOLF_PDF_PARALLEL_MIN")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|&n| n > 0)
        .unwrap_or(6)
}

/// A reusable PDF pipeline. The **primary** worker runs its models on every core,
/// so a single-page / small / image / METS input is converted at full intra-op
/// speed with no pool to load. A document with enough pages instead fans out
/// across a **pool** of narrower workers processed concurrently. Both load lazily
/// and are cached for reuse, so a one-shot conversion only pays for what it uses.
pub struct Pipeline {
    /// Full-intra worker for the serial path; loaded on first serial use.
    primary: Option<Worker>,
    /// Narrower workers (≈cores/`target_workers` threads each) for the parallel
    /// path; loaded on first multi-page use and cached.
    pool: Vec<Worker>,
    /// The single TableFormer instance every worker shares (see [`TfSlot`]).
    tables: SharedTables,
    /// Desired pool size for multi-page documents.
    target_workers: usize,
    /// Page count at/above which the parallel pool is worth its load cost.
    parallel_min: usize,
    /// Skip loading/running TableFormer; table regions fall back to geometric
    /// reconstruction. See [`Pipeline::no_table_former`].
    no_table_former: bool,
    /// Skip layout, OCR, and TableFormer entirely. See [`Pipeline::no_ocr`].
    no_ocr: bool,
}

impl Pipeline {
    /// Construct the pipeline. Models load lazily on first use (full-intra primary
    /// for serial inputs, the helper pool for multi-page PDFs), so nothing is
    /// loaded that a given document doesn't need.
    pub fn new() -> Result<Self, PdfError> {
        Ok(Self {
            primary: None,
            pool: Vec::new(),
            tables: Arc::new(Mutex::new(TfSlot::Unloaded)),
            target_workers: pdf_worker_count(),
            parallel_min: pdf_parallel_min(),
            no_table_former: false,
            no_ocr: false,
        })
    }

    /// Skip loading and running the TableFormer table-structure model. Table
    /// regions still get emitted, but reconstructed geometrically from cell
    /// positions instead of via the ONNX model's predicted structure — faster
    /// (no model load, no per-table inference) at the cost of table fidelity.
    /// No effect if a worker is already loaded; set this before the first
    /// conversion.
    pub fn no_table_former(mut self, disable: bool) -> Self {
        self.no_table_former = disable;
        self
    }

    /// Skip layout detection, OCR, and TableFormer entirely — no model load, no
    /// inference of any kind. The PDF's embedded text cells are grouped by line
    /// and emitted as plain paragraphs in reading order: no headings, lists,
    /// tables, code blocks, or pictures, since that structure comes from the
    /// layout model. The fastest possible PDF path, but pages with no embedded
    /// text layer (scanned/image-only PDFs) yield no text at all — convert those
    /// without this flag. Implies `no_table_former`. No effect if a worker is
    /// already loaded; set this before the first conversion.
    pub fn no_ocr(mut self, disable: bool) -> Self {
        self.no_ocr = disable;
        self
    }

    /// The shared TableFormer slot handed to each worker, or `None` when the
    /// pipeline options skip TableFormer entirely.
    fn tables_slot(&self) -> Option<SharedTables> {
        if self.no_table_former || self.no_ocr {
            None
        } else {
            Some(Arc::clone(&self.tables))
        }
    }

    /// The full-intra serial worker, loaded on first use.
    fn primary(&mut self) -> Result<&mut Worker, PdfError> {
        if self.primary.is_none() {
            self.primary = Some(Worker::load(
                intra_threads(),
                self.tables_slot(),
                self.no_ocr,
            )?);
        }
        Ok(self.primary.as_mut().unwrap())
    }

    /// Convert a PDF (bytes) to a [`DoclingDocument`]. A document with fewer than
    /// `parallel_min` pages (or a pool size of 1) streams through the full-intra
    /// primary; a larger one renders on this thread (pdfium is not thread-safe) and
    /// fans the pages out across the worker pool, reassembled in page order so the
    /// output is byte-identical to the serial path.
    pub fn convert(
        &mut self,
        bytes: &[u8],
        password: Option<&str>,
        name: &str,
    ) -> Result<DoclingDocument, PdfError> {
        let pages = pdfium_backend::page_count(bytes, password)?;
        let doc = if self.target_workers >= 2 && pages >= self.parallel_min {
            self.convert_parallel(bytes, password, name)?
        } else {
            self.convert_serial(bytes, password, name)?
        };
        timing::report();
        Ok(doc)
    }

    /// Stream pages one at a time through the primary worker — render → process →
    /// drop — so the document holds ~one page bitmap (~5 MB) at a time.
    fn convert_serial(
        &mut self,
        bytes: &[u8],
        password: Option<&str>,
        name: &str,
    ) -> Result<DoclingDocument, PdfError> {
        let mut doc = DoclingDocument::new(name);
        let render_image = !self.no_ocr;
        let worker = self.primary()?;
        pdfium_backend::for_each_page(bytes, password, render_image, |n, _total, mut page| {
            let (nodes, links) = worker.process(n, &mut page)?;
            doc.nodes.extend(nodes);
            doc.links.extend(links);
            Ok::<(), PdfError>(())
        })?;
        assemble::merge_continuations(&mut doc.nodes);
        Ok(doc)
    }

    /// Render pages serially on this thread (pdfium) and process them in parallel
    /// across the worker pool. A bounded channel applies backpressure so only a
    /// handful of page bitmaps are resident at once; results carry their page
    /// index and are reassembled in order, so the output is byte-identical to the
    /// serial path.
    fn convert_parallel(
        &mut self,
        bytes: &[u8],
        password: Option<&str>,
        name: &str,
    ) -> Result<DoclingDocument, PdfError> {
        self.ensure_pool()?;
        let n_workers = self.pool.len();
        let render_image = !self.no_ocr;
        let (work_tx, work_rx) = sync_channel::<(usize, PdfPage)>(n_workers * 2);
        let work_rx: Arc<Mutex<Receiver<(usize, PdfPage)>>> = Arc::new(Mutex::new(work_rx));
        let results: Arc<Mutex<Vec<(usize, PageOut)>>> = Arc::new(Mutex::new(Vec::new()));
        let first_err: Arc<Mutex<Option<PdfError>>> = Arc::new(Mutex::new(None));

        // Move the pool into the scope so each worker gets an exclusive `&mut`.
        let mut workers = std::mem::take(&mut self.pool);
        std::thread::scope(|s| {
            for worker in workers.iter_mut() {
                let work_rx = Arc::clone(&work_rx);
                let results = Arc::clone(&results);
                let first_err = Arc::clone(&first_err);
                s.spawn(move || loop {
                    // Hold the receiver lock only for the recv; release before the
                    // (long) per-page work so other workers can pull concurrently.
                    let item = work_rx.lock().unwrap().recv();
                    let Ok((idx, mut page)) = item else { break };
                    match worker.process(idx, &mut page) {
                        Ok(out) => results.lock().unwrap().push((idx, out)),
                        Err(e) => {
                            let mut slot = first_err.lock().unwrap();
                            if slot.is_none() {
                                *slot = Some(e);
                            }
                        }
                    }
                });
            }
            // Render on this thread and feed the workers; backpressure blocks here
            // when the channel is full. Dropping `work_tx` afterwards signals the
            // workers (recv → Err) to finish.
            let render =
                pdfium_backend::for_each_page(bytes, password, render_image, |i, _total, page| {
                    work_tx
                        .send((i, page))
                        .map_err(|_| PdfError::Pdfium("page-worker channel closed".into()))
                });
            drop(work_tx);
            if let Err(e) = render {
                let mut slot = first_err.lock().unwrap();
                if slot.is_none() {
                    *slot = Some(e);
                }
            }
        });
        // Threads have joined; restore the pool for the next conversion.
        self.pool = workers;

        if let Some(e) = first_err.lock().unwrap().take() {
            return Err(e);
        }
        let mut results = Arc::try_unwrap(results)
            .unwrap_or_else(|arc| Mutex::new(arc.lock().unwrap().clone()))
            .into_inner()
            .unwrap();
        results.sort_by_key(|(idx, _)| *idx);
        let mut doc = DoclingDocument::new(name);
        for (_, (nodes, links)) in results {
            doc.nodes.extend(nodes);
            doc.links.extend(links);
        }
        assemble::merge_continuations(&mut doc.nodes);
        Ok(doc)
    }

    /// Convert a PDF in **streaming** mode: `emit` is called with each finalized,
    /// in-document-order batch of nodes (and that span's recovered links) as pages
    /// complete, so a caller can serialize Markdown page by page instead of waiting
    /// for the whole document. The batches are exactly the buffered [`convert`]'s
    /// nodes, split at safe block boundaries by [`assemble::StreamAssembler`] — the
    /// parallel path reorders pages back into document order before emitting, so
    /// the output is identical regardless of worker scheduling.
    ///
    /// `emit` runs on the calling thread (never a worker), so it needn't be `Send`
    /// and its backpressure throttles the whole pipeline. Returning `Err` from
    /// `emit` aborts the conversion with that error.
    pub fn convert_streaming<F>(
        &mut self,
        bytes: &[u8],
        password: Option<&str>,
        name: &str,
        emit: F,
    ) -> Result<(), PdfError>
    where
        F: FnMut(Vec<Node>, Vec<(String, String)>) -> Result<(), PdfError>,
    {
        let _ = name; // page nodes carry no name; the caller owns the document name.
        let pages = pdfium_backend::page_count(bytes, password)?;
        let r = if self.target_workers >= 2 && pages >= self.parallel_min {
            self.convert_streaming_parallel(bytes, password, emit)
        } else {
            self.convert_streaming_serial(bytes, password, emit)
        };
        timing::report();
        r
    }

    /// Serial streaming: render → process → emit, one page at a time, holding back
    /// only the tail that might still merge into the next page.
    fn convert_streaming_serial<F>(
        &mut self,
        bytes: &[u8],
        password: Option<&str>,
        mut emit: F,
    ) -> Result<(), PdfError>
    where
        F: FnMut(Vec<Node>, Vec<(String, String)>) -> Result<(), PdfError>,
    {
        let mut asm = assemble::StreamAssembler::new();
        let render_image = !self.no_ocr;
        let worker = self.primary()?;
        pdfium_backend::for_each_page(bytes, password, render_image, |n, _total, mut page| {
            let (nodes, links) = worker.process(n, &mut page)?;
            emit(asm.push(nodes), links)
        })?;
        emit(asm.finish(), Vec::new())
    }

    /// Parallel streaming: pages render serially on a dedicated thread (pdfium is
    /// not thread-safe) and process across the worker pool; results carry their
    /// page index and are reordered on the calling thread into a
    /// [`assemble::StreamAssembler`], which emits each page in document order as
    /// soon as its predecessors have arrived. Bounded channels keep only a handful
    /// of pages resident and let `emit`'s backpressure reach the renderer.
    fn convert_streaming_parallel<F>(
        &mut self,
        bytes: &[u8],
        password: Option<&str>,
        mut emit: F,
    ) -> Result<(), PdfError>
    where
        F: FnMut(Vec<Node>, Vec<(String, String)>) -> Result<(), PdfError>,
    {
        self.ensure_pool()?;
        let n_workers = self.pool.len();
        let render_image = !self.no_ocr;
        let (work_tx, work_rx) = sync_channel::<(usize, PdfPage)>(n_workers * 2);
        let work_rx: Arc<Mutex<Receiver<(usize, PdfPage)>>> = Arc::new(Mutex::new(work_rx));
        // Workers and the renderer report here; the calling thread drains it in
        // page order. Bounded so workers block (bounding resident bitmaps) when the
        // consumer falls behind.
        let (res_tx, res_rx) = sync_channel::<Result<(usize, PageOut), PdfError>>(n_workers * 2);

        let mut workers = std::mem::take(&mut self.pool);
        let mut asm = assemble::StreamAssembler::new();
        let mut first_err: Option<PdfError> = None;

        std::thread::scope(|s| {
            // Workers: pull a page, process it, report (index-tagged) result.
            for worker in workers.iter_mut() {
                let work_rx = Arc::clone(&work_rx);
                let res_tx = res_tx.clone();
                s.spawn(move || loop {
                    let item = work_rx.lock().unwrap().recv();
                    let Ok((idx, mut page)) = item else { break };
                    let out = worker.process(idx, &mut page).map(|o| (idx, o));
                    if res_tx.send(out).is_err() {
                        break; // consumer gone
                    }
                });
            }
            // Renderer: feed pages to the pool on its own thread (pdfium stays on a
            // single thread); report a render error through the same channel.
            {
                let res_tx = res_tx.clone();
                s.spawn(move || {
                    let render = pdfium_backend::for_each_page(
                        bytes,
                        password,
                        render_image,
                        |i, _total, page| {
                            work_tx
                                .send((i, page))
                                .map_err(|_| PdfError::Pdfium("page-worker channel closed".into()))
                        },
                    );
                    drop(work_tx); // signal workers to finish
                    if let Err(e) = render {
                        let _ = res_tx.send(Err(e));
                    }
                });
            }
            // Drop our own sender so the channel closes once the threads finish.
            drop(res_tx);

            // Collector (this thread): reorder into document order and emit.
            let mut buffer: BTreeMap<usize, PageOut> = BTreeMap::new();
            let mut next = 0usize;
            for msg in res_rx.iter() {
                match msg {
                    Err(e) => {
                        if first_err.is_none() {
                            first_err = Some(e);
                        }
                    }
                    Ok((idx, out)) => {
                        buffer.insert(idx, out);
                        if first_err.is_some() {
                            continue; // keep draining so the threads can exit
                        }
                        while let Some((nodes, links)) = buffer.remove(&next) {
                            if let Err(e) = emit(asm.push(nodes), links) {
                                first_err = Some(e);
                                break;
                            }
                            next += 1;
                        }
                    }
                }
            }
        });
        // Threads have joined; restore the pool for the next conversion.
        self.pool = workers;

        if let Some(e) = first_err {
            return Err(e);
        }
        emit(asm.finish(), Vec::new())
    }

    /// Lazily grow the pool to `target_workers`, loading the new workers
    /// concurrently (model load is mostly I/O + mmap, so N loads overlap to roughly
    /// one load's wall-time). Cached for reuse across documents.
    fn ensure_pool(&mut self) -> Result<(), PdfError> {
        let need = self.target_workers.saturating_sub(self.pool.len());
        if need == 0 {
            return Ok(());
        }
        let intra = pdf_intra();
        let no_ocr = self.no_ocr;
        let tables = self.tables_slot();
        let loaded: Vec<Result<Worker, PdfError>> = std::thread::scope(|s| {
            let handles: Vec<_> = (0..need)
                .map(|_| {
                    let tables = tables.clone();
                    s.spawn(move || Worker::load(intra, tables, no_ocr))
                })
                .collect();
            handles.into_iter().map(|h| h.join().unwrap()).collect()
        });
        for w in loaded {
            self.pool.push(w?);
        }
        Ok(())
    }

    /// Convert a standalone image (PNG/JPEG/TIFF/WebP/…) as a single page —
    /// docling routes images through the same layout+OCR pipeline as a PDF page.
    pub fn convert_image(&mut self, bytes: &[u8], name: &str) -> Result<DoclingDocument, PdfError> {
        let image = image::load_from_memory(bytes)
            .map_err(|e| PdfError::Pdfium(format!("image: {e}")))?
            .into_rgb8();
        let (w, h) = image.dimensions();
        // The image is its own page rendered at 1 px per "point" (scale 1.0); a
        // standalone image has no text layer, so OCR supplies the cells.
        let page = PdfPage {
            width: w as f32,
            height: h as f32,
            scale: 1.0,
            cells: Vec::new(),
            code_cells: Vec::new(),
            word_cells: Vec::new(),
            image,
            links: Vec::new(),
        };
        self.process_pages(vec![page], name)
    }

    /// Run layout (+ OCR for cell-less pages) and assemble each already-rendered
    /// page (image / METS inputs, which are small and already materialised).
    fn process_pages(
        &mut self,
        mut pages: Vec<PdfPage>,
        name: &str,
    ) -> Result<DoclingDocument, PdfError> {
        let mut doc = DoclingDocument::new(name);
        let worker = self.primary()?;
        for (n, page) in pages.iter_mut().enumerate() {
            let (nodes, links) = worker.process(n, page)?;
            doc.nodes.extend(nodes);
            doc.links.extend(links);
        }
        assemble::merge_continuations(&mut doc.nodes);
        Ok(doc)
    }
}

/// Convenience one-shot conversion (loads the pipeline per call). Errors are
/// detailed and surfaced (never silently skipped).
pub fn convert(
    bytes: &[u8],
    password: Option<&str>,
    name: &str,
) -> Result<DoclingDocument, PdfError> {
    convert_with_options(bytes, password, name, false, false)
}

/// Like [`convert`], but optionally skips loading/running TableFormer (see
/// [`Pipeline::no_table_former`]) and/or layout+OCR+TableFormer entirely (see
/// [`Pipeline::no_ocr`]).
pub fn convert_with_options(
    bytes: &[u8],
    password: Option<&str>,
    name: &str,
    no_table_former: bool,
    no_ocr: bool,
) -> Result<DoclingDocument, PdfError> {
    Pipeline::new()?
        .no_table_former(no_table_former)
        .no_ocr(no_ocr)
        .convert(bytes, password, name)
}

/// Convenience one-shot image conversion (loads the pipeline per call).
pub fn convert_image(bytes: &[u8], name: &str) -> Result<DoclingDocument, PdfError> {
    convert_image_with_options(bytes, name, false, false)
}

/// Like [`convert_image`], but optionally skips loading/running TableFormer (see
/// [`Pipeline::no_table_former`]) and/or layout+OCR+TableFormer entirely (see
/// [`Pipeline::no_ocr`]).
pub fn convert_image_with_options(
    bytes: &[u8],
    name: &str,
    no_table_former: bool,
    no_ocr: bool,
) -> Result<DoclingDocument, PdfError> {
    Pipeline::new()?
        .no_table_former(no_table_former)
        .no_ocr(no_ocr)
        .convert_image(bytes, name)
}

/// Convert pre-segmented pages (image + already-known text cells, e.g. METS/hOCR
/// scans) through the shared layout + assembly pipeline.
pub fn convert_pages(pages: Vec<PdfPage>, name: &str) -> Result<DoclingDocument, PdfError> {
    convert_pages_with_options(pages, name, false, false)
}

/// Like [`convert_pages`], but optionally skips loading/running TableFormer (see
/// [`Pipeline::no_table_former`]) and/or layout+OCR+TableFormer entirely (see
/// [`Pipeline::no_ocr`]).
pub fn convert_pages_with_options(
    pages: Vec<PdfPage>,
    name: &str,
    no_table_former: bool,
    no_ocr: bool,
) -> Result<DoclingDocument, PdfError> {
    Pipeline::new()?
        .no_table_former(no_table_former)
        .no_ocr(no_ocr)
        .process_pages(pages, name)
}