feat(simulation): atlas layer-proxy handler — cache hit/miss -> queue (#969, D-225)

handle_atlas_request: cache hit -> serialize Layer1Output, reply Ready; miss ->
resolve the source heightmap (BodySourceResolver) + enqueue an Immediate
AnalyzeBody on the #968 background queue, reply Pending (client re-requests; the
drain system populates the cache so a later request hits); unknown/no-terrain ->
NotFound. Adds the AtlasLayerRequest/Response/Status wire types (+ Serialize on
CascadeLayer). Pure handler; the bridge routing is the proxy's other half.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

server/src/atlas/cascade.rs

@@ -17,6 +17,8 @@
 
 use std::path::Path;
 
+use serde::{Deserialize, Serialize};
+
 use crate::atlas::body_world_state::{BodyWorldState, RiverNetwork};
 use crate::atlas::heightmap::{self, BodyHeightmap, HeightmapLoadError};
 use crate::atlas::layer1::{self, Layer1Output};
@@ -25,7 +27,7 @@ use crate::seed::SeedChain;
 /// Cascade layers in execution order (D-200). [`run_cascade`] runs every layer
 /// up to and including the requested one. Append new layers as they are built;
 /// the `Ord` derive relies on declaration order, so only ever append.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub enum CascadeLayer {
     /// Layer 0 — load the pre-baked 16-bit `heightmap.png` (D-202).
     Heightmap,

server/src/atlas/layer_proxy.rs

@@ -0,0 +1,239 @@
+//! Atlas layer-stream proxy handler (#969, D-225).
+//!
+//! Serves a body's generation-cascade layer data to the client, compute-on-
+//! demand and mod-first:
+//! - **Cache hit** → serialize the cached `Layer1Output` and reply `Ready`.
+//! - **Cache miss** → resolve the body's source heightmap ([`BodySourceResolver`]),
+//!   enqueue an `Immediate` `AnalyzeBody` on the background queue (#968), and
+//!   reply `Pending` (the client re-requests; the drain system populates the
+//!   cache, so a later request hits).
+//!
+//! Pure handler logic; the bridge wiring (message routing) is the proxy's other
+//! half. No baking — the heightmap is the only source of truth (D-225).
+
+use serde::{Deserialize, Serialize};
+
+use crate::atlas::body_world_state::{BodyWorldStateCache, SimTick};
+use crate::atlas::cascade::CascadeLayer;
+use crate::atlas::gen_queue::{GenPriority, GenWorkItem, GenerationQueue};
+use crate::atlas::layer1::Layer1Output;
+use crate::atlas::source_resolver::{BodySourceResolver, SourceResolveError};
+use crate::seed::SeedChain;
+
+/// Fallback sea level when the heightmap PNG carries no `sea_level` tEXt chunk
+/// (the loader prefers the chunk; this is only the floor).
+const DEFAULT_SEA_LEVEL: f32 = 0.3;
+
+/// A client request for a body's generation layers (D-225). `up_to` is a
+/// forward-compat seam — v1 always runs the Layer-1 (Topography) cascade.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AtlasLayerRequest {
+    pub body_id: String,
+    pub up_to: CascadeLayer,
+}
+
+/// Status of a layer response (D-225).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum AtlasLayerStatus {
+    /// Layer data is ready (`layer1` is populated).
+    Ready,
+    /// Analysis was enqueued; the client should re-request shortly.
+    Pending,
+    /// The body is unknown or has no source terrain — re-requesting won't help.
+    NotFound,
+    /// Resolution / IO failure (message for the client log).
+    Error(String),
+}
+
+/// A layer response: the computed `Layer1Output`, or a non-ready status.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AtlasLayerResponse {
+    pub body_id: String,
+    pub status: AtlasLayerStatus,
+    pub layer1: Option<Layer1Output>,
+}
+
+/// Serve one layer request (D-225). `current_tick` stamps the cache LRU on hit;
+/// `world_seed` derives the body's `SeedChain` for the enqueued analysis.
+pub fn handle_atlas_request(
+    req: &AtlasLayerRequest,
+    cache: &mut BodyWorldStateCache,
+    queue: &GenerationQueue,
+    resolver: &BodySourceResolver,
+    world_seed: u64,
+    current_tick: SimTick,
+) -> AtlasLayerResponse {
+    // Cache hit — serve immediately.
+    if let Some(state) = cache.get(&req.body_id, current_tick) {
+        let layer1 = Layer1Output {
+            body_id: state.body_id.clone(),
+            river_network: state.river_network.clone(),
+            drainage_basins: state.drainage_basins.clone(),
+            attractors: state.attractors.clone(),
+        };
+        return AtlasLayerResponse {
+            body_id: req.body_id.clone(),
+            status: AtlasLayerStatus::Ready,
+            layer1: Some(layer1),
+        };
+    }
+
+    // Miss — resolve the source heightmap and enqueue background analysis.
+    match resolver.resolve(&req.body_id) {
+        Ok(heightmap_path) => {
+            queue.submit(
+                GenWorkItem::AnalyzeBody {
+                    body_id: req.body_id.clone(),
+                    heightmap_path,
+                    sea_level: DEFAULT_SEA_LEVEL,
+                    body_seed: SeedChain::for_body(world_seed, &req.body_id),
+                },
+                GenPriority::Immediate,
+            );
+            AtlasLayerResponse {
+                body_id: req.body_id.clone(),
+                status: AtlasLayerStatus::Pending,
+                layer1: None,
+            }
+        }
+        // Unknown / no terrain → re-requesting won't help.
+        Err(SourceResolveError::UnknownBody(_))
+        | Err(SourceResolveError::NoTerrainReference { .. }) => AtlasLayerResponse {
+            body_id: req.body_id.clone(),
+            status: AtlasLayerStatus::NotFound,
+            layer1: None,
+        },
+        Err(e) => AtlasLayerResponse {
+            body_id: req.body_id.clone(),
+            status: AtlasLayerStatus::Error(e.to_string()),
+            layer1: None,
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::atlas::body_world_state::{BodyWorldState, RiverNetwork, CACHE_CAPACITY};
+    use crate::atlas::gen_queue::GenCompletion;
+    use rusqlite::Connection;
+    use std::path::{Path, PathBuf};
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::time::Duration;
+
+    static SEQ: AtomicU32 = AtomicU32::new(0);
+
+    fn req(body_id: &str) -> AtlasLayerRequest {
+        AtlasLayerRequest {
+            body_id: body_id.to_string(),
+            up_to: CascadeLayer::Topography,
+        }
+    }
+
+    const REL: &str = "wiki/star-systems/GJ-1/bodies/GJ1c/heightmap.png";
+
+    /// systems.db with one bodies row, + a base root containing a tiny 16-bit
+    /// heightmap PNG at the body's terrain_reference. Returns (db, resolver).
+    fn resolver_with_body(body_id: &str) -> (PathBuf, BodySourceResolver) {
+        let n = SEQ.fetch_add(1, Ordering::Relaxed);
+        let db = std::env::temp_dir().join(format!("sr_proxy_{}_{n}.db", std::process::id()));
+        let _ = std::fs::remove_file(&db);
+        let conn = Connection::open(&db).unwrap();
+        conn.execute(
+            "CREATE TABLE bodies (body_id TEXT PRIMARY KEY, terrain_reference TEXT)",
+            [],
+        )
+        .unwrap();
+        conn.execute(
+            "INSERT INTO bodies (body_id, terrain_reference) VALUES (?1, ?2)",
+            rusqlite::params![body_id, REL],
+        )
+        .unwrap();
+
+        let root = std::env::temp_dir().join(format!("sr_proxyroot_{}_{n}", std::process::id()));
+        write_tiny_heightmap(&root.join(REL));
+
+        let resolver = BodySourceResolver::open(&db, vec![root]).unwrap();
+        (db, resolver)
+    }
+
+    fn write_tiny_heightmap(path: &Path) {
+        use std::io::BufWriter;
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        let file = std::fs::File::create(path).unwrap();
+        let mut enc = png::Encoder::new(BufWriter::new(file), 32, 16);
+        enc.set_color(png::ColorType::Grayscale);
+        enc.set_depth(png::BitDepth::Sixteen);
+        let mut w = enc.write_header().unwrap();
+        let data: Vec<u8> = (0..32u32 * 16)
+            .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
+            .collect();
+        w.write_image_data(&data).unwrap();
+    }
+
+    fn empty_resolver() -> (PathBuf, BodySourceResolver) {
+        let n = SEQ.fetch_add(1, Ordering::Relaxed);
+        let db = std::env::temp_dir().join(format!("sr_proxye_{}_{n}.db", std::process::id()));
+        let _ = std::fs::remove_file(&db);
+        let conn = Connection::open(&db).unwrap();
+        conn.execute(
+            "CREATE TABLE bodies (body_id TEXT, terrain_reference TEXT)",
+            [],
+        )
+        .unwrap();
+        let resolver = BodySourceResolver::open(&db, vec![std::env::temp_dir()]).unwrap();
+        (db, resolver)
+    }
+
+    #[test]
+    fn cache_hit_is_ready() {
+        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
+        cache.insert(BodyWorldState {
+            body_id: "GJ1c".into(),
+            heightmap: vec![0.0; 4],
+            heightmap_width: 2,
+            heightmap_height: 2,
+            river_network: RiverNetwork::default(),
+            drainage_basins: vec![],
+            attractors: vec![],
+            last_accessed: 0,
+        });
+        let (_db, resolver) = empty_resolver();
+        let queue = GenerationQueue::with_threads(1);
+
+        let resp = handle_atlas_request(&req("GJ1c"), &mut cache, &queue, &resolver, 42, 1);
+        assert_eq!(resp.status, AtlasLayerStatus::Ready);
+        assert_eq!(resp.layer1.expect("layer1").body_id, "GJ1c");
+    }
+
+    #[test]
+    fn cache_miss_enqueues_and_pends_then_analyzes() {
+        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
+        let (_db, resolver) = resolver_with_body("GJ1c");
+        let queue = GenerationQueue::with_threads(1);
+
+        let resp = handle_atlas_request(&req("GJ1c"), &mut cache, &queue, &resolver, 42, 1);
+        assert_eq!(resp.status, AtlasLayerStatus::Pending);
+        assert!(resp.layer1.is_none());
+
+        // The enqueued analysis runs the real cascade and completes.
+        std::thread::sleep(Duration::from_millis(150));
+        let completions = queue.drain_completions();
+        assert!(
+            completions.iter().any(
+                |c| matches!(c, GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "GJ1c")
+            ),
+            "miss should enqueue an AnalyzeBody that completes: {completions:?}"
+        );
+    }
+
+    #[test]
+    fn unknown_body_is_not_found() {
+        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
+        let (_db, resolver) = empty_resolver();
+        let queue = GenerationQueue::with_threads(1);
+
+        let resp = handle_atlas_request(&req("ghost"), &mut cache, &queue, &resolver, 42, 1);
+        assert_eq!(resp.status, AtlasLayerStatus::NotFound);
+    }
+}

server/src/atlas/mod.rs

@@ -13,6 +13,7 @@ pub mod features;
 pub mod gen_queue;
 pub mod heightmap;
 pub mod layer1;
+pub mod layer_proxy;
 pub mod plugin;
 pub mod skeleton_gen;
 pub mod source_resolver;