feat(simulation): serve atlas layer requests over the bridge — completes #969 (D-225)

Closes the server-side layer-stream loop. serve_atlas_requests (PreInput) drains
the AtlasRequestBuffer and runs each through handle_atlas_request (cache hit ->
Ready; miss -> resolve via BodySourceResolver + enqueue an Immediate AnalyzeBody
-> Pending), buffering AtlasLayerResponses. send_atlas_responses (PostSnapshot)
flushes them to the client. main.rs wires BodySourceResolverResource (base root
= systems.db's 3rd ancestor; mod roots layer on later). Misses flow through the
#968 background tier and a re-request hits the now-warm cache.

Full path now live server-side: client request -> receive() demux -> serve ->
proxy -> (cache | queue+cascade) -> response -> client. The client half (send
request, decode response, render overlays) is #960.

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

server/src/atlas/plugin.rs

@@ -13,6 +13,11 @@ use bevy_ecs::schedule::IntoScheduleConfigs;
 
 use crate::atlas::body_world_state::{BodyWorldStateCache, CACHE_CAPACITY};
 use crate::atlas::gen_queue::{GenCompletion, GenerationQueue};
+use crate::atlas::layer_proxy::{handle_atlas_request, AtlasLayerResponse, AtlasLayerStatus};
+use crate::atlas::source_resolver::BodySourceResolverResource;
+use crate::bridge::{AtlasRequestBuffer, AtlasResponseBuffer};
+use crate::simulation::rng::SimRng;
+use crate::simulation::time::SimulationTime;
 use crate::tick_phases::TickPhase;
 
 /// Wires the D-206 background generation tier into the app (#968).
@@ -25,7 +30,39 @@ impl Plugin for GenerationPlugin {
             .add_systems(
                 Update,
                 drain_generation_completions.in_set(TickPhase::PreInput),
-            );
+            )
+            .add_systems(Update, serve_atlas_requests.in_set(TickPhase::PreInput));
+    }
+}
+
+/// Drain inbound atlas layer requests and serve each through the proxy (#969,
+/// D-225): cache hit → Ready, miss → resolve + enqueue + Pending. Responses are
+/// buffered for the bridge to flush in `PostSnapshot`.
+fn serve_atlas_requests(
+    mut requests: ResMut<AtlasRequestBuffer>,
+    mut responses: ResMut<AtlasResponseBuffer>,
+    mut cache: ResMut<BodyWorldStateCache>,
+    queue: Res<GenerationQueue>,
+    resolver: Option<Res<BodySourceResolverResource>>,
+    rng: Option<Res<SimRng>>,
+    time: Option<Res<SimulationTime>>,
+) {
+    if requests.0.is_empty() {
+        return;
+    }
+    let world_seed = rng.as_ref().map(|r| r.seed()).unwrap_or(0);
+    let tick = time.as_ref().map(|t| t.tick).unwrap_or(0);
+    let pending: Vec<_> = requests.0.drain(..).collect();
+    for req in pending {
+        let resp = match resolver.as_ref() {
+            Some(r) => handle_atlas_request(&req, &mut cache, &queue, &r.0, world_seed, tick),
+            None => AtlasLayerResponse {
+                body_id: req.body_id.clone(),
+                status: AtlasLayerStatus::Error("no body source resolver".to_string()),
+                layer1: None,
+            },
+        };
+        responses.0.push(resp);
     }
 }
 
@@ -113,4 +150,32 @@ mod tests {
             "drain system should insert the analyzed body into the cache"
         );
     }
+
+    #[test]
+    fn serve_drains_requests_into_responses() {
+        use crate::atlas::cascade::CascadeLayer;
+        use crate::atlas::layer_proxy::AtlasLayerRequest;
+
+        let mut world = World::new();
+        world.insert_resource(AtlasRequestBuffer(vec![AtlasLayerRequest {
+            body_id: "GJ1c".to_string(),
+            up_to: CascadeLayer::Topography,
+        }]));
+        world.insert_resource(AtlasResponseBuffer::default());
+        world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY));
+        world.insert_resource(GenerationQueue::with_threads(1));
+        // No resolver / SimRng / SimulationTime — all optional in the system.
+
+        let mut sched = Schedule::default();
+        sched.add_systems(serve_atlas_requests);
+        sched.run(&mut world);
+
+        let responses = world.resource::<AtlasResponseBuffer>();
+        assert_eq!(responses.0.len(), 1, "request should produce one response");
+        assert_eq!(responses.0[0].body_id, "GJ1c");
+        // No resolver wired → Error status (exercises the drain + push path).
+        assert!(matches!(responses.0[0].status, AtlasLayerStatus::Error(_)));
+        // The request buffer was drained.
+        assert!(world.resource::<AtlasRequestBuffer>().0.is_empty());
+    }
 }

server/src/bridge/mod.rs

@@ -265,6 +265,25 @@ impl Default for ServerRunning {
 #[derive(Resource, Default)]
 pub struct AtlasRequestBuffer(pub Vec<AtlasLayerRequest>);
 
+/// Outbound atlas layer responses, filled by the proxy serve system and flushed
+/// to the client in `PostSnapshot` (#969, D-225).
+#[derive(Resource, Default)]
+pub struct AtlasResponseBuffer(pub Vec<AtlasLayerResponse>);
+
+/// Flush buffered atlas responses to the client (#969, D-225). A failed send is
+/// logged but not fatal — an atlas response is not load-bearing like a snapshot.
+pub fn send_atlas_responses(
+    bridge: Option<Res<BridgeResource>>,
+    mut buffer: ResMut<AtlasResponseBuffer>,
+) {
+    let Some(bridge) = bridge else { return };
+    for resp in buffer.0.drain(..) {
+        if let Err(e) = bridge.send_atlas_response(&resp) {
+            tracing::warn!("failed to send atlas response for {}: {}", resp.body_id, e);
+        }
+    }
+}
+
 /// Bridge plugin for client-server communication
 /// Abstracts transport layer (LocalBridge/NetworkBridge)
 pub struct BridgePlugin;
@@ -281,9 +300,11 @@ impl Plugin for BridgePlugin {
             .init_resource::<DebugEnabled>()
             .init_resource::<crate::perception::query::ActivePerceptionMode>()
             .init_resource::<AtlasRequestBuffer>()
+            .init_resource::<AtlasResponseBuffer>()
             // Bridge I/O — PreInput (receive) and PostSnapshot (send)
             .add_systems(Update, receive_bridge_inputs.in_set(TickPhase::PreInput))
             .add_systems(Update, send_bridge_snapshot.in_set(TickPhase::PostSnapshot))
+            .add_systems(Update, send_atlas_responses.in_set(TickPhase::PostSnapshot))
             // Debug commands — Snapshot phase
             .add_systems(
                 Update,

server/src/main.rs

@@ -174,6 +174,30 @@ fn main() {
         }
     }
 
+    // Mod-first body source resolver for the atlas layer proxy (#969, D-225).
+    // terrain_reference is repo-root-relative; the repo root is systems.db's
+    // 3rd ancestor (<repo>/server/data/systems.db).
+    let world_root = systems_db_path
+        .canonicalize()
+        .ok()
+        .and_then(|p| p.ancestors().nth(3).map(std::path::Path::to_path_buf))
+        .unwrap_or_else(|| std::path::PathBuf::from(".."));
+    match settled_reach_server::atlas::source_resolver::BodySourceResolver::open(
+        &systems_db_path,
+        vec![world_root.clone()],
+    ) {
+        Ok(resolver) => {
+            tracing::info!("Body source resolver opened (root: {:?})", world_root);
+            app.insert_resource(
+                settled_reach_server::atlas::source_resolver::BodySourceResolverResource(resolver),
+            );
+        }
+        Err(e) => tracing::warn!(
+            "Body source resolver unavailable ({}). Atlas layer requests will error.",
+            e
+        ),
+    }
+
     // Initialize SQLite settings store (#627).
     // Path: alongside save files in the server's working directory.
     let settings_path = std::path::PathBuf::from("settings.db");