//! 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 bevy_ecs::prelude::Resource;
use serde::{Deserialize, Serialize};

use crate::atlas::body_params_reader::BodyParamsReader;
use crate::atlas::body_world_state::{BodyWorldState, BodyWorldStateCache, SimTick};
use crate::atlas::cascade::CascadeLayer;
use crate::atlas::city_context_reader::CityContextReader;
use crate::atlas::district_profile::{BodyParams, DistrictPos};
use crate::atlas::gen_queue::{GenPriority, GenWorkItem, GenerationQueue};
use crate::atlas::layer1::Layer1Output;
use crate::atlas::road_graph::RoadNodeKind;
use crate::atlas::scale::DISTRICT_M;
use crate::atlas::source_resolver::{BodySourceResolver, SourceResolveError};
use crate::bridge::ConnectionId;
use crate::seed::{SeedChain, SeedDomain};
use crate::simulation::generator::{AttractorType, DistrictType, MaintenanceAuthority, ZoningType};

/// 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;

/// Hard server-side clamp on [`AtlasLayerRequest::window_n`] (D-226 T-1124
/// amendment §4, binding numbers). 64×64 districts ≈ 131 km per side — the
/// same window size `aliveness_probe --render`'s default already proved out
/// server-side (T-1123). **Never trust `window_n` from the wire** — every
/// caller clamps to `[1, DISTRICT_WINDOW_MAX_N]` before deriving.
pub const DISTRICT_WINDOW_MAX_N: u32 = 64;

/// A client request for a body's generation layers (D-225), extended with an
/// optional district-resolution window query (D-226 T-1124 amendment §1, T-1137).
///
/// `up_to` is a forward-compat seam that is **not yet honored**: `run_work_item`
/// (`gen_queue.rs`) currently runs the cascade through `CascadeLayer::Region`
/// (the terminal layer, T-1113) unconditionally on every request, ignoring this
/// field. Wiring per-request depth (and the partial caching it implies) is
/// deferred to #1021.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AtlasLayerRequest {
    pub body_id: String,
    pub up_to: CascadeLayer,
    /// District-window centre (D-226 T-1124 amendment §1, T-1137). `None` = no
    /// window requested (whole-body layers only — today's behavior, byte-unchanged
    /// for every existing caller thanks to `#[serde(default)]`).
    #[serde(default)]
    pub window_center: Option<DistrictPos>,
    /// Window side length in districts. Ignored when `window_center` is `None`.
    /// Clamped server-side to `[1, DISTRICT_WINDOW_MAX_N]` — **never trusted
    /// from the wire** (D-226 T-1124 amendment §4).
    #[serde(default)]
    pub window_n: u32,
}

/// 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),
}

/// The cascade's coarse district grid, surfaced for the Atlas generation overlay
/// (T-1046, D-226). This is the **planetary-scale** view — one cell per coarse
/// grid square (`grid_w/cols` heightmap pixels) — not the on-demand 2 km districts
/// (those derive only when a player enters a settlement, Phase 5). `morphology`
/// and `elev_q` are row-major (`rows × cols`); `morphology[i]` is a `MorphologyZone`
/// discriminant (D-239 §6, `repr(u8)`), `elev_q[i]` is 0–100 elevation for relief
/// shading. The client maps `cols × rows` onto the displayed heightmap.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DistrictGridLayer {
    pub cols: u32,
    pub rows: u32,
    pub morphology: Vec<u8>,
    pub elev_q: Vec<u8>,
}

/// A layer response: the computed `Layer1Output` + the coarse district grid
/// (D-225, T-1046) + the road-graph and settlement overlays (T-960 §1/§2) +
/// the region climate grid (T-1113) + the quarter-footprint overlay (T-1112,
/// T-1119) + the district-resolution window query (T-1124, T-1137), or a
/// non-ready status.
///
/// Growth ceiling (governance-bounded): the one-`Option`-field-per-layer
/// pattern tops out at six fields for the **dense whole-body layer family**
/// (`district_grid`, `road_graph`, `settlements`, `region_grid`,
/// `quarter_footprints` — each a compute-once, cache-per-body snapshot) —
/// D-226's 2026-07-13 amendment (d) rules out any L5/tile Atlas layer ever,
/// and `quarter_footprints` below is the last candidate the 2026-07-16 T-1112
/// amendment named. **That budget is now consumed:** a seventh *whole-body*
/// field is not a naming exercise like the six before it — a future
/// generation-layer addition needs its own governance, not a drive-by field.
///
/// The D-226 T-1124 amendment (2026-07-18) RESOLVED what carries the next
/// addition, and it is NOT this family: a **windowed viewport query** is a
/// categorically different payload (keyed on the *request* `(body, center, n)`,
/// re-fetched per pan, not a per-body snapshot). `district_window` (wired here,
/// T-1137) rides on `AtlasLayerResponse` but is explicitly OUTSIDE the
/// whole-body family and does not count against the six-field ceiling above
/// (D-226 T-1124 §2). The windowed family has its own hard cap: exactly ONE
/// windowed-query field; a second windowed query (a second viewport, a
/// windowed chunk-preview) is a dedicated response message by rule, not a
/// second `Option` here (D-226 T-1124 §2, symmetric with the request-side
/// five-shape demux ceiling in `bridge/mod.rs`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AtlasLayerResponse {
    pub body_id: String,
    pub status: AtlasLayerStatus,
    pub layer1: Option<Layer1Output>,
    /// The coarse district/morphology grid for the Atlas overlay (T-1046).
    /// `Some` on a cache hit once the DistrictProfile layer has run; `None` otherwise.
    pub district_grid: Option<DistrictGridLayer>,
    /// The inter-settlement road/rail graph overlay (T-960 §1, T-1038).
    /// `Some` on a cache hit once the RoadGraph layer has run; `None` otherwise
    /// (including a body with zero placed settlements — an empty graph has no
    /// nodes to draw, so it collapses to `None` the same way `district_grid`
    /// does for an unrun layer).
    pub road_graph: Option<RoadGraphLayer>,
    /// The settlement-placement overlay (T-960 §2, #955). `Some` on a cache hit
    /// once the Settlement layer has placed at least one city; `None` otherwise.
    pub settlements: Option<SettlementLayer>,
    /// The region climate grid for the Atlas overlay (D-243 §3, T-1113).
    /// `Some` on a cache hit once the Region layer has run; `None` otherwise.
    pub region_grid: Option<RegionGridLayer>,
    /// The requested district window (D-226 T-1124 amendment, T-1137), or
    /// `None` when the request carried no `window_center` / no window data is
    /// cached yet for a pending derive. Distinct from the five layers above:
    /// keyed on the REQUEST `(body, center, n)`, not on the body alone — see
    /// the struct-level doc.
    pub district_window: Option<DistrictWindowLayer>,
    /// The quarter-footprint overlay (D-226 T-1112 amendment, T-1119). `Some`
    /// on a cache hit once at least one settlement's quarter skeleton has been
    /// generated (`state.quarters` non-empty); `None` otherwise, including a
    /// body with placed settlements whose quarters haven't finished the async
    /// `GenerateSkeleton` pass yet (skeleton generation runs at `Low` priority
    /// after the body's own `Ready` snapshot is cached — see `plugin.rs`).
    pub quarter_footprints: Option<QuarterFootprintLayer>,
}

/// Build the coarse [`DistrictGridLayer`] from a body's cached state (T-1046).
/// Returns `None` when the DistrictProfile layer has not run (empty `districts`).
/// The grid is dense `[0, cols) × [0, rows)` (the cascade tiles the full
/// heightmap), so the extent comes from the maximum `DistrictPos`.
pub fn build_district_grid(
    state: &crate::atlas::body_world_state::BodyWorldState,
) -> Option<DistrictGridLayer> {
    if state.districts.is_empty() {
        return None;
    }
    let cols = state.districts.keys().map(|(x, _)| *x).max().unwrap_or(0) as u32 + 1;
    let rows = state.districts.keys().map(|(_, y)| *y).max().unwrap_or(0) as u32 + 1;
    let n = (cols * rows) as usize;
    let mut morphology = vec![0u8; n];
    let mut elev_q = vec![0u8; n];
    for (&(x, y), profile) in &state.districts {
        if x < 0 || y < 0 {
            continue;
        }
        let i = (y as u32 * cols + x as u32) as usize;
        if i < n {
            morphology[i] = profile.morphology_zone as u8;
            elev_q[i] = profile.elev_q.clamp(0, 100) as u8;
        }
    }
    Some(DistrictGridLayer {
        cols,
        rows,
        morphology,
        elev_q,
    })
}

// ---------------------------------------------------------------------------
// RegionGridLayer (T-1113, D-243 §3)
// ---------------------------------------------------------------------------

/// The ~205 km region climate grid for the Atlas overlay (T-1113), dense
/// row-major like [`DistrictGridLayer`] (the T-1046 encoding precedent).
/// Serves the **mean-state** `RegionClock` fields only — the Q-105 tick-phase
/// callbacks are deferred, so what ships is the static climate context.
///
/// Wire encoding is all-integer (D-010 wire discipline):
/// - `season[i]` / `weather[i]` — the `repr(u8)` discriminants of
///   `SeasonPhase` / `WeatherState` (pinned, append-only).
/// - `mean_temp_dc[i]` — mean-annual temperature baseline in **deci-°C**
///   (×10, `round`ed; 0.1 °C is ample for a map overlay). `i16::MIN` is the
///   sentinel for "no atmosphere → no temperature" (airless bodies carry
///   `mean_temp_c: None`); real values are class-band-clamped far inside
///   i16 range.
/// - `moisture_q[i]` — the 0–100 region moisture primitive.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RegionGridLayer {
    pub cols: u32,
    pub rows: u32,
    pub season: Vec<u8>,
    pub weather: Vec<u8>,
    pub mean_temp_dc: Vec<i16>,
    pub moisture_q: Vec<u8>,
}

/// Sentinel for "airless body — no temperature baseline" in
/// [`RegionGridLayer::mean_temp_dc`].
pub const REGION_TEMP_NONE_DC: i16 = i16::MIN;

/// Build the [`RegionGridLayer`] from a body's cached state (T-1113).
/// Returns `None` when the Region layer has not run (empty `regions`).
/// The stored region set is the dense covering grid `[0, cols) × [0, rows)`
/// (see `cascade::LayerRegionOutput` — no blend-padding ring), so the extent
/// comes from the maximum `RegionPos`, mirroring [`build_district_grid`].
pub fn build_region_grid(
    state: &crate::atlas::body_world_state::BodyWorldState,
) -> Option<RegionGridLayer> {
    if state.regions.is_empty() {
        return None;
    }
    let cols = state.regions.keys().map(|(x, _)| *x).max().unwrap_or(0) as u32 + 1;
    let rows = state.regions.keys().map(|(_, y)| *y).max().unwrap_or(0) as u32 + 1;
    let n = (cols * rows) as usize;
    let mut season = vec![0u8; n];
    let mut weather = vec![0u8; n];
    let mut mean_temp_dc = vec![REGION_TEMP_NONE_DC; n];
    let mut moisture_q = vec![0u8; n];
    for (&(x, y), profile) in &state.regions {
        if x < 0 || y < 0 {
            continue;
        }
        let i = (y as u32 * cols + x as u32) as usize;
        if i < n {
            season[i] = profile.clock.season as u8;
            weather[i] = profile.clock.weather as u8;
            mean_temp_dc[i] = match profile.clock.mean_temp_c {
                Some(t) => {
                    ((t * 10.0).round() as i32).clamp(i16::MIN as i32 + 1, i16::MAX as i32) as i16
                }
                None => REGION_TEMP_NONE_DC,
            };
            moisture_q[i] = profile.moisture_q.clamp(0, 100) as u8;
        }
    }
    Some(RegionGridLayer {
        cols,
        rows,
        season,
        weather,
        mean_temp_dc,
        moisture_q,
    })
}

// ---------------------------------------------------------------------------
// DistrictWindowLayer (D-226 T-1124 amendment, T-1137)
// ---------------------------------------------------------------------------

/// The requested district window: an `n × n` grid of TRUE 2 km districts
/// centred on `center`, derived on-demand via `district_profile::derive_district`
/// (D-226 T-1124 amendment §2). **Echoes `center`/`n` back** — this is the
/// client's race-condition guard, not a convenience field: because
/// `derive_district` is pure and deterministic (D-227), the same `(center, n)`
/// query always yields the same payload, so the echoed tuple *is* the
/// cache/staleness key the client compares against its most recently requested
/// window (`body_id` disambiguation rides the enclosing `AtlasLayerResponse`,
/// not the echo — see the amendment).
///
/// All six arrays are dense row-major `n × n` (`i = row * n + col`), matching
/// the `DistrictGridLayer`/`RegionGridLayer` indexing convention. Per-cell wire
/// cost is 7 bytes (1+1+2+1+1+1) before MessagePack framing overhead (D-226
/// T-1124 amendment §4).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DistrictWindowLayer {
    pub center: DistrictPos,
    pub n: u32,
    /// `MorphologyZone` discriminant, the frozen 17-zone vocabulary (D-239 §6).
    pub morphology: Vec<u8>,
    /// 0-100, matches `DistrictGridLayer.elev_q` encoding.
    pub elev_q: Vec<u8>,
    /// Deci-°C, [`REGION_TEMP_NONE_DC`] sentinel — the SAME scheme as
    /// `RegionGridLayer.mean_temp_dc`, deliberately not a separate
    /// district-tier quantization (one temperature colorizer spans both zoom
    /// levels, D-226 T-1124 amendment §2).
    pub temp_dc: Vec<i16>,
    /// 0-100, matches `DistrictGridLayer` precedent.
    pub moisture_q: Vec<u8>,
    /// `VegetationClass` discriminant, 0-6 including `Marine = 6` (T-1126) —
    /// any client palette MUST be exhaustive over `Marine` (D-226 T-1124
    /// amendment §3, non-negotiable — the ocean-blind-vegetation bug this
    /// field caught at the district tier).
    pub vegetation: Vec<u8>,
    /// `GlaciationGrade` discriminant, 0-4 (T-1127).
    pub glaciation: Vec<u8>,
}

/// Key for the server-side window derive cache (T-1137): `(body_id, center, n)`.
/// D-227 purity means a cached window is valid forever for a given body+seed —
/// no staleness/TTL invalidation is needed, only a bound on unbounded growth
/// (see [`DistrictWindowCache`]).
pub type DistrictWindowKey = (String, DistrictPos, u32);

/// Bounded LRU-ish cache of completed district-window derives (T-1137), a
/// sibling to [`BodyWorldStateCache`] rather than a field on it: windows are
/// keyed on the *request* `(body, center, n)`, not the body alone (see the
/// struct-level doc on [`AtlasLayerResponse`]), so they don't fit the
/// per-body cache's keying at all. Eviction is capacity-only FIFO-by-insertion
/// (not access-recency LRU like `BodyWorldStateCache`) — window requests are
/// comparatively rare and cheap to re-derive on a genuine miss (a background
/// re-submit, never a stall), so exact recency tracking isn't worth the
/// bookkeeping; a simple bound against unbounded growth is enough.
#[derive(Resource, Debug, Default)]
pub struct DistrictWindowCache {
    entries: std::collections::BTreeMap<DistrictWindowKey, DistrictWindowLayer>,
    /// Insertion order, oldest first — the eviction queue.
    order: std::collections::VecDeque<DistrictWindowKey>,
    capacity: usize,
}

/// Default capacity for [`DistrictWindowCache`] — generous relative to
/// `BodyWorldStateCache::CACHE_CAPACITY` (50 bodies) since each entry here is
/// far smaller (a handful of `Vec<u8>`/`Vec<i16>` at `n ≤ 64`, ≤ 28 KiB raw vs.
/// `BodyWorldState`'s full heightmap + districts + regions), and several
/// windows can legitimately be live per body (a player panning around).
pub const DISTRICT_WINDOW_CACHE_CAPACITY: usize = 256;

impl DistrictWindowCache {
    pub fn new(capacity: usize) -> Self {
        Self {
            entries: std::collections::BTreeMap::new(),
            order: std::collections::VecDeque::new(),
            capacity,
        }
    }

    /// Look up a cached window by its full key. Never mutates — window
    /// validity has no time component (D-227), so there is nothing to bump.
    pub fn get(&self, key: &DistrictWindowKey) -> Option<&DistrictWindowLayer> {
        self.entries.get(key)
    }

    /// Insert a completed window derive, evicting the oldest entry first if
    /// at capacity. Re-inserting an existing key replaces the value without
    /// moving it in the eviction order (D-227: the value can only ever be
    /// identical, so this is a no-op in practice, but stays correct either way).
    pub fn insert(&mut self, key: DistrictWindowKey, layer: DistrictWindowLayer) {
        if !self.entries.contains_key(&key) {
            if self.entries.len() >= self.capacity {
                if let Some(victim) = self.order.pop_front() {
                    self.entries.remove(&victim);
                }
            }
            self.order.push_back(key.clone());
        }
        self.entries.insert(key, layer);
    }

    pub fn len(&self) -> usize {
        self.entries.len()
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }
}

/// Build a [`DistrictWindowLayer`] by deriving every district in the
/// `n × n` window around `center` (T-1137). Mirrors
/// `aliveness_probe::render_window_panels`'s derive loop exactly (the probe
/// this design promotes to a served layer, D-226 T-1124 amendment §2) — same
/// row-major indexing, same `derive_district` call per cell.
///
/// `n` MUST already be clamped to `[1, DISTRICT_WINDOW_MAX_N]` by the caller —
/// this function trusts it verbatim (the clamp is `handle_atlas_request`'s
/// job, applied once at the wire boundary, not re-checked on every internal
/// caller per the existing codebase convention of clamping at the edge).
pub fn build_district_window_layer(
    seed: SeedChain,
    body_id: &str,
    params: &crate::atlas::district_profile::BodyParams,
    ta: &crate::atlas::features::TerrainAnalysis,
    center: DistrictPos,
    n: u32,
    climate: &crate::atlas::district_profile::ClimateConstants,
) -> DistrictWindowLayer {
    let n_i = n as i32;
    let half = n_i / 2;
    let cells = (n * n) as usize;
    let mut morphology = vec![0u8; cells];
    let mut elev_q = vec![0u8; cells];
    let mut temp_dc = vec![REGION_TEMP_NONE_DC; cells];
    let mut moisture_q = vec![0u8; cells];
    let mut vegetation = vec![0u8; cells];
    let mut glaciation = vec![0u8; cells];
    for row in 0..n_i {
        for col in 0..n_i {
            // Row 0 = northmost, matching aliveness_probe's render_window_panels
            // (derive_district maps negative wy to negative lat_frac = north).
            let dp = (center.0 - half + col, center.1 - half + row);
            let prof = crate::atlas::district_profile::derive_district(
                seed, body_id, params, ta, dp, climate,
            );
            let i = (row * n_i + col) as usize;
            morphology[i] = prof.morphology_zone as u8;
            elev_q[i] = prof.elev_q.clamp(0, 100) as u8;
            temp_dc[i] = match prof.temperature_c {
                Some(t) => {
                    ((t * 10.0).round() as i32).clamp(i16::MIN as i32 + 1, i16::MAX as i32) as i16
                }
                None => REGION_TEMP_NONE_DC,
            };
            moisture_q[i] = prof.moisture_q.clamp(0, 100) as u8;
            vegetation[i] = prof.vegetation_class as u8;
            glaciation[i] = prof.glaciation_grade as u8;
        }
    }
    DistrictWindowLayer {
        center,
        n,
        morphology,
        elev_q,
        temp_dc,
        moisture_q,
        vegetation,
        glaciation,
    }
}

// ---------------------------------------------------------------------------
// QuarterFootprintLayer (D-226 T-1112 amendment, T-1119)
// ---------------------------------------------------------------------------

/// Per-settlement aggregate over one quarter's 4×4 `BlockSkeleton` grid, for
/// the Atlas quarter-footprint overlay (D-226 T-1112 amendment §1). Five
/// scalar fields earn their place per the amendment's hard ceiling (§2): no
/// per-block zoning/street/tag detail ever reaches the wire, and no
/// chunk/tile/voxel data is touched.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QuarterFootprintEntry {
    pub city_id: u64,
    /// Basis-point mean of `BlockSkeleton.density_pct` across the 16 blocks
    /// (integer division, D-010 — no `f32` on the wire).
    pub density_avg_pct: u8,
    /// Mode `DistrictType` across the 16 blocks; ties resolve to the lowest
    /// declaration-order variant (the `Ord` derive on `DistrictType`, T-994).
    pub dominant_district_type: DistrictType,
    /// Mode `ZoningType` across the 16 blocks; same tie rule (the `Ord`
    /// derive added on `ZoningType` for this ticket, T-1119).
    pub dominant_zoning: ZoningType,
    /// Count of blocks with `landmark: Some(_)` across the 16 blocks (max 16).
    /// Tooltip/sidebar-only per the D-226(d) ceiling — never a map-visible
    /// channel (§2).
    pub landmark_count: u8,
    /// `QuarterSkeleton.corridors.len()`, clamped to `u8`. Tooltip/sidebar-only,
    /// same ceiling as `landmark_count`.
    pub corridor_count: u8,
}

/// The quarter-footprint overlay for one body (D-226 T-1112 amendment §1),
/// keyed by `city_id` — a quarter carries no independent spatial position of
/// its own (`QuarterId` is a content-addressable hash, not a coordinate), so
/// the layer anchors at the existing L3 settlement position client-side and
/// this map only needs to answer "does this settlement have quarter data, and
/// if so what does it aggregate to".
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QuarterFootprintLayer {
    /// `BTreeMap` for D-010 determinism, matching `RegionGridLayer`'s and the
    /// source `QuarterWorldState.block_tags`' own `BTreeMap` precedent.
    pub entries: std::collections::BTreeMap<u64, QuarterFootprintEntry>,
}

/// Aggregate one quarter's 16 `BlockSkeleton`s into a [`QuarterFootprintEntry`]
/// for `city_id`.
fn aggregate_quarter_footprint(
    city_id: u64,
    skeleton: &crate::simulation::generator::QuarterSkeleton,
) -> QuarterFootprintEntry {
    let blocks: Vec<&crate::simulation::generator::BlockSkeleton> =
        skeleton.blocks.iter().flatten().collect();
    let n = blocks.len() as u32; // always 16 (the fixed 4×4 grid) — computed
                                 // rather than hardcoded so the mean formula
                                 // stays correct if the grid shape ever changes.

    let density_sum: u32 = blocks.iter().map(|b| b.density_pct as u32).sum();
    let density_avg_pct = if n == 0 { 0 } else { (density_sum / n) as u8 };

    let dominant_district_type = mode_by_declaration_order(blocks.iter().map(|b| &b.district_type))
        .cloned()
        .unwrap_or_default();
    let dominant_zoning = mode_by_declaration_order(blocks.iter().map(|b| &b.zoning))
        .cloned()
        .unwrap_or_default();

    let landmark_count = blocks.iter().filter(|b| b.landmark.is_some()).count() as u8;
    let corridor_count = skeleton.corridors.len().min(u8::MAX as usize) as u8;

    QuarterFootprintEntry {
        city_id,
        density_avg_pct,
        dominant_district_type,
        dominant_zoning,
        landmark_count,
        corridor_count,
    }
}

/// Mode of an `Ord` value over an iterator, tie-broken by lowest declaration
/// order (i.e. the `Ord`-smallest value among the tied-for-max-count values).
/// `None` for an empty iterator.
///
/// **Not** `counts.into_iter().max_by_key(...)`: `Iterator::max_by_key`
/// returns the *last* maximum on a tie (documented behaviour), which is the
/// opposite of what's needed here. `BTreeMap` iterates keys in ascending
/// `Ord` order (= declaration order for these enums), so walking forward and
/// only replacing the running best on a *strictly greater* count keeps the
/// first-seen — i.e. lowest-declaration-order — winner on every tie.
fn mode_by_declaration_order<'a, T: Ord + 'a>(
    values: impl Iterator<Item = &'a T>,
) -> Option<&'a T> {
    let mut counts: std::collections::BTreeMap<&'a T, u32> = std::collections::BTreeMap::new();
    for v in values {
        *counts.entry(v).or_insert(0) += 1;
    }
    let mut best: Option<(&'a T, u32)> = None;
    for (v, count) in counts {
        match best {
            Some((_, best_count)) if count <= best_count => {}
            _ => best = Some((v, count)),
        }
    }
    best.map(|(v, _)| v)
}

/// Build the [`QuarterFootprintLayer`] from a body's cached state (T-1119).
/// Returns `None` when the Quarter-skeleton layer has not run for any
/// settlement (empty `state.quarters`).
///
/// `state.quarters` carries no independent spatial position — the only
/// spatial anchor a quarter has is the `city_id` it was generated for
/// (D-226 T-1112 amendment §1). So this recomputes the same deterministic
/// `QuarterId` derivation the L3→L4 dispatch path uses
/// (`SeedChain::for_body(world_seed, body_id).derive(SeedDomain::Layer4Quarter,
/// city_id).seed()`, `plugin.rs::build_skeleton_work_item`) for every placed
/// settlement and looks it up in `state.quarters`. A placement whose derived
/// id isn't found (skeleton generation is async, dispatched at `Low` priority
/// after the body's `Ready` snapshot is already cached — `plugin.rs`) is
/// skipped, not defaulted: an absent quarter is not a zero-footprint quarter.
pub fn build_quarter_footprint_layer(
    state: &BodyWorldState,
    world_seed: u64,
) -> Option<QuarterFootprintLayer> {
    if state.quarters.is_empty() {
        return None;
    }
    let body_chain = SeedChain::for_body(world_seed, &state.body_id);
    let mut entries = std::collections::BTreeMap::new();
    for placement in &state.placements {
        let quarter_id = body_chain
            .derive(SeedDomain::Layer4Quarter, placement.city_id)
            .seed();
        if let Some(quarter_state) = state.quarters.get(&quarter_id) {
            entries.insert(
                placement.city_id,
                aggregate_quarter_footprint(placement.city_id, &quarter_state.skeleton),
            );
        }
    }
    Some(QuarterFootprintLayer { entries })
}

// ---------------------------------------------------------------------------
// RoadGraphLayer (T-960 §1, T-1038)
// ---------------------------------------------------------------------------

/// One node in the [`RoadGraphLayer`] overlay — a settlement junction or a
/// waypoint. Trimmed from the internal [`crate::atlas::road_graph::RoadNode`]:
/// `degree` and `parent_edge` are internal bookkeeping a planetary-map overlay
/// doesn't need (degree is trivially re-derivable client-side by counting
/// edges per node index if a renderer wants junction highlighting).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RoadGraphNode {
    /// Position in working-heightmap-grid coordinates `(row, col)` — the same
    /// space as `Layer1Output` attractors/rivers and `SettlementLayer` positions.
    pub position: (u16, u16),
    pub kind: RoadNodeKind,
    /// The settlement's `city_id` (cross-references `SettlementLayer`), or
    /// `None` for a waypoint.
    pub city_id: Option<u64>,
}

/// One edge in the [`RoadGraphLayer`] overlay — a routed road or rail segment.
/// Trimmed from [`crate::atlas::road_graph::RoadEdge`]: `length_cells` is an
/// internal A* routing-grid measure with no meaning outside that grid's scale
/// (the polyline `path` is what an overlay actually draws).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RoadGraphEdge {
    /// `nodes` indices of the endpoint settlements (`from < to`).
    pub from: usize,
    pub to: usize,
    /// Routed polyline in working-heightmap-grid coordinates `(row, col)`.
    pub path: Vec<(u16, u16)>,
    pub maintenance: MaintenanceAuthority,
    /// `true` if this edge is a railroad; `false` is a road.
    pub is_rail: bool,
    /// Joined `systems.db` named-route id, if any (empty pool today — D-223).
    pub named_route_id: Option<String>,
}

/// The inter-settlement road/rail graph, trimmed for the Atlas planetary-map
/// overlay (T-960 §1, D-211, T-1038). See [`RoadGraphNode`]/[`RoadGraphEdge`]
/// for what was dropped from the internal `RoadGraph`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RoadGraphLayer {
    pub nodes: Vec<RoadGraphNode>,
    pub edges: Vec<RoadGraphEdge>,
}

/// Build the [`RoadGraphLayer`] from a body's cached state (T-960 §1).
/// Returns `None` when the RoadGraph layer has not run, which coincides
/// exactly with "no settlements placed" (`build_road_graph` returns an empty
/// graph for zero placements, and every placement yields at least one node).
pub fn build_road_graph_layer(state: &BodyWorldState) -> Option<RoadGraphLayer> {
    if state.road_graph.nodes.is_empty() {
        return None;
    }
    let nodes = state
        .road_graph
        .nodes
        .iter()
        .map(|n| RoadGraphNode {
            position: n.position,
            kind: n.kind,
            city_id: n.city_id,
        })
        .collect();
    let edges = state
        .road_graph
        .edges
        .iter()
        .map(|e| RoadGraphEdge {
            from: e.from,
            to: e.to,
            path: e.path.clone(),
            maintenance: e.maintenance,
            is_rail: e.is_rail,
            named_route_id: e.named_route_id.clone(),
        })
        .collect();
    Some(RoadGraphLayer { nodes, edges })
}

// ---------------------------------------------------------------------------
// SettlementLayer (T-960 §2, #955)
// ---------------------------------------------------------------------------

/// Coarse settlement size class for the Atlas overlay (T-960 §2), derived from
/// raw population using the same Tier A/B population cutoffs the D-211
/// placement pipeline already uses (`attractor_matching::match_cities`):
/// Tier A (≥ 1,000,000 or `NameLocked`) settlements are `Major`, Tier B
/// (50,000–999,999) are `Standard`, and everything else (Tier C / synthetic
/// overflow) is `Minor`. A display bucket, not new simulation truth.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SettlementSizeClass {
    Major,
    Standard,
    Minor,
}

impl SettlementSizeClass {
    /// Bucket a raw population using the D-211 Tier A/B cutoffs.
    pub fn from_population(population: i64) -> Self {
        if population >= 1_000_000 {
            SettlementSizeClass::Major
        } else if population >= 50_000 {
            SettlementSizeClass::Standard
        } else {
            SettlementSizeClass::Minor
        }
    }
}

/// One placed settlement in the [`SettlementLayer`] overlay (T-960 §2).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SettlementEntry {
    pub city_id: u64,
    pub name: String,
    /// Position in working-heightmap-grid coordinates `(row, col)` — the same
    /// space as `Layer1Output` attractors/rivers (T-960 §2: match the
    /// coordinate convention layer1 features already use so the client
    /// transforms identically).
    pub position: (u16, u16),
    pub size_class: SettlementSizeClass,
    /// Authored `atlas_city_names.kind == 'capital'` (not population-derived).
    pub is_capital: bool,
    /// Cheap derived flag: `true` if the settlement's anchoring attractor is
    /// water-adjacent (`CoastalAccess` / `RiverMouth` / `LakeShore`).
    pub is_port: bool,
}

/// The settlement-placement overlay for one body (T-960 §2, #955, D-211).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SettlementLayer {
    pub settlements: Vec<SettlementEntry>,
}

/// `true` for the water-adjacent attractor types a settlement counts as a
/// "port" for cheaply (T-960 §2). No "foothold" flag: unlike `is_port`, there
/// is no existing concept in the placement data this could derive from
/// without inventing new business logic — left out (see the T-960 report).
fn is_port_attractor(at: AttractorType) -> bool {
    matches!(
        at,
        AttractorType::CoastalAccess | AttractorType::RiverMouth | AttractorType::LakeShore
    )
}

/// Build the [`SettlementLayer`] from a body's cached state (T-960 §2).
/// Returns `None` when the Settlement layer has not placed any city yet.
pub fn build_settlement_layer(state: &BodyWorldState) -> Option<SettlementLayer> {
    if state.placements.is_empty() {
        return None;
    }
    let settlements = state
        .placements
        .iter()
        .map(|p| SettlementEntry {
            city_id: p.city_id,
            name: p.name.clone(),
            position: p.position,
            size_class: SettlementSizeClass::from_population(p.population),
            is_capital: p.is_capital,
            is_port: is_port_attractor(p.attractor_type),
        })
        .collect();
    Some(SettlementLayer { settlements })
}

/// Normalize a wire-supplied district-window centre against the body's
/// physical geometry (T-1142 — a letterbox-click bug sent `window_center`
/// wildly out of a body's valid range; the server accepted it, derived
/// clamped garbage inside `derive_district`, and CACHED that garbage under
/// the raw un-normalized key). Mirrors `district_profile::derive_district`'s
/// own forward mapping (`district_profile.rs:1426-1451`) EXACTLY, so a window
/// centre that survives this normalization derives identically to how
/// `derive_district` would have resolved it anyway — this function only
/// closes the gap between "derive_district silently clamps/wraps internally"
/// and "the SERVING path (cache key, coalescing key) saw the raw value".
///
/// **Column (longitude) wraps** — `rem_euclid` against the body's
/// circumference in districts, mirroring the forward map's
/// `(wx / circumference_m).rem_euclid(1.0)` (longitude is periodic; a click
/// at column 12276 on a body whose circumference is a few hundred districts
/// wide is the SAME point as some small in-range column, not garbage).
///
/// **Row (latitude) clamps** — to `±half_meridian_districts`, mirroring the
/// forward map's `(wy / meridian_m).clamp(-0.5, 0.5)` (latitude is NOT
/// periodic; it terminates at the poles, so out-of-range rows collapse to the
/// nearest pole rather than wrapping — same asymmetry `derive_district`
/// itself already encodes).
///
/// Both bounds are derived from `body_radius_km` via the SAME `DISTRICT_M`
/// (2 048 m, D-243) constant the forward map uses — no independent magic
/// numbers that could silently drift out of sync with `derive_district`.
///
/// **No radius (`body_radius_km` absent/non-positive — tiny test bodies
/// only, per `BodyParams`'s own doc: "e.g. tiny test bodies"; every real
/// `systems.db` body row carries a radius):** identity, no wrap/clamp. The
/// forward map's own no-radius branch has no periodicity concept either (it
/// clamps the FRACTIONAL PIXEL position directly against the heightmap's
/// working-grid dimensions, which aren't known at request-serving time — only
/// inside the Rayon work item once the heightmap is loaded); a real letterbox
/// click can never hit this branch, so it is out of this fix's scope.
fn normalize_window_center(params: &BodyParams, center: DistrictPos) -> DistrictPos {
    let (dx, dy) = center;
    match params.body_radius_km {
        Some(r_km) if r_km > 0.0 => {
            let circumference_m = std::f64::consts::TAU * r_km * 1000.0;
            let meridian_m = std::f64::consts::PI * r_km * 1000.0;
            // Whole districts per full circumference / per half-meridian —
            // rounded (not truncated) so the bound matches the forward map's
            // continuous fraction as closely as an integer district grid can.
            let districts_per_circumference =
                (circumference_m / DISTRICT_M as f64).round().max(1.0) as i32;
            let half_meridian_districts = (meridian_m / DISTRICT_M as f64 / 2.0).round() as i32;

            let wrapped_dx = dx.rem_euclid(districts_per_circumference);
            let clamped_dy = dy.clamp(-half_meridian_districts, half_meridian_districts);
            (wrapped_dx, clamped_dy)
        }
        // No radius: derive_district's own fallback has no wrap/clamp concept
        // at the DistrictPos level (see the doc above) — identity.
        _ => center,
    }
}

/// Resolve `req`'s district-window query, if any (D-226 T-1124 amendment,
/// T-1137). Returns `None` immediately when `req.window_center` is absent (no
/// window requested — the common case, zero cost).
///
/// **Independent of the whole-body cache state** (the amendment is explicit:
/// "the window derivation depends only on `TerrainAnalysis` + `BodyParams`
/// being resolvable for the body ... not on which whole-body layers the
/// cascade has cached") — so this runs whether `handle_atlas_request` is
/// about to take its cache-hit or cache-miss branch, sharing neither's control
/// flow.
///
/// **`window_center` is normalized via [`normalize_window_center`] BEFORE the
/// `DistrictWindowCache` key AND the `submit_window` coalescing key are built**
/// (T-1142) — this is why `body_params` is read HERE, unconditionally,
/// rather than only inside the former miss-branch: normalization needs
/// `body_radius_km` to compute the wrap/clamp bounds, and it must happen
/// before either key exists, or an insane request and its sane normalized
/// twin would land in different cache entries (exactly the bug this fix
/// closes — a garbage `window_center` was cached standalone instead of
/// collapsing onto its valid twin). The coalescing key itself is
/// `(ConnectionId, body_id)` — it never carried `center`, so coalescing
/// was never at risk of diverging per-center; normalizing before
/// `submit_window` matters only so the work item DERIVES (and echoes) the
/// canonical center. The one-time cost
/// (a single indexed `bodies` row read) is paid on every window request now,
/// not just on a cache miss — a request whose normalized center hits the
/// cache still needed this read to know WHICH key to check.
///
/// Cache hit (`(body_id, normalized_center, n)` already in `window_cache`) →
/// `Some` immediately, no queue submission (D-227: a previously-derived
/// window for this body+seed is valid forever, no staleness check needed).
/// Cache miss → submit a `DeriveWindow` work item (queue-based, per the
/// amendment's binding serving model — never inline here) and return `None`;
/// the *next* request for this `(body, normalized_center, n)` re-checks the
/// cache and finds it populated once `drain_generation_completions` has
/// processed the completion (the existing D-225 poll-and-recheck-cache
/// pattern every other layer already uses, not a push).
///
/// `window_n` is clamped to `[1, DISTRICT_WINDOW_MAX_N]` here — the ONE place
/// that clamp is applied; nothing downstream re-checks the wire value.
#[allow(clippy::too_many_arguments)]
fn serve_district_window(
    req: &AtlasLayerRequest,
    window_cache: &mut DistrictWindowCache,
    queue: &GenerationQueue,
    resolver: &BodySourceResolver,
    body_params_reader: Option<&BodyParamsReader>,
    world_seed: u64,
    conn_id: ConnectionId,
) -> Option<DistrictWindowLayer> {
    let raw_center = req.window_center?;
    let n = req.window_n.clamp(1, DISTRICT_WINDOW_MAX_N);

    // body_params is needed to normalize the centre BEFORE either cache key
    // exists (T-1142) — read it first, unconditionally (not gated on a cache
    // miss like the former structure). A read failure here can't distinguish
    // "insane vs. sane center" for the key, so it's a hard skip for the whole
    // window (window stays None on this response), same failure posture the
    // former miss-only read already had.
    let Some(reader) = body_params_reader else {
        tracing::warn!(
            body_id = %req.body_id,
            "district window request: no body_params_reader wired — window stays None"
        );
        return None;
    };
    let body_params = match reader.read_body_params(&req.body_id) {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!(
                body_id = %req.body_id,
                error = %e,
                "district window request: body_params read failed — window stays None"
            );
            return None;
        }
    };

    let center = normalize_window_center(&body_params, raw_center);
    if center != raw_center {
        tracing::debug!(
            body_id = %req.body_id,
            raw = ?raw_center,
            normalized = ?center,
            "district window request: out-of-range window_center normalized (T-1142)"
        );
    }

    let key: DistrictWindowKey = (req.body_id.clone(), center, n);
    if let Some(layer) = window_cache.get(&key) {
        return Some(layer.clone());
    }

    // Miss — resolve the heightmap and submit a background derive.
    // Read/resolve failures are non-fatal for the window (log + skip): the
    // window simply stays None on this response, same as an unrun whole-body
    // layer, rather than failing the entire AtlasLayerResponse.
    let heightmap_path = match resolver.resolve(&req.body_id) {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!(
                body_id = %req.body_id,
                error = %e,
                "district window request: heightmap resolve failed — window stays None"
            );
            return None;
        }
    };

    queue.submit_window(
        GenWorkItem::DeriveWindow {
            body_id: req.body_id.clone(),
            conn_id,
            heightmap_path,
            sea_level: DEFAULT_SEA_LEVEL,
            body_seed: SeedChain::for_body(world_seed, &req.body_id),
            body_params: Box::new(body_params),
            center,
            n,
        },
        GenPriority::Immediate,
    );
    None
}

/// 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.
///
/// `city_reader` supplies the body's settlements for Layer-3 placement (#955),
/// read on a cache miss. `None` (or a read failure) places no cities — the
/// cascade still runs Layer 1; the body just gets no settlement placements.
///
/// `body_params_reader` supplies the body's physical parameters for the
/// DistrictProfile carrier layer (T-1032, D-239 §1), read on a cache miss.
/// `None` (or a read failure) passes `body_params: None` to the work item,
/// causing the cascade to stop at `CascadeLayer::Settlement` (pre-T-1032
/// behaviour). A successful read passes `Some(Box::new(params))`, enabling
/// the full `CascadeLayer::DistrictProfile` path.
///
/// `window_cache` + `conn_id` serve the optional district-window query
/// (D-226 T-1124 amendment, T-1137) via [`serve_district_window`] — see that
/// function for the caching/coalescing model. `conn_id` is used ONLY as the
/// window request's coalescing key; nothing else in this function is
/// connection-aware (the D-254 §2 convention this proxy already follows).
#[allow(clippy::too_many_arguments)]
pub fn handle_atlas_request(
    req: &AtlasLayerRequest,
    cache: &mut BodyWorldStateCache,
    window_cache: &mut DistrictWindowCache,
    queue: &GenerationQueue,
    resolver: &BodySourceResolver,
    city_reader: Option<&CityContextReader>,
    body_params_reader: Option<&BodyParamsReader>,
    world_seed: u64,
    current_tick: SimTick,
    conn_id: ConnectionId,
) -> AtlasLayerResponse {
    let district_window = serve_district_window(
        req,
        window_cache,
        queue,
        resolver,
        body_params_reader,
        world_seed,
        conn_id,
    );

    // 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(),
            // The cascade ran on the downsampled heightmap, so its dims are the
            // working grid all Layer-1 positions are expressed in (#960).
            grid_w: state.heightmap_width,
            grid_h: state.heightmap_height,
            // district_basin_dirs is transient — it is aggregated during run_layer1
            // and consumed by derive_all_districts before being stored on
            // BodyWorldState. When reconstructing Layer1Output from the cache for
            // the client response, the per-district direction is already encoded in
            // DistrictProfile.basin_direction (BodyWorldState.districts) and is not
            // needed again here. Supply an empty map.
            district_basin_dirs: std::collections::BTreeMap::new(),
        };
        let district_grid = build_district_grid(state);
        let road_graph = build_road_graph_layer(state);
        let settlements = build_settlement_layer(state);
        let region_grid = build_region_grid(state);
        let quarter_footprints = build_quarter_footprint_layer(state, world_seed);
        return AtlasLayerResponse {
            body_id: req.body_id.clone(),
            status: AtlasLayerStatus::Ready,
            layer1: Some(layer1),
            district_grid,
            road_graph,
            settlements,
            region_grid,
            district_window,
            quarter_footprints,
        };
    }

    // Miss — resolve the source heightmap and enqueue background analysis.
    match resolver.resolve(&req.body_id) {
        Ok(heightmap_path) => {
            // Pre-resolve this body's settlements + system faction so the Rayon
            // work item stays DB-free (#955/#956, D-225). Read failures are
            // non-fatal: log and fall back (no cities / no faction → frontier).
            let (cities, dominant_faction) = match city_reader {
                Some(reader) => {
                    let cities = reader
                        .read_body_settlements(&req.body_id)
                        .unwrap_or_else(|e| {
                            tracing::warn!(
                                body_id = %req.body_id,
                                error = %e,
                                "settlement read failed; placing no cities"
                            );
                            Vec::new()
                        });
                    let faction = reader
                        .read_body_dominant_faction(&req.body_id)
                        .unwrap_or_else(|e| {
                            tracing::warn!(
                                body_id = %req.body_id,
                                error = %e,
                                "dominant_faction read failed; defaulting to frontier"
                            );
                            None
                        });
                    (cities, faction)
                }
                None => (Vec::new(), None),
            };
            // Pre-resolve body physical params so the Rayon work item stays
            // DB-free (D-225 pattern). Read failures are non-fatal: log and
            // fall back to None (cascade stops at Settlement, pre-T-1032
            // behaviour, rather than aborting the entire analysis).
            let body_params = match body_params_reader {
                Some(reader) => reader
                    .read_body_params(&req.body_id)
                    .map(|p| Some(Box::new(p)))
                    .unwrap_or_else(|e| {
                        tracing::warn!(
                            body_id = %req.body_id,
                            error = %e,
                            "body_params read failed; district layer skipped"
                        );
                        None
                    }),
                None => None,
            };
            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),
                    cities,
                    dominant_faction,
                    body_params,
                },
                GenPriority::Immediate,
            );
            AtlasLayerResponse {
                body_id: req.body_id.clone(),
                status: AtlasLayerStatus::Pending,
                layer1: None,
                district_grid: None,
                road_graph: None,
                settlements: None,
                region_grid: None,
                district_window,
                quarter_footprints: 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,
            district_grid: None,
            road_graph: None,
            settlements: None,
            region_grid: None,
            district_window,
            quarter_footprints: None,
        },
        Err(e) => AtlasLayerResponse {
            body_id: req.body_id.clone(),
            status: AtlasLayerStatus::Error(e.to_string()),
            layer1: None,
            district_grid: None,
            road_graph: None,
            settlements: None,
            region_grid: None,
            district_window,
            quarter_footprints: 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);

    #[test]
    fn district_grid_built_from_cached_districts() {
        use crate::atlas::district_profile::{
            DistrictProfile, GlaciationGrade, PrecipitationClass, TectonicClass, VegetationClass,
        };
        use crate::simulation::generator::MorphologyZone;
        let dp = |zone: MorphologyZone, elev: i32| DistrictProfile {
            morphology_zone: zone,
            tectonic_class: TectonicClass::Stable,
            glaciation_grade: GlaciationGrade::None,
            precipitation_class: PrecipitationClass::Arid,
            slope_q: 0,
            elev_q: elev,
            ocean_fraction_q: 0,
            river_threshold: 200,
            temperature_c: Some(10.0),
            moisture_q: 50,
            vegetation_class: VegetationClass::Barren,
            basin_direction: crate::atlas::scale::BasinDirection::default(),
        };
        let mut state = BodyWorldState {
            body_id: "GJ1c".into(),
            heightmap: vec![],
            heightmap_width: 16,
            heightmap_height: 8,
            river_network: RiverNetwork::default(),
            drainage_basins: vec![],
            attractors: vec![],
            placements: vec![],
            road_graph: crate::atlas::road_graph::RoadGraph::default(),
            quarters: std::collections::BTreeMap::new(),
            districts: std::collections::BTreeMap::new(),
            regions: std::collections::BTreeMap::new(),
            last_accessed: 0,
        };
        // 3×2 grid with two distinct zones at the corners.
        state
            .districts
            .insert((0, 0), dp(MorphologyZone::AlluvialPlain, 10));
        state
            .districts
            .insert((2, 1), dp(MorphologyZone::Alpine, 90));

        let grid = build_district_grid(&state).expect("districts present → Some grid");
        assert_eq!((grid.cols, grid.rows), (3, 2));
        assert_eq!(grid.morphology.len(), 6);
        assert_eq!(grid.morphology[0], MorphologyZone::AlluvialPlain as u8);
        assert_eq!(
            grid.morphology[grid.cols as usize + 2],
            MorphologyZone::Alpine as u8
        );
        assert_eq!(grid.elev_q[grid.cols as usize + 2], 90);

        // Empty districts → None (DistrictProfile layer hasn't run).
        state.districts.clear();
        assert!(build_district_grid(&state).is_none());
    }

    /// T-1113: the region climate grid mirrors the district-grid encoding —
    /// `None` when the Region layer hasn't run; dense row-major with the
    /// integer wire quantization (deci-°C temp, `i16::MIN` airless sentinel)
    /// when it has.
    #[test]
    fn build_region_grid_encodes_dense_quantized_climate() {
        use crate::atlas::region_profile::{RegionClock, RegionProfile, SeasonPhase, WeatherState};

        let mut state = blank_state("GJ1c");
        // Empty regions → None (the Region layer hasn't run).
        assert!(build_region_grid(&state).is_none());

        // A 2×1 covering grid: one temperate region, one airless-style region
        // (mean_temp_c = None → the sentinel).
        state.regions.insert(
            (0, 0),
            RegionProfile {
                pos: (0, 0),
                clock: RegionClock {
                    season: SeasonPhase::Summer,
                    weather: WeatherState::Clear,
                    mean_temp_c: Some(12.34),
                },
                latitude_deg: 45.0,
                moisture_q: 80,
            },
        );
        state.regions.insert(
            (1, 0),
            RegionProfile {
                pos: (1, 0),
                clock: RegionClock {
                    season: SeasonPhase::Winter,
                    weather: WeatherState::Snow,
                    mean_temp_c: None,
                },
                latitude_deg: -10.0,
                moisture_q: 5,
            },
        );

        let grid = build_region_grid(&state).expect("regions present → Some grid");
        assert_eq!((grid.cols, grid.rows), (2, 1));
        assert_eq!(grid.season.len(), 2);
        assert_eq!(grid.season[0], SeasonPhase::Summer as u8);
        assert_eq!(grid.weather[0], WeatherState::Clear as u8);
        // 12.34 °C → 123 deci-°C (rounded).
        assert_eq!(grid.mean_temp_dc[0], 123);
        assert_eq!(grid.moisture_q[0], 80);
        assert_eq!(grid.season[1], SeasonPhase::Winter as u8);
        assert_eq!(grid.weather[1], WeatherState::Snow as u8);
        assert_eq!(
            grid.mean_temp_dc[1], REGION_TEMP_NONE_DC,
            "airless None maps to the sentinel"
        );
        assert_eq!(grid.moisture_q[1], 5);
    }

    // -----------------------------------------------------------------------
    // DistrictWindowLayer (D-226 T-1124 amendment, T-1137)
    // -----------------------------------------------------------------------

    /// Minimal deterministic heightmap fixture, mirroring
    /// `district_profile::tests::test_hm` (T-1137: the window path shares the
    /// same on-demand `derive_district` call, so it earns the same fixture
    /// shape).
    fn window_test_hm() -> crate::atlas::heightmap::BodyHeightmap {
        use crate::atlas::heightmap::BodyHeightmap;
        let (w, h) = (64u32, 32u32);
        let n = (w * h) as usize;
        let data = (0..n)
            .map(|i| {
                let r = (i / w as usize) as f32 / h as f32;
                let c = (i % w as usize) as f32 / w as f32;
                (r * 0.6 + c * 0.4).min(1.0)
            })
            .collect();
        BodyHeightmap {
            body_id: "test".into(),
            width: w,
            height: h,
            data,
            sea_level: 0.3,
        }
    }

    fn window_test_ta(
        hm: &crate::atlas::heightmap::BodyHeightmap,
    ) -> crate::atlas::features::TerrainAnalysis {
        use crate::atlas::drainage;
        use crate::atlas::features::TerrainAnalysis;
        let dr = drainage::analyze(&hm.data, hm.width, hm.height, hm.sea_level);
        TerrainAnalysis::analyze(hm, &dr)
    }

    fn window_test_params() -> crate::atlas::district_profile::BodyParams {
        crate::atlas::district_profile::BodyParams {
            hydrosphere: Some("ocean".into()),
            atmosphere: Some("breathable".into()),
            planet_class: Some("temperate".into()),
            body_radius_km: Some(6371.0),
            ..Default::default()
        }
    }

    /// `build_district_window_layer` produces a dense `n × n` row-major grid
    /// (the `DistrictGridLayer`/`RegionGridLayer` indexing convention) whose
    /// cell count and per-array lengths match `n`, and whose values are
    /// pulled straight from the corresponding `derive_district` profile field
    /// (T-1137).
    #[test]
    fn build_district_window_layer_produces_dense_n_by_n_grid() {
        let hm = window_test_hm();
        let ta = window_test_ta(&hm);
        let params = window_test_params();
        let climate = crate::atlas::district_profile::ClimateConstants::default();
        let seed = SeedChain::root(42).derive(SeedDomain::Body, 1);

        let n = 4u32;
        let layer =
            build_district_window_layer(seed, "test_body", &params, &ta, (10, -5), n, &climate);

        assert_eq!(layer.center, (10, -5));
        assert_eq!(layer.n, n);
        let cells = (n * n) as usize;
        assert_eq!(layer.morphology.len(), cells);
        assert_eq!(layer.elev_q.len(), cells);
        assert_eq!(layer.temp_dc.len(), cells);
        assert_eq!(layer.moisture_q.len(), cells);
        assert_eq!(layer.vegetation.len(), cells);
        assert_eq!(layer.glaciation.len(), cells);

        // Spot-check one cell against a direct derive_district call — the
        // window builder must not transform the profile's values, only pack
        // them (row 0, col 0 → district (center.0 - n/2, center.1 - n/2)).
        let half = (n / 2) as i32;
        let dp = (10 - half, -5 - half);
        let prof = crate::atlas::district_profile::derive_district(
            seed,
            "test_body",
            &params,
            &ta,
            dp,
            &climate,
        );
        assert_eq!(layer.morphology[0], prof.morphology_zone as u8);
        assert_eq!(layer.elev_q[0], prof.elev_q.clamp(0, 100) as u8);
        assert_eq!(layer.moisture_q[0], prof.moisture_q.clamp(0, 100) as u8);
        assert_eq!(layer.vegetation[0], prof.vegetation_class as u8);
        assert_eq!(layer.glaciation[0], prof.glaciation_grade as u8);
    }

    /// Clamped-window edge: `n = 1` is the minimum valid window (a single
    /// district) — no panic, no empty output, exactly one cell per array.
    #[test]
    fn build_district_window_layer_handles_n_equals_one() {
        let hm = window_test_hm();
        let ta = window_test_ta(&hm);
        let params = window_test_params();
        let climate = crate::atlas::district_profile::ClimateConstants::default();
        let seed = SeedChain::root(1).derive(SeedDomain::Body, 1);

        let layer =
            build_district_window_layer(seed, "test_body", &params, &ta, (0, 0), 1, &climate);
        assert_eq!(layer.n, 1);
        assert_eq!(layer.morphology.len(), 1);
        assert_eq!(layer.elev_q.len(), 1);
        assert_eq!(layer.temp_dc.len(), 1);
        assert_eq!(layer.moisture_q.len(), 1);
        assert_eq!(layer.vegetation.len(), 1);
        assert_eq!(layer.glaciation.len(), 1);
    }

    /// Determinism spot-check (D-010, T-1123 precedent promoted to a real
    /// test per the ticket): two full derive passes over the SAME window are
    /// byte-identical, at a window size large enough to exercise many cells
    /// (mirrors `aliveness_probe --render`'s own two-pass proof, now pinned
    /// as a unit test rather than a probe-only demonstration).
    #[test]
    fn build_district_window_layer_two_passes_are_byte_identical() {
        let hm = window_test_hm();
        let ta = window_test_ta(&hm);
        let params = window_test_params();
        let climate = crate::atlas::district_profile::ClimateConstants::default();
        let seed = SeedChain::root(7).derive(SeedDomain::Body, 3);

        let n = 8u32;
        let first =
            build_district_window_layer(seed, "test_body", &params, &ta, (3, -2), n, &climate);
        let second =
            build_district_window_layer(seed, "test_body", &params, &ta, (3, -2), n, &climate);
        assert_eq!(
            first, second,
            "two full derive passes over the same (center, n) must be byte-identical (D-010/D-227)"
        );
    }

    /// FULL-PATH determinism (PR #187 review — Tyre C3, binding, load-bearing
    /// for save-file lineage under D-227): the test above reuses ONE `ta` for
    /// both passes, which only proves `build_district_window_layer` (the
    /// packer) is a pure function of its arguments — it says nothing about
    /// whether re-running `run_layer1` itself (the D8 drainage pass +
    /// `TerrainAnalysis::analyze`) is deterministic, which is exactly what the
    /// production `DeriveWindow` path depends on (T-1137's `TerrainAnalysis`
    /// re-derive / `TerrainAnalysisCache::get_or_derive` on a cache miss, and
    /// `aliveness_probe --render`'s workaround before it).
    ///
    /// This test runs `run_layer1` TWICE, independently, from the SAME
    /// `(seed, heightmap)` inputs — no shared `ta` — and asserts the two
    /// COMPLETE `DistrictWindowLayer` outputs (derive AND pack) are
    /// byte-identical. D-227's save-file guarantee ("same seed/body/position →
    /// same derived output, always") is only as strong as the weakest link in
    /// that chain; this closes the gap the packer-only test left open.
    #[test]
    fn full_path_two_independent_run_layer1_passes_produce_identical_window() {
        let hm = window_test_hm();
        let params = window_test_params();
        let climate = crate::atlas::district_profile::ClimateConstants::default();
        let seed = SeedChain::root(11).derive(SeedDomain::Body, 5);
        let n = 6u32;
        let center = (4, -1);

        // Two INDEPENDENT calls to run_layer1 — each re-runs D8 drainage +
        // TerrainAnalysis::analyze from scratch on the SAME heightmap, exactly
        // mirroring what a cold TerrainAnalysisCache miss does on the real
        // DeriveWindow path (or a second body eviction re-pay).
        let (_, ta_pass1) = crate::atlas::layer1::run_layer1(&hm);
        let (_, ta_pass2) = crate::atlas::layer1::run_layer1(&hm);

        // Confirm the two independent TerrainAnalysis derivations themselves
        // agree field-by-field — a precise failure signal if drainage/analyze
        // ever introduces nondeterminism (unordered iteration, uninitialized
        // memory, etc.) BEFORE the packer even runs.
        assert_eq!(ta_pass1.ocean_mask, ta_pass2.ocean_mask);
        assert_eq!(ta_pass1.lake_mask, ta_pass2.lake_mask);
        assert_eq!(ta_pass1.water_dist, ta_pass2.water_dist);
        assert_eq!(ta_pass1.slope_deg, ta_pass2.slope_deg);
        assert_eq!(ta_pass1.elev_pct, ta_pass2.elev_pct);

        // Now the FULL path: pack a DistrictWindowLayer from each independent
        // TerrainAnalysis and confirm the complete served payload agrees.
        let window_from_pass1 =
            build_district_window_layer(seed, "test_body", &params, &ta_pass1, center, n, &climate);
        let window_from_pass2 =
            build_district_window_layer(seed, "test_body", &params, &ta_pass2, center, n, &climate);
        assert_eq!(
            window_from_pass1, window_from_pass2,
            "two independent run_layer1 derivations from the same (seed, heightmap) \
             must pack to a byte-identical DistrictWindowLayer end to end (D-227)"
        );
    }

    /// [`DistrictWindowCache`] insert/get round-trips, and a capacity-1 cache
    /// evicts the oldest entry FIFO — mirroring `BodyWorldStateCache`'s own
    /// `evicts_lru_on_overflow` precedent, adapted to this cache's
    /// capacity-only insertion-order eviction (no access-recency tracking,
    /// per the struct doc: D-227 means a cached window has no staleness to
    /// track, only unbounded growth to bound).
    #[test]
    fn district_window_cache_insert_get_and_evict() {
        let mut cache = DistrictWindowCache::new(2);
        let key_a: DistrictWindowKey = ("Alpha".into(), (0, 0), 4);
        let key_b: DistrictWindowKey = ("Beta".into(), (1, 1), 4);
        let key_c: DistrictWindowKey = ("Gamma".into(), (2, 2), 4);
        let mk = |center, n| DistrictWindowLayer {
            center,
            n,
            morphology: vec![0; (n * n) as usize],
            elev_q: vec![0; (n * n) as usize],
            temp_dc: vec![REGION_TEMP_NONE_DC; (n * n) as usize],
            moisture_q: vec![0; (n * n) as usize],
            vegetation: vec![0; (n * n) as usize],
            glaciation: vec![0; (n * n) as usize],
        };

        assert!(cache.get(&key_a).is_none());
        cache.insert(key_a.clone(), mk((0, 0), 4));
        cache.insert(key_b.clone(), mk((1, 1), 4));
        assert_eq!(cache.len(), 2);
        assert!(cache.get(&key_a).is_some());
        assert!(cache.get(&key_b).is_some());

        // Cache at capacity (2): inserting a third entry evicts key_a (oldest).
        cache.insert(key_c.clone(), mk((2, 2), 4));
        assert_eq!(cache.len(), 2);
        assert!(
            cache.get(&key_a).is_none(),
            "key_a should have been evicted"
        );
        assert!(cache.get(&key_b).is_some());
        assert!(cache.get(&key_c).is_some());
    }

    /// `handle_atlas_request`'s window branch clamps `window_n` server-side to
    /// `[1, DISTRICT_WINDOW_MAX_N]` — a request claiming an oversized `n` on
    /// the wire never reaches `build_district_window_layer` un-clamped. This
    /// exercises the full request→submit→drain→cache→re-request loop with an
    /// out-of-range `window_n`, confirming the CACHED layer (once the
    /// background derive completes) carries the CLAMPED `n`, not the
    /// requested one.
    #[test]
    fn handle_atlas_request_clamps_oversized_window_n() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);
        let (_db, resolver, params_reader, _root) = resolver_and_params_reader("GJ1c");
        let queue = GenerationQueue::with_threads(1);

        let oversized_req = AtlasLayerRequest {
            body_id: "GJ1c".to_string(),
            up_to: CascadeLayer::Topography,
            window_center: Some((0, 0)),
            window_n: DISTRICT_WINDOW_MAX_N * 10, // wildly over the wire — must clamp, not trust
        };

        let resp = handle_atlas_request(
            &oversized_req,
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            1,
            test_conn_id(),
        );
        // First request: window not yet cached → None, but a DeriveWindow
        // must have been submitted (checked via the drain below).
        assert!(resp.district_window.is_none());

        // Wait for the Rayon DeriveWindow work item to complete.
        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        let window_completion = completions.into_iter().find_map(|c| {
            if let GenCompletion::WindowDerived { body_id, layer } = c {
                if body_id == "GJ1c" {
                    return Some(layer);
                }
            }
            None
        });
        let layer = window_completion.expect("DeriveWindow must complete for GJ1c");
        assert_eq!(
            layer.n, DISTRICT_WINDOW_MAX_N,
            "server must clamp window_n to DISTRICT_WINDOW_MAX_N, never trust the wire value"
        );
    }

    // -------------------------------------------------------------------
    // normalize_window_center (T-1142 — letterbox-click out-of-range bug)
    // -------------------------------------------------------------------

    /// A small-body fixture (500 km radius) whose bounds are hand-checkable:
    /// `districts_per_circumference = round(TAU*500_000/2048) = 1534`,
    /// `half_meridian_districts = round(PI*500_000/2048/2) = 383`. The
    /// literal column/row this fixture uses (`12276`, `3021`) are the exact
    /// values the reported T-1142 letterbox-click bug sent — at THIS radius
    /// they genuinely overflow both bounds (at Earth radius, coincidentally,
    /// they wouldn't — the bounds are tens of thousands of districts wide),
    /// so this is a faithful small-body reproduction, not just an
    /// arbitrarily-chosen out-of-range pair.
    fn small_body_params() -> BodyParams {
        BodyParams {
            hydrosphere: Some("ocean".into()),
            atmosphere: Some("breathable".into()),
            planet_class: Some("temperate".into()),
            body_radius_km: Some(500.0),
            ..Default::default()
        }
    }

    /// Out-of-range COLUMN wraps (longitude is periodic) to its in-range
    /// canonical twin via `rem_euclid` — mirroring `derive_district`'s own
    /// `(wx / circumference_m).rem_euclid(1.0)` forward map.
    #[test]
    fn normalize_window_center_wraps_out_of_range_column() {
        let params = small_body_params();
        // districts_per_circumference = 1534 (hand-computed above).
        // 12276 rem_euclid 1534 = 4 (12276 = 8*1534 + 4).
        assert_eq!(
            12276_i32.rem_euclid(1534),
            4,
            "sanity: the hand-computed wrap"
        );
        let (nx, ny) = normalize_window_center(&params, (12276, 0));
        assert_eq!(
            nx, 4,
            "out-of-range column wraps to its canonical in-range twin"
        );
        assert_eq!(ny, 0, "an in-range row is untouched");

        // The canonical twin normalizes to itself (idempotent).
        let (nx2, _) = normalize_window_center(&params, (4, 0));
        assert_eq!(nx2, 4);

        // Negative columns wrap too (rem_euclid, not truncating rem) —
        // longitude has no sign discontinuity.
        let (nx3, _) = normalize_window_center(&params, (-1, 0));
        assert_eq!(
            nx3, 1533,
            "negative column wraps to the top of the range, not a negative remainder"
        );
    }

    /// Beyond-pole ROW clamps (latitude terminates, does not wrap) to
    /// `±half_meridian_districts` — mirroring `derive_district`'s own
    /// `(wy / meridian_m).clamp(-0.5, 0.5)` forward map. This is the
    /// asymmetry the coordinator's fix explicitly calls out: columns wrap,
    /// rows clamp — never the other way around.
    #[test]
    fn normalize_window_center_clamps_beyond_pole_row() {
        let params = small_body_params();
        // half_meridian_districts = 383 (hand-computed above).
        let (_, ny) = normalize_window_center(&params, (0, 3021));
        assert_eq!(
            ny, 383,
            "beyond-pole row clamps to the pole boundary, not wraps"
        );

        let (_, ny_neg) = normalize_window_center(&params, (0, -9000));
        assert_eq!(ny_neg, -383, "clamping is symmetric at both poles");

        // A row exactly at the boundary is untouched.
        let (_, ny_boundary) = normalize_window_center(&params, (0, 383));
        assert_eq!(ny_boundary, 383);
    }

    /// An in-range center (well inside both bounds) is returned UNCHANGED —
    /// normalization must be a no-op for the overwhelming common case (every
    /// legitimate click), not just a defensive clamp that happens to also
    /// preserve valid input.
    #[test]
    fn normalize_window_center_leaves_in_range_center_unchanged() {
        let params = small_body_params();
        let center = (100, -50);
        assert_eq!(normalize_window_center(&params, center), center);

        // (0, 0) — the origin — is always in range regardless of body size.
        assert_eq!(normalize_window_center(&params, (0, 0)), (0, 0));
    }

    /// No-radius bodies (tiny test-body fallback, `body_radius_km: None`) get
    /// IDENTITY — `derive_district`'s own no-radius branch has no
    /// wrap/clamp-in-district-space concept (see the function doc); an
    /// extreme center here is out of this fix's scope by design, not an
    /// oversight.
    #[test]
    fn normalize_window_center_no_radius_is_identity() {
        let params = BodyParams::default(); // body_radius_km: None
        let extreme = (999_999, -999_999);
        assert_eq!(normalize_window_center(&params, extreme), extreme);
    }

    /// End-to-end (the coordinator's core ask): an out-of-range
    /// `window_center` and its already-normalized twin, requested through the
    /// REAL `handle_atlas_request` path, land in the SAME `DistrictWindowCache`
    /// entry and produce a byte-identical `DistrictWindowLayer` — normalization
    /// happens BEFORE the cache key is built, so an insane request and its
    /// sane twin never diverge into separate cache entries (the bug this fix
    /// closes: the insane request was cached STANDALONE).
    #[test]
    fn insane_and_sane_twin_requests_share_one_cache_entry() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);
        let (_db, resolver, params_reader, _root) =
            resolver_and_params_reader_with_radius("SmallMoon", 500.0);
        let queue = GenerationQueue::with_threads(1);

        // The insane request — the reported T-1142 letterbox-click values.
        let insane_req = AtlasLayerRequest {
            body_id: "SmallMoon".to_string(),
            up_to: CascadeLayer::Topography,
            window_center: Some((12276, 3021)),
            window_n: 4,
        };
        let resp1 = handle_atlas_request(
            &insane_req,
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            1,
            test_conn_id(),
        );
        assert!(
            resp1.district_window.is_none(),
            "first request — cache miss, DeriveWindow submitted"
        );

        // Wait for the background derive to complete and drain it into the cache
        // (mirrors handle_atlas_request_clamps_oversized_window_n's pattern).
        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        for c in completions {
            if let GenCompletion::WindowDerived { body_id, layer } = c {
                if body_id == "SmallMoon" {
                    window_cache.insert((body_id, layer.center, layer.n), *layer);
                }
            }
        }
        // Exactly ONE entry must exist in the window cache after the insane
        // request's derive completes — normalization means it was keyed on
        // the canonical (4, 383), not the raw (12276, 3021).
        assert_eq!(
            window_cache.len(),
            1,
            "the insane request's derive must be cached under its NORMALIZED key"
        );

        // The "sane twin" — the already-normalized canonical center — hits
        // the SAME cache entry the insane request just populated.
        let sane_twin_req = AtlasLayerRequest {
            body_id: "SmallMoon".to_string(),
            up_to: CascadeLayer::Topography,
            window_center: Some((4, 383)), // the hand-computed canonical twin
            window_n: 4,
        };
        let resp2 = handle_atlas_request(
            &sane_twin_req,
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            2,
            test_conn_id(),
        );
        let twin_layer = resp2.district_window.expect(
            "the sane twin must hit the cache the insane request populated — no new derive needed",
        );
        assert_eq!(
            window_cache.len(),
            1,
            "the sane twin must NOT create a second cache entry"
        );

        // Re-request the ORIGINAL insane center too — it must ALSO now hit
        // the same populated cache entry (both requests normalize to the
        // same key).
        let resp3 = handle_atlas_request(
            &insane_req,
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            3,
            test_conn_id(),
        );
        let insane_layer_second_try = resp3
            .district_window
            .expect("the insane request, re-requested, must ALSO hit the shared cache entry");

        assert_eq!(
            twin_layer, insane_layer_second_try,
            "the insane request and its sane twin must resolve to a BYTE-IDENTICAL layer"
        );
        assert_eq!(
            window_cache.len(),
            1,
            "still exactly one entry — neither re-request created a second one"
        );
    }

    /// The echoed `center` on `DistrictWindowLayer` is the NORMALIZED value,
    /// not the raw wire value — the client's D-227 staleness guard (D-226
    /// T-1124 amendment §2) must see what was ACTUALLY derived, so it can
    /// correctly match this response against its own (now also normalized,
    /// per the T-1142 fix note to the client team) cache key.
    #[test]
    fn echoed_center_is_the_normalized_value_not_the_raw_wire_value() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);
        let (_db, resolver, params_reader, _root) =
            resolver_and_params_reader_with_radius("SmallMoon2", 500.0);
        let queue = GenerationQueue::with_threads(1);

        let insane_req = AtlasLayerRequest {
            body_id: "SmallMoon2".to_string(),
            up_to: CascadeLayer::Topography,
            window_center: Some((12276, 3021)), // raw, out-of-range
            window_n: 4,
        };
        handle_atlas_request(
            &insane_req,
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            1,
            test_conn_id(),
        );

        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        let window_completion = completions.into_iter().find_map(|c| {
            if let GenCompletion::WindowDerived { body_id, layer } = c {
                if body_id == "SmallMoon2" {
                    return Some(layer);
                }
            }
            None
        });
        let layer = window_completion.expect("DeriveWindow must complete");
        assert_eq!(
            layer.center,
            (4, 383),
            "the completed/echoed layer.center is the NORMALIZED value, not the raw (12276, 3021)"
        );
        assert_ne!(
            layer.center,
            (12276, 3021),
            "the raw out-of-range wire value must never be echoed back"
        );
    }

    /// `serve_district_window` returns `None` (no window requested) when the
    /// request carries no `window_center` — the common case, and the ONLY
    /// path every pre-T-1137 caller takes (wire back-compat: an old client's
    /// `{body_id, up_to}` frame decodes with `window_center: None` via
    /// `#[serde(default)]`).
    #[test]
    fn handle_atlas_request_no_window_center_leaves_district_window_none() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);
        let (_db, resolver) = empty_resolver();
        let queue = GenerationQueue::with_threads(1);

        let resp = handle_atlas_request(
            &req("GJ1c"), // window_center: None
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
            test_conn_id(),
        );
        assert!(resp.district_window.is_none());
        assert!(
            window_cache.is_empty(),
            "no window requested → no DeriveWindow submitted, cache stays empty"
        );
    }

    /// `AtlasLayerResponse.district_window` survives a MessagePack round trip
    /// (mirrors the existing `atlas_layer_response_with_new_layers_round_trips_msgpack`
    /// precedent) — the wire shape every field the amendment specifies:
    /// echoed `center`/`n`, all six parallel arrays including the
    /// `REGION_TEMP_NONE_DC` sentinel and `VegetationClass::Marine = 6`.
    #[test]
    fn district_window_layer_round_trips_msgpack_inside_response() {
        let window = DistrictWindowLayer {
            center: (10, -5),
            n: 2,
            morphology: vec![0, 8, 14, 16],
            elev_q: vec![0, 45, 98, 60],
            temp_dc: vec![205, 150, REGION_TEMP_NONE_DC, 80],
            moisture_q: vec![90, 55, 0, 100],
            vegetation: vec![6, 3, 0, 5], // includes Marine = 6
            glaciation: vec![0, 0, 4, 1],
        };
        let resp = AtlasLayerResponse {
            body_id: "GJ1c".into(),
            status: AtlasLayerStatus::Ready,
            layer1: None,
            district_grid: None,
            road_graph: None,
            settlements: None,
            region_grid: None,
            district_window: Some(window.clone()),
            quarter_footprints: None,
        };

        let bytes = rmp_serde::to_vec_named(&resp).expect("encode");
        let decoded: AtlasLayerResponse = rmp_serde::from_slice(&bytes).expect("decode");

        let dw = decoded
            .district_window
            .expect("district_window survives round trip");
        assert_eq!(dw, window);
        assert_eq!(dw.center, (10, -5));
        assert_eq!(dw.n, 2);
        assert_eq!(
            dw.temp_dc[2], REGION_TEMP_NONE_DC,
            "airless sentinel preserved"
        );
        assert_eq!(dw.vegetation[0], 6, "Marine discriminant preserved");
    }

    /// Wire back-compat (D-226 T-1124 amendment §1): a pre-T-1137 request
    /// frame carrying only `{body_id, up_to}` — no `window_center`/`window_n`
    /// keys at all — decodes cleanly via `#[serde(default)]`, byte-unchanged
    /// for every existing caller.
    #[test]
    fn old_request_frame_without_window_fields_decodes_with_none() {
        #[derive(serde::Serialize)]
        struct OldAtlasLayerRequest {
            body_id: String,
            up_to: CascadeLayer,
        }
        let old = OldAtlasLayerRequest {
            body_id: "GJ1c".into(),
            up_to: CascadeLayer::Topography,
        };
        let bytes = rmp_serde::to_vec_named(&old).expect("encode old-shape frame");
        let decoded: AtlasLayerRequest = rmp_serde::from_slice(&bytes).expect("decode");
        assert_eq!(decoded.body_id, "GJ1c");
        assert_eq!(decoded.up_to, CascadeLayer::Topography);
        assert_eq!(decoded.window_center, None);
        assert_eq!(decoded.window_n, 0);
    }

    /// T-1119: `build_quarter_footprint_layer` returns `None` when no
    /// settlement's quarter skeleton has been generated (`state.quarters`
    /// empty), mirroring `build_district_grid`/`build_region_grid`'s
    /// "unrun layer → None" contract.
    #[test]
    fn quarter_footprint_layer_none_when_quarters_empty() {
        let state = blank_state("GJ1c");
        assert!(build_quarter_footprint_layer(&state, 42).is_none());
    }

    /// A `BlockSkeleton` fixture builder for quarter-footprint tests — only
    /// the fields the aggregate reads are wired; the rest default.
    fn block(
        zoning: crate::simulation::generator::ZoningType,
        district_type: DistrictType,
        density_pct: u8,
        landmark: Option<crate::simulation::generator::LandmarkSlot>,
    ) -> crate::simulation::generator::BlockSkeleton {
        crate::simulation::generator::BlockSkeleton {
            zoning,
            district_type,
            density_pct,
            landmark,
            ..Default::default()
        }
    }

    /// T-1119: a populated quarter aggregates correctly — the density mean,
    /// the dominant-mode fields (including a tie resolving to the lowest
    /// declaration-order variant per the D-226 T-1112 amendment §1), the
    /// landmark count, and the corridor count.
    #[test]
    fn quarter_footprint_layer_aggregates_populated_quarter() {
        use crate::atlas::attractor_matching::CityPlacement;
        use crate::simulation::generator::{
            ArrangementPattern, FoundingOrientation, PoliticalArchetype, ZoningType,
        };

        let mut state = blank_state("GJ1c");
        let placement = CityPlacement {
            city_id: 7,
            name: "Millbrook".into(),
            position: (30, 40),
            attractor_type: AttractorType::ValleyFloor,
            score: 500,
            synthetic: false,
            political_archetype: PoliticalArchetype::Commission,
            arrangement_pattern: ArrangementPattern::RadialCore,
            founding_orientation: FoundingOrientation::Cardinal,
            population: 200_000,
            is_capital: false,
            is_standalone_hq: false,
        };
        state.placements = vec![placement.clone()];

        let world_seed = 42;
        let quarter_id = SeedChain::for_body(world_seed, "GJ1c")
            .derive(SeedDomain::Layer4Quarter, placement.city_id)
            .seed();

        // 16 blocks: 10 Commercial/Commercial, 6 Industrial/Industrial — a
        // clean (non-tied) mode on both district_type and zoning, plus a
        // known density mean and landmark/corridor counts.
        let mut blocks: [[crate::simulation::generator::BlockSkeleton; 4]; 4] = Default::default();
        let mut flat: Vec<&mut crate::simulation::generator::BlockSkeleton> =
            blocks.iter_mut().flatten().collect();
        for (i, b) in flat.iter_mut().enumerate() {
            if i < 10 {
                **b = block(ZoningType::Commercial, DistrictType::Commercial, 60, None);
            } else {
                **b = block(
                    ZoningType::Industrial,
                    DistrictType::Industrial,
                    20,
                    Some("landmark".to_string()),
                );
            }
        }
        // 3 landmarks among the Industrial blocks (indices 10, 11, 12).
        *flat[10] = block(
            ZoningType::Industrial,
            DistrictType::Industrial,
            20,
            Some("A".to_string()),
        );
        *flat[11] = block(
            ZoningType::Industrial,
            DistrictType::Industrial,
            20,
            Some("B".to_string()),
        );
        *flat[12] = block(
            ZoningType::Industrial,
            DistrictType::Industrial,
            20,
            Some("C".to_string()),
        );
        for b in flat.iter_mut().skip(13) {
            **b = block(ZoningType::Industrial, DistrictType::Industrial, 20, None);
        }

        // (10 * 60 + 6 * 20) / 16 = 720 / 16 = 45.
        state.quarters.insert(
            quarter_id,
            crate::simulation::generator::QuarterWorldState {
                skeleton: crate::simulation::generator::QuarterSkeleton {
                    quarter_id,
                    blocks,
                    corridors: vec![
                        crate::simulation::generator::CorridorSpine {
                            from: 0,
                            to: 1,
                            path: vec![(0, 0), (4, 4)],
                        },
                        crate::simulation::generator::CorridorSpine {
                            from: 1,
                            to: 2,
                            path: vec![(4, 4), (8, 8)],
                        },
                    ],
                    ..Default::default()
                },
                block_tags: Default::default(),
            },
        );

        let layer =
            build_quarter_footprint_layer(&state, world_seed).expect("populated quarters → Some");
        let entry = layer.entries.get(&7).expect("city_id 7 entry present");
        assert_eq!(entry.city_id, 7);
        assert_eq!(entry.density_avg_pct, 45);
        assert_eq!(entry.dominant_district_type, DistrictType::Commercial);
        assert_eq!(entry.dominant_zoning, ZoningType::Commercial);
        assert_eq!(entry.landmark_count, 3);
        assert_eq!(entry.corridor_count, 2);
    }

    /// T-1119: the mode tie-break resolves to the lowest declaration-order
    /// variant (the `Ord` derive), per the D-226 T-1112 amendment §1's
    /// explicit tie rule — this is the reason `ZoningType` gained
    /// `PartialOrd`/`Ord` in this same ticket.
    #[test]
    fn quarter_footprint_layer_tie_breaks_by_declaration_order() {
        use crate::atlas::attractor_matching::CityPlacement;
        use crate::simulation::generator::{
            ArrangementPattern, FoundingOrientation, PoliticalArchetype, ZoningType,
        };

        let mut state = blank_state("GJ1c");
        let placement = CityPlacement {
            city_id: 3,
            name: "Farmstead Rell".into(),
            position: (50, 60),
            attractor_type: AttractorType::PlainCenter,
            score: 100,
            synthetic: false,
            political_archetype: PoliticalArchetype::Commission,
            arrangement_pattern: ArrangementPattern::RadialCore,
            founding_orientation: FoundingOrientation::Cardinal,
            population: 8_000,
            is_capital: false,
            is_standalone_hq: false,
        };
        state.placements = vec![placement.clone()];

        let world_seed = 99;
        let quarter_id = SeedChain::for_body(world_seed, "GJ1c")
            .derive(SeedDomain::Layer4Quarter, placement.city_id)
            .seed();

        // 8 blocks Industrial, 8 blocks Commercial — an exact tie. Declaration
        // order on both DistrictType and ZoningType lists Commercial before
        // Industrial, so the tie-broken dominant must be Commercial on both.
        let mut blocks: [[crate::simulation::generator::BlockSkeleton; 4]; 4] = Default::default();
        for (i, b) in blocks.iter_mut().flatten().enumerate() {
            *b = if i < 8 {
                block(ZoningType::Industrial, DistrictType::Industrial, 50, None)
            } else {
                block(ZoningType::Commercial, DistrictType::Commercial, 50, None)
            };
        }
        state.quarters.insert(
            quarter_id,
            crate::simulation::generator::QuarterWorldState {
                skeleton: crate::simulation::generator::QuarterSkeleton {
                    quarter_id,
                    blocks,
                    ..Default::default()
                },
                block_tags: Default::default(),
            },
        );

        let layer =
            build_quarter_footprint_layer(&state, world_seed).expect("populated quarters → Some");
        let entry = layer.entries.get(&3).expect("city_id 3 entry present");
        assert_eq!(
            entry.dominant_district_type,
            DistrictType::Commercial,
            "tie resolves to Commercial (declared before Industrial)"
        );
        assert_eq!(
            entry.dominant_zoning,
            ZoningType::Commercial,
            "tie resolves to Commercial (declared before Industrial)"
        );
    }

    /// T-1119: a placement whose deterministically-derived `quarter_id` is
    /// NOT yet in `state.quarters` (skeleton generation is async, dispatched
    /// after the body's own snapshot is cached — D-226 T-1112 amendment §1)
    /// is skipped, not defaulted. `entries` is a subset of `placements`.
    #[test]
    fn quarter_footprint_layer_skips_placement_without_matching_quarter() {
        use crate::atlas::attractor_matching::CityPlacement;
        use crate::simulation::generator::{
            ArrangementPattern, FoundingOrientation, PoliticalArchetype, ZoningType,
        };

        let mut state = blank_state("GJ1c");
        let has_quarter = CityPlacement {
            city_id: 1,
            name: "Port Aldren".into(),
            position: (12, 58),
            attractor_type: AttractorType::CoastalAccess,
            score: 1000,
            synthetic: false,
            political_archetype: PoliticalArchetype::Commission,
            arrangement_pattern: ArrangementPattern::RadialCore,
            founding_orientation: FoundingOrientation::Cardinal,
            population: 2_000_000,
            is_capital: true,
            is_standalone_hq: false,
        };
        let no_quarter_yet = CityPlacement {
            city_id: 2,
            name: "Farmstead Rell".into(),
            ..has_quarter.clone()
        };
        state.placements = vec![has_quarter.clone(), no_quarter_yet];

        let world_seed = 42;
        let quarter_id = SeedChain::for_body(world_seed, "GJ1c")
            .derive(SeedDomain::Layer4Quarter, has_quarter.city_id)
            .seed();
        let mut blocks: [[crate::simulation::generator::BlockSkeleton; 4]; 4] = Default::default();
        for b in blocks.iter_mut().flatten() {
            *b = block(ZoningType::Mixed, DistrictType::MixedUse, 10, None);
        }
        state.quarters.insert(
            quarter_id,
            crate::simulation::generator::QuarterWorldState {
                skeleton: crate::simulation::generator::QuarterSkeleton {
                    quarter_id,
                    blocks,
                    ..Default::default()
                },
                block_tags: Default::default(),
            },
        );

        let layer =
            build_quarter_footprint_layer(&state, world_seed).expect("populated quarters → Some");
        assert_eq!(
            layer.entries.len(),
            1,
            "only the placement with a matching quarter gets an entry"
        );
        assert!(layer.entries.contains_key(&1));
        assert!(
            !layer.entries.contains_key(&2),
            "city_id 2 has no generated quarter yet — must be absent, not defaulted"
        );
    }

    /// T-1119 (D-010): building the layer twice from identical state produces
    /// byte-identical output — the `city_id → quarter_id` derivation and the
    /// mode aggregation are pure functions of their inputs.
    #[test]
    fn quarter_footprint_layer_is_deterministic() {
        use crate::atlas::attractor_matching::CityPlacement;
        use crate::simulation::generator::{
            ArrangementPattern, FoundingOrientation, PoliticalArchetype, ZoningType,
        };

        let mut state = blank_state("GJ1c");
        let placement = CityPlacement {
            city_id: 5,
            name: "Groombridge".into(),
            position: (1, 1),
            attractor_type: AttractorType::PlainCenter,
            score: 300,
            synthetic: false,
            political_archetype: PoliticalArchetype::Commission,
            arrangement_pattern: ArrangementPattern::RadialCore,
            founding_orientation: FoundingOrientation::Cardinal,
            population: 60_000,
            is_capital: false,
            is_standalone_hq: false,
        };
        state.placements = vec![placement.clone()];

        let world_seed = 7;
        let quarter_id = SeedChain::for_body(world_seed, "GJ1c")
            .derive(SeedDomain::Layer4Quarter, placement.city_id)
            .seed();
        let mut blocks: [[crate::simulation::generator::BlockSkeleton; 4]; 4] = Default::default();
        for (i, b) in blocks.iter_mut().flatten().enumerate() {
            *b = block(
                ZoningType::Residential,
                DistrictType::Residential,
                (i as u8) * 5,
                None,
            );
        }
        state.quarters.insert(
            quarter_id,
            crate::simulation::generator::QuarterWorldState {
                skeleton: crate::simulation::generator::QuarterSkeleton {
                    quarter_id,
                    blocks,
                    ..Default::default()
                },
                block_tags: Default::default(),
            },
        );

        let a = build_quarter_footprint_layer(&state, world_seed);
        let b = build_quarter_footprint_layer(&state, world_seed);
        assert_eq!(a, b, "identical state must produce identical output");
    }

    /// A blank `BodyWorldState` for tests that only care about one field —
    /// callers overwrite `placements`/`road_graph`/etc. as needed.
    fn blank_state(body_id: &str) -> BodyWorldState {
        BodyWorldState {
            body_id: body_id.into(),
            heightmap: vec![],
            heightmap_width: 16,
            heightmap_height: 8,
            river_network: RiverNetwork::default(),
            drainage_basins: vec![],
            attractors: vec![],
            placements: vec![],
            road_graph: crate::atlas::road_graph::RoadGraph::default(),
            quarters: std::collections::BTreeMap::new(),
            districts: std::collections::BTreeMap::new(),
            regions: std::collections::BTreeMap::new(),
            last_accessed: 0,
        }
    }

    /// T-960 §1: `build_road_graph_layer` trims the internal `RoadGraph` (drops
    /// `degree`/`parent_edge`/`length_cells`) while keeping everything a
    /// planetary-map overlay needs (positions, kind, polyline, maintenance,
    /// rail flag, named-route id).
    #[test]
    fn road_graph_layer_built_from_cached_state() {
        use crate::atlas::road_graph::{RoadEdge, RoadGraph, RoadNode};
        use crate::simulation::generator::MaintenanceAuthority;

        let mut state = blank_state("GJ1c");
        state.road_graph = RoadGraph {
            nodes: vec![
                RoadNode {
                    city_id: Some(1),
                    position: (10, 20),
                    kind: RoadNodeKind::Settlement,
                    degree: 1,
                    parent_edge: None,
                    is_hub: false,
                },
                RoadNode {
                    city_id: None,
                    position: (15, 25),
                    kind: RoadNodeKind::Waypoint,
                    degree: 0,
                    parent_edge: Some(0),
                    is_hub: false,
                },
            ],
            edges: vec![RoadEdge {
                from: 0,
                to: 1,
                path: vec![(10, 20), (15, 25)],
                length_cells: 20, // internal routing-grid measure — dropped
                maintenance: MaintenanceAuthority::Trade,
                named_route_id: Some("split/hwy-1".into()),
                is_rail: true,
            }],
        };

        let layer = build_road_graph_layer(&state).expect("populated road_graph → Some");
        assert_eq!(layer.nodes.len(), 2);
        assert_eq!(layer.nodes[0].position, (10, 20));
        assert_eq!(layer.nodes[0].kind, RoadNodeKind::Settlement);
        assert_eq!(layer.nodes[0].city_id, Some(1));
        assert_eq!(layer.nodes[1].kind, RoadNodeKind::Waypoint);
        assert_eq!(layer.nodes[1].city_id, None);
        assert_eq!(layer.edges.len(), 1);
        assert_eq!(layer.edges[0].path, vec![(10, 20), (15, 25)]);
        assert_eq!(layer.edges[0].maintenance, MaintenanceAuthority::Trade);
        assert!(layer.edges[0].is_rail);
        assert_eq!(
            layer.edges[0].named_route_id.as_deref(),
            Some("split/hwy-1")
        );

        // Layer hasn't run (or zero settlements) → None, mirroring district_grid.
        let unrun = blank_state("GJ1c");
        assert!(build_road_graph_layer(&unrun).is_none());
    }

    /// T-960 §2: `build_settlement_layer` derives `size_class` from population
    /// using the D-211 Tier A/B cutoffs, threads `is_capital` straight through,
    /// and derives `is_port` cheaply from the anchoring attractor type.
    #[test]
    fn settlement_layer_built_from_cached_placements() {
        use crate::atlas::attractor_matching::CityPlacement;
        use crate::simulation::generator::{
            ArrangementPattern, FoundingOrientation, PoliticalArchetype,
        };

        let mk = |city_id: u64,
                  name: &str,
                  pos: (u16, u16),
                  population: i64,
                  is_capital: bool,
                  attractor_type: AttractorType| CityPlacement {
            city_id,
            name: name.to_string(),
            position: pos,
            attractor_type,
            score: 1000,
            synthetic: false,
            political_archetype: PoliticalArchetype::Commission,
            arrangement_pattern: ArrangementPattern::RadialCore,
            founding_orientation: FoundingOrientation::Cardinal,
            population,
            is_capital,
            is_standalone_hq: false,
        };

        let mut state = blank_state("GJ1c");
        state.placements = vec![
            mk(
                1,
                "Port Aldren",
                (12, 58),
                2_000_000,
                true,
                AttractorType::CoastalAccess,
            ),
            mk(
                2,
                "Millbrook",
                (30, 40),
                200_000,
                false,
                AttractorType::ValleyFloor,
            ),
            mk(
                3,
                "Farmstead Rell",
                (50, 60),
                8_000,
                false,
                AttractorType::PlainCenter,
            ),
        ];

        let layer = build_settlement_layer(&state).expect("populated placements → Some");
        assert_eq!(layer.settlements.len(), 3);

        let capital = layer.settlements.iter().find(|s| s.city_id == 1).unwrap();
        assert_eq!(capital.name, "Port Aldren");
        assert_eq!(capital.position, (12, 58));
        assert_eq!(capital.size_class, SettlementSizeClass::Major);
        assert!(capital.is_capital);
        assert!(capital.is_port, "CoastalAccess must read as a port");

        let mid = layer.settlements.iter().find(|s| s.city_id == 2).unwrap();
        assert_eq!(mid.size_class, SettlementSizeClass::Standard);
        assert!(!mid.is_capital);
        assert!(!mid.is_port, "ValleyFloor is not a port attractor");

        let small = layer.settlements.iter().find(|s| s.city_id == 3).unwrap();
        assert_eq!(small.size_class, SettlementSizeClass::Minor);
        assert!(!small.is_port);

        // No placements → None.
        let unrun = blank_state("GJ1c");
        assert!(build_settlement_layer(&unrun).is_none());
    }

    /// T-960 / T-1119: the new layers survive a MessagePack round trip inside
    /// `AtlasLayerResponse` — the same wire path the bridge uses
    /// (`rmp_serde::to_vec_named` / `from_slice`, matching `layer1`/
    /// `district_grid`'s existing serialization).
    #[test]
    fn atlas_layer_response_with_new_layers_round_trips_msgpack() {
        use crate::atlas::attractor_matching::CityPlacement;
        use crate::atlas::road_graph::{RoadEdge, RoadGraph, RoadNode};
        use crate::simulation::generator::{
            ArrangementPattern, BlockSkeleton, DistrictType, FoundingOrientation,
            MaintenanceAuthority, PoliticalArchetype, QuarterSkeleton, QuarterWorldState,
            ZoningType,
        };

        let mut state = blank_state("GJ1c");
        state.placements = vec![CityPlacement {
            city_id: 1,
            name: "Port Aldren".into(),
            position: (12, 58),
            attractor_type: AttractorType::CoastalAccess,
            score: 1000,
            synthetic: false,
            political_archetype: PoliticalArchetype::Commission,
            arrangement_pattern: ArrangementPattern::RadialCore,
            founding_orientation: FoundingOrientation::Cardinal,
            population: 2_000_000,
            is_capital: true,
            is_standalone_hq: false,
        }];
        state.road_graph = RoadGraph {
            nodes: vec![RoadNode {
                city_id: Some(1),
                position: (12, 58),
                kind: RoadNodeKind::Settlement,
                degree: 0,
                parent_edge: None,
                is_hub: false,
            }],
            edges: vec![RoadEdge {
                from: 0,
                to: 0,
                path: vec![(12, 58)],
                length_cells: 0,
                maintenance: MaintenanceAuthority::Administrative,
                named_route_id: None,
                is_rail: false,
            }],
        };
        let world_seed = 42;
        let quarter_id = SeedChain::for_body(world_seed, "GJ1c")
            .derive(SeedDomain::Layer4Quarter, 1)
            .seed();
        let mut block = BlockSkeleton {
            zoning: ZoningType::Commercial,
            district_type: DistrictType::Commercial,
            density_pct: 40,
            ..Default::default()
        };
        block.position = (0, 0);
        state.quarters.insert(
            quarter_id,
            QuarterWorldState {
                skeleton: QuarterSkeleton {
                    quarter_id,
                    blocks: std::array::from_fn(|_| std::array::from_fn(|_| block.clone())),
                    ..Default::default()
                },
                block_tags: Default::default(),
            },
        );

        let resp = AtlasLayerResponse {
            body_id: "GJ1c".into(),
            status: AtlasLayerStatus::Ready,
            layer1: None,
            district_grid: None,
            road_graph: build_road_graph_layer(&state),
            settlements: build_settlement_layer(&state),
            region_grid: build_region_grid(&state),
            district_window: None,
            quarter_footprints: build_quarter_footprint_layer(&state, world_seed),
        };

        let bytes = rmp_serde::to_vec_named(&resp).expect("encode");
        let decoded: AtlasLayerResponse = rmp_serde::from_slice(&bytes).expect("decode");

        assert_eq!(decoded.body_id, "GJ1c");
        let rg = decoded.road_graph.expect("road_graph survives round trip");
        assert_eq!(rg.nodes[0].position, (12, 58));
        assert_eq!(
            rg.edges[0].maintenance,
            MaintenanceAuthority::Administrative
        );
        let qf = decoded
            .quarter_footprints
            .expect("quarter_footprints survives round trip");
        let entry = qf.entries.get(&1).expect("city_id 1 entry present");
        assert_eq!(entry.density_avg_pct, 40);
        assert_eq!(entry.dominant_district_type, DistrictType::Commercial);
        assert_eq!(entry.dominant_zoning, ZoningType::Commercial);
        let settlements = decoded
            .settlements
            .expect("settlements survives round trip");
        assert_eq!(settlements.settlements[0].name, "Port Aldren");
        assert_eq!(
            settlements.settlements[0].size_class,
            SettlementSizeClass::Major
        );
        assert!(settlements.settlements[0].is_capital);
        assert!(settlements.settlements[0].is_port);
    }

    fn req(body_id: &str) -> AtlasLayerRequest {
        AtlasLayerRequest {
            body_id: body_id.to_string(),
            up_to: CascadeLayer::Topography,
            window_center: None,
            window_n: 0,
        }
    }

    fn test_conn_id() -> ConnectionId {
        ConnectionId(1)
    }

    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![],
            placements: vec![],
            road_graph: crate::atlas::road_graph::RoadGraph::default(),
            quarters: std::collections::BTreeMap::new(),
            districts: std::collections::BTreeMap::new(),
            regions: std::collections::BTreeMap::new(),
            last_accessed: 0,
        });
        let (_db, resolver) = empty_resolver();
        let queue = GenerationQueue::with_threads(1);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
            test_conn_id(),
        );
        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 mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
            test_conn_id(),
        );
        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 mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);

        let resp = handle_atlas_request(
            &req("ghost"),
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
            test_conn_id(),
        );
        assert_eq!(resp.status, AtlasLayerStatus::NotFound);
    }

    /// Build a DB with the columns needed by both `BodySourceResolver` and
    /// `BodyParamsReader` for the same body (Earth radius, 6371 km), plus a
    /// tiny heightmap root.
    ///
    /// Returns (db_path, resolver, body_params_reader, _root_kept_alive).
    fn resolver_and_params_reader(
        body_id: &str,
    ) -> (
        PathBuf,
        BodySourceResolver,
        crate::atlas::body_params_reader::BodyParamsReader,
        PathBuf, // root dir — must stay alive for the test duration
    ) {
        resolver_and_params_reader_with_radius(body_id, 6371.0)
    }

    /// Same as [`resolver_and_params_reader`] with a caller-chosen
    /// `body_radius_km` (T-1142: the window-centre normalization tests need a
    /// SMALL body — at Earth radius the district-circumference/half-meridian
    /// bounds are tens of thousands of districts wide, too large for a
    /// hand-checkable out-of-range test value).
    fn resolver_and_params_reader_with_radius(
        body_id: &str,
        r_km: f64,
    ) -> (
        PathBuf,
        BodySourceResolver,
        crate::atlas::body_params_reader::BodyParamsReader,
        PathBuf, // root dir — must stay alive for the test duration
    ) {
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let db = std::env::temp_dir().join(format!("sr_proxybp_{}_{n}.db", std::process::id()));
        let _ = std::fs::remove_file(&db);
        let conn = Connection::open(&db).unwrap();

        conn.execute_batch(
            "CREATE TABLE star_systems (
                 system_id TEXT PRIMARY KEY,
                 spectral_class TEXT,
                 star_type TEXT
             );
             CREATE TABLE bodies (
                 body_id TEXT PRIMARY KEY,
                 system_id TEXT,
                 terrain_reference TEXT,
                 hydrosphere TEXT,
                 atmosphere TEXT,
                 planet_class TEXT,
                 body_radius_km REAL,
                 orbital_period_days REAL,
                 axial_tilt_deg REAL
             );",
        )
        .unwrap();

        conn.execute(
            "INSERT INTO star_systems (system_id, spectral_class, star_type) VALUES ('GJ-1', 'G', 'main_sequence')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO bodies (body_id, system_id, terrain_reference, hydrosphere, atmosphere, planet_class, body_radius_km, orbital_period_days, axial_tilt_deg)
             VALUES (?1, 'GJ-1', ?2, 'ocean', 'breathable', 'temperate', ?3, 365.25, 23.5)",
            rusqlite::params![body_id, REL, r_km],
        )
        .unwrap();
        drop(conn);

        let root = std::env::temp_dir().join(format!("sr_proxybproot_{}_{n}", std::process::id()));
        write_tiny_heightmap(&root.join(REL));

        let resolver = BodySourceResolver::open(&db, vec![root.clone()]).unwrap();
        let params_reader = crate::atlas::body_params_reader::BodyParamsReader::open(&db).unwrap();

        (db, resolver, params_reader, root)
    }

    /// With body_params_reader wired, a cache miss enqueues an AnalyzeBody that
    /// completes with populated `districts` (DistrictProfile layer ran) AND
    /// populated `regions` (Region layer ran — the production terminal,
    /// T-1113). Then the completed state served back through
    /// `handle_atlas_request` carries a `region_grid` — closing the full
    /// dispatch → Ready → region_grid loop (PR #179 F4).
    #[test]
    fn body_params_reader_wired_produces_populated_regions() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let (_db, resolver, params_reader, _root) = resolver_and_params_reader("GJ1c");
        let queue = GenerationQueue::with_threads(1);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            1,
            test_conn_id(),
        );
        assert_eq!(resp.status, AtlasLayerStatus::Pending);

        // Wait for the Rayon work item to complete.
        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        let body_state = completions
            .into_iter()
            .find_map(|c| {
                if let GenCompletion::BodyAnalyzed { body_id, state } = c {
                    if body_id == "GJ1c" {
                        return Some(state);
                    }
                }
                None
            })
            .expect("AnalyzeBody must complete for GJ1c");

        assert!(
            !body_state.districts.is_empty(),
            "districts must be populated when body_params_reader is wired (T-1032 dispatch path)"
        );
        assert!(
            !body_state.regions.is_empty(),
            "regions must be populated when body_params_reader is wired (T-1113 dispatch path)"
        );

        // Serve the completed state back through the proxy: the cache-hit
        // branch must build and include the region grid.
        cache.insert(body_state);
        let ready = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            2,
            test_conn_id(),
        );
        assert_eq!(ready.status, AtlasLayerStatus::Ready);
        assert!(
            ready.region_grid.is_some(),
            "a Ready response for a Region-populated body must carry region_grid"
        );
    }

    /// Without body_params_reader (None), districts is empty — pre-T-1032 behaviour.
    #[test]
    fn no_body_params_reader_leaves_regions_empty() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let (_db, resolver) = resolver_with_body("GJ1c");
        let queue = GenerationQueue::with_threads(1);
        let mut window_cache = DistrictWindowCache::new(DISTRICT_WINDOW_CACHE_CAPACITY);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &mut window_cache,
            &queue,
            &resolver,
            None,
            None, // no body_params_reader
            42,
            1,
            test_conn_id(),
        );
        assert_eq!(resp.status, AtlasLayerStatus::Pending);

        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        let body_state = completions
            .into_iter()
            .find_map(|c| {
                if let GenCompletion::BodyAnalyzed { body_id, state } = c {
                    if body_id == "GJ1c" {
                        return Some(state);
                    }
                }
                None
            })
            .expect("AnalyzeBody must complete for GJ1c");

        assert!(
            body_state.districts.is_empty(),
            "districts must remain empty when no body_params_reader is wired"
        );
        assert!(
            body_state.regions.is_empty(),
            "regions must remain empty when no body_params_reader is wired \
             (the Region layer gates on body_params, T-1113)"
        );
    }
}