diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs
index 890971b..bf37e9d 100644
--- a/crates/lance-context-master/src/config.rs
+++ b/crates/lance-context-master/src/config.rs
@@ -16,6 +16,8 @@ pub struct MasterConfig {
     pub append: crate::rollout_append::AppendConfig,
     #[command(flatten)]
     pub wal_tail: crate::wal_tail::WalTailConfig,
+    #[command(flatten)]
+    pub planner: crate::planner::PlannerConfig,
     /// Data directory / object-store prefix shared with the data-plane server.
     #[arg(long, env = "DATA_DIR", default_value = "./data")]
     pub data_dir: String,
diff --git a/crates/lance-context-master/src/lib.rs b/crates/lance-context-master/src/lib.rs
index cbbc866..6d2aafd 100644
--- a/crates/lance-context-master/src/lib.rs
+++ b/crates/lance-context-master/src/lib.rs
@@ -4,6 +4,7 @@ pub mod admission;
 pub mod catchup;
 pub mod config;
 pub mod demand_publish;
+pub mod planner;
 pub use lance_context_merge::demand;
 pub mod discovery;
 pub mod eligibility;
diff --git a/crates/lance-context-master/src/planner.rs b/crates/lance-context-master/src/planner.rs
new file mode 100644
index 0000000..e210c97
--- /dev/null
+++ b/crates/lance-context-master/src/planner.rs
@@ -0,0 +1,593 @@
+//! Leader-elected demand planner, shadow mode (P1.5).
+//!
+//! Spec: `docs/design/scheduler-p1-data-model.md` §1.2, §3. One master holds
+//! `P/leader` under a lease and folds every `P/demand-events/<hex>/<shard>`
+//! record into a per-table `P/demand/<hex>` cache. The cache is a pure
+//! function of the stored events (see `demand::TableDemand::fold`), so it can
+//! be rebuilt from scratch by any leader and is never a source of truth.
+//!
+//! This module **places nothing and removes nothing**. Every existing loop
+//! keeps running for every table. Its only outputs are the `demand/` cache,
+//! the `/scheduler/demand` view and metrics. Scores, classes and shadow
+//! placements come in later PRs once this cache is trusted.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use etcd_client::{Compare, CompareOp, GetOptions, PutOptions, Txn, TxnOp};
+use lance_context_core::generate_id;
+use lance_context_merge::demand::{DemandEvent, TableDemand};
+use serde::{Deserialize, Serialize};
+
+use crate::state::MasterState;
+
+#[derive(Clone, Debug, clap::Args)]
+pub struct PlannerConfig {
+    /// Run the demand planner (leader election + demand cache). Shadow only:
+    /// nothing is scheduled from it yet.
+    #[arg(long, env = "PLANNER_ENABLED", default_value_t = false)]
+    pub planner_enabled: bool,
+    /// Full reconcile interval; between ticks the leader reacts to changes.
+    #[arg(long, env = "PLANNER_RECONCILE_SECS", default_value_t = 30)]
+    pub planner_reconcile_secs: u64,
+}
+
+impl Default for PlannerConfig {
+    fn default() -> Self {
+        Self {
+            planner_enabled: false,
+            planner_reconcile_secs: 30,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct LeaderRecord {
+    pub v: u32,
+    pub token: String,
+    pub instance: String,
+    pub since_ms: i64,
+}
+
+fn now_ms() -> i64 {
+    chrono::Utc::now().timestamp_millis()
+}
+
+fn hex(target: &str) -> String {
+    target.bytes().map(|b| format!("{b:02x}")).collect()
+}
+
+pub(crate) fn unhex(encoded: &str) -> Option<String> {
+    if !encoded.len().is_multiple_of(2) {
+        return None;
+    }
+    let bytes: Option<Vec<u8>> = (0..encoded.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&encoded[i..i + 2], 16).ok())
+        .collect();
+    String::from_utf8(bytes?).ok()
+}
+
+pub(crate) struct Keys {
+    prefix: String,
+}
+
+impl Keys {
+    pub(crate) fn new(prefix: &str) -> Self {
+        Self {
+            prefix: prefix.trim_end_matches('/').to_string(),
+        }
+    }
+    pub(crate) fn leader(&self) -> String {
+        format!("{}/leader", self.prefix)
+    }
+    /// Bare token beside the record so planner writes can compare on it.
+    pub(crate) fn leader_token(&self) -> String {
+        format!("{}/leader-token", self.prefix)
+    }
+    pub(crate) fn events_prefix(&self) -> String {
+        format!("{}/demand-events/", self.prefix)
+    }
+    pub(crate) fn demand_prefix(&self) -> String {
+        format!("{}/demand/", self.prefix)
+    }
+    pub(crate) fn demand(&self, target: &str) -> String {
+        format!("{}/demand/{}", self.prefix, hex(target))
+    }
+    /// `demand-events/<hex>/<shard>` → (target, shard).
+    fn parse_event_key(&self, key: &str) -> Option<(String, String)> {
+        let rest = key.strip_prefix(&self.events_prefix())?;
+        let (encoded, shard) = rest.split_once('/')?;
+        Some((unhex(encoded)?, shard.to_string()))
+    }
+}
+
+/// Fold every stored event into per-table records. Pure; the planner's one
+/// unit of work. Events whose key or body cannot be read are counted and
+/// skipped, never fatal.
+pub(crate) fn fold_events<'a>(
+    keys: &Keys,
+    kvs: impl Iterator<Item = (&'a [u8], &'a [u8], i64)>,
+    now: i64,
+) -> (BTreeMap<String, TableDemand>, usize) {
+    let mut tables: BTreeMap<String, TableDemand> = BTreeMap::new();
+    let mut skipped = 0usize;
+    for (key, value, revision) in kvs {
+        let Some((target, _shard)) =
+            keys.parse_event_key(std::str::from_utf8(key).unwrap_or_default())
+        else {
+            skipped += 1;
+            continue;
+        };
+        let Ok(event) = serde_json::from_slice::<DemandEvent>(value) else {
+            skipped += 1;
+            continue;
+        };
+        let table = tables.entry(target).or_default();
+        if let Err(lance_context_merge::demand::Ignored::UnknownVersion(_)) =
+            table.fold(&event, revision, now)
+        {
+            skipped += 1;
+        }
+    }
+    (tables, skipped)
+}
+
+pub(crate) fn spawn(state: &Arc<MasterState>) {
+    if !state.config.planner.planner_enabled {
+        return;
+    }
+    let weak = Arc::downgrade(state);
+    tokio::spawn(async move {
+        loop {
+            let Some(state) = weak.upgrade() else { return };
+            match lead(&state).await {
+                Ok(_) => {}
+                Err(error) => {
+                    tracing::warn!(%error, "planner leadership loop ended");
+                    metrics::counter!("planner_leader_errors_total").increment(1);
+                }
+            }
+            drop(state);
+            tokio::time::sleep(Duration::from_secs(2)).await;
+        }
+    });
+}
+
+/// Campaign for leadership; while leader, run reconcile cycles until the
+/// lease is lost or the process drains. Returns `Ok(true)` if this call held
+/// leadership at all, `Ok(false)` if the campaign was lost.
+async fn lead(state: &Arc<MasterState>) -> Result<bool, String> {
+    let keys = Keys::new(&state.config.etcd.etcd_prefix);
+    let mut client = state.task_store.etcd_client().clone();
+    let ttl = state.config.etcd_lease_ttl_secs.max(5);
+    let lease = client
+        .lease_grant(ttl, None)
+        .await
+        .map_err(|e| e.to_string())?
+        .id();
+    let token = generate_id();
+    let record = LeaderRecord {
+        v: 1,
+        token: token.clone(),
+        instance: state.admission.status().executor_id,
+        since_ms: now_ms(),
+    };
+    let won = client
+        .txn(
+            Txn::new()
+                .when(vec![Compare::version(
+                    keys.leader().as_str(),
+                    CompareOp::Equal,
+                    0,
+                )])
+                .and_then(vec![
+                    TxnOp::put(
+                        keys.leader(),
+                        serde_json::to_vec(&record).unwrap(),
+                        Some(PutOptions::new().with_lease(lease)),
+                    ),
+                    TxnOp::put(
+                        keys.leader_token(),
+                        token.clone(),
+                        Some(PutOptions::new().with_lease(lease)),
+                    ),
+                ]),
+        )
+        .await
+        .map_err(|e| e.to_string())?
+        .succeeded();
+    if !won {
+        let _ = client.lease_revoke(lease).await;
+        metrics::gauge!("planner_is_leader").set(0.0);
+        return Ok(false);
+    }
+    metrics::gauge!("planner_is_leader").set(1.0);
+    tracing::info!(token = %token, "planner leadership acquired");
+    let (mut keeper, mut stream) = client
+        .lease_keep_alive(lease)
+        .await
+        .map_err(|e| e.to_string())?;
+    let keepalive_every = Duration::from_secs((ttl as u64 / 3).max(1));
+    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 result = loop {
+        tokio::select! {
+            _ = keepalive.tick() => {
+                if keeper.keep_alive().await.is_err() {
+                    break Err("lease keepalive failed".to_string());
+                }
+                match tokio::time::timeout(Duration::from_secs(5), stream.message()).await {
+                    Ok(Ok(Some(response))) if response.ttl() > 0 => {}
+                    _ => break Err("lease expired or keepalive stream closed".to_string()),
+                }
+            }
+            _ = reconcile.tick() => {
+                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);
+                }
+            }
+        }
+    };
+    metrics::gauge!("planner_is_leader").set(0.0);
+    drop(keeper);
+    let _ = client.lease_revoke(lease).await;
+    tracing::info!(token = %token, ?result, "planner leadership released");
+    result
+}
+
+/// One reconcile: read every event, fold, write every changed table record
+/// under a leader-token compare, delete records for tables with no events.
+pub(crate) async fn reconcile_once(
+    state: &Arc<MasterState>,
+    keys: &Keys,
+    leader_token: &str,
+) -> 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 now = now_ms();
+    let (tables, skipped) = fold_events(
+        keys,
+        events
+            .kvs()
+            .iter()
+            .map(|kv| (kv.key(), kv.value(), kv.mod_revision())),
+        now,
+    );
+    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 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);
+        }
+    }
+    let mut ops = Vec::new();
+    let mut written = 0usize;
+    let mut total_pending = 0u64;
+    for (target, mut table) in tables {
+        total_pending += table.pending_generations();
+        let key = keys.demand(&target);
+        // Only the fold's inputs decide equality; the clock does not.
+        let unchanged = existing_by_key
+            .remove(key.as_bytes())
+            .is_some_and(|mut old| {
+                old.updated_ms = 0;
+                let mut cmp = table.clone();
+                cmp.updated_ms = 0;
+                old == cmp
+            });
+        if unchanged {
+            continue;
+        }
+        table.updated_ms = now;
+        ops.push(TxnOp::put(key, serde_json::to_vec(&table).unwrap(), None));
+        written += 1;
+    }
+    // Tables that no longer have any events: drop the cache row.
+    for stale_key in existing_by_key.into_keys() {
+        ops.push(TxnOp::delete(stale_key, None));
+        written += 1;
+    }
+    // Always confirm leadership, even with nothing to write, so a deposed
+    // leader learns it is deposed on the very next tick rather than when it
+    // next has a change.
+    let still_leader = client
+        .get(keys.leader_token(), None)
+        .await
+        .map_err(|e| e.to_string())?
+        .kvs()
+        .first()
+        .is_some_and(|kv| kv.value() == leader_token.as_bytes());
+    if !still_leader {
+        return Err("leadership lost during reconcile".into());
+    }
+    // etcd caps a txn at 128 ops by default; chunk, each guarded by the
+    // leader token so a deposed leader cannot write a stale cache.
+    for chunk in ops.chunks(100) {
+        let response = client
+            .txn(
+                Txn::new()
+                    .when(vec![Compare::value(
+                        keys.leader_token().as_str(),
+                        CompareOp::Equal,
+                        leader_token,
+                    )])
+                    .and_then(chunk.to_vec()),
+            )
+            .await
+            .map_err(|e| e.to_string())?;
+        if !response.succeeded() {
+            return Err("leadership lost during reconcile".into());
+        }
+    }
+    metrics::gauge!("planner_tables").set(existing.kvs().len() 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);
+    tracing::debug!(
+        written,
+        skipped,
+        revision = header_revision,
+        "planner reconcile"
+    );
+    Ok(written)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use lance_context_merge::demand::{EventSource, SCHEMA_VERSION};
+
+    fn event(shard: &str, sealed: u64, merged: Option<u64>) -> Vec<u8> {
+        serde_json::to_vec(&DemandEvent {
+            v: SCHEMA_VERSION,
+            shard: shard.into(),
+            sealed_through: sealed,
+            sealed_bytes_through: 0,
+            merged_through: merged,
+            flushed_at_ms: 1,
+            writer_epoch: 1,
+            source: EventSource::Writer,
+            merged_epoch: None,
+            sealed_times: Default::default(),
+        })
+        .unwrap()
+    }
+
+    #[test]
+    fn fold_events_groups_by_table_and_skips_garbage() {
+        let keys = Keys::new("/p");
+        let a1 = format!("/p/demand-events/{}/s1", hex("alpha"));
+        let a2 = format!("/p/demand-events/{}/s2", hex("alpha"));
+        let b1 = format!("/p/demand-events/{}/s1", hex("generic:beta"));
+        let bad_key = "/p/demand-events/zz/s1".to_string();
+        let bad_body = format!("/p/demand-events/{}/s9", hex("alpha"));
+        let future = serde_json::json!({"v": 999, "shard": "s", "sealed_through": 1,
+            "flushed_at_ms": 0, "writer_epoch": 1, "source": "writer"})
+        .to_string();
+        let future_key = format!("/p/demand-events/{}/s1", hex("gamma"));
+        let kvs: Vec<(Vec<u8>, Vec<u8>, i64)> = vec![
+            (a1.into_bytes(), event("s1", 10, Some(4)), 1),
+            (a2.into_bytes(), event("s2", 3, None), 2),
+            (b1.into_bytes(), event("s1", 7, Some(7)), 3),
+            (bad_key.into_bytes(), event("s1", 1, None), 4),
+            (bad_body.into_bytes(), b"not json".to_vec(), 5),
+            (future_key.into_bytes(), future.into_bytes(), 6),
+        ];
+        let (tables, skipped) = fold_events(
+            &keys,
+            kvs.iter().map(|(k, v, r)| (k.as_slice(), v.as_slice(), *r)),
+            0,
+        );
+        assert_eq!(skipped, 3, "bad key, bad body, future version");
+        assert_eq!(tables.len(), 3, "alpha, generic:beta, gamma (empty)");
+        assert_eq!(tables["alpha"].pending_generations(), 6 + 3);
+        assert_eq!(tables["alpha"].shards.len(), 2);
+        assert_eq!(tables["generic:beta"].pending_generations(), 0);
+        assert_eq!(tables["alpha"].observed_revision, 2);
+        assert!(tables["gamma"].shards.is_empty());
+    }
+
+    #[test]
+    fn hex_round_trips_and_rejects_odd_input() {
+        for s in ["alpha", "generic:beta", "", "ünïcödé"] {
+            assert_eq!(unhex(&hex(s)).as_deref(), Some(s));
+        }
+        assert_eq!(unhex("abc"), None);
+        assert_eq!(unhex("zz"), None);
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn single_leader_builds_a_cache_equal_to_a_direct_fold_and_deposed_leader_cannot_write() {
+        use crate::state::MasterState;
+        use clap::Parser;
+        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-test/{}", generate_id());
+        cfg.planner.planner_enabled = true;
+        cfg.planner.planner_reconcile_secs = 1;
+        cfg.stats_scan_interval_secs = 0;
+        cfg.compaction_interval_secs = 0;
+        cfg.merge_wal_interval_secs = 0;
+        let first = MasterState::new(cfg.clone()).await.unwrap();
+        let second = MasterState::new(cfg.clone()).await.unwrap();
+        let keys = Keys::new(&cfg.etcd.etcd_prefix);
+
+        // Two tables, one with two shards, published through the real path.
+        let coordinator = first.task_store.merge_coordinator();
+        for (target, shard, sealed, merged) in [
+            ("hot", "a", 40u64, Some(10u64)),
+            ("hot", "b", 12, None),
+            ("cold", "a", 3, Some(3)),
+        ] {
+            coordinator
+                .publish_demand_event(
+                    target,
+                    &DemandEvent {
+                        v: SCHEMA_VERSION,
+                        shard: shard.into(),
+                        sealed_through: sealed,
+                        sealed_bytes_through: sealed * 100,
+                        merged_through: merged,
+                        flushed_at_ms: 1000,
+                        writer_epoch: 1,
+                        source: EventSource::Writer,
+                        merged_epoch: None,
+                        sealed_times: Default::default(),
+                    },
+                )
+                .await
+                .unwrap();
+        }
+
+        // Both campaign; exactly one wins. The loser returns Ok(false)
+        // promptly; the winner keeps leading until aborted.
+        let a = tokio::spawn({
+            let first = first.clone();
+            async move { lead(&first).await }
+        });
+        let b = tokio::spawn({
+            let second = second.clone();
+            async move { lead(&second).await }
+        });
+        let mut client = first.task_store.etcd_client().clone();
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        let cache = loop {
+            let got = client
+                .get(keys.demand_prefix(), Some(GetOptions::new().with_prefix()))
+                .await
+                .unwrap();
+            if got.kvs().len() == 2 {
+                break got;
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "planner never wrote the cache"
+            );
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        };
+        let leader_kv = client.get(keys.leader(), None).await.unwrap();
+        assert_eq!(leader_kv.kvs().len(), 1, "exactly one leader");
+        let leader: LeaderRecord = serde_json::from_slice(leader_kv.kvs()[0].value()).unwrap();
+
+        // Cache equals a direct fold of the events.
+        let events = client
+            .get(keys.events_prefix(), Some(GetOptions::new().with_prefix()))
+            .await
+            .unwrap();
+        let (direct, skipped) = fold_events(
+            &keys,
+            events
+                .kvs()
+                .iter()
+                .map(|kv| (kv.key(), kv.value(), kv.mod_revision())),
+            0,
+        );
+        assert_eq!(skipped, 0);
+        for kv in cache.kvs() {
+            let mut stored: TableDemand = serde_json::from_slice(kv.value()).unwrap();
+            let target = unhex(
+                String::from_utf8_lossy(kv.key())
+                    .strip_prefix(&keys.demand_prefix())
+                    .unwrap(),
+            )
+            .unwrap();
+            stored.updated_ms = 0;
+            let mut expect = direct[&target].clone();
+            expect.updated_ms = 0;
+            assert_eq!(stored, expect, "{target}");
+        }
+        let hot: TableDemand = serde_json::from_slice(
+            client.get(keys.demand("hot"), None).await.unwrap().kvs()[0].value(),
+        )
+        .unwrap();
+        assert_eq!(hot.pending_generations(), 30 + 12);
+
+        // Via the HTTP view.
+        let axum::Json(report) = crate::routes::scheduler_demand(
+            axum::extract::State(first.clone()),
+            axum::extract::Query(crate::routes::DemandQuery {
+                target: Some("hot".into()),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            report.leader.as_ref().map(|l| &l.token),
+            Some(&leader.token)
+        );
+        assert_eq!(report.tables.len(), 1);
+        assert_eq!(report.tables[0].pending_generations, 42);
+        assert_eq!(report.tables[0].shards, 2);
+
+        // A deposed leader's reconcile is refused: forge a different token.
+        let err = reconcile_once(&first, &keys, "not-the-leader")
+            .await
+            .unwrap_err();
+        assert!(err.contains("leadership lost"), "{err}");
+
+        // Dropping the cold table's events removes its cache row on the next
+        // reconcile; hot stays.
+        client
+            .delete(
+                format!("{}{}/", keys.events_prefix(), hex("cold")),
+                Some(etcd_client::DeleteOptions::new().with_prefix()),
+            )
+            .await
+            .unwrap();
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            let got = client
+                .get(keys.demand_prefix(), Some(GetOptions::new().with_prefix()))
+                .await
+                .unwrap();
+            if got.kvs().len() == 1 {
+                break;
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "stale cache row not dropped"
+            );
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        // Exactly one campaigner lost (finished with Ok(false)); the other is
+        // still leading. If both led, both would still be running.
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        let finished = [a.is_finished(), b.is_finished()];
+        assert_eq!(
+            finished.iter().filter(|f| **f).count(),
+            1,
+            "exactly one campaigner must have lost and returned (finished={finished:?})"
+        );
+        let (loser, winner) = if finished[0] { (a, b) } else { (b, a) };
+        assert_eq!(loser.await.unwrap(), Ok(false));
+        winner.abort();
+    }
+}
diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs
index 9af5fdb..bdd2a12 100644
--- a/crates/lance-context-master/src/routes.rs
+++ b/crates/lance-context-master/src/routes.rs
@@ -572,6 +572,93 @@ pub async fn list_cooldowns(
         .map_err(MasterError::from_lance)
 }
 
+#[derive(Deserialize, Default)]
+pub struct DemandQuery {
+    pub target: Option<String>,
+}
+
+/// One table's demand as the planner sees it.
+#[derive(Debug, Serialize)]
+pub struct DemandView {
+    pub target: String,
+    pub pending_generations: u64,
+    pub pending_bytes: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub oldest_pending_ms: Option<i64>,
+    pub shards: usize,
+    pub observed_revision: i64,
+    pub updated_ms: i64,
+    pub record: lance_context_merge::demand::TableDemand,
+}
+
+#[derive(Debug, Serialize)]
+pub struct DemandReport {
+    /// Whether this master currently holds planner leadership.
+    pub leader: Option<crate::planner::LeaderRecord>,
+    pub tables: Vec<DemandView>,
+}
+
+/// `GET /api/v1/scheduler/demand[?target=]` — the planner's demand cache,
+/// folded from per-shard watermark events. Read-only; this is what the
+/// planner would schedule from. Absent while no planner has run.
+pub async fn scheduler_demand(
+    State(state): State<Arc<MasterState>>,
+    Query(query): Query<DemandQuery>,
+) -> Result<Json<DemandReport>, MasterError> {
+    let keys = crate::planner::Keys::new(&state.config.etcd.etcd_prefix);
+    let mut client = state.task_store.etcd_client().clone();
+    let leader = client
+        .get(keys.leader(), None)
+        .await
+        .map_err(|e| MasterError::Internal(e.to_string()))?
+        .kvs()
+        .first()
+        .and_then(|kv| serde_json::from_slice(kv.value()).ok());
+    let response = match &query.target {
+        Some(target) => {
+            if target.is_empty() || target.len() > 512 {
+                return Err(MasterError::InvalidRequest("invalid target".into()));
+            }
+            client.get(keys.demand(target), None).await
+        }
+        None => {
+            client
+                .get(
+                    keys.demand_prefix(),
+                    Some(etcd_client::GetOptions::new().with_prefix()),
+                )
+                .await
+        }
+    }
+    .map_err(|e| MasterError::Internal(e.to_string()))?;
+    let prefix = keys.demand_prefix();
+    let mut tables = Vec::new();
+    for kv in response.kvs() {
+        let Ok(record) =
+            serde_json::from_slice::<lance_context_merge::demand::TableDemand>(kv.value())
+        else {
+            continue;
+        };
+        let key = String::from_utf8_lossy(kv.key());
+        let target = key
+            .strip_prefix(prefix.as_str())
+            .and_then(crate::planner::unhex)
+            .unwrap_or_else(|| key.to_string());
+        tables.push(DemandView {
+            target,
+            pending_generations: record.pending_generations(),
+            pending_bytes: record.pending_bytes(),
+            oldest_pending_ms: record.oldest_pending_ms(),
+            shards: record.shards.len(),
+            observed_revision: record.observed_revision,
+            updated_ms: record.updated_ms,
+            record,
+        });
+    }
+    tables.sort_by_key(|t| std::cmp::Reverse(t.pending_generations));
+    Ok(Json(DemandReport { leader, tables }))
+}
+
 /// `GET /api/v1/scheduler/repairs` — base-table repairs the master performed,
 /// most recent first: which fragments were dropped from which store, when,
 /// and how many rows they held.
@@ -835,6 +922,7 @@ pub fn api_router() -> Router<Arc<MasterState>> {
             post(crate::catchup::trigger).get(crate::catchup::status),
         )
         .route("/scheduler/cooldowns", get(list_cooldowns))
+        .route("/scheduler/demand", get(scheduler_demand))
         .route("/scheduler/repairs", get(list_repairs))
         .route("/registry/diff", get(registry_diff))
         .route("/registry/backfill", post(registry_backfill))
@@ -901,6 +989,7 @@ mod tests {
             append: Default::default(),
             catchup: Default::default(),
             wal_tail: Default::default(),
+            planner: Default::default(),
             maintenance: Default::default(),
             merge_rollout: Default::default(),
             data_dir: dir.path().to_string_lossy().to_string(),
diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs
index d6d75ba..15b4d2a 100644
--- a/crates/lance-context-master/src/scheduler.rs
+++ b/crates/lance-context-master/src/scheduler.rs
@@ -979,6 +979,7 @@ pub(crate) async fn enqueue_merge_request(
 pub fn spawn_scheduler(state: &Arc<MasterState>) -> JoinHandle<()> {
     crate::wal_tail::spawn(state);
     crate::resident_recovery::spawn(state);
+    crate::planner::spawn(state);
     // Retry metadata independently of coarse stats sweeps. Every replica may
     // enqueue; the existing task dedupe/claim transaction elects one executor.
     let retry_state = state.clone();
@@ -1292,6 +1293,7 @@ mod tests {
             append: Default::default(),
             catchup: Default::default(),
             wal_tail: Default::default(),
+            planner: Default::default(),
             maintenance: Default::default(),
             merge_rollout: lance_context_merge::rollout::MergeRollout {
                 owned_targets: ["exp", "generic:gs", "broken"]
diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs
index 9779c91..e4d0c12 100644
--- a/crates/lance-context-master/src/state.rs
+++ b/crates/lance-context-master/src/state.rs
@@ -322,6 +322,7 @@ mod tests {
             append: Default::default(),
             catchup: Default::default(),
             wal_tail: Default::default(),
+            planner: Default::default(),
             maintenance: Default::default(),
             merge_rollout: Default::default(),
             data_dir: dir.path().to_string_lossy().to_string(),
diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs
index b301943..935b2eb 100644
--- a/crates/lance-context-master/src/task_store.rs
+++ b/crates/lance-context-master/src/task_store.rs
@@ -2309,6 +2309,7 @@ mod tests {
             append: Default::default(),
             catchup: Default::default(),
             wal_tail: Default::default(),
+            planner: Default::default(),
             maintenance: Default::default(),
             merge_rollout: Default::default(),
             data_dir: dir.path().to_string_lossy().to_string(),
diff --git a/docs/design/scheduler-p1-data-model.md b/docs/design/scheduler-p1-data-model.md
index 8443839..d6b35f7 100644
--- a/docs/design/scheduler-p1-data-model.md
+++ b/docs/design/scheduler-p1-data-model.md
@@ -167,11 +167,17 @@ executor, garbage-collected after 10 min. No executor binds anything in P1.
 ## 3. Leader
 
 ```
-P/leader   (leased)   { "v": 1, "token": "<uuid>", "instance": "...", "since_ms": ... }
+P/leader         (leased)  { "v": 1, "token": "<uuid>", "instance": "...", "since_ms": ... }
+P/leader-token   (leased)  "<uuid>"      // bare token so writes can compare on it
 ```
 
-Campaign with `Version(P/leader) == 0`; every planner write compares
-`P/leader == own token`. Followers watch `demand-events/`, `demand/`,
+Campaign with `Version(P/leader) == 0`, writing both keys under one lease in
+one txn. Every planner write txn compares `P/leader-token == own token`, and
+every reconcile reads it first even when there is nothing to write, so a
+deposed leader notices on its next tick, not on its next change. Keepalive
+every `TTL/3`; a failed or expired keepalive ends leadership and the loop
+re-campaigns after a short pause. `PLANNER_ENABLED` (default off) gates all of
+this; `PLANNER_RECONCILE_SECS` (default 30) sets the full-fold cadence. Followers watch `demand-events/`, `demand/`,
 `executors/`, `assignments/`, `merge-executions/`, `target-locks/` to keep a
 warm cache. The leader's cycle is triggered by those watches plus a 1 s tick.