bywind-viz 0.2.0

use std::sync::mpsc::{Receiver, TryRecvError};

use bywind::{
    BoatConfig, EnsembleMode, GenerateConfig, LonLatBbox, MapBounds, RealizationRun, SearchConfig,
    SearchError, SearchPhase, SearchProgressEvent, SearchResult, SearchWeights, TimedWindMap,
    WindInput, route_evolution_match, run_realizations, run_search_blocking,
    run_search_blocking_with_baked, run_time_reopt_blocking,
};

use crate::config::{EditorState, Tool, ViewState};
use crate::search::{ReoptMsg, SearchOutputs};

/// We derive Deserialize/Serialize so we can persist app state on shutdown.
#[derive(serde::Deserialize, serde::Serialize)]
#[serde(default)] // if we add new fields, give them default values when deserializing old state
pub struct BywindApp {
    pub(crate) editor: EditorState,
    pub(crate) view: ViewState,
    pub(crate) generate: GenerateConfig,
    pub(crate) search: SearchConfig,
    pub(crate) boat: BoatConfig,
    pub(crate) outputs: SearchOutputs,

    #[serde(skip)]
    pub(crate) wind_map: Option<TimedWindMap>,

    /// Background sailing-search worker.
    #[serde(skip)]
    pub(crate) search_job: AsyncJob<Result<SearchResult, SearchError>>,

    /// Wall-clock instant the running search started, so the Cancel
    /// button can render an elapsed-time suffix. `Some` only while
    /// `search_job` is in `Running`; cleared when the result arrives or
    /// the user cancels.
    #[serde(skip)]
    pub(crate) search_started_at: Option<std::time::Instant>,

    /// Background time-only PSO fired on Waypoint-Edit drag-release.
    /// Each new drag overwrites the slot, cancelling the prior worker.
    #[serde(skip)]
    pub(crate) reopt_job: AsyncJob<ReoptMsg>,

    /// Background per-realization PSO cohort. Fired from the
    /// "Run per realization" button; on success its payload replaces
    /// `outputs.realization_runs`. Independent of `search_job` so a
    /// main search can run alongside (though the UI gates them
    /// sequentially today to keep CPU contention predictable).
    #[serde(skip)]
    pub(crate) realization_job: AsyncJob<Result<Vec<RealizationRun>, SearchError>>,

    /// Wall-clock instant the running realization cohort started,
    /// mirroring `search_started_at` so the button can show elapsed
    /// time while the K-search loop is running.
    #[serde(skip)]
    pub(crate) realization_started_at: Option<std::time::Instant>,

    /// Background decoder for the embedded `wind_av1` sample dataset.
    /// Spawned at startup when `assets/sample_wind.wcav` was present at
    /// build time; the decoded `TimedWindMap` slots into `wind_map` as
    /// soon as it's ready.
    #[serde(skip)]
    pub(crate) bundled_sample_job: AsyncJob<Result<TimedWindMap, String>>,

    /// Side channel carrying in-progress events from the search worker
    /// (e.g. the A* benchmark, which is computed before the main PSO
    /// and worth showing immediately). Separate from `search_job` so
    /// that abstraction stays single-shot for the terminal result.
    /// `None` between searches.
    #[serde(skip)]
    pub(crate) search_progress_rx: Option<std::sync::mpsc::Receiver<SearchProgressEvent>>,

    /// Latest [`SearchPhase`] event received over `search_progress_rx`.
    /// `Some(_)` while a search is running so the UI can show
    /// "baking…" / "computing benchmark…" / "searching…" instead of a
    /// generic spinner; cleared back to `None` on the terminal `Done`.
    #[serde(skip)]
    pub(crate) current_search_phase: Option<SearchPhase>,

    /// Background worker + log buffer for `File → Fetch from AWS…`.
    /// Streams `bywind::fetch::FetchProgress` events back via mpsc and
    /// holds the shared cancel flag.
    #[cfg(not(target_arch = "wasm32"))]
    #[serde(skip)]
    pub(crate) fetch_job: crate::fetch::FetchJob,

    /// Background worker + log buffer for `File → Fetch GEFS
    /// Ensemble…`. Same shape as [`Self::fetch_job`] but iterates
    /// over K members and writes one `.wcav` per member to disk;
    /// on success its poll yields the output directory so the app
    /// can drop it straight into `SearchConfig::ensemble_path`.
    #[cfg(not(target_arch = "wasm32"))]
    #[serde(skip)]
    pub(crate) fetch_ensemble_job: crate::fetch::FetchEnsembleJob,

    /// User-facing error rendered as a toast. Cleared on Dismiss.
    #[serde(skip)]
    pub(crate) last_error: Option<String>,
}

// Every field uses its type's `Default`, including `wind_map: None`.
// `clippy::derivable_impls` would have us `#[derive(Default)]` instead,
// but the explicit impl is the natural place to call out *why* there's
// no up-front wind data: the embedded `wind_av1` sample (when present)
// decodes asynchronously and slots in within a few seconds, and
// without a sample the user reaches for `Advanced → Generate Wind Map`
// or `File → Load GRIB2…`. The prior behaviour of synthesising a
// random grid up-front was just a distractor that always got replaced.
#[expect(
    clippy::derivable_impls,
    reason = "documents the no-wind-map startup choice"
)]
impl Default for BywindApp {
    fn default() -> Self {
        Self {
            editor: EditorState::default(),
            view: ViewState::default(),
            generate: GenerateConfig::default(),
            search: SearchConfig::default(),
            boat: BoatConfig::default(),
            outputs: SearchOutputs::default(),
            wind_map: None,
            search_job: AsyncJob::default(),
            search_started_at: None,
            reopt_job: AsyncJob::default(),
            realization_job: AsyncJob::default(),
            realization_started_at: None,
            search_progress_rx: None,
            current_search_phase: None,
            bundled_sample_job: AsyncJob::default(),
            #[cfg(not(target_arch = "wasm32"))]
            fetch_job: crate::fetch::FetchJob::default(),
            #[cfg(not(target_arch = "wasm32"))]
            fetch_ensemble_job: crate::fetch::FetchEnsembleJob::default(),
            last_error: None,
        }
    }
}

impl BywindApp {
    /// Log `message` and surface it as a UI toast.
    pub(crate) fn report_error(&mut self, message: String) {
        log::error!("{message}");
        self.last_error = Some(message);
    }

    /// While CTRL is held, force `Tool::Pointer` so the user can pan /
    /// inspect without changing their selected tool. Release restores.
    pub(crate) fn apply_ctrl_pointer_override(&mut self, ctx: &egui::Context) {
        let ctrl = ctx.input(|i| i.modifiers.ctrl);
        if ctrl {
            if self.editor.pre_ctrl_tool.is_none() && self.editor.selected_tool != Tool::Pointer {
                self.editor.pre_ctrl_tool = Some(self.editor.selected_tool);
                self.editor.selected_tool = Tool::Pointer;
            }
        } else if let Some(prev) = self.editor.pre_ctrl_tool.take() {
            self.editor.selected_tool = prev;
        }
    }

    pub fn new(cc: &eframe::CreationContext<'_>) -> Self {
        let mut app: Self = if let Some(storage) = cc.storage {
            eframe::get_value(storage, eframe::APP_KEY).unwrap_or_default()
        } else {
            Default::default()
        };
        // Kick off the sample loader on every cold start. The random
        // wind map from `Default::default()` stays visible while the
        // worker runs — embed-decode (~12 s) or cache-decode is fast;
        // first-launch download adds the network round-trip on top.
        #[cfg(not(target_arch = "wasm32"))]
        app.start_bundled_sample_decode(&cc.egui_ctx);
        app
    }

    /// Spawn a worker that loads the `wind_av1` sample dataset —
    /// going through embed → cache → download per `bundled_sample`
    /// — then decodes the bytes into a `TimedWindMap`. Replaces an
    /// already-running job so the menu entry can re-trigger after the
    /// user has loaded other data. No-op on wasm.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn start_bundled_sample_decode(&mut self, ctx: &egui::Context) {
        let ctx = ctx.clone();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let result = crate::bundled_sample::load_sample_bytes()
                .and_then(|bytes| bywind::wind_av1::decode(&bytes[..]).map_err(|e| e.to_string()));
            drop(tx.send(result));
            ctx.request_repaint();
        });
        self.bundled_sample_job.set_running(rx);
    }

    /// Spawn a background sailing search. Validates up-front so a bad
    /// slider doesn't waste a worker thread silently.
    pub(crate) fn run_search(&mut self, ctx: &egui::Context) {
        if self.search_job.is_running() {
            return;
        }
        // Missing endpoints would silently fall back to the wind-map
        // bbox corners (see `MapBounds::resolve_endpoints`). The result
        // is rarely what the user wants, and they have no visual cue
        // that the fallback was used. Refuse the click, animate the
        // Set Endpoints tool, and let the user place them.
        if self.editor.start_waypoint.is_none() || self.editor.end_waypoint.is_none() {
            self.editor.highlight_endpoint_tool = true;
            ctx.request_repaint();
            return;
        }
        if let Err(e) = self.boat.validate() {
            self.report_error(format!("Invalid boat config — {e}"));
            return;
        }
        if let Err(e) = self.search.validate() {
            self.report_error(format!("Invalid search config — {e}"));
            return;
        }
        // Ensemble dispatch: when the user has set an ensemble dir in
        // advanced settings, the GUI doesn't need an in-memory wind
        // map — the worker thread will load the ensemble itself. We
        // still need *some* map bounds to derive the bake / route
        // window; the user-drawn route bbox is the only source we
        // have without paying the load cost on the UI thread, so
        // require it.
        let ensemble_dispatch = self.search.ensemble_path.clone();
        let map_bounds = if let Some(_ensemble) = &ensemble_dispatch {
            let Some(bbox) = self.editor.route_bbox else {
                self.report_error(
                    "Ensemble search needs a route bbox to size the bake. \
                     Draw one with the Route Bounds tool, then click Run Search."
                        .to_owned(),
                );
                return;
            };
            // `EditorState::route_bbox` is `(lon_min, lon_max, lat_min,
            // lat_max)` per io.rs:286 (the TOML layout convention).
            // Naming each field at the construction site makes a
            // permutation visible to a reader — commit aa57635 fixed
            // a bug where the tuple components were silently reordered
            // into a positional `LonLatBbox::new(...)` call.
            MapBounds {
                bbox: LonLatBbox {
                    lon_min: bbox.0,
                    lon_max: bbox.1,
                    lat_min: bbox.2,
                    lat_max: bbox.3,
                },
            }
        } else {
            let Some(wind_map) = &self.wind_map else {
                return;
            };
            let Some(mb) = MapBounds::from_wind_map(wind_map) else {
                return;
            };
            mb
        };
        let bounds = map_bounds.clamp_to(self.editor.route_bbox);
        if !bounds.is_non_degenerate() {
            self.report_error(
                "Route Bounds rectangle does not overlap the wind map; clear it or redraw."
                    .to_owned(),
            );
            return;
        }

        let (origin, destination) =
            bounds.resolve_endpoints(self.editor.start_waypoint, self.editor.end_waypoint);
        // Both endpoints are present by the early-return above; clear
        // the "set endpoints" prompt now that the worker is launching.
        self.editor.highlight_endpoint_tool = false;
        let route_bounds = bounds.to_route_bounds_with_step_fraction(
            origin,
            destination,
            self.search.step_distance_fraction,
        );
        let bake_bounds = bounds.to_bake_bounds(self.search.bake_step_deg);
        let sdf_resolution = self.search.sdf_resolution_deg;
        let fine_sdf_resolution = self.search.fine_sdf_resolution_deg;

        let ctx = ctx.clone();
        let waypoint_count = self.search.waypoint_count;
        let weights = SearchWeights {
            time_weight: self.search.time_weight,
            fuel_weight: self.search.fuel_weight,
            land_weight: self.search.land_weight,
        };
        // Resolve the seed up front so the Advanced Params UI can show
        // what the search actually ran with — including when the user
        // left "Deterministic seed" unchecked (we still pick a u64 here
        // and pass it through as `Some(_)` so the run is reproducible
        // after the fact via the displayed value).
        let mut search_settings = self.search.to_search_settings();
        let effective_seed = search_settings.seed.unwrap_or_else(rand::random);
        search_settings.seed = Some(effective_seed);
        self.outputs.last_search_seed = Some(effective_seed);
        let ship = self.boat.to_boat();
        let ensemble_mode = self.search.ensemble_mode;
        let (tx, rx) = std::sync::mpsc::channel();
        // Side channel for in-progress events (phase transitions and
        // the early-emitted A* benchmark). Stored on the app so the
        // per-frame poll can drain it; cleared when the terminal
        // `Done` event arrives.
        let (progress_tx, progress_rx) = std::sync::mpsc::channel::<SearchProgressEvent>();
        self.search_progress_rx = Some(progress_rx);
        // Seed the phase indicator before any event arrives so the
        // status label switches off "Searching…" the moment the
        // worker spawns rather than after the first Phase event.
        // The worker overwrites this within milliseconds.
        self.current_search_phase = Some(if ensemble_dispatch.is_some() {
            SearchPhase::Loading
        } else {
            SearchPhase::Baking
        });
        // Drop any benchmark from a previous search so the dashed
        // overlay doesn't linger as a stale reference while the new
        // search runs. The new bench lands via `BenchmarkReady` as
        // soon as the worker's A* + time-PSO step completes.
        self.outputs.benchmark = None;
        // Same for any leftover live gbest from a previous search.
        // Per-iteration snapshots from the new search will repopulate
        // it once the PSO loop starts.
        self.outputs.live_gbest = None;

        if let Some(ensemble_path) = ensemble_dispatch {
            // Ensemble path: load + bake + search all in the worker
            // thread so the UI stays responsive during the ~3 s load.
            std::thread::spawn(move || {
                let ctx_progress = ctx.clone();
                let mut progress = move |ev: SearchProgressEvent| {
                    drop(progress_tx.send(ev));
                    ctx_progress.request_repaint();
                };
                let result = run_ensemble_search(
                    &ensemble_path,
                    bake_bounds,
                    route_bounds,
                    waypoint_count,
                    search_settings,
                    ship,
                    weights,
                    sdf_resolution,
                    fine_sdf_resolution,
                    ensemble_mode,
                    &mut progress,
                );
                drop(tx.send(result));
                ctx.request_repaint();
            });
        } else {
            // Single-deterministic path: existing flow. `wind_map` is
            // guaranteed `Some` here because the ensemble dispatch
            // branch above is the only way to skip the `wind_map`
            // requirement, and we wouldn't be on this branch if it
            // had fired.
            let wind_map_snapshot: TimedWindMap = self
                .wind_map
                .as_ref()
                .expect("wind_map presence is checked at the top of run_search")
                .clone();
            std::thread::spawn(move || {
                let ctx_progress = ctx.clone();
                let mut progress = move |ev: SearchProgressEvent| {
                    drop(progress_tx.send(ev));
                    ctx_progress.request_repaint();
                };
                let result = run_search_blocking(
                    &wind_map_snapshot,
                    bake_bounds,
                    route_bounds,
                    waypoint_count,
                    search_settings,
                    ship,
                    weights,
                    sdf_resolution,
                    fine_sdf_resolution,
                    &mut progress,
                );
                drop(tx.send(result));
                ctx.request_repaint();
            });
        }

        self.search_job.set_running(rx);
        self.search_started_at = Some(std::time::Instant::now());
    }

    /// Per-realization PSO cohort. Runs K independent
    /// single-deterministic searches, one per ensemble member viewed
    /// as a separate weather realization, so the user can see what
    /// alternate routes each realization's wind would produce —
    /// complement to the ensemble spread, which only scores the chosen
    /// gbest against each member. Requires `ensemble_path` and a route
    /// bbox to be set; otherwise reports a toast and bails. The K
    /// searches run sequentially in the worker to keep CPU usage
    /// predictable (each search is already internally rayon-parallel).
    pub(crate) fn run_realizations(&mut self, ctx: &egui::Context) {
        if self.realization_job.is_running() {
            return;
        }
        let Some(ensemble_path) = self.search.ensemble_path.clone() else {
            self.report_error(
                "Per-realization run needs an ensemble dir set in Advanced Params.".to_owned(),
            );
            return;
        };
        let Some(route_bbox) = self.editor.route_bbox else {
            self.report_error(
                "Per-realization run needs a route bbox to size the bake. Draw one with \
                 the Route Bounds tool, then click Run per realization."
                    .to_owned(),
            );
            return;
        };
        if let Err(e) = self.boat.validate() {
            self.report_error(format!("Invalid boat config — {e}"));
            return;
        }
        if let Err(e) = self.search.validate() {
            self.report_error(format!("Invalid search config — {e}"));
            return;
        }
        let map_bounds = MapBounds {
            bbox: LonLatBbox {
                lon_min: route_bbox.0,
                lon_max: route_bbox.1,
                lat_min: route_bbox.2,
                lat_max: route_bbox.3,
            },
        };
        let bounds = map_bounds.clamp_to(Some(route_bbox));
        if !bounds.is_non_degenerate() {
            self.report_error(
                "Route Bounds rectangle is degenerate; clear it or redraw.".to_owned(),
            );
            return;
        }
        let (origin, destination) =
            bounds.resolve_endpoints(self.editor.start_waypoint, self.editor.end_waypoint);
        self.editor.highlight_endpoint_tool = false;
        let route_bounds = bounds.to_route_bounds_with_step_fraction(
            origin,
            destination,
            self.search.step_distance_fraction,
        );
        let bake_bounds = bounds.to_bake_bounds(self.search.bake_step_deg);
        let sdf_resolution = self.search.sdf_resolution_deg;
        let fine_sdf_resolution = self.search.fine_sdf_resolution_deg;
        let waypoint_count = self.search.waypoint_count;
        let weights = SearchWeights {
            time_weight: self.search.time_weight,
            fuel_weight: self.search.fuel_weight,
            land_weight: self.search.land_weight,
        };
        // Resolve the seed up front like `run_search`. We then derive
        // per-realization seeds as `base + k` inside the worker so
        // distinct realizations aren't initialised with the exact same
        // swarm cloud — identical seeds would let two near-identical
        // winds collapse to bit-identical routes, defeating the point
        // of an overlay.
        let mut search_settings = self.search.to_search_settings();
        let base_seed = search_settings.seed.unwrap_or_else(rand::random);
        search_settings.seed = Some(base_seed);
        let ship = self.boat.to_boat();
        let ctx = ctx.clone();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let result = run_realization_searches(
                &ensemble_path,
                bake_bounds,
                route_bounds,
                waypoint_count,
                search_settings,
                base_seed,
                ship,
                weights,
                sdf_resolution,
                fine_sdf_resolution,
            );
            drop(tx.send(result));
            ctx.request_repaint();
        });
        self.realization_job.set_running(rx);
        self.realization_started_at = Some(std::time::Instant::now());
    }

    /// Re-run only the time PSO with the user-edited xy fixed. Each
    /// drag-release overwrites the slot so the prior reopt is cancelled.
    pub(crate) fn start_time_reopt(&mut self, ctx: &egui::Context) {
        let Some(baked_ref) = self.outputs.baked_wind_map.as_ref() else {
            return;
        };
        let Some(rb) = self.outputs.route_bounds else {
            return;
        };
        let Some(re) = self.outputs.route_evolution.as_ref() else {
            return;
        };

        let iteration = self.outputs.iteration;
        let weights = SearchWeights {
            time_weight: self.search.time_weight,
            fuel_weight: self.search.fuel_weight,
            land_weight: self.search.land_weight,
        };
        let settings = self.search.to_search_settings();
        let ship = self.boat.to_boat();
        let sdf_resolution = self.search.sdf_resolution_deg;
        let fine_sdf_resolution = self.search.fine_sdf_resolution_deg;

        let baked = baked_ref.clone();
        let ctx = ctx.clone();
        let (tx, rx) = std::sync::mpsc::channel::<ReoptMsg>();

        // The macro dispatches over the gbest path's const-generic N
        // so a typed `Path<N>` can move into the closure.
        route_evolution_match!(re, |evo| {
            let frames = evo.frames();
            let iter_idx = iteration.min(frames.len().saturating_sub(1));
            let Some(particles) = frames.get(iter_idx) else {
                return;
            };
            let Some(best) = particles.iter().max_by(|a, b| {
                a.best_fit
                    .partial_cmp(&b.best_fit)
                    .unwrap_or(std::cmp::Ordering::Equal)
            }) else {
                return;
            };
            let fixed_path = best.best_pos;
            std::thread::spawn(move || {
                let path = run_time_reopt_blocking(
                    &baked,
                    rb,
                    settings,
                    &ship,
                    fixed_path,
                    weights,
                    sdf_resolution,
                    fine_sdf_resolution,
                );
                // Drop silently if a later drag-release overwrote the slot.
                drop(tx.send(ReoptMsg {
                    iteration: iter_idx,
                    new_times: path.t.0.0.to_vec(),
                }));
                ctx.request_repaint();
            });
        });

        self.reopt_job.set_running(rx);
    }

    /// Splice a reopt result's `new_times` into the gbest particle's `best_pos.t`
    /// at the recorded iteration. The length check inside `apply_reopt_times`
    /// drops the message if the path's N has changed since dispatch (e.g. the
    /// user re-ran the search with a different waypoint count between drag-
    /// release and result arrival).
    pub(crate) fn apply_reopt_result(&mut self, msg: &ReoptMsg) {
        if let Some(re) = self.outputs.route_evolution.as_mut() {
            re.apply_reopt_times(msg.iteration, &msg.new_times);
        }
    }
}

/// Single-shot background-job slot. Wraps the `Option<Receiver<T>>` +
/// `try_recv` polling pattern used by the search worker and the time-reopt
/// worker.
///
/// Idle by default. `set_running` parks a fresh receiver; `poll` drains the
/// channel, transitioning back to `Idle` on either `Ok` or `Disconnected`
/// so the caller doesn't have to distinguish "result arrived" from "worker
/// died". Lives in `app.rs` because both call sites — and the abstraction
/// itself — are tied to `BywindApp`'s update loop.
#[derive(Default)]
pub(crate) enum AsyncJob<T> {
    #[default]
    Idle,
    Running(Receiver<T>),
}

impl<T> AsyncJob<T> {
    pub(crate) fn is_running(&self) -> bool {
        matches!(self, Self::Running(_))
    }

    /// Replaces any in-flight receiver. The prior one's eventual
    /// `tx.send` becomes a no-op — used by time-reopt to cancel the
    /// prior worker on each drag-release.
    pub(crate) fn set_running(&mut self, rx: Receiver<T>) {
        *self = Self::Running(rx);
    }

    /// Drop the receiver and go back to `Idle` without waiting on the
    /// worker. Used by the Cancel button: the spawned thread keeps
    /// running until it finishes naturally, but its eventual `tx.send`
    /// becomes a no-op so the UI is responsive immediately.
    pub(crate) fn cancel(&mut self) {
        *self = Self::Idle;
    }

    /// `Some(v)` once, then `Idle` on either `Ok` or `Disconnected`.
    pub(crate) fn poll(&mut self) -> Option<T> {
        let rx = match self {
            Self::Idle => return None,
            Self::Running(rx) => rx,
        };
        match rx.try_recv() {
            Ok(value) => {
                *self = Self::Idle;
                Some(value)
            }
            Err(TryRecvError::Disconnected) => {
                *self = Self::Idle;
                None
            }
            Err(TryRecvError::Empty) => None,
        }
    }
}

impl eframe::App for BywindApp {
    /// Per-frame orchestrator. Panel render order matches egui's
    /// requirement: top/bottom → side → central.
    fn ui(&mut self, ui: &mut egui::Ui, _frame: &mut eframe::Frame) {
        self.apply_ctrl_pointer_override(ui.ctx());

        // Drain any in-flight search-progress events before polling
        // the terminal result. Today the only progress event is
        // `BenchmarkReady`, which lands `outputs.benchmark` as soon
        // as the worker's A* + time-PSO step finishes — well before
        // the main PSO completes — so the dashed reference overlay
        // paints right away instead of waiting for the whole search.
        if let Some(rx) = self.search_progress_rx.as_ref() {
            loop {
                match rx.try_recv() {
                    Ok(SearchProgressEvent::Phase(phase)) => {
                        self.current_search_phase = Some(phase);
                    }
                    Ok(SearchProgressEvent::BenchmarkReady(b)) => {
                        self.outputs.benchmark = Some(b);
                    }
                    Ok(SearchProgressEvent::Iteration {
                        iter_idx,
                        total_iters,
                        gbest_xs,
                        gbest_ys,
                        gbest_ts,
                        best_fit,
                    }) => {
                        self.outputs.live_gbest = Some(crate::search::LiveGbest {
                            iter_idx,
                            total_iters,
                            xs: gbest_xs,
                            ys: gbest_ys,
                            ts: gbest_ts,
                            best_fit,
                        });
                    }
                    // `SearchProgressEvent` is `#[non_exhaustive]`;
                    // future event kinds land here and get ignored
                    // until we hook them up.
                    Ok(_) => {}
                    Err(std::sync::mpsc::TryRecvError::Empty) => break,
                    Err(std::sync::mpsc::TryRecvError::Disconnected) => {
                        // Worker dropped its sender — drop our receiver
                        // too so subsequent frames skip the poll.
                        self.search_progress_rx = None;
                        break;
                    }
                }
            }
        }

        if let Some(msg) = self.search_job.poll() {
            self.search_started_at = None;
            // Search is done; drop the phase indicator regardless of
            // outcome so the status label stops claiming the search
            // is still in flight.
            self.current_search_phase = None;
            // The real `route_evolution` (or an error toast) is about
            // to take over — clear the live snapshot so the central
            // panel's overlay routes through the terminal data instead
            // of stale per-iter state.
            self.outputs.live_gbest = None;
            match msg {
                Ok(SearchResult {
                    route_evolution,
                    route_bounds,
                    baked,
                    boat,
                    benchmark,
                    bake_duration,
                    search_duration,
                    ensemble,
                }) => {
                    self.outputs.iteration = route_evolution.iter_count().saturating_sub(1);
                    self.outputs.route_evolution = Some(route_evolution);
                    self.outputs.route_bounds = Some(route_bounds);
                    self.outputs.baked_wind_map = Some(baked);
                    self.outputs.boat = Some(boat);
                    self.outputs.benchmark = benchmark;
                    self.outputs.bake_duration = Some(bake_duration);
                    self.outputs.search_duration = Some(search_duration);
                    self.outputs.ensemble = ensemble;
                    // Drop any prior realization cohort — it was scored
                    // against the previous route bounds / endpoints
                    // and would draw stale routes over the new gbest.
                    self.outputs.realization_runs.clear();
                    // Reset the Summary selector — a stale
                    // `Realization(k)` would point into a now-empty
                    // cohort, and `Gbest` is the only thing that
                    // makes sense to show right after a new gbest.
                    self.outputs.summary_selection = crate::search::SummarySelection::Gbest;
                }
                Err(e) => {
                    // Prior outputs stay displayed — discarding them
                    // would hide a search the user may still be looking at.
                    self.report_error(format!("Search failed — {e}"));
                }
            }
        }

        // Reopt result is spliced into the gbest path's `t` so segment stats
        // and totals refresh on the next render.
        if let Some(msg) = self.reopt_job.poll() {
            self.apply_reopt_result(&msg);
        }

        // Realization cohort lands as a `Vec<RealizationRun>` — assign
        // wholesale, overwriting any prior cohort. Failures surface a
        // toast and leave the previous cohort visible (same policy as
        // the main search arm).
        if let Some(result) = self.realization_job.poll() {
            self.realization_started_at = None;
            match result {
                Ok(runs) => {
                    // Clamp a stale `Realization(k)` selection back to
                    // `Gbest` if the new cohort doesn't have a `k`-th
                    // entry. K typically matches across reruns, but
                    // re-fetching the ensemble dir with a different
                    // member count would otherwise leave a dangling
                    // index pointing past the new vector's end.
                    if let crate::search::SummarySelection::Realization(k) =
                        self.outputs.summary_selection
                        && k >= runs.len()
                    {
                        self.outputs.summary_selection = crate::search::SummarySelection::Gbest;
                    }
                    self.outputs.realization_runs = runs;
                }
                Err(e) => {
                    self.report_error(format!("Per-realization run failed — {e}"));
                }
            }
        }

        // The embedded-sample decoder runs once at startup. When it
        // lands we slot the bundled wind map into `wind_map` and
        // invalidate the view's grid-derived cache. Waypoints,
        // bbox, and any other editor state are deliberately
        // preserved: the bundled sample is an *ambient* async load
        // (no explicit user action), so it races against actions the
        // user takes during startup. The bad case it used to lose:
        // user clicks `File → Load Scenario` before the sample
        // arrives, the scenario populates start/end/bbox, then the
        // sample lands and silently wipes them. The persisted-state-
        // from-prior-session concern that motivated the old clears
        // is real but rare, and the waypoints are real-world
        // (lon, lat) coordinates — they remain geographically
        // meaningful even if they sit awkwardly on the new map.
        if let Some(result) = self.bundled_sample_job.poll() {
            match result {
                Ok(map) => {
                    self.wind_map = Some(map);
                    self.view.view_lon0 = None;
                    self.view.view_lat0 = None;
                    // Invalidate any wrap-region cache pinned to the
                    // pre-swap wind map's grid layout.
                    self.view.synthesized_frame = None;
                }
                Err(e) => {
                    self.report_error(format!("failed to load bundled wind sample: {e}"));
                }
            }
        }

        // Drain progress events from the AWS-fetch worker (if any) and
        // swap the resulting map in when the worker reports `Done`.
        #[cfg(not(target_arch = "wasm32"))]
        if let Some(map) = self.fetch_job.poll() {
            self.wind_map = Some(map);
            self.editor.start_waypoint = None;
            self.editor.end_waypoint = None;
            self.editor.route_bbox = None;
            self.view.view_lon0 = None;
            self.view.view_lat0 = None;
            self.view.synthesized_frame = None;
        }

        // Ensemble fetch: on Done(Ok(dir)), point `ensemble_path` at
        // the freshly-written directory so the next Run Search picks
        // ensemble mode automatically without the user having to copy
        // the path into Advanced Params.
        #[cfg(not(target_arch = "wasm32"))]
        if let Some(dir) = self.fetch_ensemble_job.poll() {
            self.search.ensemble_path = Some(dir);
        }

        // Clamp persisted UI state against the (possibly reloaded) wind map.
        if let Some(wind_map) = &self.wind_map {
            let max_frame = wind_map.frame_count().saturating_sub(1);
            self.editor.current_frame = self.editor.current_frame.min(max_frame);
        }

        // Re-derive the route bbox before panel render so the new
        // rectangle draws this frame, not next.
        self.update_auto_route_bbox();

        self.render_menu_bar(ui);
        self.render_stats_panel(ui);
        self.render_tools_panel(ui);
        self.render_central_panel(ui);
        self.render_grib2_load_dialog(ui);
        #[cfg(not(target_arch = "wasm32"))]
        self.render_fetch_dialog(ui);
        #[cfg(not(target_arch = "wasm32"))]
        self.render_fetch_ensemble_dialog(ui);
        self.render_advanced_settings_window(ui.ctx());
        self.render_generate_window(ui.ctx());
        self.render_error_toast(ui);
    }

    /// Called by the framework to save state before shutdown.
    fn save(&mut self, storage: &mut dyn eframe::Storage) {
        eframe::set_value(storage, eframe::APP_KEY, self);
    }
}

/// Worker-side ensemble search. Loads the K members from disk, bakes
/// them in parallel, wraps them in the right [`WindInput`] variant
/// based on `ensemble_mode`, and dispatches to
/// `run_search_blocking_with_baked`.
///
/// Returns the same `Result<SearchResult, SearchError>` shape as
/// [`run_search_blocking`] so the UI-side polling code doesn't need to
/// know which path produced the result.
#[expect(
    clippy::too_many_arguments,
    reason = "the search worker is a glue function with no natural shared bundle"
)]
fn run_ensemble_search(
    ensemble_path: &std::path::Path,
    bake_bounds: bywind::wind_map::BakeBounds,
    route_bounds: bywind::RouteBounds,
    waypoint_count: bywind::WaypointCount,
    search_settings: bywind::SearchSettings,
    ship: bywind::Boat,
    weights: SearchWeights,
    sdf_resolution_deg: f64,
    fine_sdf_resolution_deg: Option<f64>,
    ensemble_mode: EnsembleMode,
    progress: &mut dyn FnMut(SearchProgressEvent),
) -> Result<SearchResult, SearchError> {
    progress(SearchProgressEvent::Phase(SearchPhase::Loading));
    let ensemble = match bywind::TimedEnsembleWindMap::load_dir(ensemble_path) {
        Ok(e) => e,
        Err(e) => {
            log::error!("ensemble load failed: {e}");
            return Err(SearchError::NoFeasibleRoute { best_fit: f64::NAN });
        }
    };
    // K member bakes + the `mean()` reduction inside
    // `ensemble_full` / `ensemble_fast_mean` are all "rasterise to the
    // search grid" work from the user's perspective, so they share the
    // single Baking phase label.
    progress(SearchProgressEvent::Phase(SearchPhase::Baking));
    let baked = ensemble.bake(bake_bounds);
    let wind_input = match ensemble_mode {
        EnsembleMode::Full => WindInput::ensemble_full(baked),
        EnsembleMode::FastMean => WindInput::ensemble_fast_mean(baked),
    };
    run_search_blocking_with_baked(
        wind_input,
        route_bounds,
        waypoint_count,
        search_settings,
        ship,
        weights,
        sdf_resolution_deg,
        fine_sdf_resolution_deg,
        progress,
    )
}

/// Worker-side per-realization cohort. Loads + bakes the ensemble
/// from disk, then delegates the actual K-search loop to
/// [`bywind::run_realizations`] so the testable logic lives in core.
/// The `.wcav`-decode failure is converted to a [`SearchError`] here
/// since the inner function takes an already-baked ensemble.
#[expect(
    clippy::too_many_arguments,
    reason = "matches the inner search-blocking signature one-for-one"
)]
fn run_realization_searches(
    ensemble_path: &std::path::Path,
    bake_bounds: bywind::wind_map::BakeBounds,
    route_bounds: bywind::RouteBounds,
    waypoint_count: bywind::WaypointCount,
    search_settings: bywind::SearchSettings,
    base_seed: u64,
    ship: bywind::Boat,
    weights: SearchWeights,
    sdf_resolution_deg: f64,
    fine_sdf_resolution_deg: Option<f64>,
) -> Result<Vec<RealizationRun>, SearchError> {
    let ensemble = match bywind::TimedEnsembleWindMap::load_dir(ensemble_path) {
        Ok(e) => e,
        Err(e) => {
            log::error!("ensemble load failed: {e}");
            return Err(SearchError::NoFeasibleRoute { best_fit: f64::NAN });
        }
    };
    let baked = ensemble.bake(bake_bounds);
    run_realizations(
        &baked,
        route_bounds,
        waypoint_count,
        search_settings,
        base_seed,
        ship,
        weights,
        sdf_resolution_deg,
        fine_sdf_resolution_deg,
    )
}

#[cfg(test)]
mod async_job_tests {
    use super::AsyncJob;
    use std::sync::mpsc;

    #[test]
    fn idle_poll_stays_idle() {
        let mut job: AsyncJob<i32> = AsyncJob::default();
        assert!(!job.is_running());
        assert_eq!(job.poll(), None);
        assert!(!job.is_running());
    }

    #[test]
    fn running_with_no_message_polls_none_and_stays_running() {
        let (_tx, rx) = mpsc::channel::<i32>();
        let mut job = AsyncJob::default();
        job.set_running(rx);
        assert!(job.is_running());
        assert_eq!(job.poll(), None);
        assert!(
            job.is_running(),
            "an empty channel must not transition to Idle"
        );
    }

    #[test]
    fn running_with_message_returns_value_then_becomes_idle() {
        let (tx, rx) = mpsc::channel();
        let mut job = AsyncJob::default();
        job.set_running(rx);
        tx.send(42).unwrap();
        assert_eq!(job.poll(), Some(42));
        assert!(!job.is_running());
        assert_eq!(job.poll(), None, "second poll on idle returns None");
    }

    #[test]
    fn disconnected_sender_transitions_to_idle() {
        let (tx, rx) = mpsc::channel::<i32>();
        let mut job = AsyncJob::default();
        job.set_running(rx);
        drop(tx);
        assert_eq!(job.poll(), None);
        assert!(!job.is_running());
    }

    #[test]
    fn set_running_overwrites_previous_receiver() {
        // Time-reopt cancellation: replacing the slot drops `rx1` and
        // its queued value.
        let (tx1, rx1) = mpsc::channel();
        tx1.send(1).unwrap();
        let (tx2, rx2) = mpsc::channel();
        let mut job = AsyncJob::default();
        job.set_running(rx1);
        job.set_running(rx2);
        tx2.send(2).unwrap();
        assert_eq!(job.poll(), Some(2));
    }
}