feat(simulation): activate background generation tier end-to-end (#968, D-206/D-225)

The D-206 queue (#924) and D-203 cache (#917) were built but never connected to
the running app — the whole background tier was inert. This closes the loop:
submit -> Rayon -> cascade -> completion -> drain -> cache.

- gen_queue.rs: GenWorkItem::AnalyzeBody now carries enqueuer-resolved inputs
  { body_id, heightmap_path, sea_level, body_seed: SeedChain } (D-225 boundary —
  run_work_item stays pure compute, no path/DB resolution). run_work_item runs
  the real cascade: load heightmap.png -> downsample to GRID_W×GRID_H working
  grid (D-202) -> run_layer1 -> BodyWorldState; load failure -> Failed.
  GenCompletion::BodyAnalyzed carries the computed BodyWorldState.
- cascade.rs: CascadeSnapshot::into_body_world_state() conversion.
- plugin.rs (new): GenerationPlugin registers GenerationQueue +
  BodyWorldStateCache and adds a PreInput drain system that inserts BodyAnalyzed
  states into the cache (off the Rayon workers — a cheap channel drain, never
  the ~45ms cascade). Wired into both the production and test app setups.

Tests run the real cascade on a tiny temp heightmap PNG (no committed fixture);
the plugin test proves the full submit->...->cache loop. The proxy (#969) is the
production submitter.

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

server/src/atlas/cascade.rs

@@ -17,6 +17,7 @@
 
 use std::path::Path;
 
+use crate::atlas::body_world_state::{BodyWorldState, RiverNetwork};
 use crate::atlas::heightmap::{self, BodyHeightmap, HeightmapLoadError};
 use crate::atlas::layer1::{self, Layer1Output};
 use crate::seed::SeedChain;
@@ -48,6 +49,29 @@ pub struct CascadeSnapshot {
     pub layer1: Option<Layer1Output>,
 }
 
+impl CascadeSnapshot {
+    /// Convert into a [`BodyWorldState`] for the D-203 cache (#968). Moves the
+    /// heightmap raster and the Layer-1 outputs in; `last_accessed` starts at 0
+    /// (the cache stamps it on read). A snapshot that stopped at Layer 0 yields
+    /// empty river/basin/attractor data.
+    pub fn into_body_world_state(self) -> BodyWorldState {
+        let (river_network, drainage_basins, attractors) = match self.layer1 {
+            Some(l1) => (l1.river_network, l1.drainage_basins, l1.attractors),
+            None => (RiverNetwork::default(), Vec::new(), Vec::new()),
+        };
+        BodyWorldState {
+            body_id: self.body_id,
+            heightmap: self.heightmap.data,
+            heightmap_width: self.heightmap.width,
+            heightmap_height: self.heightmap.height,
+            river_network,
+            drainage_basins,
+            attractors,
+            last_accessed: 0,
+        }
+    }
+}
+
 /// Run the cascade from a heightmap already in memory, up to `up_to`.
 ///
 /// Pure (no I/O); this is the testable core. `body_seed` is this body's

server/src/atlas/gen_queue.rs

@@ -21,11 +21,17 @@
 //!
 //! **Thread count (D-206):** `available_parallelism - 2`, minimum 1.
 
+use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 
 use bevy_ecs::prelude::Resource;
 use crossbeam_channel::{Receiver, Sender};
 
+use crate::atlas::body_world_state::BodyWorldState;
+use crate::atlas::cascade::{run_cascade_from_heightmap, CascadeLayer};
+use crate::atlas::heightmap::{load_heightmap_png, GRID_H, GRID_W};
+use crate::seed::SeedChain;
+
 // ---------------------------------------------------------------------------
 // Priority
 // ---------------------------------------------------------------------------
@@ -50,8 +56,16 @@ pub enum GenPriority {
 /// A unit of background generation work (D-206).
 #[derive(Debug, Clone)]
 pub enum GenWorkItem {
-    /// Run D8 drainage analysis + attractor extraction for this body.
+    /// Run the Layer-1 cascade (drainage → features → sub-biome) for this body.
-    AnalyzeBody { body_id: String },
+    /// The enqueuer resolves the inputs (D-225): `heightmap_path` is the
+    /// mod-resolved source PNG, `body_seed` is this body's SeedChain position.
+    /// `run_work_item` is pure compute — it does no path/DB resolution.
+    AnalyzeBody {
+        body_id: String,
+        heightmap_path: PathBuf,
+        sea_level: f32,
+        body_seed: SeedChain,
+    },
     /// Generate a Phase 1 DistrictSkeleton for this city.
     GenerateSkeleton { city_id: u64 },
     /// Pre-fill a chunk in an existing district.
@@ -63,7 +77,7 @@ pub enum GenWorkItem {
 
 impl GenWorkItem {
     pub fn body_id(&self) -> Option<&str> {
-        if let GenWorkItem::AnalyzeBody { body_id } = self {
+        if let GenWorkItem::AnalyzeBody { body_id, .. } = self {
             Some(body_id)
         } else {
             None
@@ -80,6 +94,8 @@ impl GenWorkItem {
 pub enum GenCompletion {
     BodyAnalyzed {
         body_id: String,
+        /// The computed world state, ready for `BodyWorldStateCache::insert`.
+        state: BodyWorldState,
     },
     SkeletonGenerated {
         city_id: u64,
@@ -286,15 +302,38 @@ impl Default for GenerationQueue {
 // Work execution stub
 // ---------------------------------------------------------------------------
 
-/// Execute one work item. This is the Rayon task body.
+/// Execute one work item. This is the Rayon task body (off the tick thread).
 ///
-/// Currently a stub — real implementations will call `drainage::analyze()`,
+/// `AnalyzeBody` runs the real Layer-1 cascade (#968, D-225). `GenerateSkeleton`
-/// the attractor pipeline, and the district skeleton generator. Stubs return
+/// and `FillChunk` remain stubs — their layers (#957 / #959) are not built yet —
-/// immediate success to allow the queue infrastructure to be tested independently.
+/// returning immediate success so the queue infrastructure stays testable.
 fn run_work_item(item: &GenWorkItem) -> GenCompletion {
     match item {
-        GenWorkItem::AnalyzeBody { body_id } => GenCompletion::BodyAnalyzed {
+        GenWorkItem::AnalyzeBody {
-            body_id: body_id.clone(),
+            body_id,
+            heightmap_path,
+            sea_level,
+            body_seed,
+        } => match load_heightmap_png(heightmap_path, body_id, *sea_level) {
+            Ok(hm) => {
+                // Layer 1 runs at the GRID_W×GRID_H working resolution (D-202):
+                // downsample the higher-res stored heightmap first.
+                let working = if hm.width > GRID_W || hm.height > GRID_H {
+                    hm.downsample(GRID_W, GRID_H)
+                } else {
+                    hm
+                };
+                let snapshot =
+                    run_cascade_from_heightmap(*body_seed, working, CascadeLayer::Topography);
+                GenCompletion::BodyAnalyzed {
+                    body_id: body_id.clone(),
+                    state: snapshot.into_body_world_state(),
+                }
+            }
+            Err(e) => GenCompletion::Failed {
+                item: item.clone(),
+                reason: format!("heightmap load failed: {e}"),
+            },
         },
         GenWorkItem::GenerateSkeleton { city_id } => {
             GenCompletion::SkeletonGenerated { city_id: *city_id }
@@ -322,22 +361,51 @@ mod tests {
         GenerationQueue::with_threads(2)
     }
 
+    /// Write a tiny 16-bit grayscale heightmap PNG to a unique temp path so the
+    /// real cascade can run in `run_work_item` without a committed fixture.
+    fn test_heightmap_path() -> std::path::PathBuf {
+        use std::io::BufWriter;
+        use std::sync::atomic::{AtomicU32, Ordering};
+        static SEQ: AtomicU32 = AtomicU32::new(0);
+        let n = SEQ.fetch_add(1, Ordering::Relaxed);
+        let path = std::env::temp_dir().join(format!("sr_genq_{}_{n}.png", std::process::id()));
+        let file = std::fs::File::create(&path).expect("create test heightmap");
+        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().expect("png header");
+        let data: Vec<u8> = (0..32u32 * 16)
+            .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
+            .collect();
+        w.write_image_data(&data).expect("png data");
+        path
+    }
+
+    /// Build an `AnalyzeBody` work item pointing at a tiny test heightmap.
+    fn analyze(body_id: &str) -> GenWorkItem {
+        GenWorkItem::AnalyzeBody {
+            body_id: body_id.to_string(),
+            heightmap_path: test_heightmap_path(),
+            sea_level: 0.3,
+            body_seed: SeedChain::for_body(42, body_id),
+        }
+    }
+
     #[test]
     fn submit_and_drain() {
         let q = make_queue();
-        q.submit(
+        q.submit(analyze("TestBody"), GenPriority::Medium);
-            GenWorkItem::AnalyzeBody {
+        // Give Rayon time to load the heightmap and run the cascade.
-                body_id: "TestBody".to_string(),
+        std::thread::sleep(Duration::from_millis(100));
-            },
-            GenPriority::Medium,
-        );
-        // Give Rayon time to complete the (stub) task.
-        std::thread::sleep(Duration::from_millis(50));
         let completions = q.drain_completions();
         assert_eq!(completions.len(), 1);
+        // The work item ran the real cascade and produced a populated state.
         assert!(matches!(
             &completions[0],
-            GenCompletion::BodyAnalyzed { body_id } if body_id == "TestBody"
+            GenCompletion::BodyAnalyzed { body_id, state }
+                if body_id == "TestBody"
+                    && state.heightmap_width == 32
+                    && state.heightmap_height == 16
         ));
     }
 
@@ -345,18 +413,8 @@ mod tests {
     fn dedup_analyze_body() {
         let q = make_queue();
         // Submit the same body twice before it can complete.
-        q.submit(
+        q.submit(analyze("Dup"), GenPriority::Low);
-            GenWorkItem::AnalyzeBody {
+        q.submit(analyze("Dup"), GenPriority::Low);
-                body_id: "Dup".to_string(),
-            },
-            GenPriority::Low,
-        );
-        q.submit(
-            GenWorkItem::AnalyzeBody {
-                body_id: "Dup".to_string(),
-            },
-            GenPriority::Low,
-        );
         std::thread::sleep(Duration::from_millis(50));
         let completions = q.drain_completions();
         // Should have completed exactly once.
@@ -399,18 +457,8 @@ mod tests {
         //     Vec while "BodyA" is in-flight (in_flight_count = 1 = n_threads).
         //   - When "BodyA" completes, drain_completions() calls dispatch_next()
         //     which picks index 0 = "BodyB" (Immediate).
-        q.submit(
+        q.submit(analyze("BodyA"), GenPriority::Low);
-            GenWorkItem::AnalyzeBody {
+        q.submit(analyze("BodyB"), GenPriority::Immediate);
-                body_id: "BodyA".to_string(),
-            },
-            GenPriority::Low,
-        );
-        q.submit(
-            GenWorkItem::AnalyzeBody {
-                body_id: "BodyB".to_string(),
-            },
-            GenPriority::Immediate,
-        );
         // Wait for BodyA to complete.
         std::thread::sleep(Duration::from_millis(50));
         // drain_completions dispatches BodyB (Immediate, index 0 of pending).
@@ -421,9 +469,11 @@ mod tests {
 
         assert_eq!(first.len(), 1);
         assert_eq!(second.len(), 1);
-        assert!(matches!(&first[0], GenCompletion::BodyAnalyzed { body_id } if body_id == "BodyA"));
         assert!(
-            matches!(&second[0], GenCompletion::BodyAnalyzed { body_id } if body_id == "BodyB")
+            matches!(&first[0], GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "BodyA")
+        );
+        assert!(
+            matches!(&second[0], GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "BodyB")
         );
     }
 

server/src/atlas/mod.rs

@@ -13,6 +13,9 @@ pub mod features;
 pub mod gen_queue;
 pub mod heightmap;
 pub mod layer1;
+pub mod plugin;
 pub mod skeleton_gen;
 pub mod subbiome;
 pub mod tile_condition;
+
+pub use plugin::GenerationPlugin;

server/src/atlas/plugin.rs

@@ -0,0 +1,116 @@
+//! Generation tier plugin (#968, D-206) — wires the background generation queue
+//! and the per-body world-state cache into the running app.
+//!
+//! Registers [`GenerationQueue`] and [`BodyWorldStateCache`] as resources and
+//! adds a `PreInput` system that drains completed work each tick and inserts the
+//! computed [`BodyWorldState`](crate::atlas::body_world_state::BodyWorldState)
+//! into the cache. The queue's *submitter* is the atlas layer-stream proxy
+//! (#969, D-225); this plugin closes the submit→Rayon→cascade→drain→cache loop.
+
+use bevy_app::prelude::*;
+use bevy_ecs::prelude::*;
+use bevy_ecs::schedule::IntoScheduleConfigs;
+
+use crate::atlas::body_world_state::{BodyWorldStateCache, CACHE_CAPACITY};
+use crate::atlas::gen_queue::{GenCompletion, GenerationQueue};
+use crate::tick_phases::TickPhase;
+
+/// Wires the D-206 background generation tier into the app (#968).
+pub struct GenerationPlugin;
+
+impl Plugin for GenerationPlugin {
+    fn build(&self, app: &mut App) {
+        app.insert_resource(GenerationQueue::new())
+            .insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY))
+            .add_systems(
+                Update,
+                drain_generation_completions.in_set(TickPhase::PreInput),
+            );
+    }
+}
+
+/// Drain finished background work each tick and apply it to the cache (D-206).
+///
+/// Runs in `PreInput` (off the Rayon workers, on the main thread): a cheap
+/// channel drain + cache insert, never the ~45 ms cascade itself.
+fn drain_generation_completions(
+    queue: Res<GenerationQueue>,
+    mut cache: ResMut<BodyWorldStateCache>,
+) {
+    for completion in queue.drain_completions() {
+        match completion {
+            GenCompletion::BodyAnalyzed { state, .. } => cache.insert(state),
+            GenCompletion::Failed { item, reason } => {
+                tracing::warn!(?item, %reason, "background generation work item failed");
+            }
+            // Produced once the later layers land (#957 / #959); no consumer yet.
+            GenCompletion::SkeletonGenerated { .. } | GenCompletion::ChunkFilled { .. } => {}
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::atlas::gen_queue::{GenPriority, GenWorkItem};
+    use crate::seed::SeedChain;
+    use bevy_ecs::schedule::Schedule;
+    use std::time::Duration;
+
+    /// Tiny 16-bit grayscale heightmap PNG at a unique temp path, so the real
+    /// cascade can run without a committed fixture.
+    fn test_heightmap_path() -> std::path::PathBuf {
+        use std::io::BufWriter;
+        use std::sync::atomic::{AtomicU32, Ordering};
+        static SEQ: AtomicU32 = AtomicU32::new(0);
+        let n = SEQ.fetch_add(1, Ordering::Relaxed);
+        let path =
+            std::env::temp_dir().join(format!("sr_genplugin_{}_{n}.png", std::process::id()));
+        let file = std::fs::File::create(&path).expect("create test heightmap");
+        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().expect("png header");
+        let data: Vec<u8> = (0..32u32 * 16)
+            .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
+            .collect();
+        w.write_image_data(&data).expect("png data");
+        path
+    }
+
+    #[test]
+    fn drain_system_populates_cache() {
+        // The full loop: submit → Rayon cascade → completion → drain → cache.
+        let mut world = World::new();
+        world.insert_resource(GenerationQueue::new());
+        world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY));
+
+        world.resource::<GenerationQueue>().submit(
+            GenWorkItem::AnalyzeBody {
+                body_id: "PlanetX".to_string(),
+                heightmap_path: test_heightmap_path(),
+                sea_level: 0.3,
+                body_seed: SeedChain::for_body(42, "PlanetX"),
+            },
+            GenPriority::Immediate,
+        );
+
+        let mut sched = Schedule::default();
+        sched.add_systems(drain_generation_completions);
+
+        // Rayon runs the cascade asynchronously; the drain runs each schedule pass.
+        let mut found = false;
+        for _ in 0..100 {
+            sched.run(&mut world);
+            if world.resource::<BodyWorldStateCache>().contains("PlanetX") {
+                found = true;
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert!(
+            found,
+            "drain system should insert the analyzed body into the cache"
+        );
+    }
+}

server/src/main.rs

@@ -154,6 +154,7 @@ fn main() {
     app.add_plugins(settled_reach_server::storyteller::StorytellerPlugin);
     app.add_plugins(settled_reach_server::settings::SettingsPlugin);
     app.add_plugins(settled_reach_server::bookmark::BookmarkPlugin::default());
+    app.add_plugins(settled_reach_server::atlas::GenerationPlugin);
 
     // Initialize culture resolver (#679, D-128).
     // systems.db is shipped read-only alongside the binary.
@@ -391,6 +392,7 @@ fn dump_schedule_graph() {
     app.add_plugins(settled_reach_server::storyteller::StorytellerPlugin);
     app.add_plugins(settled_reach_server::settings::SettingsPlugin);
     app.add_plugins(settled_reach_server::bookmark::BookmarkPlugin::default());
+    app.add_plugins(settled_reach_server::atlas::GenerationPlugin);
     app.insert_resource(settled_reach_server::simulation::rng::SimRng::new(0));
 
     // Access Schedules resource directly — schedules are populated by plugins