brep_render/

runner.rs

//! The history-run seam — the async RUN machine behind a trait, mirroring the
//! platform-seam shape of `brep-app/src/store.rs`'s `ModelStore` (a trait with
//! impls behind it, async surfaced via a poll idiom).
//!
//! A history run is split SUBMIT → POLL/APPLY: [`HistoryRunner::submit_run`]
//! kicks off a run tagged with a monotonic generation, and the completed
//! [`RunReply`] is drained later via [`HistoryRunner::poll_run`]. The runner OWNS
//! the [`SceneRunner`](crate::pipeline::SceneRunner) — so a future thread/worker
//! impl owns the resident registry that `execute_history` populates — and holds
//! the delta baseline across reruns.
//!
//! This slice ships the DEFAULT [`InlineRunner`]: it runs on `submit_run` and
//! stashes the reply for an immediate `poll_run`, so the run stays synchronous
//! and byte-identical to the pre-seam in-process run. A native-thread impl (M2b)
//! and a wasm-worker impl (M3) slot in behind the SAME trait — `submit_run`
//! defers the work and `poll_run` surfaces it a frame (or many) later, so the
//! `EngineState::pump` caller never changes.

/// A completed history run, tagged with the [`generation`](Self::generation) it
/// was submitted under so the applier can drop stale replies (a newer run that
/// finished first). The [`output`](Self::output) is the [`SceneRunner`] delta to
/// apply to the display scene.
///
/// [`SceneRunner`]: crate::pipeline::SceneRunner
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RunReply {
    /// The monotonic generation this run was submitted under.
    pub generation: u64,
    /// The delta snapshot + report the run produced.
    pub output: crate::pipeline::RunOutput,
}

/// Which exact measurement a [`MeasureQuery`] wants, mirroring the object-info
/// kinds ([`crate::metadata`]): a whole solid, one named face, or one named edge.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum MeasureKind {
    Solid,
    Face,
    Edge,
}

/// A per-object MEASUREMENT request routed to the runner (which owns the warm
/// registry). The runner resolves `owner` to its resident handle and measures by
/// [`kind`](Self::kind); the reply is the object-info JSON fragment MINUS the
/// main-injected `name`/`creatingFeature` fields. Tagged with a monotonic
/// [`id`](Self::id) so the main side can pair the reply with its pending request.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct MeasureQuery {
    /// Monotonic request id (main pairs the reply back by it).
    pub id: u64,
    /// The measurement to run.
    pub kind: MeasureKind,
    /// The OWNING solid name (a solid is its own owner; a face/edge names its solid).
    pub owner: String,
    /// The face/edge NAME to measure (ignored for a whole-solid query).
    pub entity: String,
    /// The density (mass per mm³) to scale a solid's weight (ignored for face/edge).
    pub density: f64,
}

/// A completed [`MeasureQuery`]: the object-info measurement fields as a JSON
/// fragment (WITHOUT `name`/`creatingFeature`, which the main thread injects from
/// its eager provenance), tagged with the request [`id`](Self::id).
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct MeasureReply {
    /// The request id this answers.
    pub id: u64,
    /// The measurement JSON fragment (see [`measure_json`]).
    pub result: String,
}

/// Mesh encoding accepted by the off-thread reconstruction channel.
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub enum MeshImportFormat {
    Stl,
    Obj,
}

/// A mesh reconstruction request. The byte payload moves to the native thread
/// or is serialized to the browser worker; no parsing or fitting happens on UI.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct MeshImportRequest {
    pub id: u64,
    pub format: MeshImportFormat,
    pub bytes: Vec<u8>,
}

/// Completed off-thread reconstruction. Success carries validated STEP text,
/// which the main side appends through the ordinary IMPORT3D history lane.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct MeshImportReply {
    pub id: u64,
    pub result: Result<String, String>,
}

/// The history-run seam: SUBMIT a run, POLL for its completed reply, RESET the
/// delta baseline, plus a QUERY channel for per-object measurements (routed to the
/// runner so the warm registry answers them, never the cold main-side one). The
/// Inline impl runs everything synchronously; a later thread/worker impl defers
/// the work and surfaces the replies through the same poll idiom.
pub trait HistoryRunner {
    /// Submit a history run tagged with a monotonic generation. The runner executes
    /// it (immediately for Inline; on a background thread later) and makes the reply
    /// available via [`poll_run`](Self::poll_run).
    fn submit_run(&mut self, request: brep_kernel::HistoryRequest, generation: u64);
    /// Non-blocking: the next completed run reply, if any (drained each frame).
    fn poll_run(&mut self) -> Option<RunReply>;
    /// Submit a per-object measurement query (answered against the runner's warm
    /// registry). Inline computes it immediately; a thread impl computes it on the
    /// runner thread and surfaces it via [`poll_query`](Self::poll_query).
    fn submit_query(&mut self, query: MeasureQuery);
    /// Non-blocking: the next completed measurement reply, if any.
    fn poll_query(&mut self) -> Option<MeasureReply>;
    /// Submit RANSAC mesh reconstruction to the runner's thread/worker.
    fn submit_mesh_import(&mut self, request: MeshImportRequest);
    /// Non-blocking: the next completed mesh reconstruction, if any.
    fn poll_mesh_import(&mut self) -> Option<MeshImportReply>;
    /// Drop the delta baseline (document switch → full rebuild).
    fn reset(&mut self);

    /// Bring the runner's resident PARTS LIBRARY up to `revision`, calling
    /// `fetch` ONLY when it is not already there. Called immediately before
    /// every [`submit_run`](Self::submit_run).
    ///
    /// This is the whole point of the library channel: the library is sent
    /// when it CHANGES (insert, document load, refresh), not on every run.
    /// Cheap to call — the revision comparison is an integer, and `fetch`
    /// (which clones the store) never runs in the steady state. The default is
    /// a no-op for [`InlineRunner`], which shares the caller's kernel store.
    fn sync_parts_library(
        &mut self,
        _revision: u64,
        _fetch: &mut dyn FnMut() -> brep_kernel::PartsLibraryMap,
    ) {
    }

    /// True when the runner REFUSED a run because its resident parts library
    /// could not serve it (see [`Reply::NeedPartsLibrary`]). It has already
    /// forgotten its copy, so the next `sync_parts_library` reinstalls; the
    /// caller must re-submit the run. Never true for a runner that shares the
    /// caller's store.
    fn poll_library_request(&mut self) -> bool {
        false
    }
}

/// Measure `query` against `runner`'s resident geometry and emit the object-info
/// JSON FRAGMENT — EXACTLY the fields [`crate::metadata::EngineState::object_info_json`]
/// emits for that kind EXCEPT `name` and `creatingFeature` (the main thread injects
/// those from its eager provenance, so the merged output is byte-identical to the
/// pre-seam in-process result). Shared verbatim by the Inline and thread runners so
/// both produce the identical fragment. A missing handle / kernel error yields
/// `{ "ok": false, "message": .. }` (main injects `name`).
fn measure_json(runner: &crate::pipeline::SceneRunner, query: &MeasureQuery) -> String {
    let Some(handle) = runner.handle_of(&query.owner) else {
        return serde_json::json!({
            "ok": false,
            "message": format!("solid '{}' has no resident geometry", query.owner),
        })
        .to_string();
    };
    match query.kind {
        MeasureKind::Solid => {
            match brep_kernel::mass_properties_handle_native(handle, query.density) {
                Ok(properties) => {
                    let edge_total =
                        brep_kernel::solid_edge_length_total_native(handle).unwrap_or(0.0);
                    serde_json::json!({
                        "ok": true,
                        "kind": "solid",
                        "volume": properties.volume,
                        "surfaceArea": properties.surface_area,
                        "edgeLengthTotal": edge_total,
                        "density": properties.density,
                        "weight": properties.mass,
                    })
                    .to_string()
                }
                Err(error) => serde_json::json!({ "ok": false, "message": error }).to_string(),
            }
        }
        MeasureKind::Face => match brep_kernel::face_measurements_native(handle, &query.entity) {
            Ok((area, edge_total, surface_type)) => serde_json::json!({
                "ok": true,
                "kind": "face",
                "solid": query.owner,
                "surfaceType": surface_type,
                "area": area,
                "edgeLengthTotal": edge_total,
            })
            .to_string(),
            Err(error) => serde_json::json!({ "ok": false, "message": error }).to_string(),
        },
        MeasureKind::Edge => match brep_kernel::edge_length_native(handle, &query.entity) {
            Ok(length) => serde_json::json!({
                "ok": true,
                "kind": "edge",
                "solid": query.owner,
                "length": length,
            })
            .to_string(),
            Err(error) => serde_json::json!({ "ok": false, "message": error }).to_string(),
        },
    }
}

/// The default, SYNCHRONOUS runner: runs on submit, stashes the reply for an
/// immediate poll. Behavior-identical to the pre-seam in-process run.
pub struct InlineRunner {
    /// The scene-free history runner it owns (delta baseline lives here).
    runner: crate::pipeline::SceneRunner,
    /// Completed replies awaiting a poll. For Inline this holds exactly one entry
    /// between a `submit_run` and the immediately-following `poll_run`.
    pending: std::collections::VecDeque<RunReply>,
    /// Completed measurement replies awaiting a poll (computed synchronously on
    /// `submit_query`, popped on `poll_query`) — the synchronous mirror of the
    /// thread runner's query buffer, so the object-info path resolves same-call.
    query_pending: std::collections::VecDeque<MeasureReply>,
    mesh_import_pending: std::collections::VecDeque<MeshImportReply>,
}

impl InlineRunner {
    pub fn new() -> Self {
        Self {
            runner: crate::pipeline::SceneRunner::new(),
            pending: std::collections::VecDeque::new(),
            query_pending: std::collections::VecDeque::new(),
            mesh_import_pending: std::collections::VecDeque::new(),
        }
    }
}

impl HistoryRunner for InlineRunner {
    fn submit_run(&mut self, request: brep_kernel::HistoryRequest, generation: u64) {
        let output = self.runner.run(&request, None);
        self.pending.push_back(RunReply { generation, output });
    }

    /// Nothing to send: Inline runs on the CALLER's thread against the CALLER's
    /// kernel store, so the library the caller would hand over is already the
    /// one this run resolves against. (Which is also why Inline never needs the
    /// preflight — there is only one store, and the caller seeds it directly.)
    fn sync_parts_library(
        &mut self,
        _revision: u64,
        _fetch: &mut dyn FnMut() -> brep_kernel::PartsLibraryMap,
    ) {
    }

    fn poll_run(&mut self) -> Option<RunReply> {
        self.pending.pop_front()
    }

    fn submit_query(&mut self, query: MeasureQuery) {
        let result = measure_json(&self.runner, &query);
        self.query_pending.push_back(MeasureReply { id: query.id, result });
    }

    fn poll_query(&mut self) -> Option<MeasureReply> {
        self.query_pending.pop_front()
    }

    fn submit_mesh_import(&mut self, request: MeshImportRequest) {
        self.mesh_import_pending
            .push_back(reconstruct_mesh(request));
    }

    fn poll_mesh_import(&mut self) -> Option<MeshImportReply> {
        self.mesh_import_pending.pop_front()
    }

    fn reset(&mut self) {
        self.runner.reset();
    }
}

impl Default for InlineRunner {
    fn default() -> Self {
        Self::new()
    }
}

// ===========================================================================
// Shared run/query PROTOCOL (M3b): the `Command`/`Reply` message pair and the
// `process_command` step. Both async runners speak it — the native `ThreadRunner`
// ships the enums over `mpsc` (no serialization), and the wasm `WorkerRunner`
// serializes them to JSON for `postMessage`. Hence the serde derives (the enums
// carry serde types: `HistoryRequest`/`MeasureQuery`/`RunReply`/`MeasureReply`)
// and `process_command` being un-gated + shared so both drivers do the SAME work.
// ===========================================================================

/// A command sent main → runner (thread channel OR worker `postMessage`) over one
/// ordered stream (so a `Reset` before a `Run` stays before it). `Run` carries the
/// whole request; the driver coalesces consecutive `Run`s to shed a slider drag's
/// backlog (the thread in `thread_main`, the worker's main side in `WorkerRunner`).
#[derive(serde::Serialize, serde::Deserialize)]
pub enum Command {
    Run {
        request: brep_kernel::HistoryRequest,
        generation: u64,
        /// The [`brep_kernel::parts_library_revision`] this run was built
        /// against. The runner refuses to execute a run stamped for a library
        /// it does not hold — see [`process_command`].
        #[serde(default)]
        parts_library_revision: u64,
    },
    /// Install the parts library on the runner's OWN kernel store. Sent only
    /// when the library actually changes (insert, document load, refresh),
    /// never per run: it carries the whole embedded-part payload, which for an
    /// imported STEP assembly is megabytes, and stringifying that on the
    /// browser's main thread once per edit is what froze the UI.
    ///
    /// Deliberately its OWN command rather than an optional field on `Run`:
    /// both drivers COALESCE consecutive runs (see [`thread_main`] and
    /// `WorkerRunner::submit_run`), so a library riding on a run could be
    /// dropped with it. A `SetPartsLibrary` is never coalesced away, and the
    /// stream is ordered, so it always lands before the run that needs it.
    SetPartsLibrary {
        library: brep_kernel::PartsLibraryMap,
        revision: u64,
    },
    Query(MeasureQuery),
    MeshImport(MeshImportRequest),
    Reset,
}

/// A reply sent runner → main; the main side demuxes it into per-kind buffers.
#[derive(serde::Serialize, serde::Deserialize)]
pub enum Reply {
    Run(RunReply),
    Query(MeasureReply),
    MeshImport(MeshImportReply),
    /// The runner REFUSED a run because its resident parts library could not
    /// serve it (a part the request references is missing, or the run was
    /// stamped for a different library revision). No geometry was touched; the
    /// main side re-sends the library and re-submits. Refusing is the whole
    /// point — running with the wrong library would be silently wrong
    /// geometry, which is far worse than one extra round trip.
    NeedPartsLibrary,
}

fn reconstruct_mesh(request: MeshImportRequest) -> MeshImportReply {
    let result = (|| {
        use brep_reconstruction::stl_conversion::{
            binary_stl_coordinate_precision_tolerance, convert_stl_mesh_to_step,
            StlConversionOptions,
        };
        use brep_reconstruction::{Mesh, Vec3};

        let (mesh, positions, indices, coordinate_precision_tolerance) = match request.format {
            MeshImportFormat::Stl => {
                use brep_reconstruction::stl::{parse_stl_bytes, StlFormat, StlReadOptions};
                let imported = parse_stl_bytes(&request.bytes, &StlReadOptions::default())
                    .map_err(|error| format!("STL import failed: {error}"))?;
                let positions = imported
                    .mesh
                    .vertices
                    .iter()
                    .flat_map(|point| [point.x, point.y, point.z])
                    .collect::<Vec<_>>();
                let indices = imported
                    .mesh
                    .triangles
                    .iter()
                    .flatten()
                    .copied()
                    .collect::<Vec<_>>();
                let precision = if imported.format == StlFormat::Binary {
                    binary_stl_coordinate_precision_tolerance(&imported.mesh)
                } else {
                    0.0
                };
                (imported.mesh, positions, indices, precision)
            }
            MeshImportFormat::Obj => {
                let text = std::str::from_utf8(&request.bytes)
                    .map_err(|_| "OBJ import failed: file is not UTF-8 text".to_string())?;
                let obj = brep_kernel::read_obj(text)
                    .map_err(|error| format!("OBJ import failed: {error}"))?;
                let vertices = obj
                    .positions
                    .chunks_exact(3)
                    .map(|point| Vec3::new(point[0], point[1], point[2]))
                    .collect::<Vec<_>>();
                let triangles = obj
                    .indices
                    .chunks_exact(3)
                    .map(|triangle| [triangle[0], triangle[1], triangle[2]])
                    .collect::<Vec<_>>();
                (
                    Mesh::new(vertices, triangles),
                    obj.positions,
                    obj.indices,
                    0.0,
                )
            }
        };
        let mut options = StlConversionOptions::default();
        options.coordinate_precision_tolerance = coordinate_precision_tolerance;
        convert_stl_mesh_to_step(
            &mesh,
            &positions,
            Some(&indices),
            &options,
            "Imported mesh",
            "MM",
            "",
        )
        .map(|output| output.step_text)
        .map_err(|error| format!("RANSAC reconstruction failed: {error}"))
    })();
    MeshImportReply {
        id: request.id,
        result,
    }
}

/// Execute ONE [`Command`] against `runner` and return the [`Reply`] it produces,
/// if any. The shared step both async drivers run: the native [`thread_main`]
/// calls it for each (post-coalescing) command on the runner thread; the wasm
/// `WorkerRunner`'s `worker_entry` calls it per `postMessage` on the worker.
/// `Run` → a [`Reply::Run`]; `Query` → a [`Reply::Query`]; `Reset` drops the delta
/// baseline AND the runner's OWN kernel history cache (the resident registry lives
/// with the runner — thread or worker — not on main) and yields no reply.
pub fn process_command(
    runner: &mut crate::pipeline::SceneRunner,
    command: Command,
) -> Option<Reply> {
    match command {
        Command::Run {
            request,
            generation,
            parts_library_revision,
        } => {
            // PREFLIGHT (see `Reply::NeedPartsLibrary`). Two independent
            // checks, because neither alone is enough:
            //
            // * the revision stamp catches CHANGED content the sender knows
            //   about but this store has not received;
            // * `missing_library_parts` catches content that is simply GONE
            //   here — the orphan GC at the end of every run drops entries no
            //   ACOMP in THAT run referenced, so an undo to zero components
            //   empties this store while the sender's (which never ran) keeps
            //   everything and its revision never moves. Only looking at the
            //   actual content sees that.
            let stale = match runner.parts_library_revision {
                Some(installed) => installed != parts_library_revision,
                // Nothing installed yet. Accept a run stamped 0 — that is
                // either an empty library or a caller driving `submit_run`
                // directly (the runner tests); the content preflight below is
                // what actually protects the run. A non-zero stamp with nothing
                // installed IS a gap: refuse it.
                None => parts_library_revision != 0,
            };
            // Only a part the sender DID send and this store has since lost is
            // worth asking for again. One it never sent is a dangling reference
            // in the document itself: run it, so the ACOMP feature reports it
            // the way it always has. (An install inserts every incoming name,
            // so right after one this set can only be empty — which is what
            // makes the ask-and-retry terminate.)
            let recoverable = brep_kernel::missing_library_parts(&request)
                .iter()
                .any(|name| runner.parts_library_names.contains(name));
            if stale || recoverable {
                runner.parts_library_revision = None;
                return Some(Reply::NeedPartsLibrary);
            }
            let output = runner.run(&request, None);
            Some(Reply::Run(RunReply { generation, output }))
        }
        Command::SetPartsLibrary { library, revision } => {
            runner.parts_library_names = library.keys().cloned().collect();
            brep_kernel::install_parts_library(&library);
            runner.parts_library_revision = Some(revision);
            None
        }
        Command::Query(query) => {
            let result = measure_json(runner, &query);
            Some(Reply::Query(MeasureReply { id: query.id, result }))
        }
        Command::MeshImport(request) => Some(Reply::MeshImport(reconstruct_mesh(request))),
        Command::Reset => {
            // A document switch: drop the delta baseline AND this runner's OWN kernel
            // history cache (the resident registry lives here, not on main), mirroring
            // `set_history_json`'s main-thread clear so a new model rebuilds fully and
            // the old model's handles are freed.
            runner.reset();
            brep_kernel::clear_history_cache();
            None
        }
    }
}

// ===========================================================================
// ThreadRunner (M2b): a persistent std::thread that OWNS the SceneRunner, so a
// history run — and per-object measurement queries — execute OFF the main thread
// and the native UI stays responsive during a run AND during selection. Native
// only: `std::thread` + `std::sync::mpsc` do not exist on wasm32 (M3b lands a
// worker impl behind this same trait), so the whole thing is cfg-gated out there.
// ===========================================================================

/// The persistent-thread runner. `submit_*`/`reset` push [`Command`]s down the
/// channel; `poll_*` first DRAIN every ready [`Reply`] into the two demux buffers,
/// then pop the matching one. The `SceneRunner` (and thus the kernel's resident
/// registry it warms) lives ENTIRELY on the thread — it is never shared — so the
/// only cross-thread traffic is the `Send` command/reply payloads.
#[cfg(not(target_arch = "wasm32"))]
pub struct ThreadRunner {
    /// Main → thread. `Option` so [`Drop`] can take + drop it, ending the thread's
    /// blocking `recv` (the run loop exits, the join completes).
    tx: Option<std::sync::mpsc::Sender<Command>>,
    /// Thread → main.
    rx: std::sync::mpsc::Receiver<Reply>,
    /// The runner thread's join handle (joined on drop for a clean shutdown).
    handle: Option<std::thread::JoinHandle<()>>,
    /// Demuxed completed run replies awaiting `poll_run`.
    run_buf: std::collections::VecDeque<RunReply>,
    /// Demuxed completed measurement replies awaiting `poll_query`.
    query_buf: std::collections::VecDeque<MeasureReply>,
    mesh_import_buf: std::collections::VecDeque<MeshImportReply>,
    /// The parts-library revision last SENT down the channel (`None` = never,
    /// or the thread dropped it). Lives here rather than on the caller so
    /// "reset forgets the library" is a local property of this object.
    sent_library_revision: Option<u64>,
    /// The thread refused a run for want of its library (drained by
    /// [`HistoryRunner::poll_library_request`]).
    library_requested: bool,
}

#[cfg(not(target_arch = "wasm32"))]
impl ThreadRunner {
    pub fn new() -> Self {
        let (tx, cmd_rx) = std::sync::mpsc::channel::<Command>();
        let (reply_tx, rx) = std::sync::mpsc::channel::<Reply>();
        let handle = std::thread::Builder::new()
            .name("brep-history-runner".to_string())
            .spawn(move || thread_main(cmd_rx, reply_tx))
            .expect("spawn brep-history-runner thread");
        Self {
            tx: Some(tx),
            rx,
            handle: Some(handle),
            run_buf: std::collections::VecDeque::new(),
            query_buf: std::collections::VecDeque::new(),
            mesh_import_buf: std::collections::VecDeque::new(),
            sent_library_revision: None,
            library_requested: false,
        }
    }

    /// Pull every ready reply off the channel and demux it into the run/query
    /// buffers (so a `poll_run` never swallows a query reply and vice versa).
    fn drain(&mut self) {
        while let Ok(reply) = self.rx.try_recv() {
            match reply {
                Reply::Run(run) => self.run_buf.push_back(run),
                Reply::Query(query) => self.query_buf.push_back(query),
                Reply::MeshImport(reply) => self.mesh_import_buf.push_back(reply),
                Reply::NeedPartsLibrary => {
                    // The thread dropped (or never had) the library this run
                    // needs. Forget what we believe it holds so the next
                    // `sync_parts_library` re-installs, and flag the refused
                    // run for the caller to re-submit.
                    self.sent_library_revision = None;
                    self.library_requested = true;
                }
            }
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl Default for ThreadRunner {
    fn default() -> Self {
        Self::new()
    }
}

/// The runner thread's whole life: block on the next command, batch it with every
/// other command already queued, COALESCE consecutive `Run`s (a `Run` immediately
/// followed by another `Run` — with no `Query`/`Reset` between — is dropped; only
/// the last of each consecutive group runs), then process the batch IN ORDER so a
/// `Reset` or a `Query` interleaved between two runs keeps its place. Exits when
/// the command sender is dropped (`recv` errors) or the reply receiver is gone (a
/// `send` errors — the main side went away).
#[cfg(not(target_arch = "wasm32"))]
fn thread_main(
    cmd_rx: std::sync::mpsc::Receiver<Command>,
    reply_tx: std::sync::mpsc::Sender<Reply>,
) {
    let mut runner = crate::pipeline::SceneRunner::new();
    while let Ok(first) = cmd_rx.recv() {
        // Gather this command plus everything else already waiting.
        let mut batch = vec![first];
        loop {
            match cmd_rx.try_recv() {
                Ok(command) => batch.push(command),
                Err(_) => break, // Empty or Disconnected — process what we have.
            }
        }
        // A Run immediately followed by another Run is coalesced away (its result
        // would be overwritten before anything observed it); a Run followed by a
        // Query/Reset/end still runs, so interleaved work sees the right geometry.
        let mut run_here: Vec<bool> = vec![true; batch.len()];
        for i in 0..batch.len() {
            if matches!(batch[i], Command::Run { .. })
                && matches!(batch.get(i + 1), Some(Command::Run { .. }))
            {
                run_here[i] = false;
            }
        }
        // Process the KEPT commands in order through the SHARED `process_command`
        // (byte-identical work to the wasm worker's per-message step), sending each
        // reply it produces; a coalesced-away Run is skipped without touching the
        // runner, so interleaved Query/Reset still see the right geometry.
        for (i, command) in batch.into_iter().enumerate() {
            if !run_here[i] {
                continue;
            }
            if let Some(reply) = process_command(&mut runner, command) {
                if reply_tx.send(reply).is_err() {
                    return; // The reply receiver is gone — the main side went away.
                }
            }
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl HistoryRunner for ThreadRunner {
    fn submit_run(&mut self, request: brep_kernel::HistoryRequest, generation: u64) {
        if let Some(tx) = &self.tx {
            let _ = tx.send(Command::Run {
                request,
                generation,
                parts_library_revision: self.sent_library_revision.unwrap_or(0),
            });
        }
    }

    fn sync_parts_library(
        &mut self,
        revision: u64,
        fetch: &mut dyn FnMut() -> brep_kernel::PartsLibraryMap,
    ) {
        if self.sent_library_revision == Some(revision) {
            return;
        }
        if let Some(tx) = &self.tx {
            let _ = tx.send(Command::SetPartsLibrary {
                library: fetch(),
                revision,
            });
            self.sent_library_revision = Some(revision);
        }
    }

    fn poll_library_request(&mut self) -> bool {
        self.drain();
        std::mem::take(&mut self.library_requested)
    }

    fn poll_run(&mut self) -> Option<RunReply> {
        self.drain();
        self.run_buf.pop_front()
    }

    fn submit_query(&mut self, query: MeasureQuery) {
        if let Some(tx) = &self.tx {
            let _ = tx.send(Command::Query(query));
        }
    }

    fn poll_query(&mut self) -> Option<MeasureReply> {
        self.drain();
        self.query_buf.pop_front()
    }

    fn submit_mesh_import(&mut self, request: MeshImportRequest) {
        if let Some(tx) = &self.tx {
            let _ = tx.send(Command::MeshImport(request));
        }
    }

    fn poll_mesh_import(&mut self) -> Option<MeshImportReply> {
        self.drain();
        self.mesh_import_buf.pop_front()
    }

    fn reset(&mut self) {
        if let Some(tx) = &self.tx {
            let _ = tx.send(Command::Reset);
        }
        // A reset is a wholesale document switch: any replies still buffered from
        // the old model are stale — drop them (a fresh run/query supersedes).
        self.run_buf.clear();
        self.query_buf.clear();
        self.mesh_import_buf.clear();
        // `Command::Reset` clears the thread's kernel store, library included,
        // so what we believe it holds is void.
        self.sent_library_revision = None;
        self.library_requested = false;
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl Drop for ThreadRunner {
    fn drop(&mut self) {
        // Dropping the sender ends the thread's blocking `recv`; then join so the
        // thread (and its resident registry) is fully torn down before we return.
        self.tx.take();
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

// ===========================================================================
// ThreadRunner integration tests (native only — the async run/query machine the
// wasm WorkerRunner (M3) reuses). Driven end-to-end: submit → spin poll → assert.
// ===========================================================================
#[cfg(all(test, not(target_arch = "wasm32")))]
mod thread_tests {
    use super::*;

    /// A one-feature history: a P.CU cube `Box` of side 20 (volume 8000).
    fn box_request() -> brep_kernel::HistoryRequest {
        serde_json::from_str(
            r#"{
                "expressions": "", "configurator": {},
                "features": [{
                    "type": "P.CU",
                    "inputParams": {
                        "id": "Box", "sizeX": 20.0, "sizeY": 20.0, "sizeZ": 20.0,
                        "transform": { "position": [0,0,0], "rotationEuler": [0,0,0], "scale": [1,1,1] },
                        "boolean": { "targets": [], "operation": "NONE" }
                    },
                    "persistentData": {}
                }]
            }"#,
        )
        .unwrap()
    }

    fn spin_run(runner: &mut ThreadRunner) -> RunReply {
        for _ in 0..3000 {
            if let Some(reply) = runner.poll_run() {
                return reply;
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        panic!("thread run did not complete within the spin budget");
    }

    fn spin_query(runner: &mut ThreadRunner) -> MeasureReply {
        for _ in 0..3000 {
            if let Some(reply) = runner.poll_query() {
                return reply;
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        panic!("thread query did not complete within the spin budget");
    }

    /// A submitted run executes ON THE THREAD and its reply is drained back with
    /// the right generation + a freshly-tessellated snapshot entry for the box.
    #[test]
    fn thread_runner_runs_and_replies() {
        let mut runner = ThreadRunner::new();
        runner.submit_run(box_request(), 1);
        let reply = spin_run(&mut runner);
        assert_eq!(reply.generation, 1, "reply carries its submit generation");
        assert_eq!(reply.output.snapshot.len(), 1, "one solid (the box)");
        let (name, _handle, display) = &reply.output.snapshot[0];
        assert_eq!(name, "Box");
        assert!(display.is_some(), "first emit is a fresh tessellation, not a reuse");
        assert_eq!(reply.output.provenance, vec![("Box".to_string(), "Box".to_string())]);
    }

    /// A measurement query routes to the thread's WARM registry (populated by the
    /// preceding run) and returns the exact object-info fragment.
    #[test]
    fn thread_runner_measures_after_run() {
        let mut runner = ThreadRunner::new();
        runner.submit_run(box_request(), 1);
        let _ = spin_run(&mut runner);
        runner.submit_query(MeasureQuery {
            id: 7,
            kind: MeasureKind::Solid,
            owner: "Box".to_string(),
            entity: String::new(),
            density: 1.0,
        });
        let reply = spin_query(&mut runner);
        assert_eq!(reply.id, 7);
        let info: serde_json::Value = serde_json::from_str(&reply.result).unwrap();
        assert_eq!(info["ok"], true, "measured on the thread: {}", reply.result);
        assert!(
            (info["volume"].as_f64().unwrap() - 8000.0).abs() < 1.0,
            "box volume {} != 8000",
            info["volume"]
        );
    }

    /// A re-submitted IDENTICAL run replays the thread's incremental cache: the
    /// second reply is a REUSE (`None` in the snapshot), proving the runner's
    /// baseline + the kernel cache both persist across submits on the thread.
    #[test]
    fn thread_runner_reuses_on_identical_resubmit() {
        let mut runner = ThreadRunner::new();
        runner.submit_run(box_request(), 1);
        let first = spin_run(&mut runner);
        assert!(first.output.snapshot[0].2.is_some(), "first is fresh");

        runner.submit_run(box_request(), 2);
        let second = spin_run(&mut runner);
        assert_eq!(second.generation, 2);
        assert!(
            second.output.snapshot[0].2.is_none(),
            "identical resubmit replays as a REUSE (unchanged handle)"
        );
    }

    // --- the PARTS-LIBRARY channel ---------------------------------------
    //
    // The library is sent when it CHANGES, not with every run (sending it per
    // run stringified megabytes of embedded part payload on the browser's main
    // thread for every edit). These cover the two ways that can go wrong.

    /// A one-instance assembly: an ACOMP referencing part `widget`, plus the
    /// library entry that defines it.
    fn widget_part() -> (brep_kernel::PartsLibraryMap, brep_kernel::HistoryRequest) {
        let document = serde_json::json!({
            "expressions": "", "configurator": {},
            "features": [{
                "type": "P.CU",
                "inputParams": {
                    "id": "Part", "sizeX": 4.0, "sizeY": 4.0, "sizeZ": 4.0,
                    "transform": { "position": [0,0,0], "rotationEuler": [0,0,0], "scale": [1,1,1] },
                    "boolean": { "targets": [], "operation": "NONE" }
                },
                "persistentData": {}
            }]
        });
        let mut library = brep_kernel::PartsLibraryMap::new();
        library.insert(
            "widget".to_string(),
            brep_kernel::PartsLibraryEntry {
                document,
                ..Default::default()
            },
        );
        let request = serde_json::from_str(
            r#"{
                "expressions": "", "configurator": {},
                "features": [{
                    "type": "ACOMP",
                    "inputParams": {
                        "id": "ACOMP1", "partName": "widget",
                        "transform": { "position": [0,0,0], "rotationEuler": [0,0,0], "scale": [1,1,1] }
                    },
                    "persistentData": {}
                }]
            }"#,
        )
        .unwrap();
        (library, request)
    }

    /// Spin until the thread either completes the run or REFUSES it for want of
    /// its parts library (`None`).
    fn spin_reply(runner: &mut ThreadRunner) -> Option<RunReply> {
        for _ in 0..3000 {
            if runner.poll_library_request() {
                return None;
            }
            if let Some(reply) = runner.poll_run() {
                return Some(reply);
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        panic!("thread run neither completed nor asked for the library");
    }

    /// The steady state: the library is installed ONCE and every later run
    /// resolves its ACOMP against the thread's resident copy — no run carries
    /// the payload, and the fetch closure is called exactly once.
    #[test]
    fn thread_runner_installs_the_library_once_and_reuses_it() {
        let (library, request) = widget_part();
        let mut runner = ThreadRunner::new();
        let mut fetches = 0;
        for generation in 1..=3 {
            runner.sync_parts_library(1, &mut || {
                fetches += 1;
                library.clone()
            });
            runner.submit_run(request.clone(), generation);
            let reply = spin_reply(&mut runner).expect("the run executed");
            assert_eq!(reply.generation, generation);
            assert_eq!(
                reply.output.snapshot.len(),
                1,
                "the ACOMP resolved against the resident library"
            );
        }
        assert_eq!(fetches, 1, "the library is sent once, not once per run");
    }

    /// THE DRIFT CASE, and why the runner preflights CONTENT and not just a
    /// revision. The kernel's orphan GC drops entries no ACOMP in the run
    /// referenced, so a componentless run empties the THREAD's library while the
    /// sender's store — and therefore its revision — never moves. The next run
    /// that does reference the part must be refused, not executed against the
    /// emptied store, and must recover.
    #[test]
    fn thread_runner_recovers_after_its_library_is_gc_d() {
        let (library, request) = widget_part();
        let empty: brep_kernel::HistoryRequest =
            serde_json::from_str(r#"{"expressions":"","configurator":{},"features":[]}"#).unwrap();
        let mut runner = ThreadRunner::new();

        runner.sync_parts_library(1, &mut || library.clone());
        runner.submit_run(request.clone(), 1);
        assert!(spin_reply(&mut runner).is_some(), "first run executes");

        // A componentless run: the thread's GC drops `widget`. Nothing about the
        // revision changed — only the thread's CONTENT did.
        runner.sync_parts_library(1, &mut || library.clone());
        runner.submit_run(empty, 2);
        assert!(spin_reply(&mut runner).is_some(), "the empty run executes");

        // Back to the assembly. The content preflight must catch the gap.
        runner.sync_parts_library(1, &mut || library.clone());
        runner.submit_run(request.clone(), 3);
        assert!(
            spin_reply(&mut runner).is_none(),
            "a run whose part the thread GC'd must be REFUSED, never run against \
             an empty library"
        );

        // Recovery: the refusal forgot the sent revision, so this re-installs.
        let mut refetched = 0;
        runner.sync_parts_library(1, &mut || {
            refetched += 1;
            library.clone()
        });
        assert_eq!(refetched, 1, "the refusal forces a re-send");
        runner.submit_run(request, 4);
        let reply = spin_reply(&mut runner).expect("the re-driven run executes");
        assert_eq!(reply.generation, 4);
        assert_eq!(reply.output.snapshot.len(), 1, "the component is back");
    }

    /// The revision half of the preflight: a run stamped for a library the
    /// runner never received is refused rather than executed.
    #[test]
    fn a_run_stamped_for_an_unknown_library_is_refused() {
        let (_library, request) = widget_part();
        let mut scene = crate::pipeline::SceneRunner::new();
        let refusal = process_command(
            &mut scene,
            Command::Run {
                request,
                generation: 1,
                parts_library_revision: 7,
            },
        );
        assert!(
            matches!(refusal, Some(Reply::NeedPartsLibrary)),
            "an unknown library stamp must refuse the run"
        );
        assert_eq!(
            scene.parts_library_revision, None,
            "the refusal drops what the runner believed it held"
        );
    }

    /// The ask-and-retry must TERMINATE. A document referencing a part its own
    /// library never had is a dangling reference, not a lost cache entry: the
    /// run goes ahead so the ACOMP feature reports it, exactly as before the
    /// library moved off the request.
    #[test]
    fn a_part_the_library_never_had_errors_instead_of_looping() {
        let (_library, request) = widget_part();
        let mut runner = ThreadRunner::new();
        // Install an EMPTY library at revision 1 — `widget` was never sent.
        runner.sync_parts_library(1, &mut brep_kernel::PartsLibraryMap::new);
        runner.submit_run(request, 1);
        let reply = spin_reply(&mut runner)
            .expect("a dangling part name must not refuse the run forever");
        assert!(
            !reply.output.report.feature_errors.is_empty(),
            "the ACOMP reports the missing part: {:?}",
            reply.output.report
        );
    }

    /// A `reset` clears the thread's delta baseline AND its kernel cache, so the
    /// next run rebuilds fresh (a new handle ⇒ a `Some` snapshot, not a reuse).
    #[test]
    fn thread_runner_reset_forces_full_rebuild() {
        let mut runner = ThreadRunner::new();
        runner.submit_run(box_request(), 1);
        let _ = spin_run(&mut runner);
        runner.reset();
        runner.submit_run(box_request(), 2);
        let reply = spin_run(&mut runner);
        assert!(
            reply.output.snapshot[0].2.is_some(),
            "after reset the box re-tessellates (baseline dropped)"
        );
    }

    /// The wasm worker WIRE FORMAT: a `Command::Run` — carrying a `HistoryRequest`
    /// whose brand-new `Serialize` must honor its field renames (`type`,
    /// `inputParams`, …) — survives the serde-JSON round trip the `WorkerRunner`'s
    /// `postMessage` uses, and the round-tripped request still executes. (The
    /// `RunOutput`/`RunReply` reply half already has its own round-trip test.)
    #[test]
    fn run_command_round_trips_through_serde() {
        let command = Command::Run {
            request: box_request(),
            generation: 42,
            parts_library_revision: 9,
        };
        let json = serde_json::to_string(&command).expect("Command serializes");
        let back: Command = serde_json::from_str(&json).expect("Command deserializes");
        match back {
            Command::Run {
                request,
                generation,
                parts_library_revision,
            } => {
                assert_eq!(generation, 42, "generation round-trips");
                assert_eq!(parts_library_revision, 9, "library stamp round-trips");
                assert_eq!(request.features.len(), 1);
                // The `type` rename survived serialize → deserialize.
                assert_eq!(request.features[0].feature_type, "P.CU");
                // The deserialized request is still executable on a fresh runner.
                let output = crate::pipeline::SceneRunner::new().run(&request, None);
                assert_eq!(output.snapshot.len(), 1);
                assert_eq!(output.snapshot[0].0, "Box");
            }
            _ => panic!("round-tripped to the wrong Command variant"),
        }
    }

    /// Browser workers receive the same mesh command as JSON. Preserve every
    /// source byte and the request id/format across that boundary.
    #[test]
    fn mesh_import_command_round_trips_through_serde() {
        let command = Command::MeshImport(MeshImportRequest {
            id: 17,
            format: MeshImportFormat::Stl,
            bytes: vec![0, 1, 127, 128, 255],
        });
        let json = serde_json::to_string(&command).expect("mesh command serializes");
        let back: Command = serde_json::from_str(&json).expect("mesh command deserializes");
        match back {
            Command::MeshImport(request) => {
                assert_eq!(request.id, 17);
                assert!(matches!(request.format, MeshImportFormat::Stl));
                assert_eq!(request.bytes, vec![0, 1, 127, 128, 255]);
            }
            _ => panic!("round-tripped to the wrong Command variant"),
        }
    }
}