diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs
index c7dfb9c2..06bfaf52 100644
--- a/crates/lance-context-core/src/store_base.rs
+++ b/crates/lance-context-core/src/store_base.rs
@@ -224,6 +224,10 @@ pub struct ShardWatermark {
     pub sealed_through: u64,
     /// Flushed generations not yet merged into the base table.
     pub pending_generations: usize,
+    /// Highest generation the base table's MemWAL index records as merged
+    /// for this shard, when the caller read the base index. `None` means
+    /// "not read", not "nothing merged".
+    pub merged_through: Option<u64>,
 }
 
 pub fn is_pending_generations_exceeded(err: &LanceError) -> bool {
@@ -948,6 +952,7 @@ impl StorageBase {
                     .max()
                     .unwrap_or(0),
                 pending_generations: manifest.flushed_generations.len(),
+                merged_through: None,
             }))
     }
 
@@ -2091,6 +2096,15 @@ impl StorageBase {
         if self.is_version_pinned() {
             return Ok(Vec::new());
         }
+        // The base table's MemWAL index is the authority on what has been
+        // merged. A scan must report it, or a lost completion event leaves
+        // a fully merged shard looking pending forever, and an old table's
+        // first contact reads generation 1001 as 1001 pending.
+        let merged = Arc::new(
+            crate::rollout_append::watermarks(&self.dataset)
+                .await
+                .unwrap_or_default(),
+        );
         let object_store = self.dataset.object_store(None).await?;
         let branch_path = self.dataset.branch_location().path.clone();
         let shard_ids = self.dataset.list_mem_wal_latest_shard_ids().await?;
@@ -2098,6 +2112,7 @@ impl StorageBase {
             .map(|shard_id| {
                 let object_store = object_store.clone();
                 let branch_path = branch_path.clone();
+                let merged = merged.clone();
                 async move {
                     let manifest_store = ShardManifestStore::new(
                         object_store,
@@ -2116,6 +2131,7 @@ impl StorageBase {
                                 .max()
                                 .unwrap_or(0),
                             pending_generations: manifest.flushed_generations.len(),
+                            merged_through: merged.get(&manifest.shard_id).copied(),
                         }
                     }))
                 }
diff --git a/crates/lance-context-master/src/demand_publish.rs b/crates/lance-context-master/src/demand_publish.rs
index 5377cef3..dd560515 100644
--- a/crates/lance-context-master/src/demand_publish.rs
+++ b/crates/lance-context-master/src/demand_publish.rs
@@ -65,7 +65,7 @@ pub(crate) fn spawn_scan_demand(
                     shard: mark.shard_id.to_string(),
                     sealed_through: mark.sealed_through,
                     sealed_bytes_through: 0,
-                    merged_through: None,
+                    merged_through: mark.merged_through,
                     flushed_at_ms: now_ms,
                     writer_epoch: mark.writer_epoch,
                     source: EventSource::Scan,
diff --git a/crates/lance-context-master/src/executors.rs b/crates/lance-context-master/src/executors.rs
index 49bceca1..4a849c3b 100644
--- a/crates/lance-context-master/src/executors.rs
+++ b/crates/lance-context-master/src/executors.rs
@@ -16,7 +16,7 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 use std::time::Duration;
 
-use etcd_client::{GetOptions, PutOptions};
+use etcd_client::PutOptions;
 use lance_context_api::TaskKind;
 use serde::{Deserialize, Serialize};
 
@@ -322,38 +322,23 @@ pub(crate) async fn load_headroom(
     client: &etcd_client::Client,
     keys: &Keys,
 ) -> Result<BTreeMap<String, Headroom>, String> {
-    let mut client = client.clone();
-    let hbs = client
-        .get(
-            keys.executors_prefix(),
-            Some(GetOptions::new().with_prefix()),
-        )
-        .await
-        .map_err(|e| e.to_string())?;
+    let (hbs, _) = crate::planner::read_prefix_paged(client, &keys.executors_prefix()).await?;
     let prefix = keys.executors_prefix();
     let heartbeats: Vec<(String, Heartbeat)> = hbs
-        .kvs()
         .iter()
         .filter_map(|kv| {
-            let id = std::str::from_utf8(kv.key())
+            let id = std::str::from_utf8(&kv.key)
                 .ok()?
                 .strip_prefix(prefix.as_str())?
                 .to_string();
-            let hb: Heartbeat = serde_json::from_slice(kv.value()).ok()?;
+            let hb: Heartbeat = serde_json::from_slice(&kv.value).ok()?;
             (hb.v == SCHEMA_VERSION).then_some((id, hb))
         })
         .collect();
-    let asg = client
-        .get(
-            keys.assignments_prefix(),
-            Some(GetOptions::new().with_prefix()),
-        )
-        .await
-        .map_err(|e| e.to_string())?;
+    let (asg, _) = crate::planner::read_prefix_paged(client, &keys.assignments_prefix()).await?;
     let assignments: Vec<Assignment> = asg
-        .kvs()
         .iter()
-        .filter_map(|kv| serde_json::from_slice::<Assignment>(kv.value()).ok())
+        .filter_map(|kv| serde_json::from_slice::<Assignment>(&kv.value).ok())
         .filter(|a| a.v == SCHEMA_VERSION)
         .collect();
     Ok(headroom(
diff --git a/crates/lance-context-master/src/planner.rs b/crates/lance-context-master/src/planner.rs
index b6bf1a7c..c3160800 100644
--- a/crates/lance-context-master/src/planner.rs
+++ b/crates/lance-context-master/src/planner.rs
@@ -69,6 +69,71 @@ pub(crate) fn unhex(encoded: &str) -> Option<String> {
     String::from_utf8(bytes?).ok()
 }
 
+/// A key/value/revision triple from a paged prefix read.
+pub(crate) struct Kv {
+    pub key: Vec<u8>,
+    pub value: Vec<u8>,
+    pub mod_revision: i64,
+}
+
+/// Page size for prefix reads. 13k tables x a few shards each must never go
+/// through one response: the etcd client caps a message at 4 MiB.
+const PAGE: i64 = 512;
+
+/// Read every key under `prefix` in pages, all at one revision so the
+/// result is a consistent snapshot. Returns the kvs and that revision.
+pub(crate) async fn read_prefix_paged(
+    client: &etcd_client::Client,
+    prefix: &str,
+) -> Result<(Vec<Kv>, i64), String> {
+    let mut client = client.clone();
+    let range_end = {
+        let mut end = prefix.as_bytes().to_vec();
+        for i in (0..end.len()).rev() {
+            if end[i] != u8::MAX {
+                end[i] += 1;
+                end.truncate(i + 1);
+                break;
+            }
+        }
+        end
+    };
+    let mut out = Vec::new();
+    let mut start = prefix.as_bytes().to_vec();
+    let mut revision: i64 = 0;
+    loop {
+        let mut opts = GetOptions::new()
+            .with_range(range_end.clone())
+            .with_limit(PAGE);
+        if revision > 0 {
+            opts = opts.with_revision(revision);
+        }
+        let response = client
+            .get(start.clone(), Some(opts))
+            .await
+            .map_err(|e| e.to_string())?;
+        if revision == 0 {
+            revision = response.header().map_or(0, |h| h.revision());
+        }
+        let more = response.more();
+        let last = response.kvs().last().map(|kv| kv.key().to_vec());
+        for kv in response.kvs() {
+            out.push(Kv {
+                key: kv.key().to_vec(),
+                value: kv.value().to_vec(),
+                mod_revision: kv.mod_revision(),
+            });
+        }
+        let Some(last) = last else { break };
+        if !more {
+            break;
+        }
+        start = last;
+        start.push(0);
+    }
+    Ok((out, revision))
+}
+
 pub(crate) struct Keys {
     prefix: String,
 }
@@ -213,6 +278,7 @@ async fn lead(state: &Arc<MasterState>) -> Result<bool, String> {
     let reconcile_every = Duration::from_secs(state.config.planner.planner_reconcile_secs.max(5));
     let mut reconcile = tokio::time::interval(reconcile_every);
     let mut keepalive = tokio::time::interval(keepalive_every);
+    let mut inflight: Option<tokio::task::JoinHandle<()>> = None;
     let result = loop {
         tokio::select! {
             _ = keepalive.tick() => {
@@ -228,14 +294,32 @@ async fn lead(state: &Arc<MasterState>) -> Result<bool, String> {
                 if !state.admission.status().accepting {
                     break Ok(true);
                 }
-                if let Err(error) = reconcile_once(state, &keys, &token).await {
-                    tracing::warn!(%error, "planner reconcile failed");
-                    metrics::counter!("planner_reconcile_errors_total").increment(1);
+                // Reconcile runs detached so a slow pass (13k tables, many
+                // pages) can never starve this loop's lease keepalive. At
+                // most one pass in flight; a tick that finds one running is
+                // skipped and counted.
+                if let Some(h) = &inflight {
+                    if !h.is_finished() {
+                        metrics::counter!("planner_reconcile_skipped_total").increment(1);
+                        continue;
+                    }
                 }
+                let st = state.clone();
+                let k = Keys::new(&state.config.etcd.etcd_prefix);
+                let t = token.clone();
+                inflight = Some(tokio::spawn(async move {
+                    if let Err(error) = reconcile_once(&st, &k, &t).await {
+                        tracing::warn!(%error, "planner reconcile failed");
+                        metrics::counter!("planner_reconcile_errors_total").increment(1);
+                    }
+                }));
             }
         }
     };
     metrics::gauge!("planner_is_leader").set(0.0);
+    if let Some(h) = inflight.take() {
+        h.abort();
+    }
     drop(keeper);
     let _ = client.lease_revoke(lease).await;
     tracing::info!(token = %token, ?result, "planner leadership released");
@@ -251,31 +335,27 @@ pub(crate) async fn reconcile_once(
 ) -> Result<usize, String> {
     let started = std::time::Instant::now();
     let mut client = state.task_store.etcd_client().clone();
-    let events = client
-        .get(keys.events_prefix(), Some(GetOptions::new().with_prefix()))
-        .await
-        .map_err(|e| e.to_string())?;
-    let header_revision = events.header().map_or(0, |h| h.revision());
+    let (events, header_revision) =
+        read_prefix_paged(state.task_store.etcd_client(), &keys.events_prefix()).await?;
     let now = now_ms();
     let (tables, skipped) = fold_events(
         keys,
         events
-            .kvs()
             .iter()
-            .map(|kv| (kv.key(), kv.value(), kv.mod_revision())),
+            .map(|kv| (kv.key.as_slice(), kv.value.as_slice(), kv.mod_revision)),
         now,
     );
+    metrics::gauge!("planner_events").set(events.len() as f64);
     if skipped > 0 {
         metrics::counter!("planner_events_skipped_total").increment(skipped as u64);
     }
-    let existing = client
-        .get(keys.demand_prefix(), Some(GetOptions::new().with_prefix()))
-        .await
-        .map_err(|e| e.to_string())?;
+    let (existing, _) =
+        read_prefix_paged(state.task_store.etcd_client(), &keys.demand_prefix()).await?;
+    let existing_count = existing.len();
     let mut existing_by_key: BTreeMap<Vec<u8>, TableDemand> = BTreeMap::new();
-    for kv in existing.kvs() {
-        if let Ok(record) = serde_json::from_slice::<TableDemand>(kv.value()) {
-            existing_by_key.insert(kv.key().to_vec(), record);
+    for kv in &existing {
+        if let Ok(record) = serde_json::from_slice::<TableDemand>(&kv.value) {
+            existing_by_key.insert(kv.key.clone(), record);
         }
     }
     let mut ops = Vec::new();
@@ -364,7 +444,7 @@ pub(crate) async fn reconcile_once(
             tracing::warn!(%error, "planner could not load executor headroom");
         }
     }
-    metrics::gauge!("planner_tables").set(existing.kvs().len() as f64);
+    metrics::gauge!("planner_tables").set(existing_count as f64);
     metrics::gauge!("planner_pending_generations_total").set(total_pending as f64);
     metrics::histogram!("planner_reconcile_seconds").record(started.elapsed().as_secs_f64());
     metrics::gauge!("planner_last_revision").set(header_revision as f64);
@@ -419,21 +499,18 @@ async fn shadow_pass(
     // Only Shadow records from previous passes are replaced; nothing else
     // under assignments/ is touched.
     let mut client = state.task_store.etcd_client().clone();
-    let existing = client
-        .get(
-            executors_keys.assignments_prefix(),
-            Some(GetOptions::new().with_prefix()),
-        )
-        .await
-        .map_err(|e| e.to_string())?;
+    let (existing, _) = read_prefix_paged(
+        state.task_store.etcd_client(),
+        &executors_keys.assignments_prefix(),
+    )
+    .await?;
     let mut ops: Vec<TxnOp> = existing
-        .kvs()
         .iter()
         .filter(|kv| {
-            serde_json::from_slice::<Assignment>(kv.value())
+            serde_json::from_slice::<Assignment>(&kv.value)
                 .is_ok_and(|a| a.state == AssignmentState::Shadow)
         })
-        .map(|kv| TxnOp::delete(kv.key().to_vec(), None))
+        .map(|kv| TxnOp::delete(kv.key.clone(), None))
         .collect();
     // Headroom already counts previous Shadow records; release them in the
     // arithmetic too, since they are about to be replaced.
@@ -442,8 +519,8 @@ async fn shadow_pass(
         h.bytes_reserved = 0;
         h.assignments = 0;
     }
-    for kv in existing.kvs() {
-        if let Ok(a) = serde_json::from_slice::<Assignment>(kv.value()) {
+    for kv in &existing {
+        if let Ok(a) = serde_json::from_slice::<Assignment>(&kv.value) {
             if a.state != AssignmentState::Shadow && a.holds_capacity() {
                 if let Some(h) = headroom.get_mut(&a.executor) {
                     for (k, n) in &a.reserved_slots {
@@ -462,11 +539,22 @@ async fn shadow_pass(
         if s.class == Class::Tail {
             continue; // the real sweeps do not touch tails either
         }
-        // Estimated cost: staging buffers are ~2x decoded bytes, capped at
-        // the per-pass byte limit the append path enforces.
-        let expected_bytes = (s.pending_bytes.saturating_mul(2))
-            .min(state.config.append.rollout_append_max_bytes as u64 * 2)
-            .max(1);
+        // Estimated cost. A merge pass stages at most
+        // ROLLOUT_APPEND_MAX_BYTES of decoded data and the append path
+        // reserves ~2x that for buffers, so one unit costs at most
+        // 2 * max_bytes regardless of backlog. When pending bytes are
+        // unknown (writers do not measure; only the scan does) the estimate
+        // is that full cost, never a token: an unknown must reserve the
+        // worst case or the capacity check validates nothing.
+        let per_pass = state.config.append.rollout_append_max_bytes as u64 * 2;
+        let expected_bytes = if s.pending_bytes == 0 {
+            per_pass
+        } else {
+            s.pending_bytes.saturating_mul(2).min(per_pass).max(1)
+        };
+        if s.pending_bytes == 0 {
+            metrics::counter!("planner_shadow_bytes_unknown_total").increment(1);
+        }
         let pick = headroom
             .values_mut()
             .filter(|h| h.fits(TaskKind::MergeWal, expected_bytes))
@@ -521,12 +609,37 @@ async fn shadow_pass(
         }
     }
 
-    // Disagreement: planner says place vs. reality has no task; or planner
-    // says nothing (no demand scored, or tail) vs. reality is running one.
+    // Comparison with the live system. These are **observations, not a
+    // takeover gate**: the shadow planner has only demand and capacity,
+    // while the real queue also reflects ownership, eligibility, cooldowns
+    // and executors this planner cannot see. Each metric names one precise
+    // question so the answer can be interpreted:
+    //
+    //  discovery_miss   planner scored demand on a table that has no
+    //                   MergeWal task queued or running at all - the real
+    //                   discovery loops have not noticed it yet (or it is a
+    //                   tail they ignore by design).
+    //  placement_gap    planner found no executor headroom for a due table
+    //                   that reality is in fact running - the planner's
+    //                   capacity model is tighter than the real pools.
+    //  phantom_task     reality is running/queued a MergeWal task on a table
+    //                   the planner sees no demand for - either a lost
+    //                   demand event, or work the planner would not have
+    //                   scheduled.
+    //
+    // The old single "disagreement" counter is kept for continuity but is
+    // the sum of these and should not be read on its own.
     let placed: std::collections::BTreeSet<&str> =
         would_place.iter().map(|(t, _)| t.as_str()).collect();
-    let mut disagreements = 0u64;
+    let mut discovery_miss = 0u64;
+    let mut placement_gap = 0u64;
+    let mut agree = 0u64;
+    let mut scored_targets: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
     for s in &scored {
+        scored_targets.insert(s.target.as_str());
+        if s.class == Class::Tail {
+            continue;
+        }
         let active = state
             .task_store
             .get_active_id(TaskKind::MergeWal, &s.target)
@@ -534,15 +647,45 @@ async fn shadow_pass(
             .map_err(|e| e.to_string())?
             .is_some();
         let planner_wants = placed.contains(s.target.as_str());
-        if planner_wants != active {
-            disagreements += 1;
-            tracing::debug!(
-                target = %s.target, class = ?s.class, score = s.score,
-                planner_wants, reality_active = active,
-                "planner shadow disagreement"
-            );
+        match (planner_wants, active) {
+            (true, true) | (false, false) if planner_wants => agree += 1,
+            (true, false) => {
+                discovery_miss += 1;
+                tracing::debug!(target = %s.target, class = ?s.class, score = s.score,
+                    "shadow: planner would place, reality has no task");
+            }
+            (false, true) => {
+                placement_gap += 1;
+                tracing::debug!(target = %s.target, class = ?s.class, score = s.score,
+                    "shadow: due but unplaceable, reality is running it");
+            }
+            _ => {}
         }
     }
+    // Phantom tasks: live MergeWal tasks on tables with no scored demand.
+    let mut phantom_task = 0u64;
+    if let Ok(running) = state
+        .task_store
+        .list_active_targets(TaskKind::MergeWal)
+        .await
+    {
+        for target in running {
+            if !scored_targets.contains(target.as_str()) {
+                phantom_task += 1;
+                tracing::debug!(target = %target,
+                    "shadow: reality runs a MergeWal task with no scored demand");
+            }
+        }
+    }
+    for (label, n) in [
+        ("discovery_miss", discovery_miss),
+        ("placement_gap", placement_gap),
+        ("phantom_task", phantom_task),
+        ("agree", agree),
+    ] {
+        metrics::gauge!("planner_shadow_comparison", "outcome" => label).set(n as f64);
+    }
+    let disagreements = discovery_miss + placement_gap + phantom_task;
     metrics::counter!("scheduler_shadow_disagreements_total", "kind" => "merge_wal")
         .increment(disagreements);
     metrics::gauge!("planner_shadow_disagreements_last_pass").set(disagreements as f64);
@@ -613,6 +756,168 @@ mod tests {
         assert_eq!(unhex("zz"), None);
     }
 
+    /// Round-5 finding 1 at production shape: 13,405 tables, two shards
+    /// each (about 12 MB of events, three times the etcd client's 4 MiB
+    /// message cap). A single-shot prefix read fails; the paged read must
+    /// return every record at one revision, the fold must count them all,
+    /// and a reconcile must complete while the leader keeps renewing its
+    /// lease (the slow-reconcile-starves-keepalive finding).
+    /// Seeds ~27k keys and reads them back several times; on a shared
+    /// single-node etcd this starves concurrently running tests with
+    /// `request timed out`. Run alone: `cargo test -p lance-context-master
+    /// --lib reconcile_at_production_scale -- --ignored --test-threads=1`.
+    /// Gated on `PLANNER_SCALE_TEST=1` so the ordinary `--ignored` sweep in
+    /// CI does not include it.
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS and PLANNER_SCALE_TEST=1; run alone"]
+    async fn reconcile_at_production_scale_pages_reads_and_keeps_the_lease() {
+        if std::env::var("PLANNER_SCALE_TEST").is_err() {
+            eprintln!("PLANNER_SCALE_TEST unset; skipping");
+            return;
+        }
+        use crate::state::MasterState;
+        use clap::Parser;
+        const TABLES: usize = 13_405;
+        let dir = tempfile::tempdir().unwrap();
+        let mut cfg = crate::config::MasterConfig::parse_from([
+            "test",
+            "--data-dir",
+            dir.path().to_str().unwrap(),
+        ]);
+        cfg.etcd.etcd_endpoints = std::env::var("ETCD_TEST_ENDPOINTS")
+            .unwrap()
+            .split(',')
+            .map(str::to_owned)
+            .collect();
+        cfg.etcd.etcd_prefix = format!("/planner-scale/{}", generate_id());
+        cfg.planner.planner_enabled = true;
+        cfg.planner.planner_reconcile_secs = 5;
+        cfg.etcd_lease_ttl_secs = 5;
+        cfg.stats_scan_interval_secs = 0;
+        cfg.compaction_interval_secs = 0;
+        cfg.merge_wal_interval_secs = 0;
+        let state = MasterState::new(cfg.clone()).await.unwrap();
+        let keys = Keys::new(&cfg.etcd.etcd_prefix);
+        let mut client = state.task_store.etcd_client().clone();
+
+        // Seed events directly, in batched txns, with realistic bodies.
+        let mut ops = Vec::with_capacity(128);
+        for i in 0..TABLES {
+            let target = format!("exp-{i:05}");
+            let pending = (i % 50) as u64; // 354-ish tables with real backlog is
+                                           // fewer than this; worst case is fine
+            for shard in [
+                "aaaaaaaa-0000-0000-0000-000000000001",
+                "aaaaaaaa-0000-0000-0000-000000000002",
+            ] {
+                let event = DemandEvent {
+                    v: SCHEMA_VERSION,
+                    shard: shard.into(),
+                    sealed_through: 1000 + pending,
+                    sealed_bytes_through: 64 << 20,
+                    merged_through: Some(1000),
+                    flushed_at_ms: 1_700_000_000_000,
+                    writer_epoch: 3,
+                    source: EventSource::Scan,
+                    merged_epoch: None,
+                    sealed_times: (1001..=1000 + pending)
+                        .map(|g| (g, 1_700_000_000_000 + g as i64))
+                        .collect(),
+                };
+                ops.push(TxnOp::put(
+                    format!("{}{}/{shard}", keys.events_prefix(), hex(&target)),
+                    serde_json::to_vec(&event).unwrap(),
+                    None,
+                ));
+                if ops.len() == 100 {
+                    client
+                        .txn(Txn::new().and_then(std::mem::take(&mut ops)))
+                        .await
+                        .unwrap();
+                }
+            }
+        }
+        if !ops.is_empty() {
+            client.txn(Txn::new().and_then(ops)).await.unwrap();
+        }
+
+        // A single unpaged read of this prefix exceeds the client limit.
+        let single = client
+            .get(keys.events_prefix(), Some(GetOptions::new().with_prefix()))
+            .await;
+        assert!(
+            single.is_err(),
+            "expected the unpaged read to exceed the message cap; got {} kvs",
+            single.map(|r| r.kvs().len()).unwrap_or(0)
+        );
+
+        // The paged read returns everything at one revision.
+        let (kvs, revision) = read_prefix_paged(&client, &keys.events_prefix())
+            .await
+            .unwrap();
+        assert_eq!(kvs.len(), TABLES * 2);
+        assert!(revision > 0);
+        let (tables, skipped) = fold_events(
+            &keys,
+            kvs.iter()
+                .map(|kv| (kv.key.as_slice(), kv.value.as_slice(), kv.mod_revision)),
+            0,
+        );
+        assert_eq!(skipped, 0);
+        assert_eq!(tables.len(), TABLES);
+        let expected_pending: u64 = (0..TABLES as u64).map(|i| (i % 50) * 2).sum();
+        assert_eq!(
+            tables
+                .values()
+                .map(|t| t.pending_generations())
+                .sum::<u64>(),
+            expected_pending
+        );
+
+        // Lead for real: the reconcile runs detached, so the keepalive must
+        // keep firing and the leader key must survive a full pass.
+        let leader = tokio::spawn({
+            let state = state.clone();
+            async move { lead(&state).await }
+        });
+        let deadline = std::time::Instant::now() + Duration::from_secs(120);
+        loop {
+            let (cache, _) = read_prefix_paged(&client, &keys.demand_prefix())
+                .await
+                .unwrap();
+            if cache.len() == TABLES {
+                break;
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "cache incomplete: {}",
+                cache.len()
+            );
+            assert!(
+                !leader.is_finished(),
+                "leader loop ended during reconcile: {:?}",
+                leader.await
+            );
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+        // Lease is alive well past several TTLs.
+        tokio::time::sleep(Duration::from_secs(12)).await;
+        let lk = client.get(keys.leader(), None).await.unwrap();
+        assert_eq!(
+            lk.kvs().len(),
+            1,
+            "leader lost its lease during/after the pass"
+        );
+        assert!(!leader.is_finished());
+        leader.abort();
+        let _ = client
+            .delete(
+                cfg.etcd.etcd_prefix.as_str(),
+                Some(etcd_client::DeleteOptions::new().with_prefix()),
+            )
+            .await;
+    }
+
     #[tokio::test]
     #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn single_leader_builds_a_cache_equal_to_a_direct_fold_and_deposed_leader_cannot_write() {
diff --git a/crates/lance-context-master/src/scanner.rs b/crates/lance-context-master/src/scanner.rs
index bca92bbf..f950905e 100644
--- a/crates/lance-context-master/src/scanner.rs
+++ b/crates/lance-context-master/src/scanner.rs
@@ -347,10 +347,9 @@ async fn scan_once_inner(state: &Arc<MasterState>) -> lance::Result<usize> {
                         // Demand reconciliation rides the scan: publish every
                         // shard's watermark as a scan-sourced event. Detached
                         // and bounded; the scan never waits on it.
-                        let target = match entry.kind {
-                            ScanKind::Rollout => entry.name.clone(),
-                            ScanKind::Generic => format!("generic:{}", entry.name),
-                        };
+                        // `entry.name` is already the task target: bare for
+                        // rollout, `generic:<name>` for generic stores.
+                        let target = entry.name.clone();
                         crate::demand_publish::spawn_scan_demand(
                             state.clone(),
                             target,
diff --git a/crates/lance-context-master/src/scoring.rs b/crates/lance-context-master/src/scoring.rs
index c69a7bd8..1f876c81 100644
--- a/crates/lance-context-master/src/scoring.rs
+++ b/crates/lance-context-master/src/scoring.rs
@@ -86,7 +86,12 @@ pub fn score_merge(
     if pending == 0 {
         return None;
     }
-    let wait_ms = demand.oldest_pending_ms().map(|t| (now_ms - t).max(0));
+    // Only an exact oldest-pending time may drive aging and promotion. A
+    // shard whose age is known only as an upper bound (merged past the kept
+    // flush times) contributes no wait; its count still scores.
+    let wait_ms = demand
+        .oldest_pending_exact_ms()
+        .map(|t| (now_ms - t).max(0));
     let by_gens = pending as f64 / policy.min_generations.max(1) as f64;
     let by_bytes = if policy.min_bytes > 0 {
         bytes as f64 / policy.min_bytes as f64
@@ -236,6 +241,43 @@ mod tests {
         assert_eq!(v[0].target, "b", "class before score");
     }
 
+    /// Round-5 finding 6: when the oldest pending time is only an upper
+    /// bound (merged past the kept entries), it must not promote for age
+    /// and must not inflate the score.
+    #[test]
+    fn an_age_bound_never_promotes() {
+        let p = MergePolicy {
+            max_turn_wait_secs: 10,
+            max_age_secs: 10,
+            ..Default::default()
+        };
+        let now = 1_000_000_000;
+        let mut t = demand(128, now - 500_000);
+        let _ = t.fold(
+            &DemandEvent {
+                v: SCHEMA_VERSION,
+                shard: "s".into(),
+                sealed_through: 128,
+                sealed_bytes_through: 0,
+                merged_through: Some(64),
+                flushed_at_ms: 0,
+                writer_epoch: 1,
+                source: EventSource::Executor,
+                merged_epoch: None,
+                sealed_times: Default::default(),
+            },
+            999,
+            0,
+        );
+        assert_eq!(t.pending_generations(), 64);
+        assert!(t.oldest_pending_ms().is_some(), "a bound is shown");
+        assert_eq!(t.oldest_pending_exact_ms(), None, "but is not exact");
+        let s = score_merge("x", &t, &p, now).unwrap();
+        assert_eq!(s.wait_ms, None);
+        assert_eq!(s.class, Class::Normal, "64/8 = 8 >= 1, not promoted");
+        assert!((s.score - 8.0).abs() < 1e-9, "no age boost: {}", s.score);
+    }
+
     #[test]
     fn ordering_is_total_and_stable() {
         let p = MergePolicy::default();
diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs
index 935b2ebd..fe1f9cd7 100644
--- a/crates/lance-context-master/src/task_store.rs
+++ b/crates/lance-context-master/src/task_store.rs
@@ -545,6 +545,27 @@ impl TaskStore {
 
     /// Whether the sweeps should skip this target for now because it has
     /// failed repeatedly. Manual enqueues are not gated by this.
+    /// Targets with a `kind` task queued or running right now, read in
+    /// pages. For the shadow planner's comparison against reality.
+    pub async fn list_active_targets(&self, kind: TaskKind) -> lance::Result<Vec<String>> {
+        let mut out = Vec::new();
+        for prefix in [self.inner.queue_prefix(), self.inner.running_prefix()] {
+            let (kvs, _) = crate::planner::read_prefix_paged(&self.inner.client, &prefix)
+                .await
+                .map_err(lance::Error::io)?;
+            for kv in kvs {
+                if let Ok(task) = decode_task(&kv.value, &String::from_utf8_lossy(&kv.key)) {
+                    if task.kind == kind {
+                        out.push(task.target);
+                    }
+                }
+            }
+        }
+        out.sort();
+        out.dedup();
+        Ok(out)
+    }
+
     /// Id of the queued or running task that `enqueue(kind, target)` would
     /// dedupe into, if any. Lets a sweep tell a new task from a no-op.
     pub async fn get_active_id(
diff --git a/crates/lance-context-merge/src/demand.rs b/crates/lance-context-merge/src/demand.rs
index 31f034b1..8e7a3c87 100644
--- a/crates/lance-context-merge/src/demand.rs
+++ b/crates/lance-context-merge/src/demand.rs
@@ -120,6 +120,15 @@ pub struct ShardDemand {
     pub sealed_times: BTreeMap<u64, i64>,
 }
 
+/// How much to trust an oldest-pending time.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AgeBound {
+    Exact,
+    /// The true time is at or before this; never promote on it.
+    Upper,
+}
+
 impl ShardDemand {
     pub fn pending_generations(&self) -> u64 {
         self.sealed
@@ -127,22 +136,56 @@ impl ShardDemand {
             .saturating_sub(self.merged.map_or(0, |m| m.value))
     }
 
-    /// Flush time of the oldest generation still unmerged, if known.
-    pub fn oldest_pending_ms(&self) -> Option<i64> {
+    /// Flush time of the oldest generation still unmerged, with how much
+    /// to trust it.
+    ///
+    /// The record keeps flush times for at most `SEALED_TIMES_CAP` of the
+    /// lowest pending generations. While the oldest pending generation's
+    /// own time is kept the answer is **exact**. Once merging has consumed
+    /// every kept entry (a shard that had more than the cap pending), only
+    /// the newest sealed time is known; that is an **upper** bound on the
+    /// oldest pending flush time, so it must never drive age-based
+    /// promotion. Never `None` while something is pending.
+    pub fn oldest_pending(&self) -> Option<(i64, AgeBound)> {
         if self.pending_generations() == 0 {
             return None;
         }
-        self.sealed_times.values().next().copied()
+        let oldest = self.merged.map_or(0, |m| m.value) + 1;
+        if let Some(t) = self.sealed_times.get(&oldest) {
+            return Some((*t, AgeBound::Exact));
+        }
+        // Entries at or below merged are stale and ignored. The lowest kept
+        // entry above `oldest` is an upper bound on its flush time; the
+        // newest sealed generation is always kept (see `prune_times`), so
+        // there is always one to give.
+        self.sealed_times
+            .range(oldest..)
+            .next()
+            .map(|(_, t)| (*t, AgeBound::Upper))
     }
 
-    fn prune(&mut self) {
-        if let Some(m) = self.merged {
-            self.sealed_times.retain(|g, _| *g > m.value);
-        }
-        while self.sealed_times.len() > SEALED_TIMES_CAP {
-            self.sealed_times.pop_last();
+    /// Convenience: the time only, whatever its bound.
+    pub fn oldest_pending_ms(&self) -> Option<i64> {
+        self.oldest_pending().map(|(t, _)| t)
+    }
+
+    /// Exact oldest-pending time, or `None` when only a bound is known.
+    /// What the scheduler uses for promotion: a bound never promotes.
+    pub fn oldest_pending_exact_ms(&self) -> Option<i64> {
+        match self.oldest_pending() {
+            Some((t, AgeBound::Exact)) => Some(t),
+            _ => None,
         }
     }
+
+    /// Bound the map by cap **only**. Dropping entries at or below `merged`
+    /// here would make the kept set depend on whether the merge or the
+    /// flushes arrived first (merge-first keeps 65..128, flush-first keeps
+    /// 1..63 and 128), which breaks order independence. Entries at or below
+    /// merged are ignored at read time instead.
+    fn prune(&mut self) {
+        prune_times(&mut self.sealed_times);
+    }
 }
 
 /// The per-table record under `P/demand/<hex>`.
@@ -192,6 +235,23 @@ pub enum Ignored {
     UnknownVersion(u32),
 }
 
+/// Bound a times map: the lowest `SEALED_TIMES_CAP - 1` generations plus
+/// the single highest. Both parts are functions of the set, so the result
+/// is order independent. Keeping the highest means an upper bound on every
+/// pending age survives even when merging consumes all the low entries.
+fn prune_times(times: &mut BTreeMap<u64, i64>) {
+    if times.len() <= SEALED_TIMES_CAP {
+        return;
+    }
+    let newest = times.pop_last();
+    while times.len() > SEALED_TIMES_CAP - 1 {
+        times.pop_last();
+    }
+    if let Some((g, t)) = newest {
+        times.insert(g, t);
+    }
+}
+
 /// Insert `time` for `generation`, keeping the earliest. Returns whether the
 /// map changed.
 fn note_time(times: &mut BTreeMap<u64, i64>, generation: u64, time: i64) -> bool {
@@ -238,7 +298,8 @@ impl TableDemand {
             .sum()
     }
 
-    /// Flush time of the oldest generation still unmerged on any shard.
+    /// Flush time of the oldest generation still unmerged on any shard,
+    /// whatever its bound. For display.
     pub fn oldest_pending_ms(&self) -> Option<i64> {
         self.shards
             .values()
@@ -246,6 +307,15 @@ impl TableDemand {
             .min()
     }
 
+    /// Oldest **exact** pending time across shards. What promotion uses;
+    /// a shard whose age is only an upper bound contributes nothing.
+    pub fn oldest_pending_exact_ms(&self) -> Option<i64> {
+        self.shards
+            .values()
+            .filter_map(ShardDemand::oldest_pending_exact_ms)
+            .min()
+    }
+
     /// Fold one event. Returns `Err(reason)` when the record is unchanged.
     pub fn fold(
         &mut self,
@@ -282,8 +352,11 @@ impl TableDemand {
         // Age bookkeeping: this event's own flush time plus any times it
         // carries from a stored record, min per generation, only for
         // generations still above merged.
-        let floor = shard.merged.map_or(0, |m| m.value);
-        if event.sealed_through > floor
+        // An executor's flushed_at_ms is its own observation time, not when
+        // the generation was flushed; only writer and scan events seed the
+        // age clock from their own timestamp.
+        if event.source != EventSource::Executor
+            && event.sealed_through > 0
             && note_time(
                 &mut shard.sealed_times,
                 event.sealed_through,
@@ -293,7 +366,7 @@ impl TableDemand {
             changed = true;
         }
         for (g, t) in &event.sealed_times {
-            if *g > floor && note_time(&mut shard.sealed_times, *g, *t) {
+            if note_time(&mut shard.sealed_times, *g, *t) {
                 changed = true;
             }
         }
@@ -326,18 +399,14 @@ pub fn join_events(stored: Option<&DemandEvent>, event: &DemandEvent) -> Option<
         if fresh.merged_epoch == Some(fresh.writer_epoch) {
             fresh.merged_epoch = None;
         }
-        let floor = fresh.merged_through.unwrap_or(0);
-        if fresh.sealed_through > floor {
+        if fresh.source != EventSource::Executor && fresh.sealed_through > 0 {
             note_time(
                 &mut fresh.sealed_times,
                 fresh.sealed_through,
                 fresh.flushed_at_ms,
             );
         }
-        fresh.sealed_times.retain(|g, _| *g > floor);
-        while fresh.sealed_times.len() > SEALED_TIMES_CAP {
-            fresh.sealed_times.pop_last();
-        }
+        prune_times(&mut fresh.sealed_times);
         return Some(fresh);
     };
     let s_sealed = Mark {
@@ -369,23 +438,19 @@ pub fn join_events(stored: Option<&DemandEvent>, event: &DemandEvent) -> Option<
     };
 
     // Times: union of everything known, min per generation.
-    let floor = best_merged.map_or(0, |m| m.value);
     let mut times = stored.sealed_times.clone();
     for (g, t) in &event.sealed_times {
         note_time(&mut times, *g, *t);
     }
-    for (g, t) in [
-        (stored.sealed_through, stored.flushed_at_ms),
-        (event.sealed_through, event.flushed_at_ms),
+    for (src, g, t) in [
+        (stored.source, stored.sealed_through, stored.flushed_at_ms),
+        (event.source, event.sealed_through, event.flushed_at_ms),
     ] {
-        if g > floor {
+        if src != EventSource::Executor && g > 0 {
             note_time(&mut times, g, t);
         }
     }
-    times.retain(|g, _| *g > floor);
-    while times.len() > SEALED_TIMES_CAP {
-        times.pop_last();
-    }
+    prune_times(&mut times);
     let times_changed = times != stored.sealed_times;
 
     if !sealed_wins && !merged_wins && !times_changed {
@@ -401,9 +466,9 @@ pub fn join_events(stored: Option<&DemandEvent>, event: &DemandEvent) -> Option<
     next.merged_epoch = best_merged
         .map(|m| m.epoch)
         .filter(|e| *e != next.writer_epoch);
-    if merged_wins && !sealed_wins {
-        next.source = event.source;
-    }
+    // `source` and `flushed_at_ms` describe who set `sealed_through` and
+    // when it was flushed; they stay with the sealed mark so a rebuild
+    // re-derives exactly the time entries the fold did.
     next.sealed_times = times;
     next.v = SCHEMA_VERSION;
     Some(next)
@@ -701,11 +766,21 @@ mod tests {
             Err(Ignored::NotNewer)
         );
         // A "newer epoch" with a lower generation cannot occur in Lance; if
-        // reported it loses the max.
+        // reported it loses the max. Its flush time is recorded once (the
+        // map is bounded by cap only; see ShardDemand::prune) and is inert:
+        // generation 2 is below merged and never read.
+        table
+            .fold(&ev("a", 2, None, 4, EventSource::Scan), 5, 0)
+            .unwrap();
         assert_eq!(
-            table.fold(&ev("a", 2, None, 4, EventSource::Scan), 5, 0),
+            table.fold(&ev("a", 2, None, 4, EventSource::Scan), 6, 0),
             Err(Ignored::NotNewer)
         );
+        assert_eq!(
+            table.shards["a"].oldest_pending_ms(),
+            Some(150),
+            "gen 15's time, not gen 2's"
+        );
         assert_eq!(
             table.shards["a"].sealed,
             Mark {
@@ -776,12 +851,69 @@ mod tests {
         table.fold(&merged, 4, 0).unwrap();
         assert_eq!(table.oldest_pending_ms(), Some(2000));
         let stored = join_events(Some(&stored), &merged).unwrap();
-        assert_eq!(stored.sealed_times.values().next(), Some(&2000));
+        let mut from_stored = TableDemand::default();
+        from_stored.fold(&stored, 0, 0).unwrap();
+        assert_eq!(from_stored.oldest_pending_ms(), Some(2000));
         let all = [flush(1, 1000), flush(2, 2000), flush(3, 3000), merged];
         assert_eq!(rebuild(&all).oldest_pending_ms(), Some(2000));
         assert_order_independent(&all);
     }
 
+    /// Round-5 finding 6: 128 pending, merge the first 64. The oldest
+    /// pending is generation 65, whose time the cap never kept. The answer
+    /// must not be unknown and must not drive promotion as if exact: it is
+    /// reported as an Upper bound (the newest kept or sealed time), excluded
+    /// from `oldest_pending_exact_ms`, and identical after a rebuild.
+    #[test]
+    fn age_after_merging_past_the_cap_is_a_flagged_bound() {
+        let flush = |g: u64| DemandEvent {
+            flushed_at_ms: g as i64 * 1000,
+            ..ev("s", g, None, 1, EventSource::Writer)
+        };
+        let mut events: Vec<_> = (1..=128).map(flush).collect();
+        let table = fold_all(&events);
+        assert_eq!(
+            table.shards["s"].oldest_pending(),
+            Some((1000, AgeBound::Exact))
+        );
+        assert_eq!(table.oldest_pending_exact_ms(), Some(1000));
+        events.push(ev("s", 128, Some(64), 1, EventSource::Executor));
+        let table = fold_all(&events);
+        let (t, bound) = table.shards["s"]
+            .oldest_pending()
+            .expect("64 pending: not unknown");
+        assert_eq!(bound, AgeBound::Upper, "gen 65's time was never kept");
+        assert!(
+            t >= 65_000,
+            "an upper bound is at or after the true time: {t}"
+        );
+        assert_eq!(
+            table.oldest_pending_exact_ms(),
+            None,
+            "a bound never promotes"
+        );
+        assert_eq!(table.oldest_pending_ms(), Some(t), "but is shown");
+        assert_eq!(
+            rebuild(&events).shards["s"].oldest_pending(),
+            Some((t, AgeBound::Upper))
+        );
+        let mut reversed = events.clone();
+        reversed.reverse();
+        assert_eq!(
+            fold_all(&reversed).shards["s"].oldest_pending(),
+            Some((t, AgeBound::Upper))
+        );
+        // Below the cap the answer is exact again: 10 pending, merge 3 -> gen 4.
+        let small: Vec<_> = (1..=10)
+            .map(flush)
+            .chain([ev("s", 10, Some(3), 1, EventSource::Executor)])
+            .collect();
+        assert_eq!(
+            fold_all(&small).shards["s"].oldest_pending(),
+            Some((4000, AgeBound::Exact))
+        );
+    }
+
     #[test]
     fn beyond_cap_counts_stay_exact_and_age_is_a_bound() {
         let n = (SEALED_TIMES_CAP as u64) + 20;
diff --git a/docs/design/scheduler-p1-data-model.md b/docs/design/scheduler-p1-data-model.md
index 43c19cf6..865cca5f 100644
--- a/docs/design/scheduler-p1-data-model.md
+++ b/docs/design/scheduler-p1-data-model.md
@@ -56,11 +56,17 @@ Rules:
   epoch a merged mark was reported under when it differs from the record's
   `writer_epoch`; provenance and tiebreak only.
 * `oldest_pending_ms` is the flush time of the lowest unmerged generation. The
-  stored record carries `sealed_times` (bounded to `SEALED_TIMES_CAP = 64`
-  lowest unmerged generations, min time per generation, pruned as `merged`
-  advances) so the age clock **survives a rebuild from etcd**. A newer flush
-  adds an entry and never resets the clock. Counts and bytes are exact in every
-  order; the age field is exact up to the cap and a lower bound beyond it.
+  stored record carries `sealed_times`: the lowest `SEALED_TIMES_CAP − 1 = 63`
+  recorded generations plus the newest, min time per generation, bounded by
+  cap **only** (pruning by `merged` would make the kept set depend on arrival
+  order). Entries at or below `merged` are ignored on read. The age clock
+  therefore **survives a rebuild from etcd** and a newer flush never resets
+  it. The answer carries a bound: **Exact** when the oldest pending
+  generation's own time is kept; **Upper** (the lowest kept time above it,
+  at worst the newest sealed) once merging has consumed every kept entry on a
+  shard that had more than the cap pending. Only an Exact age drives aging
+  and promotion; a bound is displayed but never promotes. Counts and bytes
+  are exact in every order.
 * Records carry `v`. A reader rejects an event it does not understand
   (`Ignored::UnknownVersion`) and a publisher **refuses to overwrite a stored
   record from a newer schema**; `v == 0` is read as version 1.
@@ -88,13 +94,23 @@ source of truth, and may be rebuilt from events + scan at any time. Folding is
 a pure function `fold(table, event) -> table` and is tested for commutativity
 and idempotency (design invariant 5a).
 
+### 1.2a Reads at scale
+
+Production has ~13,400 tables and ~27,000 shard records. A single prefix
+read of `demand-events/` exceeds the etcd client's 4 MiB message cap. Every
+planner prefix read (`demand-events/`, `demand/`, `executors/`,
+`assignments/`, `queue/`, `running/`) is paged (512 keys) **at one
+revision**, so each pass sees a consistent snapshot. The reconcile runs
+detached from the leader loop so a slow pass can never starve the lease
+keepalive; at most one pass is in flight and skipped ticks are counted.
+
 ### 1.3 What writes what in P1
 
 | Source | Writes | Change from today |
 |---|---|---|
 | Rollout server flush (`sweeper.rs::route_merge`) | `demand-events/<hex>/<shard>` **in addition to** `merge-requests/<hex>` | additive |
 | Master append merge release (`rollout_append.rs`, `merge_execution.rs`) | `merged_through` per shard in the release txn | additive |
-| Stats scanner (`scanner.rs`) | `demand-events` with `source: scan` for every shard it observes, plus `fragment_count`/`index_stale` into `demand/` | additive; existing enqueue/retire paths untouched |
+| Stats scanner (`scanner.rs`) | `demand-events` with `source: scan` for every shard it observes, carrying **both** the shard manifest's `sealed_through` and the base table's merged watermark for that shard, so a lost completion event is corrected on the next scan and an old table's first contact does not read generation 1001 as 1001 pending | additive; existing enqueue/retire paths untouched |
 | Planner (new, leader only) | `demand/<hex>` | new |
 
 No existing key is removed or re-purposed in P1.
@@ -164,10 +180,35 @@ leadership). The planner scores `MergeWal` demand only (design §4.2: classes
 Critical / Normal / Tail; class 1 oldest‑first), places against headroom, and
 writes assignments **only in shadow mode**: state `shadow`, replaced wholesale
 every pass, never read by any executor (a source‑level test enforces this), no
-binding. Tails are not placed, matching the real sweeps. Each pass then
-compares its placements with whether a real `MergeWal` task is active per
-table and increments `scheduler_shadow_disagreements_total{kind}`; the
-per‑pass count is `planner_shadow_disagreements_last_pass`. Compaction and
+binding. Tails are not placed, matching the real sweeps.
+
+Each pass then compares itself with the live system. **These are
+observations, not a takeover gate.** The shadow planner sees demand and
+capacity; the real queue also reflects ownership, eligibility, cooldowns,
+and executors the planner cannot see (legacy workers, catch‑up Jobs). The
+comparison therefore answers three precise questions, each its own gauge
+`planner_shadow_comparison{outcome}`:
+
+| outcome | meaning |
+|---|---|
+| `discovery_miss` | planner scored due demand on a table with no `MergeWal` task queued or running — the real discovery loops have not noticed it |
+| `placement_gap` | planner found no headroom for a due table that reality is running — the planner's capacity model is tighter than the real pools |
+| `phantom_task` | reality has a `MergeWal` task on a table the planner sees no demand for — a lost demand event, or work the planner would not schedule |
+| `agree` | both would run it |
+
+`scheduler_shadow_disagreements_total{kind}` is the sum of the first three,
+kept for continuity; read the components. What "approach zero" means per
+component: `discovery_miss` falling to the tail population says the demand
+feed is complete; `phantom_task` at zero says no demand events are lost;
+`placement_gap` is expected to stay non‑zero until legacy workers and
+catch‑up Jobs heartbeat as executors, and is not a defect of the planner.
+
+Byte estimates: writers do not measure generation sizes, so `pending_bytes`
+is zero for most tables. An unknown reserves the **full per‑pass cost**
+(`2 × ROLLOUT_APPEND_MAX_BYTES`), never a token — a capacity model that
+reserves 1 byte validates nothing. `planner_shadow_bytes_unknown_total`
+counts these. Measured sizes come with the scan reading generation data
+statistics, a later change. Compaction and
 index scoring, commit‑ready class 2 and `max_consecutive_turns` come with P2's
 preparation integration; catch‑up Job heartbeats come when Jobs become an
 executor kind.