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,
}

/// 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>;
    /// Drop the delta baseline (document switch → full rebuild).
    fn reset(&mut self);
}

/// 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)) => serde_json::json!({
                "ok": true,
                "kind": "face",
                "solid": query.owner,
                "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>,
}

impl InlineRunner {
    pub fn new() -> Self {
        Self {
            runner: crate::pipeline::SceneRunner::new(),
            pending: std::collections::VecDeque::new(),
            query_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 });
    }

    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 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,
    },
    Query(MeasureQuery),
    Reset,
}

/// A reply sent runner → main; the main side demuxes it into per-kind buffers so
/// `poll_run`/`poll_query` each pop their own stream.
#[derive(serde::Serialize, serde::Deserialize)]
pub enum Reply {
    Run(RunReply),
    Query(MeasureReply),
}

/// 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 } => {
            let output = runner.run(&request, None);
            Some(Reply::Run(RunReply { generation, output }))
        }
        Command::Query(query) => {
            let result = measure_json(runner, &query);
            Some(Reply::Query(MeasureReply { id: query.id, result }))
        }
        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>,
}

#[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(),
        }
    }

    /// 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),
            }
        }
    }
}

#[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 });
        }
    }

    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 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();
    }
}

#[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)"
        );
    }

    /// 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 };
        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 } => {
                assert_eq!(generation, 42, "generation 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"),
        }
    }
}