franken_ocr 0.10.0

//! `franken_ocr` — a pure-Rust, CPU-hyper-optimized runner for the Baidu
//! Unlimited-OCR model, with no general ML framework.
//!
//! See [`COMPREHENSIVE_PLAN_FOR_FRANKEN_OCR.md`] for the master plan and
//! `AGENTS.md` for the engineering doctrine. The public surface is the
//! synchronous, blocking [`OcrEngine`] (plan §3.3, G6) plus the `focr` CLI; the
//! heavy model forward, the model-specific int8/int4 kernels, and the weight
//! converter land across Phases 1–4. The end-to-end pipeline is **wired** here
//! (preprocess → vision → connector → decoder → sampler → postprocess) over the
//! [`native_engine`] modules; stages whose `.focrq` tensor accessors are not yet
//! built surface a clean [`FocrError::NotImplemented`] rather than fabricating
//! output (doctrine #1).
//!
//! [`COMPREHENSIVE_PLAN_FOR_FRANKEN_OCR.md`]: ../docs/planning/COMPREHENSIVE_PLAN_FOR_FRANKEN_OCR.md
#![cfg_attr(target_arch = "aarch64", allow(stable_features))]
#![cfg_attr(
    target_arch = "aarch64",
    feature(stdarch_neon_dotprod, stdarch_neon_i8mm)
)]
#![deny(unsafe_code)]

pub mod adaptive;
#[cfg(feature = "native")]
pub mod batch_events;
#[cfg(feature = "native")]
pub mod cli;
#[cfg(feature = "native")]
pub mod conformance;
#[cfg(feature = "native")]
pub mod dist;
#[cfg(feature = "native")]
pub mod doctor;
pub mod error;
pub mod native_engine;
pub mod page_source;
#[cfg(feature = "pdf")]
pub mod pdf;
pub mod preprocess;
pub mod progress;
pub mod quant;
#[cfg(feature = "native")]
pub mod resident;
#[cfg(feature = "native")]
pub mod robot;
pub mod simd;
#[cfg(feature = "native")]
pub mod storage;
pub mod tall;
pub mod tokenizer;

#[cfg(feature = "native")]
pub use cli::cli_main;
pub use error::{FocrError, FocrResult};
pub use native_engine::DocumentResult;
pub use native_engine::checkpoint::{ResumeKey, RunFiles};
/// Multi-model architecture descriptors + registry (the "model zoo" foundation,
/// epic bd-3jo6 / A1). Additive metadata layer; the live forward is unchanged.
pub use native_engine::model_arch;
pub use native_engine::{
    DocumentEvent, DocumentOptions, ExtractedFigure, Half, LayoutSpan, PageFinish, PageOrigin,
    PageResult, PageWindow, RecognizedDocument, ResumePolicy, StreamedPage,
};

/// The version of the document output contract, `docs/library-output.md`. A new
/// field keeps it; a changed meaning, unit or order bumps it.
pub const OUTPUT_CONTRACT_VERSION: u32 = 1;

#[cfg(feature = "native")]
use std::cell::RefCell;
#[cfg(feature = "native")]
use std::path::Path;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
#[cfg(feature = "native")]
use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError};
#[cfg(feature = "native")]
use std::time::Duration;

#[cfg(feature = "native")]
use asupersync::runtime::{Runtime, RuntimeBuilder};
#[cfg(feature = "native")]
use native_engine::OcrModel;

/// The pinned Unlimited-OCR release-artifact version every released binary
/// resolves (`unlimited-ocr.v{VERSION}.<quant>.focrq`). Lives at the crate
/// root (not in the network-gated `dist` module) because model *resolution*
/// needs it even in builds without the pull machinery (the wasm core).
pub(crate) const UNLIMITED_OCR_ARTIFACT_VERSION: &str = "0.7.0";

/// Environment override for the model artifact path (`.focrq` blob or a
/// safetensors directory). When unset, [`OcrEngine`] falls back to
/// [`DEFAULT_MODEL_PATH`].
pub const MODEL_PATH_ENV: &str = "FOCR_MODEL_PATH";

/// Source-code license notice for this crate, surfaced in the long version
/// report separately from the model-weights notice.
pub const FOCR_PROJECT_LICENSE_NOTICE: &str =
    "franken_ocr - Copyright (c) 2026 Jeffrey Emanuel, MIT License (with OpenAI/Anthropic Rider)";

/// Baidu Unlimited-OCR model-weights notice. This is the single source of truth
/// for the notice that must travel with redistributed `.focrq` artifacts and
/// agent-facing provenance surfaces (plan §2.2 / §11).
pub const FOCR_MODEL_LICENSE_NOTICE: &str =
    "Baidu Unlimited-OCR - Copyright (c) 2026 Baidu, MIT License";

/// Default model artifact location when [`MODEL_PATH_ENV`] is unset (plan §7.5).
/// A relative `models/unlimited-ocr.focrq` next to the working directory; the
/// model-gated e2e tests deliberately point this at `/nonexistent` to prove the
/// native path's clean [`FocrError::ModelNotFound`].
pub const DEFAULT_MODEL_PATH: &str = "models/unlimited-ocr.focrq";

const DEFAULT_FORWARD_STAGE_BUDGET_MS: u64 = 10 * 60 * 1000;

/// The per-input forward stage budget (`FOCR_STAGE_BUDGET_FORWARD_MS`), read
/// by the engine's stage timeout and by Unlimited-OCR-2's document loop,
/// which checks it cooperatively (U2-B6), in every build: an embedder that
/// drives [`native_engine::OcrModel`] directly gets the same cooperative
/// budget.
///
/// `None` means "no budget": `0` or the literal `unlimited` disables it
/// (GH #10), and the stage then runs until completion or cooperative
/// cancellation. Unset or unparsable values use the 10-minute default.
pub(crate) fn forward_stage_budget() -> Option<std::time::Duration> {
    let millis = match std::env::var("FOCR_STAGE_BUDGET_FORWARD_MS") {
        Ok(raw) => {
            let trimmed = raw.trim();
            if trimmed == "0" || trimmed.eq_ignore_ascii_case("unlimited") {
                return None;
            }
            trimmed
                .parse::<u64>()
                .ok()
                .filter(|&ms| ms > 0)
                .unwrap_or(DEFAULT_FORWARD_STAGE_BUDGET_MS)
        }
        Err(_) => DEFAULT_FORWARD_STAGE_BUDGET_MS,
    };
    Some(std::time::Duration::from_millis(millis))
}

/// The OCR engine handle.
///
/// Per the proven `franken_whisper` integration (plan §3.3) this **OWNS exactly
/// one** `asupersync` [`Runtime`] and exposes a **synchronous, blocking** API:
/// public methods run the heavy work via `runtime.block_on(...)`, so the async
/// runtime is an implementation detail never leaked to the host (satisfies G6).
/// The model forward fans out across the [`thread_budget`] cores via the frankentorch
/// kernel's own rayon pool, driven from a **sequential** outer page loop — never
/// nest rayon under a held lock, never nest a second runtime (doctrine #5).
///
/// The loaded [`OcrModel`] is cached behind a [`Mutex<Option<Arc<…>>>`] so the
/// 6.67 GB weight blob is read once per engine and shared across calls. The
/// global weak cache in [`native_engine`] additionally de-dups across engines in
/// one process.
// ── Cooperative shutdown (bd-223.2) ─────────────────────────────────────────
//
// The process-global shutdown flag IS the ShutdownController: Ctrl+C (the CLI
// installs the handler in `cli_main`) or an embedder's `request_shutdown()`
// sets it; every long loop in the engine — the per-page loops and every
// per-decode-step loop — polls it via [`cancel_checkpoint`] (one relaxed
// atomic load per token: unmeasurable) and returns [`FocrError::Cancelled`]
// (exit 6) at the next boundary. Cancellation is COOPERATIVE by design:
// `spawn_blocking` closures keep running on drop, so the flag is observed
// INSIDE them (doctrine #5 / the franken_whisper pattern). Per-request tokens
// for embedders who need independent cancellation of concurrent engines are a
// documented follow-up — a single flag matches the one-live-forward discipline.
static SHUTDOWN_REQUESTED: AtomicBool = AtomicBool::new(false);

/// Request cooperative shutdown: every in-flight recognition aborts with
/// [`FocrError::Cancelled`] at its next checkpoint (page boundary or decode
/// step), except an Unlimited-OCR-2 document, which returns the pages it
/// completed with `finish_reason: cancelled` (U2-E2). Idempotent; never
/// blocks.
pub fn request_shutdown() {
    SHUTDOWN_REQUESTED.store(true, Ordering::SeqCst);
}

/// Whether cooperative shutdown has been requested.
#[must_use]
pub fn shutdown_requested() -> bool {
    SHUTDOWN_REQUESTED.load(Ordering::Relaxed)
}

/// Clear the shutdown flag (tests + long-lived embedders that survive a
/// cancelled batch and start a new one).
pub fn reset_shutdown() {
    SHUTDOWN_REQUESTED.store(false, Ordering::SeqCst);
}

/// The cooperative cancellation checkpoint (bd-223.2): call at every page
/// boundary and decode step.
///
/// # Errors
/// [`FocrError::Cancelled`] once [`request_shutdown`] has been called, and
/// [`FocrError::Timeout`] once this thread's stage is past its budget.
pub fn cancel_checkpoint() -> FocrResult<()> {
    if shutdown_requested() {
        return Err(FocrError::Cancelled);
    }
    #[cfg(feature = "native")]
    if RUNNING_STAGE.with_borrow(|stage| stage.as_ref().is_some_and(|s| s.stop_requested())) {
        return Err(FocrError::Timeout("the stage's budget ran out".into()));
    }
    Ok(())
}

// ── Stage budgets (bd-3ttp) ─────────────────────────────────────────────────
//
// A budget ending does not end its closure, for the reason above, and a decode
// left running keeps every core from the next input, which then times out too.
// So a stage past its budget is told to stop and is waited for. The request
// goes through a thread-local, not the shutdown flag: it reaches only that
// stage's own decode, on the blocking thread every decode loop runs on, and
// it reads as a timeout, not as a cancel.

/// One blocking stage's stop request, and the proof that it ended.
#[cfg(feature = "native")]
#[derive(Default)]
struct StageStop {
    stop: AtomicBool,
    ended: Mutex<bool>,
    ended_changed: Condvar,
}

#[cfg(feature = "native")]
impl StageStop {
    fn stop_requested(&self) -> bool {
        self.stop.load(Ordering::Relaxed)
    }

    /// Ask the stage to stop and wait up to `grace` for it: vision and prefill
    /// run unbroken, and a stage stuck between checkpoints must not hold the
    /// timeout. Returns whether it ended.
    fn halt(&self, grace: Duration) -> bool {
        self.stop.store(true, Ordering::Relaxed);
        let ended = self.ended.lock().unwrap_or_else(PoisonError::into_inner);
        let (ended, _) = self
            .ended_changed
            .wait_timeout_while(ended, grace, |ended| !*ended)
            .unwrap_or_else(PoisonError::into_inner);
        *ended
    }
}

#[cfg(feature = "native")]
thread_local! {
    /// The stage on this thread, cleared when it ends: blocking threads are reused.
    static RUNNING_STAGE: RefCell<Option<Arc<StageStop>>> = const { RefCell::new(None) };
}

/// Holds [`RUNNING_STAGE`] for a stage's closure, and marks it ended on drop,
/// an unwind included.
#[cfg(feature = "native")]
struct RunningStage(Arc<StageStop>);

#[cfg(feature = "native")]
impl RunningStage {
    fn enter(stage: Arc<StageStop>) -> Self {
        RUNNING_STAGE.with_borrow_mut(|running| *running = Some(Arc::clone(&stage)));
        Self(stage)
    }
}

#[cfg(feature = "native")]
impl Drop for RunningStage {
    fn drop(&mut self) {
        RUNNING_STAGE.with_borrow_mut(|running| *running = None);
        *self.0.ended.lock().unwrap_or_else(PoisonError::into_inner) = true;
        self.0.ended_changed.notify_all();
    }
}

// ── The one thread/CPU budget (bd-223.2 addendum; plan §7.5) ────────────────

/// The budget, latched by [`set_thread_budget`] or the first [`thread_budget`].
static THREAD_BUDGET: OnceLock<usize> = OnceLock::new();

/// The single process-wide thread budget, read ONCE: [`set_thread_budget`]
/// (`--threads`), else `FOCR_THREADS` (env), else [`default_thread_budget`]:
/// PHYSICAL cores of the fastest tier (hyperthreads oversubscribe the int8
/// GEMMs — never `available_parallelism`). Every pool-sizing consumer (the
/// kernel rayon pool, the gauntlet fairness pins, `robot health`) reads THIS.
pub fn thread_budget() -> usize {
    *THREAD_BUDGET.get_or_init(|| {
        std::env::var("FOCR_THREADS")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .filter(|&n| n > 0)
            .unwrap_or_else(default_thread_budget)
    })
}

/// Fix the thread budget for this process, ahead of `FOCR_THREADS`. It must
/// run before anything reads the budget: the kernel pool is sized once.
///
/// # Errors
/// [`FocrError::Usage`] when the budget was already read or set to another
/// value.
pub fn set_thread_budget(threads: std::num::NonZeroUsize) -> FocrResult<()> {
    let wanted = threads.get();
    match *THREAD_BUDGET.get_or_init(|| wanted) {
        latched if latched == wanted => Ok(()),
        latched => Err(FocrError::Usage(format!(
            "--threads {wanted}: this process already runs with {latched}"
        ))),
    }
}

/// `--timing`: [`timing_enabled`] without `FOCR_TIMING` in the environment.
static TIMING: AtomicBool = AtomicBool::new(false);

/// Turn the stage-timing rows on for this process.
pub fn enable_timing() {
    TIMING.store(true, Ordering::Relaxed);
}

/// Whether stage-timing rows are on: `--timing` or `FOCR_TIMING`.
#[must_use]
pub fn timing_enabled() -> bool {
    TIMING.load(Ordering::Relaxed) || std::env::var_os("FOCR_TIMING").is_some()
}

/// The no-override default width. Physical cores (hyperthreads oversubscribe
/// the int8 GEMMs), and on a two-tier Mac only the fastest tier (bd-zex4):
/// every parallel section is a fork-join over equal shares, so the slower tier
/// sets the pace while the fast cores spin in the kernel. On an M5 Ultra (12
/// Super + 24 Performance cores, as macOS names its tiers) shared with other
/// work, all 36 took 2.4-2.6x the wall of 12 and about 6x the CPU time, 81% of
/// it in sys, for byte-identical output (`artifacts/perf/bd-zex4`).
///
/// iOS keeps its own rule, one core left to the UI: an A-series part is 2
/// performance + 4 efficiency cores, the app's own main thread has to stay
/// responsive during a minutes-long forward, and a team barrier waits for its
/// slowest member — so claiming every core makes the forward contend with the
/// UI thread that is drawing its own progress bar. It is unmeasured on device
/// (bd-v7ry), so the macOS result does not carry over.
fn default_thread_budget() -> usize {
    let physical = num_cpus::get_physical();
    if cfg!(target_os = "ios") {
        physical.saturating_sub(1).max(1)
    } else {
        fastest_tier_cores()
            .filter(|&fast| fast <= physical)
            .unwrap_or(physical)
    }
}

/// Physical cores in the fastest tier (`hw.perflevel0.physicalcpu`), on a Mac
/// that reports tiers; `None` off macOS. macOS does not call that tier the
/// same thing everywhere: on the M5 Ultra it is Super and the slower tier is
/// Performance, so "performance cores" would name the wrong ones. A one-tier
/// Mac (Intel) reports every core there or nothing, so its default is
/// unchanged either way. Read through `/usr/sbin/sysctl` like `hw.memsize`:
/// the crate denies `unsafe`.
fn fastest_tier_cores() -> Option<usize> {
    #[cfg(target_os = "macos")]
    {
        let output = std::process::Command::new("/usr/sbin/sysctl")
            .args(["-n", "hw.perflevel0.physicalcpu"])
            .output()
            .ok()?;
        if !output.status.success() {
            return None;
        }
        String::from_utf8(output.stdout)
            .ok()?
            .trim()
            .parse()
            .ok()
            .filter(|&n: &usize| n > 0)
    }
    #[cfg(not(target_os = "macos"))]
    {
        None
    }
}

/// Install the process-wide kernel rayon pool at [`thread_budget()`] width,
/// returning the width actually in force.
///
/// Two things this fixes, both invisible until you look:
///
/// 1. **The budget was documented but never installed.** `thread_budget()` says
///    every pool-sizing consumer reads it, but nothing ever handed it to rayon,
///    so the kernels ran on rayon's own default (`available_parallelism`, i.e.
///    LOGICAL cores). On a non-SMT Apple part those agree and the bug is
///    invisible; on an SMT x86 host it silently oversubscribes the int8 GEMMs
///    that doctrine §7.5 says must not be oversubscribed.
///
/// 2. **Apple demotes un-classified threads to the efficiency cores.** A thread
///    that never asks for a QoS class is fair game for the E-cores. Every
///    parallel section here is a fork-join over `par_chunks_mut`, so ONE demoted
///    worker sets the pace of the whole dispatch. Each worker asks for the same
///    class the caller's work runs at.
///
/// Idempotent and never fatal: if a global pool already exists (another
/// embedder built one first) the existing pool stands and its width is
/// returned. Call before the first forward; `kernel_pool_width()` calls it.
pub fn init_kernel_pool() -> usize {
    static INIT: OnceLock<usize> = OnceLock::new();
    *INIT.get_or_init(|| {
        let width = thread_budget();
        // An Err here means a global pool was already installed — a legitimate
        // embedder choice, not our call to override. Fall through and report.
        let _ = rayon::ThreadPoolBuilder::new()
            .num_threads(width)
            .thread_name(|i| format!("focr-kernel-{i}"))
            .start_handler(|_| apple_qos::pin_worker_to_user_initiated())
            .build_global();
        rayon::current_num_threads()
    })
}

/// Apple thread-QoS island. The only `unsafe` here is one `pthread` call that
/// takes no pointer and returns an ignored status code.
mod apple_qos {
    /// Ask for `QOS_CLASS_USER_INITIATED` on the calling thread.
    ///
    /// A no-op off Apple platforms. Errors are deliberately ignored: failing to
    /// get a QoS class costs throughput, never correctness, and a kernel worker
    /// is not a place to fail a forward from.
    #[cfg(target_vendor = "apple")]
    #[allow(unsafe_code)]
    pub(super) fn pin_worker_to_user_initiated() {
        // SAFETY: `pthread_set_qos_class_self_np` acts on the calling thread
        // only, takes a scalar class and a scalar relative priority (no
        // pointers, no borrowed memory, nothing to outlive the call), and is
        // documented as callable from any thread at any time. The return value
        // is an errno-style status we intentionally discard.
        unsafe {
            libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INITIATED, 0);
        }
    }

    #[cfg(not(target_vendor = "apple"))]
    pub(super) fn pin_worker_to_user_initiated() {}
}

/// The kernel rayon pool's CURRENT width — the diagnostic the capacity
/// certificate (bd-re8.18) records before/after the many-pages soak to prove
/// no second pool was spawned and the width never grew mid-run (the N×
/// oversubscription class doctrine #5 forbids). First call instantiates the
/// global pool at the documented budget, which is exactly what the kernels
/// themselves use.
pub fn kernel_pool_width() -> usize {
    init_kernel_pool()
}

// ── Bounded per-page result streaming (bd-223.2 scaffold) ───────────────────

/// Stream page results from a SEQUENTIAL producer to a consumer through a
/// BOUNDED channel — the bd-223.2 streaming scaffold the robot/NDJSON
/// multi-page path adopts: the producer runs on its own thread and BLOCKS
/// when the consumer lags (backpressure — memory never grows unbounded);
/// the consumer loop drains with a short `recv_timeout` so it can interleave
/// its own bookkeeping. Pages are produced STRICTLY sequentially (doctrine
/// #5 — streaming the OUTPUT of sequential pages, never concurrent
/// forwards).
///
/// `produce` yields `Some(item)` per page and `None` when exhausted;
/// `consume` receives each item in order. Returns the number of items
/// streamed.
///
/// # Errors
/// The producer's first error aborts the stream and is returned after the
/// worker joins (consumers see only the items produced before it).
#[cfg(feature = "native")]
pub fn stream_pages<T, P, C>(capacity: usize, mut produce: P, mut consume: C) -> FocrResult<usize>
where
    T: Send + 'static,
    P: FnMut() -> FocrResult<Option<T>> + Send + 'static,
    C: FnMut(T),
{
    let (tx, rx) = std::sync::mpsc::sync_channel::<T>(capacity.max(1));
    let worker = std::thread::Builder::new()
        .name("focr-page-stream".into())
        .spawn(move || -> FocrResult<()> {
            loop {
                match produce()? {
                    Some(item) => {
                        // A closed receiver means the consumer is gone —
                        // treat as cancellation, not success.
                        if tx.send(item).is_err() {
                            return Err(FocrError::Cancelled);
                        }
                    }
                    None => return Ok(()),
                }
            }
        })
        .map_err(|e| FocrError::Other(anyhow::anyhow!("page-stream worker spawn: {e}")))?;

    let mut n = 0usize;
    loop {
        match rx.recv_timeout(Duration::from_millis(40)) {
            Ok(item) => {
                consume(item);
                n += 1;
            }
            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                if worker.is_finished() {
                    // Drain anything raced in between finish and the check.
                    while let Ok(item) = rx.try_recv() {
                        consume(item);
                        n += 1;
                    }
                    break;
                }
            }
            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
        }
    }
    worker
        .join()
        .map_err(|_| FocrError::Other(anyhow::anyhow!("page-stream worker panicked")))??;
    Ok(n)
}

/// Boxed streaming sink for document decodes (bd-2z0y, U2-D5): called with
/// each [`DocumentEvent`] as the decode reaches it. `Send` because the pass
/// runs on the engine's blocking pool.
pub type DocumentSink = Box<dyn FnMut(&DocumentEvent<'_>) + Send>;

/// Told each stage a run enters: its name (`load`, `forward`) and its budget,
/// if it has one (the robot `stage` event, plan §7.3).
#[cfg(feature = "native")]
pub type StageSink = Box<dyn Fn(&'static str, Option<Duration>) + Send>;

#[cfg(feature = "native")]
pub struct OcrEngine {
    /// The single owned async runtime. All public methods block on it.
    runtime: Runtime,
    /// The lazily-loaded, shared model (one read-only weight blob per engine).
    model: Mutex<Option<Arc<OcrModel>>>,
    /// Who hears about the stages this engine enters.
    stage_sink: Mutex<Option<StageSink>>,
}

#[cfg(feature = "native")]
impl OcrEngine {
    /// Take (consume) the staff-level metadata from the most recent TrOMR
    /// music forward on this engine's cached model, if any (bd-av64.2): the
    /// recognized staves' detection indices + page-space bboxes and any
    /// per-staff skips. Returns `None` when no model is loaded, the loaded
    /// model has run no music forward since the last take, or the last
    /// forward was not a music run. The CLI uses this to emit robot `staff`
    /// events and the `--json` `staves` array.
    #[must_use]
    pub fn take_music_page_meta(&self) -> Option<native_engine::MusicPageMeta> {
        self.model
            .lock()
            .ok()
            .and_then(|slot| slot.as_ref().map(std::sync::Arc::clone))
            .and_then(|model| model.take_music_meta())
    }

    /// Take (consume) the OneChart number head and verdict of the run on this
    /// engine's cached model, if any.
    #[must_use]
    pub fn take_chart_page_meta(&self) -> Option<native_engine::ChartPageMeta> {
        self.model
            .lock()
            .ok()
            .and_then(|slot| slot.as_ref().map(std::sync::Arc::clone))
            .and_then(|model| model.take_chart_meta())
    }

    /// Take (consume) the [`native_engine::DecodeReport`] of the most recent
    /// Unlimited-OCR decode on this engine's cached model, if any (U2-X2): its
    /// `finish_reason` and, for a multi-page run, expected pages vs page
    /// bodies. The CLI turns it into `finish_reason` and `warnings[]`
    /// ([`native_engine::decode_warnings`]).
    #[must_use]
    pub fn take_last_decode_report(&self) -> Option<native_engine::DecodeReport> {
        self.model
            .lock()
            .ok()
            .and_then(|slot| slot.as_ref().map(std::sync::Arc::clone))
            .and_then(|model| model.take_last_decode())
    }

    /// Take (consume) one [`native_engine::DecodeReport`] per image of the
    /// last [`OcrEngine::recognize_batch`] on this engine's cached model, in
    /// input order; `None` where an image has none (a dense-model batch, or a
    /// page that failed before its decode). Each image's report is here
    /// instead, so after a batch `take_last_decode_report` has nothing to
    /// return (bd-iv8d).
    #[must_use]
    pub fn take_batch_decode_reports(&self) -> Vec<Option<native_engine::DecodeReport>> {
        self.model
            .lock()
            .ok()
            .and_then(|slot| slot.as_ref().map(std::sync::Arc::clone))
            .map(|model| model.take_batch_decodes())
            .unwrap_or_default()
    }

    /// Take (consume) one [`native_engine::ChartPageMeta`] per image of the
    /// last [`OcrEngine::recognize_batch`] on this engine's cached model, in
    /// input order; `None` where an image has none (not a OneChart page, or
    /// a page that failed). As with the reports, after a batch
    /// `take_chart_page_meta` has nothing to return (bd-v37k).
    #[must_use]
    pub fn take_batch_chart_page_metas(&self) -> Vec<Option<native_engine::ChartPageMeta>> {
        self.model
            .lock()
            .ok()
            .and_then(|slot| slot.as_ref().map(std::sync::Arc::clone))
            .map(|model| model.take_batch_charts())
            .unwrap_or_default()
    }

    /// The identity of the model this engine last loaded (U2-E1); `None`
    /// before any load.
    #[must_use]
    pub fn loaded_model_identity(&self) -> Option<native_engine::ModelIdentity> {
        self.model
            .lock()
            .ok()
            .and_then(|slot| slot.as_ref().map(|model| model.identity()))
    }

    /// Construct the engine, building the single owned `asupersync` runtime
    /// (plan §3.3: `worker_threads(2)`, `blocking_threads(1, 4)`,
    /// `thread_name_prefix("focr")`). The model is loaded lazily on the first
    /// [`OcrEngine::recognize`] so construction is cheap and never touches the
    /// 6.67 GB blob.
    ///
    /// # Errors
    /// [`FocrError::Other`] if the runtime fails to build (e.g. the OS refuses to
    /// spawn worker threads).
    pub fn new() -> FocrResult<Self> {
        // Install the kernel pool at the documented budget BEFORE any forward
        // can touch rayon. This is the chokepoint every library and CLI consumer
        // passes through; without it the kernels silently run on rayon's own
        // default (logical cores), which oversubscribes the int8 GEMMs on any
        // SMT host, and their workers never get an Apple QoS class.
        let _ = init_kernel_pool();
        // Small blocking pool is a guard, not the mechanism: exactly one live
        // forward at a time runs the N-core kernel fan-out (doctrine #5).
        let runtime = RuntimeBuilder::new()
            .worker_threads(2)
            .blocking_threads(1, 4)
            .thread_name_prefix("focr")
            .build()
            .map_err(|e| FocrError::Other(anyhow::anyhow!("asupersync runtime build: {e}")))?;
        Ok(Self {
            runtime,
            model: Mutex::new(None),
            stage_sink: Mutex::new(None),
        })
    }

    /// Report the stages this engine enters to `sink`, or to nobody.
    pub fn set_stage_sink(&self, sink: Option<StageSink>) {
        if let Ok(mut slot) = self.stage_sink.lock() {
            *slot = sink;
        }
    }

    fn report_stage(&self, name: &'static str, budget: Option<Duration>) {
        if let Ok(slot) = self.stage_sink.lock()
            && let Some(sink) = slot.as_ref()
        {
            sink(name, budget);
        }
    }

    /// Resolve the configured model artifact path ([`MODEL_PATH_ENV`] override,
    /// else [`DEFAULT_MODEL_PATH`]).
    #[must_use]
    pub fn model_path() -> std::path::PathBuf {
        std::env::var_os(MODEL_PATH_ENV)
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| std::path::PathBuf::from(DEFAULT_MODEL_PATH))
    }

    /// Load (or fetch the cached) [`OcrModel`] at an explicit `path`.
    ///
    /// First call for the engine reads the weights; later calls clone the cached
    /// [`Arc`]. A missing/unresolvable model yields a clean
    /// [`FocrError::ModelNotFound`] (the model-gated e2e tests rely on this path,
    /// never a panic). The cache holds at most one model per engine; if `path`
    /// differs from the cached one it is reloaded.
    ///
    /// # Errors
    /// [`FocrError::ModelNotFound`] when the artifact does not resolve; otherwise
    /// whatever [`OcrModel::load`] returns (currently [`FocrError::NotImplemented`]
    /// once a path *does* resolve — the `.focrq` reader is Phase 2).
    fn model_at(&self, path: &Path) -> FocrResult<Arc<OcrModel>> {
        {
            let guard = self.model_guard()?;
            if let Some(m) = guard.as_ref()
                && m.path() == path
            {
                return Ok(Arc::clone(m));
            }
        }

        let loaded = OcrModel::load_observed(path, || self.report_stage("load", None))?;
        let loaded_path = loaded.path().to_path_buf();

        let mut guard = self.model_guard()?;
        if let Some(m) = guard.as_ref()
            && m.path() == loaded_path
        {
            return Ok(Arc::clone(m));
        }
        *guard = Some(Arc::clone(&loaded));
        Ok(loaded)
    }

    fn model_guard(&self) -> FocrResult<MutexGuard<'_, Option<Arc<OcrModel>>>> {
        self.model
            .lock()
            .map_err(|_| FocrError::Other(anyhow::anyhow!("OcrEngine model mutex poisoned")))
    }

    /// Recognize a single document image, returning structured markdown.
    ///
    /// **Synchronous and blocking** (G6): the heavy forward runs inside the
    /// engine's owned runtime via `block_on`, with a **sequential** single-page
    /// drive (doctrine #5). The model is resolved from [`OcrEngine::model_path`]
    /// (the [`MODEL_PATH_ENV`] override, else [`DEFAULT_MODEL_PATH`]) and
    /// loaded/cached on first use; when the weights are absent this returns
    /// [`FocrError::ModelNotFound`] cleanly (not a panic) so the model-gated e2e
    /// tests can skip-with-success by pointing the fallback at `/nonexistent`.
    ///
    /// # Errors
    /// * [`FocrError::ModelNotFound`] if the model artifact is absent/unresolvable.
    /// * Otherwise whatever the forward pipeline returns (today
    ///   [`FocrError::NotImplemented`] from the first stage whose `.focrq` tensor
    ///   accessor is not yet built — the pipeline is fully wired and typed).
    pub fn recognize(&self, image_path: &Path) -> FocrResult<String> {
        self.recognize_with_model(&Self::model_path(), image_path)
    }

    /// Recognize `image_path` using the model artifact at an explicit
    /// `model_path` (the path-explicit form of [`OcrEngine::recognize`]).
    ///
    /// Used by [`OcrEngine::recognize`] (with the env-resolved default) and by
    /// callers / tests that want to pin a specific artifact without setting an
    /// environment variable. Loading happens OUTSIDE `block_on` so a missing
    /// model is the clean [`FocrError::ModelNotFound`] without ever entering the
    /// runtime.
    ///
    /// # Choosing a model
    /// The engine runs whichever `.focrq` you point it at; pick by task:
    /// * **`unlimited-ocr`** (the default; what [`OcrEngine::recognize`] resolves) —
    ///   the **fast plain-text document OCR** model for general documents & PDFs.
    ///   This is the right default for ordinary text.
    /// * **`got-ocr2`** — a heavier, **specialized structured-output** model for the
    ///   formats the default cannot produce: math (LaTeX), tables, charts, molecular
    ///   (SMILES), geometry, and sheet music. Reach for it **only when you need that
    ///   format extraction**, not as a faster general OCR.
    ///
    /// See [`native_engine::model_arch`] for the registry (id → tasks → implemented)
    /// and `docs/zoo/` for each model's spec.
    ///
    /// # Errors
    /// As [`OcrEngine::recognize`].
    pub fn recognize_with_model(&self, model_path: &Path, image_path: &Path) -> FocrResult<String> {
        let model = self.model_at(model_path)?;
        let image_path = image_path.to_path_buf();
        // One owned runtime; the per-page forward is the only blocking work and
        // is driven sequentially on the runtime blocking pool, never inline on
        // the async polling thread (no nested runtime, no concurrent forwards).
        self.run_blocking_stage_with_budget("forward", Self::forward_budget(&model, 1), move || {
            model.recognize(&image_path)
        })
    }

    /// Recognize an already-decoded in-memory [`image::DynamicImage`], returning
    /// structured markdown — the path-free form of [`OcrEngine::recognize`].
    ///
    /// This is the entry point the native PDF path uses: [`crate::pdf`] rasterizes
    /// one PDF page to a `DynamicImage` and hands it here, so a scanned PDF flows
    /// through the identical preprocess → vision → decoder → postprocess pipeline a
    /// PNG would, with no intermediate temp files. The model is resolved from
    /// [`OcrEngine::model_path`] and loaded/cached on first use.
    ///
    /// # Errors
    /// As [`OcrEngine::recognize`].
    pub fn recognize_dynamic(&self, image: image::DynamicImage) -> FocrResult<String> {
        self.recognize_dynamic_with_model(&Self::model_path(), image)
    }

    /// Recognize an in-memory [`image::DynamicImage`] against the model artifact at
    /// an explicit `model_path` (the path-explicit form of
    /// [`OcrEngine::recognize_dynamic`]).
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_with_model`].
    pub fn recognize_dynamic_with_model(
        &self,
        model_path: &Path,
        image: image::DynamicImage,
    ) -> FocrResult<String> {
        let model = self.model_at(model_path)?;
        self.run_blocking_stage_with_budget("forward", Self::forward_budget(&model, 1), move || {
            model.recognize_dynamic(image)
        })
    }

    /// Recognize a single document image, returning the markdown AND the
    /// structured layout (bounding boxes) — the structured form of
    /// [`OcrEngine::recognize`] that `focr ocr --json` / `-o out.json` uses.
    ///
    /// # Errors
    /// As [`OcrEngine::recognize`].
    pub fn recognize_with_layout(&self, image_path: &Path) -> FocrResult<RecognizedDocument> {
        self.recognize_with_layout_model(&Self::model_path(), image_path)
    }

    /// The path-explicit form of [`OcrEngine::recognize_with_layout`].
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_with_model`].
    pub fn recognize_with_layout_model(
        &self,
        model_path: &Path,
        image_path: &Path,
    ) -> FocrResult<RecognizedDocument> {
        let model = self.model_at(model_path)?;
        let image_path = image_path.to_path_buf();
        self.run_blocking_stage_with_budget("forward", Self::forward_budget(&model, 1), move || {
            model.recognize_with_layout(&image_path)
        })
    }

    /// Recognize an in-memory [`image::DynamicImage`], returning the markdown AND
    /// the structured layout — the in-memory form of
    /// [`OcrEngine::recognize_with_layout`] the native PDF JSON path uses.
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_dynamic`].
    pub fn recognize_dynamic_with_layout(
        &self,
        image: image::DynamicImage,
    ) -> FocrResult<RecognizedDocument> {
        self.recognize_dynamic_with_layout_model(&Self::model_path(), image)
    }

    /// The path-explicit form of [`OcrEngine::recognize_dynamic_with_layout`].
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_with_model`].
    pub fn recognize_dynamic_with_layout_model(
        &self,
        model_path: &Path,
        image: image::DynamicImage,
    ) -> FocrResult<RecognizedDocument> {
        let model = self.model_at(model_path)?;
        self.run_blocking_stage_with_budget("forward", Self::forward_budget(&model, 1), move || {
            model.recognize_dynamic_with_layout(image)
        })
    }

    /// Recognize a single document image, returning the markdown + layout AND the
    /// figure regions cropped out of the source image — the regions the markdown
    /// renders as `![](images/…)` placeholders. This is what `focr ocr
    /// --extract-figures` uses to write real figure files.
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_with_layout`].
    pub fn recognize_with_figures(
        &self,
        image_path: &Path,
    ) -> FocrResult<(RecognizedDocument, Vec<ExtractedFigure>)> {
        self.recognize_with_figures_model(&Self::model_path(), image_path)
    }

    /// The path-explicit form of [`OcrEngine::recognize_with_figures`].
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_with_model`].
    pub fn recognize_with_figures_model(
        &self,
        model_path: &Path,
        image_path: &Path,
    ) -> FocrResult<(RecognizedDocument, Vec<ExtractedFigure>)> {
        let model = self.model_at(model_path)?;
        let image_path = image_path.to_path_buf();
        self.run_blocking_stage_with_budget("forward", Self::forward_budget(&model, 1), move || {
            model.recognize_with_figures(&image_path)
        })
    }

    /// Recognize an in-memory [`image::DynamicImage`], returning the markdown +
    /// layout AND the cropped figure regions — the in-memory form the native PDF
    /// `--extract-figures` path uses (the page raster is the crop source).
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_dynamic_with_layout`].
    pub fn recognize_dynamic_with_figures(
        &self,
        image: image::DynamicImage,
    ) -> FocrResult<(RecognizedDocument, Vec<ExtractedFigure>)> {
        self.recognize_dynamic_with_figures_model(&Self::model_path(), image)
    }

    /// The path-explicit form of [`OcrEngine::recognize_dynamic_with_figures`].
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_with_model`].
    pub fn recognize_dynamic_with_figures_model(
        &self,
        model_path: &Path,
        image: image::DynamicImage,
    ) -> FocrResult<(RecognizedDocument, Vec<ExtractedFigure>)> {
        let model = self.model_at(model_path)?;
        self.run_blocking_stage_with_budget("forward", Self::forward_budget(&model, 1), move || {
            model.recognize_dynamic_with_figures(image)
        })
    }

    /// Recognize a batch of document images in one load-once pass, returning one
    /// [`FocrResult`] per image in input order (`result[i]` ⇄ `images[i]`).
    ///
    /// The model is resolved from [`OcrEngine::model_path`] and loaded/cached on
    /// first use; see [`OcrEngine::recognize_batch_with_model`] for the
    /// path-explicit form and the spine semantics.
    ///
    /// # Errors
    /// [`FocrError::ModelNotFound`] if the model artifact is absent/unresolvable,
    /// or [`FocrError::Timeout`] if the whole batch exceeds its budget. Per-image
    /// failures surface inside the returned `Vec`, never as the outer error.
    pub fn recognize_batch(&self, images: &[&Path]) -> FocrResult<Vec<FocrResult<String>>> {
        self.recognize_batch_with_model(&Self::model_path(), images)
    }

    /// Recognize `images` against the model artifact at an explicit `model_path`
    /// (the path-explicit form of [`OcrEngine::recognize_batch`]).
    ///
    /// The model is acquired ONCE (the per-engine `Arc` cache), then the entire
    /// batch runs inside a SINGLE blocking stage on the runtime's blocking pool —
    /// the continuous-batch decode spine is the single sequential driver, with no
    /// per-step relock (Doctrine #5). The forward budget scales with the image
    /// count. When the spine is disarmed ([`native_engine`]
    /// `FOCR_BATCH_SPINE`), [`OcrModel::recognize_batch`] falls back to the proven
    /// per-image sequential path, so the spine-off result is byte-identical to
    /// today's loop.
    ///
    /// # Errors
    /// As [`OcrEngine::recognize_batch`].
    pub fn recognize_batch_with_model(
        &self,
        model_path: &Path,
        images: &[&Path],
    ) -> FocrResult<Vec<FocrResult<String>>> {
        let model = self.model_at(model_path)?;
        let owned: Vec<std::path::PathBuf> = images.iter().map(|p| p.to_path_buf()).collect();
        let budget = Self::forward_budget(&model, owned.len());
        self.run_blocking_stage_with_budget("forward-batch", budget, move || {
            let refs: Vec<&Path> = owned.iter().map(std::path::PathBuf::as_path).collect();
            Ok(model.recognize_batch(&refs))
        })
    }

    /// Parse a document (U2-E2, D23; the reference `infer_multi` contract,
    /// bd-1gv.25): a cross-page parse where page N attends to the pages
    /// before it (OQ-13), returning one [`DocumentResult`]. v1 reads the
    /// document in one pass; Unlimited-OCR-2 reads it in windows
    /// ([`DocumentOptions`], U2-D1b). This is not
    /// [`OcrEngine::recognize_batch`] (independent pages): use it when the
    /// pages form one document whose later pages refer to earlier ones.
    ///
    /// The pages are rasterized one at a time on the forward stage's thread,
    /// each dropped once preprocessed (U2-D0); a windowed parse reads each
    /// window's pages as the window starts. `on_item` sees every page before
    /// it is read, a skipped one included; `on_event`, when given, gets each
    /// [`StreamedPage`] as its `<PAGE>` boundary is crossed in the token
    /// stream (bd-2z0y), or as its window ends, and each window as it ends
    /// ([`DocumentEvent`]). Both are boxed and `Send` because the parse runs
    /// on the blocking pool.
    ///
    /// A document that stops early (the position limit, the stage budget or
    /// a cancel) is a result with that `finish_reason` and the pages it
    /// completed, never an error. The source is consumed: a caller that
    /// retries (a first-run download) opens a fresh one.
    ///
    /// ```no_run
    /// use std::path::Path;
    ///
    /// use franken_ocr::page_source::PageSource;
    /// use franken_ocr::{DocumentEvent, DocumentOptions, OcrEngine};
    ///
    /// # fn main() -> franken_ocr::FocrResult<()> {
    /// let engine = OcrEngine::new()?;
    /// let source = PageSource::pdf(Path::new("book.pdf"))?.select(Some("1-20"))?;
    /// let document = engine.recognize_document(
    ///     &OcrEngine::model_path(),
    ///     source,
    ///     &DocumentOptions::default(),
    ///     Box::new(|_| {}),
    ///     Some(Box::new(|event| {
    ///         if let DocumentEvent::Page(page) = event {
    ///             eprintln!("page {} done", page.index);
    ///         }
    ///     })),
    /// )?;
    /// println!("{}", document.markdown);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    /// As [`crate::native_engine::OcrModel::recognize_document`], plus the
    /// model's load errors.
    pub fn recognize_document(
        &self,
        model_path: &Path,
        mut source: page_source::PageSource,
        options: &DocumentOptions,
        mut on_item: Box<dyn FnMut(&page_source::PageItem) + Send>,
        on_event: Option<DocumentSink>,
    ) -> FocrResult<DocumentResult> {
        let model = self.model_at(model_path)?;
        let budget = Self::forward_budget(&model, source.len());
        let options = options.clone();
        self.run_blocking_stage_with_budget("forward-multi-page", budget, move || match on_event {
            Some(mut sink) => {
                model.recognize_document(&mut source, &options, &mut *on_item, Some(&mut *sink))
            }
            None => model.recognize_document(&mut source, &options, &mut *on_item, None),
        })
    }

    /// Read `source` one page at a time with the model at `model_path`, each
    /// page its own forward: the walk `focr ocr book.pdf` runs, where
    /// [`OcrEngine::recognize_document`] reads a document as one decode.
    ///
    /// A page that fails is skipped and reported through `observer`, and
    /// only a failure of the whole run ends the walk
    /// ([`crate::page_source::PageSource::read_pages`]).
    ///
    /// ```no_run
    /// use std::path::Path;
    ///
    /// use franken_ocr::OcrEngine;
    /// use franken_ocr::page_source::{PageEvent, PageSource};
    ///
    /// # fn main() -> franken_ocr::FocrResult<()> {
    /// let engine = OcrEngine::new()?;
    /// let source = PageSource::pdf(Path::new("book.pdf"))?;
    /// let read = engine.recognize_pages(&OcrEngine::model_path(), source, &mut |event| {
    ///     if let PageEvent::Skipped(page, why) = event {
    ///         eprintln!("page {} skipped: {why}", page.source_page);
    ///     }
    /// })?;
    /// println!("{}", read.markdown());
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    /// As [`crate::page_source::PageSource::read_pages`].
    pub fn recognize_pages(
        &self,
        model_path: &Path,
        mut source: page_source::PageSource,
        observer: &mut dyn FnMut(page_source::PageEvent<'_>),
    ) -> FocrResult<page_source::PagesRead<RecognizedDocument>> {
        source.read_pages(
            |image| self.recognize_dynamic_with_layout_model(model_path, image),
            observer,
        )
    }

    /// The forward stage's wall-clock bound for `count` inputs to `model`: the
    /// per-input budget times the count. A model whose document loop checks
    /// the same budget cooperatively (Unlimited-OCR-2, U2-B6) stops itself
    /// first and keeps its completed pages, so its bound here adds
    /// [`native_engine::document_budget_grace`] and stays a hang backstop.
    fn forward_budget(model: &OcrModel, count: usize) -> Option<Duration> {
        let per_input = forward_stage_budget()?;
        let count = u32::try_from(count.max(1)).unwrap_or(u32::MAX);
        let budget = per_input
            .checked_mul(count)
            .unwrap_or_else(|| Duration::from_secs(u64::MAX / 2));
        Some(if model.checks_forward_budget() {
            budget.saturating_add(native_engine::document_budget_grace(budget))
        } else {
            budget
        })
    }

    fn run_blocking_stage_with_budget<T, F>(
        &self,
        stage: &'static str,
        budget: Option<Duration>,
        op: F,
    ) -> FocrResult<T>
    where
        T: Send + 'static,
        F: FnOnce() -> FocrResult<T> + Send + 'static,
    {
        self.report_stage(stage, budget);
        let Some(budget) = budget else {
            // Budget disabled (`FOCR_STAGE_BUDGET_FORWARD_MS=0`): run to
            // completion; cooperative cancellation remains the only bound.
            return self
                .runtime
                .block_on(async move { asupersync::runtime::spawn_blocking(op).await });
        };
        let stop = Arc::<StageStop>::default();
        let running = Arc::clone(&stop);
        let finished = self.runtime.block_on(async move {
            asupersync::time::timeout(
                asupersync::time::wall_now(),
                budget,
                asupersync::runtime::spawn_blocking(move || {
                    let _running = RunningStage::enter(running);
                    op()
                }),
            )
            .await
        });
        finished.unwrap_or_else(|_| {
            let grace = native_engine::document_budget_grace(budget);
            if !stop.halt(grace) {
                eprintln!(
                    "[focr] warning: the {stage} stage is still running {}s past its budget",
                    grace.as_secs()
                );
            }
            Err(FocrError::Timeout(format!(
                "{stage} stage exceeded {}ms budget",
                budget.as_millis()
            )))
        })
    }
}

#[cfg(all(test, feature = "native"))]
mod tests {
    use super::*;

    fn log_line(test: &str, phase: &str, outcome: &str, extra: &str) {
        eprintln!(
            "{{\"test\":\"{test}\",\"phase\":\"{phase}\",\"outcome\":\"{outcome}\"{}{extra}}}",
            if extra.is_empty() { "" } else { "," }
        );
    }

    /// bd-223.2: dropping the engine shuts down its owned runtime and pools.
    #[test]
    fn engine_owns_single_runtime_and_drops_clean() {
        let engine = OcrEngine::new().expect("engine builds");
        // Capture the weak handle installed on an actual runtime worker. It
        // observes this engine's runtime lifecycle without inspecting other
        // tests' process-wide threads.
        let runtime_witness = engine
            .runtime
            .block_on(engine.runtime.handle().spawn(async {
                Runtime::current_handle().expect("spawned task has a runtime handle")
            }));

        // The runtime is live: a trivial blocking stage round-trips.
        let out = engine
            .run_blocking_stage_with_budget("drop-probe", Some(Duration::from_secs(5)), || Ok(42u8))
            .expect("stage runs");
        assert_eq!(out, 42);
        log_line(
            "engine_owns_single_runtime_and_drops_clean",
            "live",
            "pass",
            "",
        );

        drop(engine); // joins the runtime workers

        assert!(
            matches!(
                runtime_witness.try_spawn(async { 42u8 }),
                Err(asupersync::runtime::state::SpawnError::RuntimeUnavailable)
            ),
            "dropped engine runtime should reject new tasks"
        );
        assert!(
            runtime_witness.spawn_blocking(|| {}).is_none(),
            "dropped engine runtime should not accept blocking tasks"
        );
        assert!(
            runtime_witness.blocking_handle().is_none(),
            "dropped engine runtime should not expose its blocking pool"
        );
        log_line(
            "engine_owns_single_runtime_and_drops_clean",
            "dropped",
            "pass",
            "",
        );
    }

    /// Serializes the tests that read a checkpoint against the one that raises
    /// the process-global shutdown flag.
    fn shutdown_flag_test() -> MutexGuard<'static, ()> {
        static FLAG: Mutex<()> = Mutex::new(());
        FLAG.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// bd-223.2: the checkpoint aborts a decode-style loop with Cancelled
    /// (exit 6) — the flag is observed INSIDE the spawn_blocking closure.
    #[test]
    fn cancellation_token_into_closure_aborts() {
        let _shutdown = shutdown_flag_test();
        reset_shutdown();
        let engine = OcrEngine::new().expect("engine builds");
        let out = engine.run_blocking_stage_with_budget(
            "cancel-probe",
            Some(Duration::from_secs(10)),
            || {
                for step in 0..1_000_000u64 {
                    cancel_checkpoint()?;
                    if step == 3 {
                        // The "Ctrl+C" arrives mid-loop, from inside the
                        // closure's world — exactly the cooperative contract.
                        request_shutdown();
                    }
                }
                Ok(0u64)
            },
        );
        reset_shutdown();
        assert!(
            matches!(out, Err(FocrError::Cancelled)),
            "expected Cancelled, got {out:?}"
        );
        assert_eq!(FocrError::Cancelled.exit_code(), 6, "exit code contract");
        log_line(
            "cancellation_token_into_closure_aborts",
            "aborted",
            "pass",
            "",
        );
    }

    /// bd-223.2: the bounded channel BLOCKS the producer when the consumer
    /// lags (backpressure) — in-flight items never exceed capacity + 1.
    #[test]
    fn bounded_stream_backpressure() {
        use std::sync::atomic::{AtomicI64, Ordering};
        static IN_FLIGHT: AtomicI64 = AtomicI64::new(0);
        static MAX_SEEN: AtomicI64 = AtomicI64::new(0);
        IN_FLIGHT.store(0, Ordering::SeqCst);
        MAX_SEEN.store(0, Ordering::SeqCst);
        let total = 24u32;
        let mut produced = 0u32;
        let n = stream_pages(
            2,
            move || {
                if produced == total {
                    return Ok(None);
                }
                produced += 1;
                let now = IN_FLIGHT.fetch_add(1, Ordering::SeqCst) + 1;
                MAX_SEEN.fetch_max(now, Ordering::SeqCst);
                Ok(Some(produced))
            },
            |item: u32| {
                // Slow consumer: the producer must stall on the bounded send.
                std::thread::sleep(Duration::from_millis(5));
                IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
                let _ = item;
            },
        )
        .expect("stream completes");
        assert_eq!(n, total as usize, "every page delivered in order");
        let max = MAX_SEEN.load(Ordering::SeqCst);
        assert!(
            max <= 4,
            "in-flight items {max} exceeded capacity(2)+channel slack — backpressure broken"
        );
        log_line(
            "bounded_stream_backpressure",
            "drained",
            "pass",
            &format!("\"max_in_flight\":{max},\"n\":{n}"),
        );
    }

    /// bd-223.2 addendum: FOCR_THREADS wins, else the PHYSICAL core count
    /// (of the fastest tier on a two-tier Mac, bd-zex4).
    #[test]
    fn thread_budget_reads_env_then_physical() {
        // The OnceLock latches per process; assert the resolved value is
        // consistent with the environment THIS process started with.
        let budget = thread_budget();
        match std::env::var("FOCR_THREADS")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
        {
            Some(env) if env > 0 => assert_eq!(budget, env, "env wins"),
            _ => assert_eq!(budget, default_thread_budget(), "the default"),
        }
        assert!(
            default_thread_budget() <= num_cpus::get_physical(),
            "the default never exceeds the physical cores"
        );
        assert!(budget > 0);
        assert!(
            budget
                <= std::thread::available_parallelism()
                    .map(|n| n.get())
                    .unwrap_or(usize::MAX),
            "physical budget cannot exceed logical width"
        );
        log_line(
            "thread_budget_reads_env_then_physical",
            "resolved",
            "pass",
            &format!("\"threads\":{budget}"),
        );
    }

    /// bd-zex4: every Apple-silicon Mac reports its fastest tier, and the
    /// default follows it; other targets keep the physical core count.
    #[test]
    fn default_budget_is_the_fastest_tier_on_apple_silicon() {
        let fast = fastest_tier_cores();
        if cfg!(all(target_os = "macos", target_arch = "aarch64")) {
            let fast = fast.expect("hw.perflevel0.physicalcpu on Apple silicon");
            assert!(fast <= num_cpus::get_physical(), "{fast} fast cores");
            assert_eq!(default_thread_budget(), fast);
        } else if cfg!(not(any(target_os = "macos", target_os = "ios"))) {
            assert_eq!(fast, None);
            assert_eq!(default_thread_budget(), num_cpus::get_physical());
        }
        log_line(
            "default_budget_is_the_fastest_tier_on_apple_silicon",
            "resolved",
            "pass",
            &format!(
                "\"fastest_tier_cores\":{},\"default\":{}",
                fast.map_or_else(|| "null".to_string(), |n| n.to_string()),
                default_thread_budget()
            ),
        );
    }

    struct TempModel(std::path::PathBuf);

    impl TempModel {
        fn write_focrq(bytes: &[u8]) -> std::io::Result<Self> {
            let nanos = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or_default();
            let path = std::env::temp_dir().join(format!(
                "franken_ocr_engine_format_mismatch_{}_{}.focrq",
                std::process::id(),
                nanos
            ));
            std::fs::write(&path, bytes)?;
            Ok(Self(path))
        }

        fn path(&self) -> &Path {
            &self.0
        }
    }

    impl Drop for TempModel {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.0);
        }
    }

    fn future_focrq_preamble() -> Vec<u8> {
        let mut blob = Vec::new();
        blob.extend_from_slice(native_engine::weights::FOCRQ_MAGIC);
        blob.extend_from_slice(&(native_engine::weights::FOCRQ_FORMAT_VERSION + 1).to_le_bytes());
        blob.push(0);
        blob.extend_from_slice(&[0u8; 32]);
        blob.extend_from_slice(&0u64.to_le_bytes());
        blob
    }

    /// The engine constructs (its single owned runtime builds) without touching
    /// the model blob — construction is cheap and lazy.
    #[test]
    fn engine_constructs_without_model() {
        let engine = OcrEngine::new().expect("runtime builds");
        // Constructing alone must not have loaded a model.
        assert!(
            engine.model_guard().expect("mutex").is_none(),
            "model must be loaded lazily, not at construction"
        );
    }

    /// `recognize_with_model` on a guaranteed-absent model path returns a clean
    /// `ModelNotFound` (exit code 3) — NOT a panic, NOT NotImplemented. This is
    /// the path the model-gated e2e tests pin (point the fallback at
    /// `/nonexistent`). We use the path-explicit form so the test never mutates
    /// the process environment (the crate root `#![deny(unsafe_code)]` rules out
    /// the `unsafe` `std::env::set_var`).
    #[test]
    fn recognize_missing_model_is_clean_model_not_found() {
        let engine = OcrEngine::new().expect("runtime builds");
        let err = engine
            .recognize_with_model(
                Path::new("/nonexistent/franken_ocr/model.focrq"),
                Path::new("/some/document.png"),
            )
            .expect_err("absent model must error");
        assert!(
            matches!(err, FocrError::ModelNotFound(_)),
            "expected ModelNotFound, got {err:?}"
        );
        assert_eq!(err.exit_code(), 3, "ModelNotFound must map to exit code 3");
    }

    /// The blocking `recognize` path (env-resolved default) also yields a clean
    /// `ModelNotFound` when `FOCR_MODEL_PATH` is unset and the default artifact is
    /// absent — proving the public entrypoint never panics without weights. (The
    /// default `models/unlimited-ocr.focrq` does not exist in the test CWD.)
    #[test]
    fn public_recognize_without_weights_is_model_not_found() {
        // Only assert when the env override is unset AND the engine's own
        // resolver finds nothing — the normal CI condition. Checking the
        // repo-relative default alone is NOT enough: since bd-3u6x the
        // default spec also resolves via the user cache
        // (~/.cache/franken_ocr/models/unlimited-ocr/), so a dev
        // box with a pulled artifact must skip rather than misfire.
        if std::env::var_os(MODEL_PATH_ENV).is_none()
            && !std::path::Path::new(DEFAULT_MODEL_PATH).exists()
            && native_engine::OcrModel::resolve_model(Path::new(DEFAULT_MODEL_PATH)).is_err()
        {
            let engine = OcrEngine::new().expect("runtime builds");
            let err = engine
                .recognize(Path::new("/some/document.png"))
                .expect_err("absent default model must error");
            assert!(matches!(err, FocrError::ModelNotFound(_)));
        }
    }

    /// The model-path resolver falls back to the documented default when the env
    /// override is unset (read-only check; no env mutation under `deny(unsafe)`).
    #[test]
    fn model_path_falls_back_to_default_when_env_unset() {
        if std::env::var_os(MODEL_PATH_ENV).is_none() {
            assert_eq!(
                OcrEngine::model_path(),
                std::path::PathBuf::from(DEFAULT_MODEL_PATH)
            );
        }
    }

    /// Calling `recognize_with_model` twice loads the model once per distinct
    /// path (the per-engine cache); a second call with the SAME absent path still
    /// returns `ModelNotFound` (the absent model is never cached as a success).
    #[test]
    fn repeated_missing_model_stays_model_not_found() {
        let engine = OcrEngine::new().expect("runtime builds");
        let p = Path::new("/nonexistent/franken_ocr/model.focrq");
        let img = Path::new("/some/document.png");
        for _ in 0..3 {
            let err = engine.recognize_with_model(p, img).expect_err("absent");
            assert!(matches!(err, FocrError::ModelNotFound(_)));
        }
    }

    /// A book read without a model ends on its first page, as the error a
    /// caller can act on, not as one skip per page.
    #[test]
    fn recognize_pages_stops_on_the_first_page_without_a_model() {
        let engine = OcrEngine::new().expect("runtime builds");
        let blank = || image::DynamicImage::new_rgb8(8, 8);
        let source = page_source::PageSource::rasters(vec![blank(), blank()]).expect("a source");
        let mut started = 0;
        let err = engine
            .recognize_pages(
                Path::new("/nonexistent/franken_ocr/model.focrq"),
                source,
                &mut |event| {
                    if matches!(event, page_source::PageEvent::Started(_)) {
                        started += 1;
                    }
                },
            )
            .expect_err("absent model must error");
        assert!(matches!(err, FocrError::ModelNotFound(_)), "got {err:?}");
        assert_eq!(started, 1);
    }

    #[test]
    fn blocking_stage_runs_on_runtime_blocking_pool() {
        let engine = OcrEngine::new().expect("runtime builds");
        let thread_name = engine
            .run_blocking_stage_with_budget("test", Some(Duration::from_secs(1)), || {
                let thread = std::thread::current();
                Ok(thread.name().unwrap_or("<unnamed>").to_string())
            })
            .expect("stage should complete");
        assert!(
            thread_name.contains("-blocking-"),
            "stage ran on {thread_name:?}, not the runtime blocking pool"
        );
    }

    #[test]
    fn blocking_stage_timeout_maps_to_stable_error() {
        let engine = OcrEngine::new().expect("runtime builds");
        let started = std::time::Instant::now();
        let err = engine
            .run_blocking_stage_with_budget("test-timeout", Some(Duration::from_millis(10)), || {
                std::thread::sleep(Duration::from_millis(100));
                Ok(())
            })
            .expect_err("slow blocking stage must time out");
        assert!(
            matches!(err, FocrError::Timeout(_)),
            "expected Timeout, got {err:?}"
        );
        assert_eq!(err.exit_code(), error::EXIT_TIMEOUT);
        assert!(
            started.elapsed() < Duration::from_millis(500),
            "timeout wrapper waited for the whole blocking closure"
        );
    }

    /// bd-3ttp: a timed-out decode has stopped by the time the timeout returns,
    /// so the next input has the model and the cores to itself.
    #[test]
    fn a_stage_past_its_budget_stops_before_the_timeout_returns() {
        struct Ended(Arc<AtomicBool>);
        impl Drop for Ended {
            fn drop(&mut self) {
                self.0.store(true, Ordering::SeqCst);
            }
        }
        let _shutdown = shutdown_flag_test();
        let engine = OcrEngine::new().expect("runtime builds");
        let ended = Arc::new(AtomicBool::new(false));
        let witness = Ended(Arc::clone(&ended));
        let err = engine
            .run_blocking_stage_with_budget(
                "test-runaway",
                Some(Duration::from_millis(10)),
                move || {
                    let _witness = witness;
                    // Bounded, so a regression fails below and does not spin on.
                    for _ in 0..3_000 {
                        cancel_checkpoint()?;
                        std::thread::sleep(Duration::from_millis(1));
                    }
                    Ok(())
                },
            )
            .expect_err("the stage must time out");
        assert!(matches!(err, FocrError::Timeout(_)), "got {err:?}");
        assert!(err.to_string().contains("test-runaway stage exceeded"));
        assert!(
            ended.load(Ordering::SeqCst),
            "the decode outlived its timeout"
        );
        // The stop dies with its stage: the blocking pool reuses this thread.
        engine
            .run_blocking_stage_with_budget(
                "test-next",
                Some(Duration::from_secs(5)),
                cancel_checkpoint,
            )
            .expect("the next stage is not stopped");
    }

    /// bd-3ttp: a stage that never reaches a checkpoint cannot hold the timeout.
    #[test]
    fn a_stage_stuck_between_checkpoints_cannot_hold_its_timeout() {
        let stuck = Arc::<StageStop>::default();
        assert!(!stuck.halt(Duration::from_millis(5)));
        assert!(stuck.stop_requested());

        let stage = Arc::<StageStop>::default();
        drop(RunningStage::enter(Arc::clone(&stage)));
        assert!(stage.halt(Duration::from_secs(5)));
    }

    /// A recognized but too-new `.focrq` must stay a public FormatMismatch
    /// through the engine entrypoint and robot event, not be collapsed into the
    /// Phase-0 generic/NotImplemented resolver path.
    #[test]
    fn public_engine_preserves_focrq_format_mismatch_robot_code()
    -> Result<(), Box<dyn std::error::Error>> {
        let model = TempModel::write_focrq(&future_focrq_preamble())?;
        let engine = OcrEngine::new()?;
        let result = engine.recognize_with_model(model.path(), Path::new("/some/document.png"));
        let Err(err) = result else {
            return Err(std::io::Error::other(
                "future .focrq version unexpectedly succeeded before forward",
            )
            .into());
        };

        assert!(
            matches!(err, FocrError::FormatMismatch(_)),
            "expected FormatMismatch, got {err:?}"
        );
        assert_eq!(err.exit_code(), error::EXIT_FORMAT_MISMATCH);

        let event = robot::run_error_event(&err);
        assert_eq!(event["event"], "run_error");
        assert_eq!(event["error_kind"], "format_mismatch");
        assert_eq!(event["code"], error::EXIT_FORMAT_MISMATCH);
        Ok(())
    }
}