perf(weather/scalar_file): single MessagePack format for both Elixir and Rust

Move ScalarFile from a dual-writer dual-format setup (ETF on the Elixir
side, materialized via NotifyListener; nothing on the Rust side) to a
single chunked-MessagePack format that both languages produce and
consume.

Elixir side:

- ScalarFile now writes gzipped MessagePack `.mp.gz` per chunk via
  Msgpax. Reader decodes with Msgpax + zlib. Atom keys are stripped to
  strings on encode and re-atomized via a whitelist on read so callers
  see the same shape regardless of which side wrote the bytes.
- DateTime values round-trip through ISO8601 strings.
- New cross-language wire-format test that decodes a hand-rolled
  payload mirroring exactly what `rmp_serde::to_vec_named` emits, so a
  format regression on either side fails the suite.

Rust side:

- New `prop_grid_rs::weather_scalar_file` module:
  - 1:1 port of `Microwaveprop.Weather.WeatherLayers.derive/1` plus
    the surface fields `Weather.build_grid_cache_row/4` adds on top
    (lapse rate, mid lapse rate, inversion strength + base, 850/700 mb
    interpolation, surface RH, surface refractivity, ducting flag).
  - `write_atomic` chunks rows into 5°×5° buckets, writes one
    `<lat_band>_<lon_band>.mp.gz` per chunk via tmp + rename, drops
    pre-existing chunks first to keep writes canonical.
  - 5 unit tests covering surface-only rows, upper-air derivation,
    write/read round-trip, chunk band boundaries, and the
    out-of-physical-range surface-temp filter.
- Pipeline integration: both `run_chain_step` (f01..f18) and the f00
  analysis path now build `Vec<ScalarRow>` in the same merged-cell
  pass that builds `CellEntry` and `Conditions`, then fire a parallel
  `spawn_blocking` write that overlaps with scoring. Awaiting the
  scalar future after the score-band writes keeps the failure-surface
  symmetric with the existing profile_future.

Result: every new forecast hour lands with the scalar artifact already
on disk, so /weather reads never fall through to ProfilesFile decode +
per-cell SoundingParams + WeatherLayers derivation. The Elixir-side
`NotifyListener.handle_propagation_ready/1` materialization remains as
an idempotent safety net for hours that pre-date this commit.

docs/weather.md

@@ -14,9 +14,7 @@ Completed so far:
 - Layer/time changes now refresh overlay tile URLs instead of pushing large `update_weather` payloads (finding #6).
 - A dedicated `Microwaveprop.Weather.ScalarFile` artifact persists pre-derived weather scalars per `valid_time` on NFS, bucketed into 5°×5° chunks. `weather_grid_at/2` and `weather_point_detail/3` read from it before falling back to raw `ProfilesFile` decoding (findings #1, #2, #3, #4, #7).
 
-Still pending:
+All listed items shipped. The `Microwaveprop.Weather.WeatherLayers.derive/1` port now lives in [`prop_grid_rs::weather_scalar_file`](/Users/graham/dev/ntms/microwaveprop/rust/prop_grid_rs/src/weather_scalar_file.rs:1) and the Rust pipeline writes `ScalarFile` artifacts inline alongside `profiles_file::write_atomic`, using the same gzipped MessagePack chunked-by-5°×5° wire format the Elixir writer produces. The Elixir reader atomizes a whitelist of keys on read so callers see the same shape regardless of which side wrote the bytes.
-
-- Have the propagation pipeline write the scalar artifact directly on the Rust side, alongside `.mp.gz`. Today, scalar materialization is triggered from Elixir's `NotifyListener.handle_propagation_ready/1` as soon as the Rust pipeline fires `NOTIFY propagation_ready`, so the gap between profile-write and scalar-availability is just one BEAM derivation pass per new `valid_time`. Closing it fully would require porting `WeatherLayers.derive` (lapse rates, inversion strength, 850/700 mb interpolation, duct summarization) to Rust, plus a compatible writer Elixir's `ScalarFile` reader can decode. Tracked as future work — the runtime cost it would save is small relative to the port effort.
 
 Available infrastructure that should shape the solution:
 
@@ -399,7 +397,7 @@ All listed phases are shipped:
 - Viewport reads only touch the chunks that overlap the current bounds — for both NFS-backed `ScalarFile` reads and the in-memory `GridCache`.
 - Point clicks read a single chunk instead of the whole grid file.
 - The browser displays server-rendered tiles; layer/time changes only swap tile URLs.
-- New forecast hours are pre-warmed: `NotifyListener.handle_propagation_ready/1` fires `Weather.materialize_scalar_file/1` in a detached Task as soon as the Rust pipeline fires `propagation_ready`, so the first interactive `/weather` reader of that hour finds the scalar artifact already on disk.
+- New forecast hours land with the scalar artifact already on disk: the Rust pipeline writes scalar chunks inline alongside the raw profile, and the Elixir-side `NotifyListener.handle_propagation_ready/1` keeps an idempotent fallback materialization for hours that pre-date the Rust deploy.
 - Legend and timeline DOM churn is gone — the legend is mounted once and the timeline only rebuilds its full HTML on data changes, otherwise just restyling existing buttons.
 
-Remaining work is a single non-runtime-critical port — move `WeatherLayers.derive` into the Rust pipeline so the BEAM doesn't pay even the one-time derivation pass per new `valid_time`. Documented as future work; the runtime savings are small relative to the port effort.
+Both languages produce and consume identical bytes on disk: gzipped MessagePack chunks under `<base>/weather_scalars/<iso>/<lat_band>_<lon_band>.mp.gz`. There is no Elixir-only or Rust-only branch in the format.

lib/microwaveprop/weather/scalar_file.ex

@@ -17,15 +17,26 @@ defmodule Microwaveprop.Weather.ScalarFile do
   a tile request can read just the chunks it needs and skip the
   `SoundingParams.derive/1` + `WeatherLayers.derive/1` work entirely.
 
+  ## Wire format
+
+  Each chunk is a gzipped MessagePack `[row_map, ...]`. MessagePack was
+  picked over ETF so the Rust `prop_grid_rs` worker can produce these
+  files inline in the propagation pipeline using `rmp-serde` and Elixir
+  reads them with `Msgpax`. Both producers write identical bytes — there
+  is no Elixir-only or Rust-only branch.
+
+  Map keys are strings on the wire (MessagePack has no atom type).
+  Whitelisted keys are atomized on read so callers see the same
+  Elixir-shaped row regardless of producer.
+
   ## Layout
 
       <base_dir>/weather_scalars/
         <iso>/                # e.g. 2026-04-28T12:00:00Z/
-          <lat_band>_<lon_band>.etf.gz
+          <lat_band>_<lon_band>.mp.gz
 
-  `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Each chunk is
+  `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Writes go
-  the compressed ETF of `[row_map, ...]`. Writes go through
+  through `rename(2)` → atomic on NFS.
-  `rename(2)` → atomic on NFS.
 
   ## Chunk size
 
@@ -39,6 +50,34 @@ defmodule Microwaveprop.Weather.ScalarFile do
   @chunk_step 5
   @subdir "weather_scalars"
 
+  # Keys produced by the writer that should land as atoms in Elixir.
+  # Anything outside this set stays a string and is ignored by callers
+  # (weather.ex / tile_renderer.ex use atom access throughout).
+  @atom_keys MapSet.new([
+               "lat",
+               "lon",
+               "valid_time",
+               "temperature",
+               "dewpoint_depression",
+               "surface_rh",
+               "surface_pressure_mb",
+               "surface_refractivity",
+               "refractivity_gradient",
+               "bl_height",
+               "pwat",
+               "temp_850mb",
+               "dewpoint_850mb",
+               "temp_700mb",
+               "dewpoint_700mb",
+               "lapse_rate",
+               "mid_lapse_rate",
+               "inversion_strength",
+               "inversion_base_m",
+               "ducting",
+               "duct_base_m",
+               "duct_strength"
+             ])
+
   @type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => term()}
   @type bounds :: %{optional(String.t()) => number()}
 
@@ -90,8 +129,8 @@ defmodule Microwaveprop.Weather.ScalarFile do
     rows
     |> Enum.group_by(&chunk_key/1)
     |> Enum.each(fn {{lat_band, lon_band}, chunk_rows} ->
-      path = Path.join(dir, "#{lat_band}_#{lon_band}.etf.gz")
+      path = Path.join(dir, "#{lat_band}_#{lon_band}.mp.gz")
-      binary = :erlang.term_to_binary(chunk_rows, [:compressed])
+      binary = encode_chunk(chunk_rows)
 
       tmp = path <> ".tmp." <> unique_suffix()
       File.write!(tmp, binary, [:binary])
@@ -222,7 +261,7 @@ defmodule Microwaveprop.Weather.ScalarFile do
     (value / @chunk_step) |> Float.floor() |> trunc()
   end
 
-  defp chunk_filename({lat_band, lon_band}), do: "#{lat_band}_#{lon_band}.etf.gz"
+  defp chunk_filename({lat_band, lon_band}), do: "#{lat_band}_#{lon_band}.mp.gz"
 
   defp chunk_intersects_bounds?(_key, nil), do: true
 
@@ -243,17 +282,73 @@ defmodule Microwaveprop.Weather.ScalarFile do
     end)
   end
 
-  defp decode_chunk(path) do
+  # Encode rows as gzipped MessagePack. We strip atoms to strings on the
-    case File.read(path) do
+  # wire (Msgpax doesn't speak atom keys natively, and Rust can't read
-      {:ok, binary} ->
+  # them anyway) and re-atomize on the read side.
-        :erlang.binary_to_term(binary)
+  defp encode_chunk(rows) do
+    rows
+    |> Enum.map(&prepare_row_for_encode/1)
+    |> Msgpax.pack!(iodata: false)
+    |> :zlib.gzip()
+  end
 
-      {:error, reason} ->
+  # Convert atom keys to strings and DateTime values to ISO8601 strings
-        Logger.warning("ScalarFile chunk read failed path=#{path} reason=#{inspect(reason)}")
+  # so the row is representable in MessagePack.
+  defp prepare_row_for_encode(row) when is_map(row) do
+    Map.new(row, fn {k, v} ->
+      key = if is_atom(k), do: Atom.to_string(k), else: k
+      {key, encode_value(v)}
+    end)
+  end
+
+  defp encode_value(%DateTime{} = dt), do: DateTime.to_iso8601(DateTime.truncate(dt, :second))
+  defp encode_value(value), do: value
+
+  defp decode_chunk(path) do
+    with {:ok, gz} <- File.read(path),
+         {:ok, binary} <- safe_gunzip(gz),
+         {:ok, decoded} <- Msgpax.unpack(binary) do
+      decoded
+      |> List.wrap()
+      |> Enum.map(&normalize_row/1)
+    else
+      error ->
+        Logger.warning("ScalarFile chunk read failed path=#{path} reason=#{inspect(error)}")
         []
     end
   end
 
+  defp safe_gunzip(binary) do
+    {:ok, :zlib.gunzip(binary)}
+  rescue
+    e -> {:error, {:gunzip, e}}
+  end
+
+  # Atomize whitelisted keys and re-parse the `valid_time` ISO8601 back
+  # into a DateTime so callers see the same shape as the in-memory rows
+  # produced by `Microwaveprop.Weather.build_grid_cache_rows/3`.
+  defp normalize_row(row) when is_map(row) do
+    Map.new(row, fn {k, v} ->
+      key = atomize_key(k)
+      {key, normalize_value(key, v)}
+    end)
+  end
+
+  defp atomize_key(k) when is_binary(k) do
+    if MapSet.member?(@atom_keys, k), do: String.to_atom(k), else: k
+  end
+
+  defp atomize_key(k), do: k
+
+  defp normalize_value(:valid_time, v) when is_binary(v) do
+    case DateTime.from_iso8601(v) do
+      {:ok, dt, _offset} -> dt
+      _ -> v
+    end
+  end
+
+  defp normalize_value(_key, v), do: v
+
   defp list_chunk_files(%DateTime{} = valid_time) do
     dir = dir_for(valid_time)
 
@@ -270,7 +365,7 @@ defmodule Microwaveprop.Weather.ScalarFile do
   end
 
   defp parse_chunk_filename(filename) do
-    with [_, lat_str, lon_str] <- Regex.run(~r/^(-?\d+)_(-?\d+)\.etf\.gz$/, filename),
+    with [_, lat_str, lon_str] <- Regex.run(~r/^(-?\d+)_(-?\d+)\.mp\.gz$/, filename),
          {lat_band, ""} <- Integer.parse(lat_str),
          {lon_band, ""} <- Integer.parse(lon_str) do
       {lat_band, lon_band}

rust/prop_grid_rs/src/lib.rs

@@ -50,3 +50,4 @@ pub mod scorer;
 pub mod scores_file;
 pub mod sounding_params;
 pub mod telemetry;
+pub mod weather_scalar_file;

rust/prop_grid_rs/src/pipeline.rs

@@ -22,6 +22,7 @@ use crate::profiles_file::{self, CellEntry};
 use crate::scorer::{self, Conditions};
 use crate::scores_file::{self, ScorePoint};
 use crate::sounding_params::{self, Level};
+use crate::weather_scalar_file::{self, ScalarRow};
 
 #[derive(Debug, thiserror::Error)]
 pub enum PipelineError {
@@ -33,6 +34,8 @@ pub enum PipelineError {
     Write(#[from] scores_file::WriteError),
     #[error("profile write: {0}")]
     ProfileWrite(#[from] profiles_file::WriteError),
+    #[error("scalar write: {0}")]
+    ScalarWrite(#[from] weather_scalar_file::WriteError),
     #[error("native duct: {0}")]
     NativeDuct(#[from] native_duct::NativeDuctError),
     #[error("nexrad: {0}")]
@@ -147,12 +150,16 @@ pub async fn run_chain_step(
     let mut prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> =
         Vec::with_capacity(merged.len());
     let mut profile_entries: Vec<CellEntry> = Vec::with_capacity(merged.len());
+    let mut scalar_rows: Vec<ScalarRow> = Vec::with_capacity(merged.len());
     for (key, cell) in merged.iter() {
         let (lat, lon) = decoder::key_to_latlon(*key);
         if let Some(conditions) = cell_to_conditions(cell, lat, lon, &valid_time) {
             let invariants = scorer::precompute_band_invariants(&conditions);
             prepared.push((lat, lon, conditions, invariants));
             profile_entries.push(cell_to_profile_entry(lat, lon, cell));
+            if let Some(row) = weather_scalar_file::derive_row(lat, lon, valid_time, cell) {
+                scalar_rows.push(row);
+            }
         }
     }
     drop(merged);
@@ -170,6 +177,21 @@ pub async fn run_chain_step(
         })
     };
 
+    // Persist the derived scalar artifact alongside the raw profile so
+    // Elixir's `/weather` reads land on a pre-derived chunk file
+    // instead of falling back to the cold ProfilesFile decode + per-cell
+    // SoundingParams + WeatherLayers derivation. Identical wire format
+    // (gzipped MessagePack chunked by 5°×5°) to what the Elixir writer
+    // produces — the reader doesn't care which side wrote the bytes.
+    let scores_dir_for_scalars = scores_dir_owned.clone();
+    let scalar_future = {
+        let rows = scalar_rows;
+        tokio::task::spawn_blocking(move || {
+            weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows)
+                .map(|_| 0u32)
+        })
+    };
+
     // Step 2: score and write every band in parallel.
     //
     // Scoring: rayon's thread pool over the 23 bands. Each band is
@@ -219,6 +241,10 @@ pub async fn run_chain_step(
         .await
         .expect("blocking join")
         .map_err(PipelineError::ProfileWrite)?;
+    scalar_future
+        .await
+        .expect("blocking join")
+        .map_err(PipelineError::ScalarWrite)?;
     let files_written = bands.len() as u32;
 
     Ok(ChainStepStats {
@@ -677,12 +703,16 @@ pub async fn run_analysis_step(
     let mut profile_entries: Vec<CellEntry> = Vec::with_capacity(merged.len());
     let mut prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> =
         Vec::with_capacity(merged.len());
+    let mut scalar_rows: Vec<ScalarRow> = Vec::with_capacity(merged.len());
     for (key, cell) in merged.iter() {
         let (lat, lon) = decoder::key_to_latlon(*key);
         if let Some(conditions) = cell_to_conditions(cell, lat, lon, &valid_time) {
             let invariants = scorer::precompute_band_invariants(&conditions);
             prepared.push((lat, lon, conditions, invariants));
             profile_entries.push(cell_to_profile_entry(lat, lon, cell));
+            if let Some(row) = weather_scalar_file::derive_row(lat, lon, valid_time, cell) {
+                scalar_rows.push(row);
+            }
         }
     }
 
@@ -699,6 +729,19 @@ pub async fn run_analysis_step(
         })
     };
 
+    // Same wire format as Elixir's `ScalarFile`. See run_chain_step
+    // (forecast hours) for the full rationale — both code paths go
+    // through the same writer so f00 and f01..f18 land identical
+    // chunked artifacts on NFS.
+    let scores_dir_for_scalars = scores_dir_owned.clone();
+    let scalar_future = {
+        let rows = scalar_rows;
+        tokio::task::spawn_blocking(move || {
+            weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows)
+                .map(|_| 0u32)
+        })
+    };
+
     // Bands and score writes run in parallel, same shape as
     // run_chain_step. Rayon across bands, try_join_all across writes.
     let bands = band_config::all_bands();
@@ -738,6 +781,10 @@ pub async fn run_analysis_step(
         .await
         .expect("blocking join")
         .map_err(PipelineError::ProfileWrite)?;
+    scalar_future
+        .await
+        .expect("blocking join")
+        .map_err(PipelineError::ScalarWrite)?;
 
     let files_written = bands.len() as u32;
     Ok(AnalysisStepStats {

rust/prop_grid_rs/src/weather_scalar_file.rs

@@ -0,0 +1,526 @@
+//! Per-cell *derived* weather scalars on disk, the cheap-read sibling
+//! of [`profiles_file`][crate::profiles_file].
+//!
+//! 1:1 wire-compatible with `Microwaveprop.Weather.ScalarFile` on the
+//! Elixir side. Chunked into 5°×5° spatial buckets so a state-sized
+//! `/weather` viewport reads only the chunks that overlap.
+//!
+//! ## Layout
+//!
+//! ```text
+//! <scores_dir>/weather_scalars/
+//!   <iso>/                           # e.g. 2026-04-29T12:00:00Z/
+//!     <lat_band>_<lon_band>.mp.gz
+//! ```
+//!
+//! `lat_band = floor(lat / 5)`, `lon_band = floor(lon / 5)`. Each chunk
+//! is gzipped MessagePack: `[ {row}, {row}, … ]`. Map keys are strings
+//! on the wire — the Elixir reader atomizes the whitelist on read, so
+//! every key emitted here must match the Elixir whitelist or it gets
+//! silently dropped by callers that use atom-style access.
+//!
+//! Scalar derivation mirrors `Microwaveprop.Weather.WeatherLayers.derive/1`
+//! plus the surface fields `Microwaveprop.Weather.build_grid_cache_row/4`
+//! adds on top. The Elixir module remains the single source of truth
+//! for any cell that hasn't been Rust-materialized yet (the
+//! `kickoff_async_scalar_materialize` fallback path).
+
+use std::collections::HashMap;
+use std::fs::File;
+use std::io::{BufWriter, Write};
+use std::path::{Path, PathBuf};
+
+use chrono::{DateTime, Utc};
+use flate2::write::GzEncoder;
+use flate2::Compression;
+use serde::Serialize;
+
+use crate::decoder::CellValues;
+use crate::fetcher;
+use crate::sounding_params::{
+    ducting_detected, min_refractivity_gradient, sat_vap_pres, surface_refractivity, Level,
+};
+
+const CHUNK_STEP: i32 = 5;
+const SUBDIR: &str = "weather_scalars";
+
+/// One row in a chunk file. Field names match the Elixir atom-key
+/// whitelist exactly — adding a field here without updating the
+/// Elixir `@atom_keys` set means the new field round-trips as a
+/// string and is invisible to atom-style callers.
+#[derive(Debug, Default, Serialize)]
+pub struct ScalarRow {
+    pub lat: f64,
+    pub lon: f64,
+    pub valid_time: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub temperature: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub dewpoint_depression: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub surface_rh: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub surface_pressure_mb: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub surface_refractivity: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub refractivity_gradient: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub bl_height: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub pwat: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub temp_850mb: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub dewpoint_850mb: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub temp_700mb: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub dewpoint_700mb: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub lapse_rate: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub mid_lapse_rate: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub inversion_strength: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub inversion_base_m: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub ducting: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub duct_base_m: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub duct_strength: Option<f64>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum WriteError {
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("encode: {0}")]
+    Encode(#[from] rmp_serde::encode::Error),
+}
+
+/// Build a `ScalarRow` from a single grid cell. Returns `None` for
+/// cells whose surface temperature is missing or out of physically
+/// plausible range — matches Elixir's `build_grid_cache_row` filter
+/// (`temp_c < -80 or temp_c > 60` are dropped).
+pub fn derive_row(
+    lat: f64,
+    lon: f64,
+    valid_time: DateTime<Utc>,
+    cell: &CellValues,
+) -> Option<ScalarRow> {
+    let temp_c = cell
+        .get("TMP:2 m above ground")
+        .map(|&v| v as f64 - 273.15)?;
+    if !temp_c.is_finite() || !(-80.0..=60.0).contains(&temp_c) {
+        return None;
+    }
+
+    let dewpoint_c = cell.get("DPT:2 m above ground").map(|&v| v as f64 - 273.15);
+    let dewpoint_depression = dewpoint_c.map(|d| temp_c - d);
+    let surface_pressure_mb = cell.get("PRES:surface").map(|&v| v as f64 / 100.0);
+    let bl_height = cell.get("HPBL:surface").map(|&v| v as f64);
+    let pwat = cell
+        .get("PWAT:entire atmosphere (considered as a single layer)")
+        .map(|&v| v as f64);
+
+    let surface_rh = dewpoint_c.map(|d| 100.0 * sat_vap_pres(d) / sat_vap_pres(temp_c));
+
+    let levels = build_levels(cell);
+    let derived_min_grad = if levels.len() >= 3 {
+        min_refractivity_gradient(levels.clone())
+    } else {
+        None
+    };
+    let native_min_grad = cell.get("native_min_gradient").map(|&v| v as f64);
+    let refractivity_gradient = derived_min_grad.or(native_min_grad);
+    let ducting = Some(ducting_detected(refractivity_gradient));
+    let surface_refractivity_val = surface_refractivity(&levels);
+
+    let mut sorted: Vec<&Level> = levels.iter().collect();
+    sorted.sort_by(|a, b| {
+        b.pres_mb
+            .partial_cmp(&a.pres_mb)
+            .unwrap_or(std::cmp::Ordering::Equal)
+    });
+
+    let temp_850mb = level_value(&sorted, 850.0, |l| Some(l.tmpc));
+    let dewpoint_850mb = level_value(&sorted, 850.0, |l| l.dwpc);
+    let temp_700mb = level_value(&sorted, 700.0, |l| Some(l.tmpc));
+    let dewpoint_700mb = level_value(&sorted, 700.0, |l| l.dwpc);
+
+    let lapse_rate = compute_lapse_rate(&sorted);
+    let mid_lapse_rate = compute_layer_lapse_rate(&sorted, 850.0, 700.0);
+    let inversion_strength = compute_inversion_strength(&sorted);
+    let inversion_base_m = compute_inversion_base_m(&sorted);
+
+    // Ducts: the f01..f18 pipeline only stores the duct-summary scalars
+    // (`max_duct_thickness_m`, `duct_count`) per cell. Duct base requires
+    // the full Duct list, which isn't threaded through `CellValues` yet —
+    // matches the current Elixir behavior on Rust-produced profiles
+    // (their `:duct_characteristics` is also nil).
+    let duct_count = cell.get("duct_count").copied().unwrap_or(0.0);
+    let duct_strength = if duct_count > 0.0 {
+        cell.get("max_duct_thickness_m").map(|&v| v as f64)
+    } else {
+        None
+    };
+
+    Some(ScalarRow {
+        lat,
+        lon,
+        valid_time: valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
+        temperature: Some(temp_c),
+        dewpoint_depression,
+        surface_rh,
+        surface_pressure_mb,
+        surface_refractivity: surface_refractivity_val,
+        refractivity_gradient,
+        bl_height,
+        pwat,
+        temp_850mb,
+        dewpoint_850mb,
+        temp_700mb,
+        dewpoint_700mb,
+        lapse_rate,
+        mid_lapse_rate,
+        inversion_strength,
+        inversion_base_m,
+        ducting,
+        duct_base_m: None,
+        duct_strength,
+    })
+}
+
+/// Absolute path the writer lands at for `valid_time`.
+pub fn dir_for(scores_dir: &Path, valid_time: DateTime<Utc>) -> PathBuf {
+    let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string();
+    scores_dir.join(SUBDIR).join(iso)
+}
+
+/// Persist `rows` for `valid_time` as 5°×5° chunked `.mp.gz` files.
+/// Mirrors the Elixir writer: the destination directory is wiped of
+/// any prior chunks first so a smaller follow-up write doesn't leave
+/// stale files behind. Each chunk is written `tmp + rename` for an
+/// atomic NFS-friendly publish.
+pub fn write_atomic(
+    scores_dir: &Path,
+    valid_time: DateTime<Utc>,
+    rows: &[ScalarRow],
+) -> Result<PathBuf, WriteError> {
+    let dir = dir_for(scores_dir, valid_time);
+    std::fs::create_dir_all(&dir)?;
+
+    if let Ok(entries) = std::fs::read_dir(&dir) {
+        for entry in entries.flatten() {
+            let _ = std::fs::remove_file(entry.path());
+        }
+    }
+
+    let mut chunks: HashMap<(i32, i32), Vec<&ScalarRow>> = HashMap::new();
+    for row in rows {
+        let key = (chunk_band(row.lat), chunk_band(row.lon));
+        chunks.entry(key).or_default().push(row);
+    }
+
+    for ((lat_band, lon_band), chunk_rows) in chunks {
+        let path = dir.join(format!("{lat_band}_{lon_band}.mp.gz"));
+
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0);
+        let pid = std::process::id();
+        let mut tmp = path.clone().into_os_string();
+        tmp.push(format!(".tmp.{nanos}.{pid}"));
+        let tmp = PathBuf::from(tmp);
+
+        {
+            let buf = rmp_serde::to_vec_named(&chunk_rows)?;
+            let file = File::create(&tmp)?;
+            let mut gz = GzEncoder::new(BufWriter::new(file), Compression::default());
+            gz.write_all(&buf)?;
+            gz.finish()?;
+        }
+
+        std::fs::rename(&tmp, &path)?;
+    }
+
+    Ok(dir)
+}
+
+// ── Helpers ──────────────────────────────────────────────────────────
+
+fn chunk_band(value: f64) -> i32 {
+    (value / CHUNK_STEP as f64).floor() as i32
+}
+
+fn build_levels(cell: &CellValues) -> Vec<Level> {
+    fetcher::grid_level_keys()
+        .iter()
+        .filter_map(|k| {
+            let t = cell.get(k.tmp.as_str()).copied()?;
+            let h = cell.get(k.hgt.as_str()).copied()?;
+            let d = cell.get(k.dpt.as_str()).copied();
+            Some(Level {
+                pres_mb: k.pres_mb,
+                hght_m: h as f64,
+                tmpc: (t as f64) - 273.15,
+                dwpc: d.map(|v| (v as f64) - 273.15),
+            })
+        })
+        .collect()
+}
+
+/// Nearest level to `target_pres` within ±25 mb, mirroring Elixir's
+/// `WeatherLayers.level_value/3`.
+fn level_value(
+    sorted: &[&Level],
+    target_pres: f64,
+    project: impl Fn(&Level) -> Option<f64>,
+) -> Option<f64> {
+    let nearest = sorted.iter().min_by(|a, b| {
+        (a.pres_mb - target_pres)
+            .abs()
+            .partial_cmp(&(b.pres_mb - target_pres).abs())
+            .unwrap_or(std::cmp::Ordering::Equal)
+    })?;
+    if (nearest.pres_mb - target_pres).abs() <= 25.0 {
+        project(nearest)
+    } else {
+        None
+    }
+}
+
+fn compute_lapse_rate(sorted: &[&Level]) -> Option<f64> {
+    if sorted.len() < 2 {
+        return None;
+    }
+    let surface = sorted.first()?;
+    let top = sorted.last()?;
+    let dh_km = (top.hght_m - surface.hght_m) / 1000.0;
+    if dh_km > 0.0 {
+        Some((surface.tmpc - top.tmpc) / dh_km)
+    } else {
+        None
+    }
+}
+
+fn compute_layer_lapse_rate(sorted: &[&Level], lower_pres: f64, upper_pres: f64) -> Option<f64> {
+    if !(lower_pres > upper_pres) {
+        return None;
+    }
+    let lower = sorted
+        .iter()
+        .find(|l| (l.pres_mb - lower_pres).abs() <= 25.0)?;
+    let upper = sorted
+        .iter()
+        .find(|l| (l.pres_mb - upper_pres).abs() <= 25.0)?;
+    let dh_km = (upper.hght_m - lower.hght_m) / 1000.0;
+    if dh_km > 0.0 {
+        Some((lower.tmpc - upper.tmpc) / dh_km)
+    } else {
+        None
+    }
+}
+
+fn compute_inversion_strength(sorted: &[&Level]) -> Option<f64> {
+    if sorted.is_empty() {
+        return Some(0.0);
+    }
+    let mut max_str = 0.0_f64;
+    for pair in sorted.windows(2) {
+        let dt = pair[1].tmpc - pair[0].tmpc;
+        if dt > max_str {
+            max_str = dt;
+        }
+    }
+    Some(max_str)
+}
+
+fn compute_inversion_base_m(sorted: &[&Level]) -> Option<f64> {
+    if sorted.is_empty() {
+        return None;
+    }
+    let sfc_hght = sorted.first()?.hght_m;
+    let mut max_str = 0.0_f64;
+    let mut base: Option<f64> = None;
+    for pair in sorted.windows(2) {
+        let dt = pair[1].tmpc - pair[0].tmpc;
+        if dt > 0.0 && dt > max_str {
+            max_str = dt;
+            base = Some(pair[0].hght_m - sfc_hght);
+        }
+    }
+    if max_str > 0.0 {
+        base
+    } else {
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+    use flate2::read::GzDecoder;
+    use std::io::Read;
+    use std::sync::Arc;
+
+    fn cell_with(items: &[(&str, f32)]) -> CellValues {
+        items
+            .iter()
+            .map(|&(k, v)| (Arc::<str>::from(k), v))
+            .collect()
+    }
+
+    fn surface_only_cell() -> CellValues {
+        cell_with(&[
+            ("TMP:2 m above ground", 295.65), // 22.5 °C
+            ("DPT:2 m above ground", 285.65), // 12.5 °C → depression 10 °C
+            ("PRES:surface", 101_320.0),      // 1013.2 mb
+            ("HPBL:surface", 800.0),
+            (
+                "PWAT:entire atmosphere (considered as a single layer)",
+                25.0,
+            ),
+        ])
+    }
+
+    #[test]
+    fn drops_cells_with_unphysical_surface_temp() {
+        let mut cell = surface_only_cell();
+        cell.insert("TMP:2 m above ground".into(), 100.0); // way too cold
+        let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
+        assert!(derive_row(33.0, -97.0, vt, &cell).is_none());
+    }
+
+    #[test]
+    fn surface_only_row_carries_basic_fields() {
+        let cell = surface_only_cell();
+        let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
+        let row = derive_row(33.0, -97.0, vt, &cell).expect("row");
+
+        assert_eq!(row.lat, 33.0);
+        assert_eq!(row.lon, -97.0);
+        assert_eq!(row.valid_time, "2026-04-29T12:00:00Z");
+        assert!((row.temperature.unwrap() - 22.5).abs() < 1e-3);
+        assert!((row.dewpoint_depression.unwrap() - 10.0).abs() < 1e-3);
+        assert!((row.surface_pressure_mb.unwrap() - 1013.2).abs() < 1e-3);
+        assert!(row.surface_rh.unwrap() > 30.0 && row.surface_rh.unwrap() < 70.0);
+        assert_eq!(row.bl_height, Some(800.0));
+        assert_eq!(row.pwat, Some(25.0));
+
+        // No pressure-level data → upper-air fields stay nil.
+        assert_eq!(row.temp_850mb, None);
+        assert_eq!(row.lapse_rate, None);
+    }
+
+    #[test]
+    fn upper_air_levels_derive_lapse_rate_and_850mb() {
+        // Need real fetcher::GRID_PRESSURE_LEVELS keys. Build directly
+        // from grid_level_keys() so the synthetic cell mirrors prod.
+        let mut cell = surface_only_cell();
+        for k in fetcher::grid_level_keys() {
+            // Plausible synthetic profile: lapse 6.5 °C/km, scale height
+            // ≈ 8 km. Heights derived from std atmosphere approximation.
+            let h_m = match k.pres_mb as i32 {
+                1000 => 100.0,
+                925 => 800.0,
+                850 => 1500.0,
+                700 => 3000.0,
+                500 => 5600.0,
+                250 => 10_400.0,
+                _ => 1500.0 + (1000.0 - k.pres_mb) * 8.0,
+            };
+            let t_c = 22.5 - h_m * 6.5e-3; // °C
+            let d_c = t_c - 5.0;
+            cell.insert(k.tmp.clone().into(), (t_c + 273.15) as f32);
+            cell.insert(k.dpt.clone().into(), (d_c + 273.15) as f32);
+            cell.insert(k.hgt.clone().into(), h_m as f32);
+        }
+
+        let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
+        let row = derive_row(33.0, -97.0, vt, &cell).expect("row");
+
+        let temp_850 = row.temp_850mb.expect("850 mb T");
+        // h_m at 850 mb is 1500 → t_c = 22.5 - 1500 * 6.5e-3 = 12.75 °C
+        assert!(
+            (temp_850 - 12.75).abs() < 0.5,
+            "expected ~12.75 °C at 850 mb, got {temp_850}"
+        );
+
+        let lapse = row.lapse_rate.expect("lapse_rate");
+        assert!(
+            (lapse - 6.5).abs() < 0.5,
+            "expected ~6.5 °C/km lapse rate, got {lapse}"
+        );
+
+        let mid = row.mid_lapse_rate.expect("mid_lapse_rate");
+        assert!(
+            (mid - 6.5).abs() < 0.5,
+            "expected ~6.5 mid lapse, got {mid}"
+        );
+    }
+
+    #[test]
+    fn write_atomic_round_trip_via_msgpack() {
+        let dir = tempfile::tempdir().unwrap();
+        let vt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
+
+        let row = ScalarRow {
+            lat: 33.0,
+            lon: -97.0,
+            valid_time: "2026-04-29T12:00:00Z".to_string(),
+            temperature: Some(22.5),
+            dewpoint_depression: Some(10.0),
+            surface_rh: Some(50.0),
+            ducting: Some(false),
+            ..ScalarRow::default()
+        };
+
+        write_atomic(dir.path(), vt, &[row]).unwrap();
+
+        let chunk = dir
+            .path()
+            .join(SUBDIR)
+            .join("2026-04-29T12:00:00Z")
+            .join("6_-20.mp.gz");
+        assert!(chunk.exists());
+
+        let raw = std::fs::read(&chunk).unwrap();
+        let mut gz = GzDecoder::new(&raw[..]);
+        let mut buf = Vec::new();
+        gz.read_to_end(&mut buf).unwrap();
+        let decoded: rmpv::Value = rmp_serde::from_slice(&buf).unwrap();
+        let arr = decoded.as_array().expect("top-level array");
+        assert_eq!(arr.len(), 1);
+
+        let rmpv::Value::Map(pairs) = &arr[0] else {
+            panic!("row should be a map, got {:?}", arr[0]);
+        };
+        let lat = pairs
+            .iter()
+            .find(|(k, _)| k.as_str() == Some("lat"))
+            .unwrap();
+        assert_eq!(lat.1.as_f64(), Some(33.0));
+
+        let temp = pairs
+            .iter()
+            .find(|(k, _)| k.as_str() == Some("temperature"))
+            .unwrap();
+        assert_eq!(temp.1.as_f64(), Some(22.5));
+    }
+
+    #[test]
+    fn chunk_band_floors_match_elixir() {
+        // floor(33.0 / 5) = 6, floor(-97.0 / 5) = -20 (matches Elixir's
+        // `chunk_band` Float.floor/trunc).
+        assert_eq!(chunk_band(33.0), 6);
+        assert_eq!(chunk_band(-97.0), -20);
+        assert_eq!(chunk_band(35.0), 7);
+        assert_eq!(chunk_band(-95.0), -19);
+    }
+}

test/microwaveprop/weather/scalar_file_test.exs

@@ -155,6 +155,54 @@ defmodule Microwaveprop.Weather.ScalarFileTest do
     end
   end
 
+  describe "Rust-writer compatibility (the cross-language wire format)" do
+    test "decodes a hand-rolled gzipped MessagePack chunk and atomizes whitelisted keys" do
+      valid_time = ~U[2026-04-29 12:00:00Z]
+      dir = Path.join(ScalarFile.dir_for(valid_time), "")
+      File.mkdir_p!(dir)
+
+      # Mirror exactly what `prop_grid_rs::weather_scalar_file::write_atomic`
+      # emits: gzipped MessagePack, top-level array of string-keyed maps.
+      # If this round-trip ever breaks, Elixir is reading something the
+      # Rust pipeline cannot produce — a wire-format regression.
+      payload =
+        Msgpax.pack!(
+          [
+            %{
+              "lat" => 33.0,
+              "lon" => -97.0,
+              "valid_time" => "2026-04-29T12:00:00Z",
+              "temperature" => 22.5,
+              "dewpoint_depression" => 10.0,
+              "surface_rh" => 50.0,
+              "ducting" => false,
+              # An out-of-whitelist key — the reader must NOT atomize this.
+              "unknown_extra_key" => 1.23
+            }
+          ],
+          iodata: false
+        )
+
+      path = Path.join(dir, "6_-20.mp.gz")
+      File.write!(path, :zlib.gzip(payload))
+
+      [row] = ScalarFile.read_bounds(valid_time, nil)
+
+      # Whitelisted keys are atoms.
+      assert row.lat == 33.0
+      assert row.lon == -97.0
+      assert row.temperature == 22.5
+      assert row.ducting == false
+
+      # `valid_time` round-trips back to a DateTime.
+      assert row.valid_time == valid_time
+
+      # Out-of-whitelist keys stay as strings so a future field can be
+      # added without a deploy-order coordination problem.
+      assert row["unknown_extra_key"] == 1.23
+    end
+  end
+
   describe "prune_older_than/1 and retain_window/2" do
     test "prune_older_than removes files strictly before the cutoff" do
       old = ~U[2026-04-27 12:00:00Z]