diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs
index aaa7a35..4ddff56 100644
--- a/crates/lance-context-core/src/generic_store.rs
+++ b/crates/lance-context-core/src/generic_store.rs
@@ -35,7 +35,9 @@ use lance_index::mem_wal::ShardManifest;
 use crate::generic_codec::{batch_to_rows, rows_to_batch, Row};
 use crate::merge_budget::MergeMemoryBudget;
 use crate::store::{CompactionConfig, CompactionStats};
-use crate::store_base::{ListSource, PreparedMerge, StorageBase, StorageBaseOptions};
+use crate::store_base::{
+    is_not_found_error, ListSource, PreparedMerge, StorageBase, StorageBaseOptions,
+};
 use lance_context_api::schema_spec::{SchemaSpec, ID_COLUMN};
 use lance_context_api::MAX_BATCH_GET_IDS;
 
@@ -444,26 +446,38 @@ impl GenericStore {
         if let Some(filter) = filter {
             // Keep point/filtered reads on this handle's view. The server's
             // refresh-on-miss path refreshes the cached handle itself; reading
-            // a newer clone here would bypass that cache update.
-            let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
-            let mut scanner = self
-                .base
-                .lsm_scanner()
-                .await?
-                .project(&refs)?
-                .filter(filter)?;
-            if limit.is_some() || offset.is_some() {
-                scanner = scanner.limit(
-                    map_i64_bound("limit", limit)?,
-                    map_i64_bound("offset", offset)?,
-                )?;
+            // a newer clone on every lookup would bypass that cache update.
+            let result = async {
+                let snapshots = self.base.wal_shard_snapshots().await?;
+                self.scan_filtered_snapshot(
+                    self.base.dataset.clone(),
+                    snapshots,
+                    filter,
+                    limit,
+                    offset,
+                    columns,
+                )
+                .await
             }
-            let mut stream = scanner.try_into_stream().await?;
-            let mut rows = Vec::new();
-            while let Some(batch) = stream.try_next().await? {
-                rows.extend(batch_to_rows(&self.spec, &batch)?);
+            .await;
+            if !self.base.is_version_pinned()
+                && matches!(&result, Err(error) if is_not_found_error(error)
+                    && !matches!(error, LanceError::InvalidInput { .. } | LanceError::SchemaMismatch { .. }))
+            {
+                // GC can remove an opened generation during the scan. An
+                // error is not an absent row: retry a fresh consistent view.
+                // Successful reads and ordinary misses retain cache semantics.
+                return self
+                    .base
+                    .read_consistent(ListSource::All, |dataset, snapshots| {
+                        self.scan_filtered_snapshot(
+                            dataset, snapshots, filter, limit, offset, columns,
+                        )
+                    })
+                    .boxed()
+                    .await;
             }
-            return Ok(rows);
+            return result;
         }
 
         // Only unfiltered lists use this compatibility recovery. Keep retries
@@ -509,6 +523,34 @@ impl GenericStore {
             .await
     }
 
+    async fn scan_filtered_snapshot(
+        &self,
+        dataset: lance::Dataset,
+        snapshots: Vec<ShardSnapshot>,
+        filter: &str,
+        limit: Option<usize>,
+        offset: Option<usize>,
+        columns: &[String],
+    ) -> LanceResult<Vec<Row>> {
+        let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
+        let mut scanner =
+            StorageBase::lsm_scanner_for_dataset(&dataset, ID_COLUMN, ListSource::All, snapshots)
+                .project(&refs)?
+                .filter(filter)?;
+        if limit.is_some() || offset.is_some() {
+            scanner = scanner.limit(
+                map_i64_bound("limit", limit)?,
+                map_i64_bound("offset", offset)?,
+            )?;
+        }
+        let mut stream = scanner.try_into_stream().await?;
+        let mut rows = Vec::new();
+        while let Some(batch) = stream.try_next().await? {
+            rows.extend(batch_to_rows(&self.spec, &batch)?);
+        }
+        Ok(rows)
+    }
+
     /// Narrow compatibility path for Lance 9's broken batched PK sidecars.
     /// Physical scans supply only candidate keys; the native point planner
     /// still decides visibility/newest-wins, including replay and tombstones.
@@ -850,6 +892,68 @@ mod tests {
     use serde_json::json;
     use tempfile::TempDir;
 
+    #[tokio::test]
+    async fn filtered_point_snapshot_recovers_peer_gc_and_preserves_filter() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let dir = TempDir::new().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let writer = GenericStore::open(uri, spec(), sealing()).await.unwrap();
+        writer
+            .add(&[
+                row(json!({"id":"wanted", "user_id":"value"})),
+                row(json!({"id":"other", "user_id":"hidden"})),
+            ])
+            .await
+            .unwrap();
+        let reader = GenericStore::open_existing(uri, Default::default())
+            .await
+            .unwrap();
+        let writer = tokio::sync::Mutex::new(writer);
+        let attempts = AtomicUsize::new(0);
+        let columns = vec!["id".to_string(), "user_id".to_string()];
+        let rows = reader
+            .base
+            .read_consistent(ListSource::All, |dataset, snapshots| {
+                let reader = &reader;
+                let writer = &writer;
+                let attempts = &attempts;
+                let columns = &columns;
+                async move {
+                    if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
+                        writer.lock().await.cleanup_wal().await.unwrap();
+                        for snapshot in &snapshots {
+                            for generation in &snapshot.sstables {
+                                let path = reader
+                                    .base
+                                    .flushed_generation_uri(snapshot.shard_id, &generation.path);
+                                tokio::time::timeout(std::time::Duration::from_secs(2), async {
+                                    while std::path::Path::new(&path).exists() {
+                                        tokio::task::yield_now().await;
+                                    }
+                                })
+                                .await
+                                .unwrap();
+                            }
+                        }
+                    }
+                    reader
+                        .scan_filtered_snapshot(
+                            dataset,
+                            snapshots,
+                            "id = 'wanted'",
+                            Some(1),
+                            None,
+                            columns,
+                        )
+                        .await
+                }
+            })
+            .await
+            .unwrap();
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(rows, vec![row(json!({"id":"wanted", "user_id":"value"}))]);
+    }
+
     fn spec() -> SchemaSpec {
         SchemaSpec::new(vec![
             (
diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs
index 40aafdc..8b0f692 100644
--- a/crates/lance-context-core/src/rollout_store.rs
+++ b/crates/lance-context-core/src/rollout_store.rs
@@ -66,7 +66,7 @@ use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use datafusion::sql::parser::{DFParser, Statement as DFStatement};
 use datafusion::sql::sqlparser::ast::Statement as SqlStatement;
-use futures::{stream, StreamExt, TryStreamExt};
+use futures::{stream, FutureExt, StreamExt, TryStreamExt};
 use lance::dataset::mem_wal::{LsmScanner, ShardManifestStore, ShardSnapshot};
 use lance::dataset::optimize::CompactionMetrics;
 use lance::dataset::scanner::MaterializationStyle;
@@ -1555,21 +1555,37 @@ impl RolloutStore {
         id: &str,
         source: ListSource,
     ) -> LanceResult<Option<RolloutRecord>> {
-        let shard_snapshots = match source {
-            ListSource::Fragments => Vec::new(),
-            ListSource::Wal | ListSource::All => self.wal_shard_snapshots().await?,
-        };
+        if source == ListSource::Fragments {
+            // Keep the cached base hit cheap. A WAL-only miss still reaches
+            // the server's existing refresh-on-miss path, updating its cache.
+            return self
+                .scan_one_by_id_snapshot(id, source, self.base.dataset.clone(), Vec::new())
+                .await;
+        }
+        self.base
+            .read_consistent(source, |dataset, snapshots| {
+                self.scan_one_by_id_snapshot(id, source, dataset, snapshots)
+            })
+            .boxed()
+            .await
+    }
+
+    async fn scan_one_by_id_snapshot(
+        &self,
+        id: &str,
+        source: ListSource,
+        dataset: Dataset,
+        shard_snapshots: Vec<ShardSnapshot>,
+    ) -> LanceResult<Option<RolloutRecord>> {
         let escaped_id = id.replace('\'', "''");
-        let columns = self
-            .non_blob_columns()
+        let columns = Self::non_blob_columns_for(&dataset)
             .into_iter()
             .filter(|column| {
                 source == ListSource::Fragments || !CLAIM_CHECK_COLUMNS.contains(&column.as_str())
             })
             .collect::<Vec<_>>();
         let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
-        let scanner = self
-            .lsm_scanner_for_source(source, shard_snapshots)
+        let scanner = StorageBase::lsm_scanner_for_dataset(&dataset, "id", source, shard_snapshots)
             .project(&refs)?
             .filter(&format!("id = '{}'", escaped_id))?;
         let mut stream = scanner.try_into_stream().await?;
@@ -2525,6 +2541,65 @@ mod tests {
         assert_eq!(attempts.load(Ordering::SeqCst), 2);
     }
 
+    #[tokio::test]
+    async fn point_snapshot_recovers_peer_gc_without_changing_wal_only_semantics() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        for source in [ListSource::All, ListSource::Wal] {
+            let dir = tempfile::tempdir().unwrap();
+            let uri = dir.path().to_str().unwrap();
+            let writer = RolloutStore::open(uri).await.unwrap();
+            writer.add(&[assistant_record("point")]).await.unwrap();
+            writer.flush().await.unwrap();
+            let reader = RolloutStore::open(uri).await.unwrap();
+            let writer = tokio::sync::Mutex::new(writer);
+            let attempts = AtomicUsize::new(0);
+            let record = reader
+                .base
+                .read_consistent(source, |dataset, snapshots| {
+                    let reader = &reader;
+                    let writer = &writer;
+                    let attempts = &attempts;
+                    async move {
+                        if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
+                            writer.lock().await.cleanup_own_shard().await.unwrap();
+                            for snapshot in &snapshots {
+                                for generation in &snapshot.sstables {
+                                    let path = reader.flushed_generation_uri(
+                                        snapshot.shard_id,
+                                        &generation.path,
+                                    );
+                                    tokio::time::timeout(
+                                        std::time::Duration::from_secs(2),
+                                        async {
+                                            while std::path::Path::new(&path).exists() {
+                                                tokio::task::yield_now().await;
+                                            }
+                                        },
+                                    )
+                                    .await
+                                    .unwrap();
+                                }
+                            }
+                        }
+                        reader
+                            .scan_one_by_id_snapshot("point", source, dataset, snapshots)
+                            .await
+                    }
+                })
+                .await
+                .unwrap();
+            assert_eq!(attempts.load(Ordering::SeqCst), 2);
+            if source == ListSource::All {
+                assert_eq!(record.unwrap().id, "point");
+            } else {
+                assert!(
+                    record.is_none(),
+                    "WAL-only must not return the merged base row"
+                );
+            }
+        }
+    }
+
     #[tokio::test]
     async fn completed_list_snapshot_survives_later_merge_commits() {
         use std::sync::atomic::{AtomicUsize, Ordering};
@@ -2692,6 +2767,71 @@ mod tests {
         assert_eq!(attempts.load(Ordering::SeqCst), 2);
     }
 
+    #[tokio::test]
+    async fn list_survives_several_progressing_peer_merges() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let writer = RolloutStore::open_with_options(
+            uri,
+            RolloutStoreOptions {
+                shard_id: Some("continuous-merge-writer".into()),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+        writer.add(&[assistant_record("row-0")]).await.unwrap();
+        writer.flush().await.unwrap();
+        let reader = RolloutStore::open(uri).await.unwrap();
+        let writer = tokio::sync::Mutex::new(writer);
+        let attempts = AtomicUsize::new(0);
+
+        let rows = reader
+            .base
+            .read_consistent(ListSource::All, |dataset, snapshots| {
+                let writer = &writer;
+                let reader = &reader;
+                let attempts = &attempts;
+                async move {
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    if attempt < 4 {
+                        // Each failed snapshot loses a DIFFERENT generation to
+                        // a real successful merge. This is useful progress, not
+                        // a permanently missing file to retry indefinitely.
+                        let mut writer = writer.lock().await;
+                        assert_eq!(writer.cleanup_own_shard().await.unwrap(), 1);
+                        for snapshot in &snapshots {
+                            for generation in &snapshot.sstables {
+                                let path = reader
+                                    .flushed_generation_uri(snapshot.shard_id, &generation.path);
+                                tokio::time::timeout(std::time::Duration::from_secs(2), async {
+                                    while std::path::Path::new(&path).exists() {
+                                        tokio::time::sleep(std::time::Duration::from_millis(1))
+                                            .await;
+                                    }
+                                })
+                                .await
+                                .expect("the actual merge GC must invalidate this snapshot");
+                            }
+                        }
+                        writer
+                            .add(&[assistant_record(&format!("row-{}", attempt + 1))])
+                            .await
+                            .unwrap();
+                        writer.flush().await.unwrap();
+                    }
+                    reader.list_non_blob_snapshot(dataset, snapshots).await
+                }
+            })
+            .await
+            .expect("healthy merge progress must not exhaust a three-attempt read limit");
+        assert_eq!(attempts.load(Ordering::SeqCst), 5);
+        let ids: HashSet<_> = rows.into_iter().map(|row| row.id).collect();
+        assert_eq!(ids, (0..5).map(|i| format!("row-{i}")).collect());
+    }
+
     #[tokio::test]
     async fn consistent_read_bounds_missing_generation_retries() {
         use std::sync::atomic::{AtomicUsize, Ordering};
diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs
index f3e01f9..54e164d 100644
--- a/crates/lance-context-core/src/store_base.rs
+++ b/crates/lance-context-core/src/store_base.rs
@@ -2299,7 +2299,10 @@ impl StorageBase {
     /// read of that snapshot. Retry missing generations even when the base
     /// version is unchanged; never return a partial result after one disappears.
     /// Ordinary WAL appends do not force retries. Pinned reads use only their
-    /// selected base version.
+    /// selected base version. A progressing merge may invalidate several fresh
+    /// views in a few milliseconds: bound refreshes separately from repeated
+    /// missing-file failures at the same base version. The retry budget limits
+    /// starting another attempt, not the duration of an otherwise valid scan.
     pub async fn read_consistent<T, F, Fut>(
         &self,
         source: ListSource,
@@ -2309,7 +2312,16 @@ impl StorageBase {
         F: FnMut(Dataset, Vec<ShardSnapshot>) -> Fut,
         Fut: std::future::Future<Output = LanceResult<T>>,
     {
-        for attempt in 0..3 {
+        const MAX_SNAPSHOT_ATTEMPTS: usize = 16;
+        const MAX_MISSING_AT_VERSION: usize = 3;
+        let started = std::time::Instant::now();
+        let retry_budget = std::time::Duration::from_secs(2);
+        let mut last_missing_version = None;
+        let mut missing_at_version = 0;
+        for attempt in 0..MAX_SNAPSHOT_ATTEMPTS {
+            if attempt != 0 && started.elapsed() >= retry_budget {
+                break;
+            }
             let mut dataset = self.dataset.clone();
             if !self.is_version_pinned() {
                 dataset.checkout_latest().await?;
@@ -2335,7 +2347,23 @@ impl StorageBase {
                 return result;
             }
             if let Err(err) = &result {
-                if !is_not_found_error(err) || attempt == 2 {
+                if !is_not_found_error(err) {
+                    return result;
+                }
+                if last_missing_version == Some(version) {
+                    missing_at_version += 1;
+                } else {
+                    last_missing_version = Some(version);
+                    missing_at_version = 1;
+                }
+                // Actual base commits justify refreshing again, but repeated
+                // failures without base progress retain the old three-attempt
+                // bound. Neither continuous commits nor missing data permit
+                // unlimited repeated payload decoding.
+                if missing_at_version >= MAX_MISSING_AT_VERSION
+                    || attempt + 1 == MAX_SNAPSHOT_ATTEMPTS
+                    || started.elapsed() >= retry_budget
+                {
                     return result;
                 }
                 // A merge may already have committed before this read began.
diff --git a/crates/lance-context-master/src/eligibility.rs b/crates/lance-context-master/src/eligibility.rs
index e6019d4..ea623c7 100644
--- a/crates/lance-context-master/src/eligibility.rs
+++ b/crates/lance-context-master/src/eligibility.rs
@@ -31,6 +31,7 @@ pub enum Skip {
 
 /// Static per-target scheduling configuration consulted by the filter.
 pub struct Policy<'a> {
+    pub auto_local_merge: bool,
     pub rollout: &'a MergeRollout,
     pub compaction_prepare_targets: &'a [String],
     pub index_prepare_targets: &'a [String],
@@ -89,10 +90,15 @@ pub fn pre_dependency_filter(
     }
     if let Some(targets) = resident_targets {
         let eligible = kind == TaskKind::MergeWal
-            && policy.rollout.owned(target)
+            && (policy.rollout.owned(target) || policy.auto_local_merge)
+            && !target.starts_with("datagen:")
+            && lance_context_core::validate_store_name(
+                target.strip_prefix("generic:").unwrap_or(target),
+            )
+            .is_ok()
             && !policy.rollout.draining(target)
             && (targets.iter().any(|t| t == target)
-                || (!is_generic(target) && listed(targets, target)));
+                || ((!is_generic(target) || policy.auto_local_merge) && listed(targets, target)));
         if !eligible {
             return Err(Skip::NotResidentEligible);
         }
@@ -183,6 +189,7 @@ mod tests {
 
     fn rollout(owned: &[&str], draining: &[&str]) -> MergeRollout {
         MergeRollout {
+            fence_legacy: false,
             owned_targets: owned.iter().map(|s| s.to_string()).collect(),
             drain_targets: draining.iter().map(|s| s.to_string()).collect(),
         }
@@ -209,6 +216,7 @@ mod tests {
             &["draining"],
         );
         let policy = Policy {
+            auto_local_merge: false,
             rollout: &rollout,
             compaction_prepare_targets: &[],
             index_prepare_targets: &[],
@@ -256,6 +264,7 @@ mod tests {
         let compact = strings(&["t", "d"]);
         let index = strings(&["*"]);
         let policy = Policy {
+            auto_local_merge: false,
             rollout: &rollout,
             compaction_prepare_targets: &compact,
             index_prepare_targets: &index,
@@ -286,6 +295,7 @@ mod tests {
         let rollout = rollout(&["t"], &[]);
         let shared = strings(&["t"]);
         let policy = Policy {
+            auto_local_merge: false,
             rollout: &rollout,
             compaction_prepare_targets: &[],
             index_prepare_targets: &[],
@@ -296,6 +306,7 @@ mod tests {
         assert!(!claim_shape(TaskKind::MergeWal, "t", &policy, true).shared_owner);
         let wildcard = strings(&["*"]);
         let policy = Policy {
+            auto_local_merge: false,
             maintenance_catchup_targets: &wildcard,
             ..policy
         };
@@ -537,4 +548,38 @@ mod tests {
             "preparation runs beside a worker execution"
         );
     }
+    #[test]
+    fn automatic_resident_filter_admits_new_rollout_and_generic_but_keeps_drains() {
+        let rollout = rollout(&[], &["draining"]);
+        let policy = Policy {
+            auto_local_merge: true,
+            rollout: &rollout,
+            compaction_prepare_targets: &[],
+            index_prepare_targets: &[],
+            maintenance_catchup_targets: &[],
+        };
+        let targets = strings(&["*"]);
+        for name in ["new-hot", "generic:new-hot"] {
+            assert_eq!(
+                pre_dependency_filter(TaskKind::MergeWal, name, |_| true, Some(&targets), &policy),
+                Ok(())
+            );
+        }
+        for name in ["draining", "datagen:new-hot", "../escape", "generic:"] {
+            assert_eq!(
+                pre_dependency_filter(TaskKind::MergeWal, name, |_| true, Some(&targets), &policy),
+                Err(Skip::NotResidentEligible)
+            );
+        }
+        assert_eq!(
+            pre_dependency_filter(
+                TaskKind::Compact,
+                "new-hot",
+                |_| true,
+                Some(&targets),
+                &policy
+            ),
+            Err(Skip::NotResidentEligible)
+        );
+    }
 }
diff --git a/crates/lance-context-master/src/local_merge.rs b/crates/lance-context-master/src/local_merge.rs
index ab1277f..0d60554 100644
--- a/crates/lance-context-master/src/local_merge.rs
+++ b/crates/lance-context-master/src/local_merge.rs
@@ -4,8 +4,15 @@ use std::sync::Arc;
 
 #[derive(Clone, Debug, Default, clap::Args)]
 pub struct LocalMergeConfig {
+    /// Route new rollout/generic demand to resident masters without a static
+    /// table allowlist. Enable only after every legacy writer/helper has drained
+    /// and all participants use persistent fenced admission. This flag is not
+    /// evidence of that initial transition; retain its deployment evidence.
+    #[arg(long, env = "MASTER_LOCAL_MERGE_AUTO", default_value_t = false)]
+    pub master_local_merge_auto: bool,
     /// Execute sealed WAL merges in the master, including generic upsert tables.
-    /// '*' selects only explicitly owned targets; it never migrates legacy writers.
+    /// '*' selects explicitly owned targets unless automatic routing is enabled.
+    /// Neither setting certifies that historical legacy writers have drained.
     #[arg(long, env = "MASTER_LOCAL_MERGE_TARGETS", value_delimiter = ',')]
     pub master_local_merge_targets: Vec<String>,
 }
@@ -23,7 +30,12 @@ pub(crate) fn enabled(config: &MasterConfig) -> bool {
 }
 
 pub(crate) fn selected(config: &MasterConfig, target: &str) -> bool {
-    config.merge_rollout.owned(target)
+    let name = target.strip_prefix("generic:").unwrap_or(target);
+    let dynamic = config.local_merge.master_local_merge_auto
+        && config.merge_rollout.fence_legacy
+        && !target.starts_with("datagen:")
+        && lance_context_core::validate_store_name(name).is_ok();
+    (config.merge_rollout.owned(target) || dynamic)
         && !config.merge_rollout.draining(target)
         && (config.append.local(target) || config.local_merge.selected(target))
 }
@@ -31,6 +43,9 @@ pub(crate) fn selected(config: &MasterConfig, target: &str) -> bool {
 /// Resolve the local selection before claiming. In particular an append wildcard
 /// cannot accidentally reserve a slot for a generic table that still needs RPCs.
 pub(crate) fn targets(config: &MasterConfig) -> Vec<String> {
+    if config.local_merge.master_local_merge_auto {
+        return vec!["*".into()];
+    }
     config
         .merge_rollout
         .owned_targets
@@ -41,6 +56,13 @@ pub(crate) fn targets(config: &MasterConfig) -> Vec<String> {
 }
 
 pub(crate) fn validate(config: &MasterConfig) -> Result<(), String> {
+    if config.local_merge.master_local_merge_auto
+        && (!config.merge_rollout.fence_legacy
+            || config.local_merge.master_local_merge_targets != ["*"]
+            || !config.maintenance.fence_legacy_maintenance)
+    {
+        return Err("automatic local merges require persistent worker fencing, fenced master maintenance and MASTER_LOCAL_MERGE_TARGETS=*; initial legacy writer drain must be verified before enabling".into());
+    }
     if config.local_merge.master_local_merge_targets.is_empty() {
         return Ok(());
     }
@@ -58,9 +80,11 @@ pub(crate) fn validate(config: &MasterConfig) -> Result<(), String> {
     {
         return Err("master local merges require stable shards, bounded passes and a sufficient shared memory budget".into());
     }
-    if targets(config)
+    if config
+        .merge_rollout
+        .owned_targets
         .iter()
-        .any(|target| config.append.enabled(target))
+        .any(|target| selected(config, target) && config.append.enabled(target))
     {
         config.append.validate()?;
         if config.append.rollout_append_local_memory_bytes < config.append.rollout_append_max_bytes
@@ -134,4 +158,25 @@ mod tests {
             config.append.rollout_append_local_memory_bytes + 1;
         assert!(validate(&config).is_err());
     }
+    #[test]
+    fn automatic_selection_requires_fenced_protocol_and_bounded_resources() {
+        let mut cfg = MasterConfig::parse_from(["test"]);
+        cfg.local_merge.master_local_merge_auto = true;
+        cfg.local_merge.master_local_merge_targets = vec!["*".into()];
+        cfg.catchup.shards = vec!["writer".into()];
+        assert!(validate(&cfg).is_err());
+        assert!(!selected(&cfg, "new-hot"));
+        cfg.merge_rollout.fence_legacy = true;
+        assert!(validate(&cfg).is_err());
+        cfg.maintenance.fence_legacy_maintenance = true;
+        assert!(validate(&cfg).is_ok());
+        assert!(selected(&cfg, "new-hot"));
+        assert!(selected(&cfg, "generic:new-hot"));
+        assert!(!selected(&cfg, "datagen:new-hot"));
+        cfg.merge_rollout.drain_targets.push("new-hot".into());
+        assert!(!selected(&cfg, "new-hot"));
+        assert_eq!(targets(&cfg), ["*"]);
+        cfg.catchup.merge_max_bytes = cfg.append.rollout_append_local_memory_bytes + 1;
+        assert!(validate(&cfg).is_err());
+    }
 }
diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs
index 730dffc..a85fb6e 100644
--- a/crates/lance-context-master/src/merge_execution.rs
+++ b/crates/lance-context-master/src/merge_execution.rs
@@ -25,18 +25,14 @@ struct Capabilities {
     idle_timeout_secs: u64,
     progress_protocol: u32,
     owned_targets: Vec<String>,
+    #[serde(default)]
+    fenced_legacy_merges: bool,
 }
 
 pub(crate) async fn run_merge_wal(
     state: &Arc<MasterState>,
     claim: &TaskClaim,
 ) -> Result<String, String> {
-    if state.config.worker_endpoints.is_empty()
-        && !state.config.merge_rollout.draining(&claim.task.target)
-        && !crate::local_merge::selected(&state.config, &claim.task.target)
-    {
-        return Err("no worker endpoints configured".into());
-    }
     let coordinator = state.task_store.merge_coordinator();
     let proof = state.task_store.merge_claim(claim);
     let target = &claim.task.target;
@@ -56,23 +52,59 @@ pub(crate) async fn run_merge_wal(
         }
         return Err("merge target draining for protocol transition".into());
     }
-    if !state.config.merge_rollout.owned(target) {
-        if coordinator.get(target).await?.is_some() {
-            return Err("owned execution still present; refusing legacy downgrade".into());
+    if !state.config.merge_rollout.owned(target) && !state.config.merge_rollout.fence_legacy {
+        if let Some(old) = coordinator.get(target).await? {
+            // Turning off new admission must never disable recovery of durable
+            // worker records. Observe useful progress before considering cancel.
+            if old.maintenance.is_some() || old.protocol != 2 {
+                return Err("owned execution still present; refusing legacy downgrade".into());
+            }
+            ensure_recovery_due(&coordinator, &old).await?;
+            if reconcile(&state.http, &coordinator, &proof, old, false)
+                .await
+                .is_err()
+            {
+                if let Some(current) = coordinator.get(target).await? {
+                    recover_execution(state, &coordinator, &proof, current).await?;
+                }
+            }
+            return Ok("reconciled recorded worker merge; new fenced admission disabled".into());
         }
         return run_legacy(state, target).await;
     }
-    if state.config.append.enabled(target) || crate::local_merge::selected(&state.config, target) {
+    if state.config.worker_endpoints.is_empty()
+        && !crate::local_merge::selected(&state.config, target)
+        && coordinator.get(target).await?.is_none()
+    {
+        return Err("no worker endpoints configured".into());
+    }
+    if (state.config.merge_rollout.owned(target) && state.config.append.enabled(target))
+        || crate::local_merge::selected(&state.config, target)
+    {
         if let Some(old) = coordinator.get(target).await? {
             ensure_recovery_due(&coordinator, &old).await?;
-            recover_execution(state, &coordinator, &proof, old).await?;
+            if old.maintenance.is_none() {
+                // A newly selected resident target may still have useful
+                // background work. Wait for its real progress and terminal
+                // acknowledgement before considering storage recovery.
+                if reconcile(&state.http, &coordinator, &proof, old, false)
+                    .await
+                    .is_err()
+                {
+                    if let Some(current) = coordinator.get(target).await? {
+                        recover_execution(state, &coordinator, &proof, current).await?;
+                    }
+                }
+            } else {
+                recover_execution(state, &coordinator, &proof, old).await?;
+            }
         }
         return crate::maintenance_execution::run_as(
             state,
             claim,
             lance_context_merge::MaintenanceKind::Catchup,
             async {
-                if state.config.append.enabled(target) {
+                if state.config.merge_rollout.owned(target) && state.config.append.enabled(target) {
                     crate::rollout_append::run(state, target).await
                 } else {
                     crate::local_merge::run(state, target).await
@@ -88,7 +120,8 @@ pub(crate) async fn run_merge_wal(
             ensure_recovery_due(&coordinator, &old).await?;
             if recovery_round > 0 || matches!(old.phase, Phase::Recovering | Phase::Uncertain) {
                 recover_execution(state, &coordinator, &proof, old).await?;
-            } else if let Err(error) = reconcile(&state.http, &coordinator, &proof, old, true).await
+            } else if let Err(error) =
+                reconcile(&state.http, &coordinator, &proof, old, false).await
             {
                 if let Some(current) = coordinator.get(&claim.task.target).await? {
                     recover_execution(state, &coordinator, &proof, current).await?;
@@ -118,6 +151,9 @@ pub(crate) async fn run_merge_wal(
 // Compatibility path is chosen only by explicit configuration, never after a
 // failed owned RPC. It retains legacy limitations until that table is drained.
 async fn run_legacy(state: &Arc<MasterState>, target: &str) -> Result<String, String> {
+    if state.config.worker_endpoints.is_empty() {
+        return Err("no worker endpoints configured".into());
+    }
     run_legacy_workers(&state.http, &state.config.worker_endpoints, target).await
 }
 
@@ -373,6 +409,9 @@ async fn run_workers(
     target: &str,
     endpoints: &[String],
 ) -> Result<String, String> {
+    if endpoints.is_empty() {
+        return Err("no worker endpoints configured after reconciling recorded execution".into());
+    }
     let mut pending = Vec::new();
     let mut deferred = Vec::new();
     for endpoint in endpoints {
@@ -481,7 +520,8 @@ async fn one(
         .map_err(|e| e.to_string())?;
     if capabilities.protocol != 2
         || capabilities.progress_protocol != 1
-        || !capabilities.owned_targets.iter().any(|t| t == target)
+        || (!capabilities.owned_targets.iter().any(|t| t == target)
+            && !capabilities.fenced_legacy_merges)
         || capabilities.timeout_secs == 0
         || capabilities.queue_timeout_secs == 0
         || capabilities.idle_timeout_secs == 0
@@ -816,6 +856,55 @@ mod tests {
         (Coordinator::new(client, prefix), proof)
     }
 
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn append_wildcard_does_not_adopt_new_tables_before_native_routing() {
+        use clap::Parser;
+        use lance_context_api::TaskKind;
+
+        let dir = tempfile::tempdir().unwrap();
+        let mut cfg = crate::config::MasterConfig::parse_from(["test"]);
+        cfg.data_dir = dir.path().to_string_lossy().into();
+        cfg.etcd.etcd_endpoints = vec![std::env::var("ETCD_TEST_ENDPOINTS").unwrap()];
+        cfg.etcd.etcd_prefix = format!(
+            "/merge-fenced-routing/{}",
+            lance_context_core::generate_id()
+        );
+        cfg.merge_rollout.fence_legacy = true;
+        cfg.append.rollout_append_targets = vec!["*".into()];
+        assert!(cfg.merge_rollout.owned_targets.is_empty());
+        assert!(!crate::local_merge::selected(&cfg, "table"));
+        let mut state = MasterState::new(cfg).await.unwrap();
+        let coordinator = state.task_store.merge_coordinator();
+        let good = Worker {
+            coordinator: coordinator.clone(),
+            calls: Arc::new(AtomicUsize::new(0)),
+            fail: false,
+            stall_first: false,
+            name: "fenced-worker",
+            events: Arc::new(Mutex::new(Vec::new())),
+        };
+        let (endpoint, server) = worker(good.clone()).await;
+        Arc::get_mut(&mut state).unwrap().config.worker_endpoints = vec![endpoint];
+        state
+            .task_store
+            .enqueue(TaskKind::MergeWal, "table", Vec::new())
+            .await
+            .unwrap();
+        let claim = state
+            .task_store
+            .claim_next_of_kinds(crate::task_store::TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        let outcome = run_merge_wal(&state, &claim).await;
+        server.abort();
+        assert!(outcome.as_ref().unwrap().contains("merged 7 generations"));
+        assert_eq!(good.calls.load(Ordering::SeqCst), 1);
+        assert!(coordinator.get("table").await.unwrap().is_none());
+        state.task_store.finish(claim, outcome).await.unwrap();
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
     async fn orphaned_merge_is_storage_fenced_and_healthy_shard_progresses() {
diff --git a/crates/lance-context-master/src/resident_recovery.rs b/crates/lance-context-master/src/resident_recovery.rs
index 2916ac8..76cf8a3 100644
--- a/crates/lance-context-master/src/resident_recovery.rs
+++ b/crates/lance-context-master/src/resident_recovery.rs
@@ -37,6 +37,8 @@ fn scope(config: &MasterConfig) -> u64 {
         sorted.hash(&mut hash);
     }
     config.append.rollout_append_local.hash(&mut hash);
+    config.local_merge.master_local_merge_auto.hash(&mut hash);
+    config.merge_rollout.fence_legacy.hash(&mut hash);
     hash.finish()
 }
 
@@ -483,6 +485,10 @@ mod tests {
     }
 
     async fn generic_fixture() -> (tempfile::TempDir, Arc<MasterState>) {
+        generic_fixture_with_auto(false).await
+    }
+
+    async fn generic_fixture_with_auto(auto: bool) -> (tempfile::TempDir, Arc<MasterState>) {
         let dir = tempfile::tempdir().unwrap();
         let mut config = MasterConfig::parse_from(["test"]);
         config.data_dir = dir.path().to_string_lossy().into();
@@ -497,6 +503,11 @@ mod tests {
         );
         config.merge_rollout.owned_targets = vec!["generic:a".into(), "generic:b".into()];
         config.local_merge.master_local_merge_targets = vec!["*".into()];
+        if auto {
+            config.merge_rollout.owned_targets.clear();
+            config.merge_rollout.fence_legacy = true;
+            config.local_merge.master_local_merge_auto = true;
+        }
         config.catchup.shards = vec!["writer".into()];
         config.catchup.merge_max_generations = 1;
         config.catchup.merge_max_bytes = 1024 * 1024;
@@ -543,6 +554,166 @@ mod tests {
         writer.flush().await.unwrap();
     }
 
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn automatic_new_table_waits_for_background_then_catches_up_across_masters() {
+        use lance_context_merge::{Execution, Phase};
+        let (_dir, state) = generic_fixture_with_auto(true).await;
+        assert!(state.config.merge_rollout.owned_targets.is_empty());
+        assert!(state.config.maintenance.fence_legacy_maintenance);
+        // Created after master startup, with no stats row or static allowlist.
+        let writer = generic_writer(&state, "new-hot").await;
+        for n in 0..17 {
+            generic_write(&writer, "shared", &n.to_string()).await;
+        }
+        let target = "generic:new-hot";
+        let c = state.task_store.merge_coordinator();
+        let mut e = Execution::new(target, "http://127.0.0.1:1", "live-worker", 60);
+        e.background = true;
+        assert!(c.reserve_background(&e).await.unwrap());
+        let running = c.start(&e).await.unwrap().unwrap();
+        assert!(c.publish_progress(&running, 0).await.unwrap());
+        let request = c.request_page(None).await.unwrap().0.remove(0);
+        crate::scheduler::enqueue_merge_request(&state, &c, &request)
+            .await
+            .unwrap();
+        let claim = state
+            .task_store
+            .claim_resident_merge(&crate::local_merge::targets(&state.config))
+            .await
+            .unwrap()
+            .unwrap();
+        let next_master = MasterState::new(state.config.clone()).await.unwrap();
+        assert!(next_master
+            .task_store
+            .claim_resident_merge(&crate::local_merge::targets(&next_master.config))
+            .await
+            .unwrap()
+            .is_none());
+        let first = tokio::spawn({
+            let state = state.clone();
+            async move {
+                let outcome = crate::merge_execution::run_merge_wal(&state, &claim).await;
+                (claim, outcome)
+            }
+        });
+        for sequence in 1..=4 {
+            tokio::time::sleep(Duration::from_millis(100)).await;
+            let current = c.get(target).await.unwrap().unwrap();
+            assert_eq!(current.id, running.id);
+            assert_eq!(
+                current.phase,
+                Phase::Running,
+                "native selection must preserve useful worker progress"
+            );
+            assert!(c.publish_progress(&running, sequence).await.unwrap());
+            assert!(!first.is_finished());
+        }
+        assert!(c.finish(&running, Ok(0)).await.unwrap());
+        let (claim, result) = tokio::time::timeout(Duration::from_secs(30), first)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(
+            result.as_ref().unwrap().contains("16 generations"),
+            "{result:?}"
+        );
+        let request = c.request_page(None).await.unwrap().0.remove(0);
+        crate::scheduler::enqueue_merge_request(&next_master, &c, &request)
+            .await
+            .unwrap();
+        assert_eq!(c.request_page(None).await.unwrap().0.len(), 1);
+        state.task_store.finish(claim, result).await.unwrap();
+        crate::scheduler::enqueue_merge_request(&next_master, &c, &request)
+            .await
+            .unwrap();
+        let claim = next_master
+            .task_store
+            .claim_resident_merge(&crate::local_merge::targets(&next_master.config))
+            .await
+            .unwrap()
+            .unwrap();
+        let result = crate::merge_execution::run_merge_wal(&next_master, &claim).await;
+        assert!(
+            result.as_ref().unwrap().contains("1 generations"),
+            "{result:?}"
+        );
+        next_master.task_store.finish(claim, result).await.unwrap();
+        assert_eq!(writer.pending_wal_generations().await.unwrap(), 0);
+        let rows = writer.list(None, None).await.unwrap();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0]["text"], "16");
+        assert_eq!(state.local_append.as_ref().unwrap().budget.reserved(), 0);
+        assert_eq!(
+            next_master.local_append.as_ref().unwrap().budget.reserved(),
+            0
+        );
+        generic_write(&writer, "shared", "after-handoff").await;
+        assert!(probe(&next_master, target).await.unwrap());
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn automatic_rollout_uses_native_merge_and_preserves_external_owner() {
+        let (_dir, initial) = generic_fixture_with_auto(true).await;
+        let mut cfg = initial.config.clone();
+        // Existing append wildcard must not send an unowned new table through
+        // the explicit-ownership staged-append protocol.
+        cfg.append.rollout_append_targets = vec!["*".into()];
+        let state = MasterState::new(cfg).await.unwrap();
+        let target = "new-rollout";
+        let writer = write(&state, target, 4).await;
+        let c = state.task_store.merge_coordinator();
+        let owner = crate::catchup::store::active_key(&state.config.etcd.etcd_prefix, target);
+        let mut client = state.task_store.etcd_client().clone();
+        client
+            .put(owner.as_str(), "fixture-helper", None)
+            .await
+            .unwrap();
+        assert!(!probe(&state, target).await.unwrap());
+        c.request_merge(target).await.unwrap();
+        let request = c.request_page(None).await.unwrap().0.remove(0);
+        crate::scheduler::enqueue_merge_request(&state, &c, &request)
+            .await
+            .unwrap();
+        assert!(state
+            .task_store
+            .claim_resident_merge(&crate::local_merge::targets(&state.config))
+            .await
+            .unwrap()
+            .is_none());
+        assert_eq!(writer.pending_wal_generations().await.unwrap(), 4);
+        // This fixture has no remote actor or commits. End its known ownership.
+        client.delete(owner, None).await.unwrap();
+        let claim = state
+            .task_store
+            .claim_resident_merge(&crate::local_merge::targets(&state.config))
+            .await
+            .unwrap()
+            .unwrap();
+        let result = crate::merge_execution::run_merge_wal(&state, &claim).await;
+        assert!(
+            result
+                .as_ref()
+                .unwrap()
+                .contains("local WAL merge reclaimed 4 generations"),
+            "{result:?}"
+        );
+        state.task_store.finish(claim, result).await.unwrap();
+        assert_eq!(writer.pending_wal_generations().await.unwrap(), 0);
+        assert_eq!(
+            lance::Dataset::open(&state.rollout_uri(target))
+                .await
+                .unwrap()
+                .count_rows(None)
+                .await
+                .unwrap(),
+            4
+        );
+        assert_eq!(state.local_append.as_ref().unwrap().budget.reserved(), 0);
+        assert!(!probe(&state, target).await.unwrap());
+    }
+
     #[tokio::test]
     #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn generic_discovery_and_continuation_preserve_updates_without_workers() {
diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs
index e787731..8ccb3bc 100644
--- a/crates/lance-context-master/src/scheduler.rs
+++ b/crates/lance-context-master/src/scheduler.rs
@@ -932,7 +932,10 @@ async fn should_probe_failure(
     state: &Arc<MasterState>,
     failure: &lance_context_merge::failure::ShardFailure,
 ) -> bool {
-    if state.config.merge_rollout.owned(&failure.target) {
+    if !state.config.merge_rollout.draining(&failure.target)
+        && (state.config.merge_rollout.owned(&failure.target)
+            || state.config.merge_rollout.fence_legacy)
+    {
         return true;
     }
     if state.config.merge_rollout.draining(&failure.target) {
@@ -956,7 +959,13 @@ pub(crate) async fn enqueue_merge_request(
     coordinator: &lance_context_merge::Coordinator,
     request: &lance_context_merge::rollout::MergeRequest,
 ) -> Result<(), String> {
-    if !state.config.merge_rollout.owned(&request.target) {
+    if !state.config.merge_rollout.owned(&request.target)
+        && !state.config.merge_rollout.fence_legacy
+        && !coordinator
+            .get(&request.target)
+            .await?
+            .is_some_and(|e| e.maintenance.is_none())
+    {
         return Ok(());
     }
     let task = enqueue(state, TaskKind::MergeWal, &request.target)
@@ -968,6 +977,23 @@ pub(crate) async fn enqueue_merge_request(
     Ok(())
 }
 
+async fn enqueue_execution_recovery(
+    state: &Arc<MasterState>,
+    coordinator: &lance_context_merge::Coordinator,
+    execution: &lance_context_merge::Execution,
+) -> Result<(), String> {
+    if execution.maintenance.is_some() {
+        return crate::maintenance_execution::enqueue_recovery(state, coordinator, execution).await;
+    }
+    // Durable worker executions survive flags, endpoint list changes and loss
+    // of their coalesced request. Task claims still serialize reconciliation.
+    crate::merge_execution::ensure_recovery_due(coordinator, execution).await?;
+    enqueue(state, TaskKind::MergeWal, &execution.target)
+        .await
+        .map_err(|e| e.to_string())?;
+    Ok(())
+}
+
 /// Spawn the scheduler pollers plus the optional periodic auto-sweep.
 ///
 /// Returns the handle of the *general* poller only. When a separate WAL-merge
@@ -1012,15 +1038,8 @@ pub fn spawn_scheduler(state: &Arc<MasterState>) -> JoinHandle<()> {
                 Ok((executions, next)) => {
                     execution_cursor = next;
                     for execution in executions {
-                        if execution.maintenance.is_none() {
-                            continue;
-                        }
-                        if let Err(error) = crate::maintenance_execution::enqueue_recovery(
-                            &retry_state,
-                            &coordinator,
-                            &execution,
-                        )
-                        .await
+                        if let Err(error) =
+                            enqueue_execution_recovery(&retry_state, &coordinator, &execution).await
                         {
                             tracing::warn!(target = %execution.target, %error, "local execution recovery enqueue failed");
                         }
@@ -1298,6 +1317,7 @@ mod tests {
             planner: Default::default(),
             maintenance: Default::default(),
             merge_rollout: lance_context_merge::rollout::MergeRollout {
+                fence_legacy: false,
                 owned_targets: ["exp", "generic:gs", "broken"]
                     .into_iter()
                     .map(str::to_string)
@@ -1875,6 +1895,104 @@ mod tests {
         assert!(coordinator.request_page(None).await.unwrap().0.is_empty());
     }
 
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn background_inventory_recovers_after_flags_and_requests_disappear() {
+        use lance_context_merge::Execution;
+        let dir = TempDir::new().unwrap();
+        let mut cfg = config(&dir);
+        cfg.merge_rollout.owned_targets.clear();
+        cfg.merge_rollout.fence_legacy = false;
+        let state = MasterState::new(cfg).await.unwrap();
+        let c = state.task_store.merge_coordinator();
+        let mut e = Execution::new("new-hot-table", "http://127.0.0.1:1", "old-worker", 60);
+        e.background = true;
+        assert!(c.reserve_background(&e).await.unwrap());
+        let e = c.start(&e).await.unwrap().unwrap();
+        assert!(c.finish(&e, Ok(19)).await.unwrap());
+        let request = c.request_page(None).await.unwrap().0.remove(0);
+        assert!(c.acknowledge_request(&request).await.unwrap());
+        let inventory = c.execution_page(None).await.unwrap().0;
+        enqueue_execution_recovery(&state, &c, &inventory[0])
+            .await
+            .unwrap();
+        let claim = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        let result = crate::merge_execution::run_merge_wal(&state, &claim)
+            .await
+            .unwrap();
+        assert!(result.contains("reconciled"));
+        assert!(c.get("new-hot-table").await.unwrap().is_none());
+        state.task_store.finish(claim, Ok(result)).await.unwrap();
+        // Ambiguous storage needs a real barrier after initiation is disabled.
+        let uri = state.rollout_uri("new-hot-table");
+        RolloutStore::open(&uri)
+            .await
+            .unwrap()
+            .close()
+            .await
+            .unwrap();
+        let version = lance::Dataset::open(&uri).await.unwrap().version().version;
+        let mut orphan = Execution::new("new-hot-table", "http://127.0.0.1:1", "lost-worker", 60);
+        orphan.background = true;
+        assert!(c.reserve_background(&orphan).await.unwrap());
+        let running = c.start(&orphan).await.unwrap().unwrap();
+        c.authorize_commit(&running, &uri, "base", version + 1)
+            .await
+            .unwrap();
+        assert!(c
+            .report_uncertain(&running, "lost response".into())
+            .await
+            .unwrap());
+        let old = c.get("new-hot-table").await.unwrap().unwrap();
+        enqueue_execution_recovery(&state, &c, &old).await.unwrap();
+        let claim = state
+            .task_store
+            .claim_next_of_kinds(TaskKinds::MERGE_WAL)
+            .await
+            .unwrap()
+            .unwrap();
+        let result = crate::merge_execution::run_merge_wal(&state, &claim)
+            .await
+            .unwrap();
+        assert!(c.get("new-hot-table").await.unwrap().is_none());
+        assert!(lance::Dataset::open(&uri).await.unwrap().version().version > version + 1);
+        assert!(c
+            .authorize_commit(&running, &uri, "base", version + 3)
+            .await
+            .is_err());
+        state.task_store.finish(claim, Ok(result)).await.unwrap();
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn fenced_new_table_demand_enqueues_without_static_ownership() {
+        let dir = TempDir::new().unwrap();
+        let mut cfg = config(&dir);
+        cfg.merge_rollout.owned_targets.clear();
+        cfg.merge_rollout.fence_legacy = true;
+        let state = MasterState::new(cfg).await.unwrap();
+        let c = state.task_store.merge_coordinator();
+        c.request_merge("new-hot-table").await.unwrap();
+        let request = c.request_page(None).await.unwrap().0.remove(0);
+        enqueue_merge_request(&state, &c, &request).await.unwrap();
+        assert!(state
+            .task_store
+            .get_active_id(TaskKind::MergeWal, "new-hot-table")
+            .await
+            .unwrap()
+            .is_some());
+        assert!(c.request_page(None).await.unwrap().0.is_empty());
+        assert!(
+            !crate::local_merge::selected(&state.config, "new-hot-table"),
+            "worker recording does not certify native ownership"
+        );
+    }
+
     /// Wait until the task reaches a terminal state, returning its final record.
     async fn await_terminal(state: &Arc<MasterState>, id: &str) -> TaskRecord {
         for _ in 0..100 {
diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs
index 1dd9872..380a246 100644
--- a/crates/lance-context-master/src/state.rs
+++ b/crates/lance-context-master/src/state.rs
@@ -135,10 +135,9 @@ pub(crate) fn worker_http_client() -> Result<reqwest::Client, reqwest::Error> {
 
 impl MasterState {
     /// Open the registry, stats dataset, and configured durable task store.
-    pub async fn new(config: MasterConfig) -> lance::Result<Arc<Self>> {
+    pub async fn new(mut config: MasterConfig) -> lance::Result<Arc<Self>> {
         config.merge_rollout.validate().map_err(lance::Error::io)?;
         config.catchup.validate().map_err(lance::Error::io)?;
-        crate::local_merge::validate(&config).map_err(lance::Error::io)?;
         config
             .append
             .validate_local(&config)
@@ -152,6 +151,15 @@ impl MasterState {
             ));
         }
         let task_store = TaskStore::open(&config).await?;
+        config.merge_rollout.fence_legacy = task_store
+            .merge_coordinator()
+            .legacy_fencing_mode(config.merge_rollout.fence_legacy)
+            .await
+            .map_err(lance::Error::io)?;
+        // Protocol activation covers every master maintenance writer as well
+        // as workers; a flag rollback must not restore unrecorded rewrites.
+        config.maintenance.fence_legacy_maintenance |= config.merge_rollout.fence_legacy;
+        crate::local_merge::validate(&config).map_err(lance::Error::io)?;
         // Serialize first-time registry/stats creation and legacy backfill in
         // etcd mode. Followers wait briefly rather than racing Lance creates.
         let dedicated = config.catchup.target.is_some();
diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs
index 339eeee..4a9ed7f 100644
--- a/crates/lance-context-master/src/task_store.rs
+++ b/crates/lance-context-master/src/task_store.rs
@@ -123,6 +123,7 @@ struct EtcdTaskStore {
     index_prepare_targets: Vec<String>,
     maintenance_catchup_targets: Vec<String>,
     rollout: lance_context_merge::rollout::MergeRollout,
+    auto_local_merge: bool,
 }
 
 /// Ownership of one running task. Dropping the claim stops lease renewal; etcd
@@ -743,6 +744,7 @@ impl EtcdTaskStore {
             index_prepare_targets: config.maintenance.index_prepare_targets.clone(),
             maintenance_catchup_targets: config.maintenance.maintenance_catchup_targets.clone(),
             rollout: config.merge_rollout.clone(),
+            auto_local_merge: config.local_merge.master_local_merge_auto,
         })
     }
 
@@ -838,6 +840,7 @@ impl EtcdTaskStore {
 
     fn eligibility_policy(&self) -> eligibility::Policy<'_> {
         eligibility::Policy {
+            auto_local_merge: self.auto_local_merge,
             rollout: &self.rollout,
             compaction_prepare_targets: &self.prepare_targets,
             index_prepare_targets: &self.index_prepare_targets,
diff --git a/crates/lance-context-merge/src/background.rs b/crates/lance-context-merge/src/background.rs
new file mode 100644
index 0000000..038e96d
--- /dev/null
+++ b/crates/lance-context-merge/src/background.rs
@@ -0,0 +1,421 @@
+//! Durable admission for worker count/timer triggers. A native handoff may
+//! close this gate, but closure alone never proves old writers are quiescent.
+use crate::{
+    encode, execution_key, execution_owner, target_lock_key, ClaimProof, Coordinator, Execution,
+    Phase, Result,
+};
+use etcd_client::{Compare, CompareOp, TxnOp};
+
+/// No lease: losing a controller cannot reopen background admission.
+pub fn admission_key(prefix: &str, target: &str) -> String {
+    execution_key(prefix, target).replace("/merge-executions/", "/background-merge-closed/")
+}
+
+impl Coordinator {
+    /// Reserve before polling any storage work. The persistent request is in
+    /// the same transaction so a crash cannot hide an admitted execution from
+    /// the master. False means contention, a closed gate, or a retry deadline;
+    /// it never authorizes an untracked fallback.
+    pub async fn reserve_background(&self, execution: &Execution) -> Result<bool> {
+        if !execution.background
+            || execution.maintenance.is_some()
+            || execution.phase != Phase::Reserved
+            || execution.protocol != 2
+            || execution.timeout_secs == 0
+            || execution.queue_timeout_secs == 0
+            || execution.idle_timeout_secs == 0
+            || execution.target.is_empty()
+            || execution.target.starts_with("datagen:")
+            || execution.instance.is_empty()
+            || uuid::Uuid::parse_str(&execution.id).is_err()
+        {
+            return Err("invalid background merge reservation".into());
+        }
+        let key = execution_key(&self.prefix, &execution.target);
+        let failure_key = self.failure_key(&execution.target, &execution.endpoint);
+        let prior = self
+            .client
+            .clone()
+            .get(failure_key.clone(), None)
+            .await
+            .map_err(|e| e.to_string())?;
+        let failure = prior.kvs().first();
+        if let Some(row) = failure {
+            let previous: crate::failure::ShardFailure =
+                serde_json::from_slice(row.value()).map_err(|e| e.to_string())?;
+            if previous.corrected().next_retry_ms > crate::failure::now_ms() {
+                return Ok(false);
+            }
+        }
+        self.transact(
+            vec![
+                Compare::version(key.as_str(), CompareOp::Equal, 0),
+                Compare::version(
+                    target_lock_key(&self.prefix, &execution.target),
+                    CompareOp::Equal,
+                    0,
+                ),
+                Compare::version(
+                    admission_key(&self.prefix, &execution.target),
+                    CompareOp::Equal,
+                    0,
+                ),
+                Compare::version(
+                    key.replace("/merge-executions/", "/catchup-active/"),
+                    CompareOp::Equal,
+                    0,
+                ),
+                Compare::version(
+                    key.replace("/merge-executions/", "/merge-claims/"),
+                    CompareOp::Equal,
+                    0,
+                ),
+                Compare::version(
+                    key.replace("/merge-executions/", "/compact-commit-ready/"),
+                    CompareOp::Equal,
+                    0,
+                ),
+                Compare::mod_revision(
+                    failure_key,
+                    CompareOp::Equal,
+                    failure.map_or(0, |kv| kv.mod_revision()),
+                ),
+            ],
+            vec![
+                TxnOp::put(key, encode(execution), None),
+                TxnOp::put(
+                    target_lock_key(&self.prefix, &execution.target),
+                    execution_owner(execution),
+                    None,
+                ),
+                TxnOp::put(
+                    self.request_key(&execution.target),
+                    execution.target.as_str(),
+                    None,
+                ),
+            ],
+        )
+        .await
+    }
+
+    /// Release only after local work and all commit leaves have drained, or a
+    /// storage barrier has completed. A stale worker cannot remove a successor.
+    /// The ordinary master release method can also reconcile these records.
+    pub async fn release_background(&self, terminal: &Execution) -> Result<bool> {
+        if !terminal.background
+            || terminal.protocol != 2
+            || !matches!(terminal.phase, Phase::Finished | Phase::Recovered)
+        {
+            return Err("background merge is not durably terminal".into());
+        }
+        let (mut compares, mut operations) = self.completion_changes(terminal).await?;
+        compares.extend([
+            // A scheduler may already be reconciling this execution. Leave
+            // its terminal record intact so ordinary release can transfer the
+            // write lock to that claim, rather than deleting its predecessor.
+            Compare::version(
+                execution_key(&self.prefix, &terminal.target)
+                    .replace("/merge-executions/", "/merge-claims/"),
+                CompareOp::Equal,
+                0,
+            ),
+            Compare::value(
+                execution_key(&self.prefix, &terminal.target),
+                CompareOp::Equal,
+                encode(terminal),
+            ),
+            Compare::value(
+                target_lock_key(&self.prefix, &terminal.target),
+                CompareOp::Equal,
+                execution_owner(terminal),
+            ),
+        ]);
+        operations.extend([
+            self.remove_permits(terminal),
+            self.remove_progress(terminal),
+            self.remove_work_progress(terminal),
+            TxnOp::delete(execution_key(&self.prefix, &terminal.target), None),
+            TxnOp::delete(target_lock_key(&self.prefix, &terminal.target), None),
+        ]);
+        self.transact(compares, operations).await
+    }
+
+    /// The canonical scheduler claim serializes this with background reserve.
+    /// This prevents NEW background work; it does not authorize native writes
+    /// or certify coverage of historical untracked writers.
+    pub async fn close_background_admission(
+        &self,
+        proof: &ClaimProof,
+        target: &str,
+    ) -> Result<bool> {
+        self.transact(
+            vec![
+                Compare::value(proof.key.as_str(), CompareOp::Equal, proof.token.as_bytes()),
+                Compare::value(
+                    target_lock_key(&self.prefix, target),
+                    CompareOp::Equal,
+                    proof.token.as_bytes(),
+                ),
+                Compare::version(execution_key(&self.prefix, target), CompareOp::Equal, 0),
+            ],
+            vec![TxnOp::put(
+                admission_key(&self.prefix, target),
+                "closed",
+                None,
+            )],
+        )
+        .await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use etcd_client::{Client, Txn};
+
+    async fn fixture() -> (Coordinator, Client, String) {
+        let endpoint = std::env::var("ETCD_TEST_ENDPOINTS").expect("isolated test etcd required");
+        let client = Client::connect([endpoint], None).await.unwrap();
+        let prefix = format!("/background-admission-test/{}", uuid::Uuid::new_v4());
+        (Coordinator::new(client.clone(), &prefix), client, prefix)
+    }
+
+    fn execution() -> Execution {
+        let mut execution = Execution::new("hot", "http://worker", "worker-boot", 60);
+        execution.background = true;
+        execution
+    }
+
+    #[test]
+    fn old_readers_cannot_cas_background_records() {
+        let background = execution();
+        let bytes = encode(&background);
+        assert_eq!(
+            serde_json::from_slice::<Execution>(&bytes).unwrap(),
+            background
+        );
+        let mut old_reader = background.clone();
+        old_reader.background = false;
+        assert_ne!(encode(&old_reader), bytes);
+        assert!(!String::from_utf8(encode(&old_reader))
+            .unwrap()
+            .contains("background"));
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"]
+    async fn concurrent_triggers_admit_one_durable_writer_and_request() {
+        let (coordinator, _, _) = fixture().await;
+        let mut tasks = Vec::new();
+        for _ in 0..20 {
+            let c = coordinator.clone();
+            tasks.push(tokio::spawn(async move {
+                let execution = execution();
+                c.reserve_background(&execution)
+                    .await
+                    .unwrap()
+                    .then_some(execution)
+            }));
+        }
+        let mut admitted = Vec::new();
+        for task in tasks {
+            admitted.extend(task.await.unwrap());
+        }
+        assert_eq!(admitted.len(), 1);
+        assert_eq!(
+            coordinator.get("hot").await.unwrap(),
+            Some(admitted[0].clone())
+        );
+        let (requests, next) = coordinator.request_page(None).await.unwrap();
+        assert_eq!(requests.len(), 1);
+        assert_eq!(requests[0].target, "hot");
+        assert!(next.is_none());
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"]
+    async fn closing_race_never_forgets_a_writer_and_survives_controller_loss() {
+        let (coordinator, mut client, prefix) = fixture().await;
+        let proof = ClaimProof {
+            key: format!("{prefix}/claim"),
+            token: "master".into(),
+            lease_id: 0,
+        };
+        client
+            .put(proof.key.as_str(), proof.token.as_str(), None)
+            .await
+            .unwrap();
+        let background = execution();
+        let master_claim = client.txn(
+            Txn::new()
+                .when([Compare::version(
+                    target_lock_key(&prefix, "hot"),
+                    CompareOp::Equal,
+                    0,
+                )])
+                .and_then([TxnOp::put(
+                    target_lock_key(&prefix, "hot"),
+                    proof.token.as_str(),
+                    None,
+                )]),
+        );
+        let (worker, master) =
+            tokio::join!(coordinator.reserve_background(&background), master_claim);
+        let worker = worker.unwrap();
+        assert_ne!(worker, master.unwrap().succeeded());
+        if worker {
+            assert!(!coordinator
+                .close_background_admission(&proof, "hot")
+                .await
+                .unwrap());
+            assert!(coordinator.release_background(&background).await.is_err());
+            let running = coordinator.start(&background).await.unwrap().unwrap();
+            assert!(coordinator.finish(&running, Ok(3)).await.unwrap());
+            assert!(coordinator
+                .release_background(&running.finished(Ok(3)))
+                .await
+                .unwrap());
+            client
+                .put(target_lock_key(&prefix, "hot"), proof.token.as_str(), None)
+                .await
+                .unwrap();
+        }
+        assert!(coordinator
+            .close_background_admission(&proof, "hot")
+            .await
+            .unwrap());
+        // Model expired controller/task ownership; only its ephemeral state
+        // disappears. The separate gate remains closed across a fresh client.
+        client.delete(proof.key, None).await.unwrap();
+        client
+            .delete(target_lock_key(&prefix, "hot"), None)
+            .await
+            .unwrap();
+        let restarted = Coordinator::new(client, prefix);
+        assert!(!restarted.reserve_background(&execution()).await.unwrap());
+        assert!(restarted.get("hot").await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"]
+    async fn uncertain_writer_requires_fencing_and_cannot_release_a_successor() {
+        let (coordinator, mut client, prefix) = fixture().await;
+        let background = execution();
+        assert!(coordinator.reserve_background(&background).await.unwrap());
+        let running = coordinator.start(&background).await.unwrap().unwrap();
+        coordinator
+            .authorize_commit(&running, "memory://dataset", "base", 7)
+            .await
+            .unwrap();
+        assert!(coordinator
+            .report_uncertain(&running, "lost PUT response".into())
+            .await
+            .unwrap());
+        let uncertain = coordinator.get("hot").await.unwrap().unwrap();
+        assert!(coordinator.release_background(&uncertain).await.is_err());
+        assert!(!coordinator.reserve_background(&execution()).await.unwrap());
+        let proof = ClaimProof {
+            key: format!("{prefix}/claim"),
+            token: "master".into(),
+            lease_id: 0,
+        };
+        client
+            .put(proof.key.as_str(), proof.token.as_str(), None)
+            .await
+            .unwrap();
+        let frozen = coordinator
+            .freeze(&proof, &uncertain)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            coordinator.watermarks(&frozen).await.unwrap().versions["base"],
+            7
+        );
+        assert!(coordinator
+            .authorize_commit(&running, "memory://dataset", "base", 8)
+            .await
+            .is_err());
+        // Only exercise metadata acknowledgement here; real object-store
+        // barrier behavior is covered by the core/master storage tests.
+        assert!(coordinator.finish_recovery(&proof, &frozen).await.unwrap());
+        let recovered = coordinator.get("hot").await.unwrap().unwrap();
+        assert!(coordinator.release(&proof, &recovered).await.unwrap());
+        let successor = Execution::new("hot", "http://successor", "next-boot", 60);
+        assert!(coordinator.reserve(&proof, &successor).await.unwrap());
+        assert!(!coordinator.release_background(&recovered).await.unwrap());
+        assert_eq!(coordinator.get("hot").await.unwrap(), Some(successor));
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"]
+    async fn terminal_background_writer_transfers_to_a_waiting_master_claim() {
+        let (coordinator, mut client, prefix) = fixture().await;
+        let background = execution();
+        assert!(coordinator.reserve_background(&background).await.unwrap());
+        let running = coordinator.start(&background).await.unwrap().unwrap();
+        let proof = ClaimProof {
+            key: format!("{prefix}/claim"),
+            token: "master".into(),
+            lease_id: 0,
+        };
+        client
+            .put(proof.key.as_str(), proof.token.as_str(), None)
+            .await
+            .unwrap();
+        client
+            .put(
+                execution_key(&prefix, "hot").replace("/merge-executions/", "/merge-claims/"),
+                proof.token.as_str(),
+                None,
+            )
+            .await
+            .unwrap();
+        assert!(coordinator.finish(&running, Ok(4)).await.unwrap());
+        let terminal = running.finished(Ok(4));
+        assert!(!coordinator.release_background(&terminal).await.unwrap());
+        assert_eq!(
+            coordinator.get("hot").await.unwrap(),
+            Some(terminal.clone())
+        );
+        assert!(coordinator.release(&proof, &terminal).await.unwrap());
+        let next = Execution::new("hot", "http://next", "next-boot", 60);
+        assert!(coordinator.reserve(&proof, &next).await.unwrap());
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated ETCD_TEST_ENDPOINTS"]
+    async fn background_failures_preserve_retry_budget_and_external_admission() {
+        let (coordinator, mut client, prefix) = fixture().await;
+        let key = execution_key(&prefix, "hot");
+        for reserved in [
+            target_lock_key(&prefix, "hot"),
+            key.replace("/merge-executions/", "/catchup-active/"),
+            key.replace("/merge-executions/", "/merge-claims/"),
+            key.replace("/merge-executions/", "/compact-commit-ready/"),
+        ] {
+            client.put(reserved.as_str(), "other", None).await.unwrap();
+            assert!(!coordinator.reserve_background(&execution()).await.unwrap());
+            client.delete(reserved, None).await.unwrap();
+        }
+        let background = execution();
+        assert!(coordinator.reserve_background(&background).await.unwrap());
+        let running = coordinator.start(&background).await.unwrap().unwrap();
+        let error = Err("storage timeout".into());
+        assert!(coordinator.finish(&running, error.clone()).await.unwrap());
+        assert!(coordinator
+            .release_background(&running.finished(error))
+            .await
+            .unwrap());
+        assert!(!coordinator.reserve_background(&execution()).await.unwrap());
+        assert_eq!(
+            coordinator
+                .failure("hot", "http://worker")
+                .await
+                .unwrap()
+                .unwrap()
+                .consecutive_attempts,
+            1
+        );
+    }
+}
diff --git a/crates/lance-context-merge/src/failure.rs b/crates/lance-context-merge/src/failure.rs
index 7a196c0..58e539c 100644
--- a/crates/lance-context-merge/src/failure.rs
+++ b/crates/lance-context-merge/src/failure.rs
@@ -183,7 +183,7 @@ impl ShardFailure {
     // Read old records with corrected semantics without resetting attempts,
     // moving the failure timestamp, or mutating ownership/durable records.
     // The next fenced completion persists the normal retry result.
-    fn corrected(mut self) -> Self {
+    pub(crate) fn corrected(mut self) -> Self {
         // One rule set: whatever `classify` says today is the effective
         // class. Only a DataOrConfiguration record can be softened; a stored
         // Retryable is never hardened on read.
@@ -227,7 +227,7 @@ impl Coordinator {
     fn failures_prefix(&self) -> String {
         format!("{}/merge-failures/", self.prefix.trim_end_matches('/'))
     }
-    fn failure_key(&self, target: &str, endpoint: &str) -> String {
+    pub(crate) fn failure_key(&self, target: &str, endpoint: &str) -> String {
         let hex = |s: &str| {
             s.as_bytes()
                 .iter()
diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs
index e89ec21..db6dad7 100644
--- a/crates/lance-context-merge/src/lib.rs
+++ b/crates/lance-context-merge/src/lib.rs
@@ -4,9 +4,11 @@
 //! Only the executor can publish its terminal outcome. Recovery first cancels
 //! and reconciles the old execution; it never clears a running fence on timeout.
 
+pub mod background;
 pub mod demand;
 pub mod failure;
 pub mod fencing;
+mod mode;
 pub mod progress;
 pub mod rollout;
 
@@ -66,6 +68,10 @@ impl MaintenanceKind {
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct Execution {
+    /// Count/timer-triggered worker execution, admitted without a scheduler
+    /// claim. Its persistent owner and commit permits are otherwise identical.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub background: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub maintenance: Option<MaintenanceKind>,
     pub id: String,
@@ -88,6 +94,10 @@ fn protocol_absent(protocol: &u32) -> bool {
     *protocol == 0
 }
 
+fn is_false(value: &bool) -> bool {
+    !value
+}
+
 fn default_queue_timeout() -> u64 {
     600
 }
@@ -95,6 +105,7 @@ fn default_queue_timeout() -> u64 {
 impl Execution {
     pub fn new(target: &str, endpoint: &str, instance: &str, timeout_secs: u64) -> Self {
         Self {
+            background: false,
             maintenance: None,
             id: uuid::Uuid::new_v4().to_string(),
             target: target.into(),
@@ -220,6 +231,7 @@ impl Coordinator {
     /// Atomic claim check and admission close the delayed-request/lease-loss race.
     pub async fn reserve(&self, claim: &ClaimProof, execution: &Execution) -> Result<bool> {
         if execution.phase != Phase::Reserved
+            || execution.background
             || execution.timeout_secs == 0
             || execution.protocol != 2
         {
diff --git a/crates/lance-context-merge/src/mode.rs b/crates/lance-context-merge/src/mode.rs
new file mode 100644
index 0000000..f018f10
--- /dev/null
+++ b/crates/lance-context-merge/src/mode.rs
@@ -0,0 +1,79 @@
+//! A protocol activation survives configuration rollback. This records the
+//! required protocol, not proof that pre-upgrade writers have drained.
+use crate::{Coordinator, Result};
+use etcd_client::{Compare, CompareOp, TxnOp};
+
+impl Coordinator {
+    pub async fn legacy_fencing_mode(&self, enable: bool) -> Result<bool> {
+        let key = format!(
+            "{}/worker-merge-admission-mode",
+            self.prefix.trim_end_matches('/')
+        );
+        let mut client = self.client.clone();
+        if enable {
+            // No lease and no API to reopen legacy admission. Do not overwrite
+            // a future protocol marker during a binary/config rollback.
+            self.transact(
+                vec![Compare::version(key.as_str(), CompareOp::Equal, 0)],
+                vec![TxnOp::put(key.as_str(), "fenced-v1", None)],
+            )
+            .await?;
+        }
+        let reply = client.get(key, None).await.map_err(|e| e.to_string())?;
+        match reply.kvs().first().map(|kv| kv.value()) {
+            None if !enable => Ok(false),
+            None => Err("worker merge protocol activation was not persisted".into()),
+            Some(b"fenced-v1") => Ok(true),
+            Some(_) => {
+                Err("unknown worker merge admission protocol; refusing legacy fallback".into())
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn activated_protocol_survives_a_new_client_and_disabled_flag() {
+        let endpoint = std::env::var("ETCD_TEST_ENDPOINTS").unwrap();
+        let prefix = format!("/merge-mode/{}", crate::Execution::new("", "", "", 1).id);
+        let first = Coordinator::new(
+            etcd_client::Client::connect([&endpoint], None)
+                .await
+                .unwrap(),
+            prefix.clone(),
+        );
+        assert!(!first.legacy_fencing_mode(false).await.unwrap());
+        assert!(first.legacy_fencing_mode(true).await.unwrap());
+        let restarted = Coordinator::new(
+            etcd_client::Client::connect([&endpoint], None)
+                .await
+                .unwrap(),
+            prefix,
+        );
+        assert!(restarted.legacy_fencing_mode(false).await.unwrap());
+    }
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn a_newer_protocol_cannot_be_replaced_by_activation() {
+        let endpoint = std::env::var("ETCD_TEST_ENDPOINTS").unwrap();
+        let prefix = format!("/merge-mode/{}", crate::Execution::new("", "", "", 1).id);
+        let mut client = etcd_client::Client::connect([endpoint], None)
+            .await
+            .unwrap();
+        client
+            .put(
+                format!("{prefix}/worker-merge-admission-mode"),
+                "fenced-v2",
+                None,
+            )
+            .await
+            .unwrap();
+        let c = Coordinator::new(client, prefix);
+        assert!(c.legacy_fencing_mode(false).await.is_err());
+        assert!(c.legacy_fencing_mode(true).await.is_err());
+    }
+}
diff --git a/crates/lance-context-merge/src/rollout.rs b/crates/lance-context-merge/src/rollout.rs
index 30d2571..7ae74cc 100644
--- a/crates/lance-context-merge/src/rollout.rs
+++ b/crates/lance-context-merge/src/rollout.rs
@@ -4,6 +4,10 @@ use etcd_client::{Compare, CompareOp, GetOptions, TxnOp};
 
 #[derive(Clone, Debug, Default, clap::Args)]
 pub struct MergeRollout {
+    /// Record new legacy-target worker merges with commit fencing. This does
+    /// not enroll targets for native catchup or fence pre-upgrade writers.
+    #[arg(long, env = "MERGE_FENCE_LEGACY", default_value_t = false)]
+    pub fence_legacy: bool,
     /// Exact scheduler targets using owned merges (generic stores: generic:name).
     /// Empty by default. Enable only after the target's legacy writers drain.
     #[arg(
@@ -56,7 +60,7 @@ pub struct MergeRequest {
 }
 
 impl Coordinator {
-    fn request_key(&self, target: &str) -> String {
+    pub(crate) fn request_key(&self, target: &str) -> String {
         execution_key(&self.prefix, target).replace("/merge-executions/", "/merge-requests/")
     }
 
diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs
index 91622e8..965dd18 100644
--- a/crates/lance-context-server/src/config.rs
+++ b/crates/lance-context-server/src/config.rs
@@ -28,6 +28,11 @@ pub struct ServerConfig {
     )]
     pub http_shutdown_response_idle_secs: u64,
 
+    /// Stable worker URL reachable by the master for execution identity and
+    /// cancellation. Required when MERGE_FENCE_LEGACY is enabled.
+    #[arg(long, env = "MERGE_EXECUTOR_ENDPOINT")]
+    pub merge_executor_endpoint: Option<String>,
+
     /// Legacy wire field; owned merges now use the real-progress idle timeout.
     #[arg(long, env = "MERGE_EXECUTION_TIMEOUT_SECS", default_value_t = 3600)]
     pub merge_execution_timeout_secs: u64,
diff --git a/crates/lance-context-server/src/main.rs b/crates/lance-context-server/src/main.rs
index d83edb0..751da9b 100644
--- a/crates/lance-context-server/src/main.rs
+++ b/crates/lance-context-server/src/main.rs
@@ -50,6 +50,9 @@ async fn main() {
         }
     };
 
+    // Resolve persisted merge protocol independently of startup and ingestion.
+    merge_execution::Executions::resolve_mode_in_background(state.clone());
+
     // Single process-wide WAL-cleanup sweeper (replaces one timer per store).
     // The handle is detached: it runs for the server's lifetime and stops when
     // the last `Arc<AppState>` is dropped.
diff --git a/crates/lance-context-server/src/merge_execution.rs b/crates/lance-context-server/src/merge_execution.rs
index fb6c96d..3b91564 100644
--- a/crates/lance-context-server/src/merge_execution.rs
+++ b/crates/lance-context-server/src/merge_execution.rs
@@ -1,5 +1,8 @@
 //! HTTP handlers only admit/cancel work. The owned executor, not the connection,
 //! holds the durable execution fence until the scoped storage future is gone.
+mod background;
+pub(crate) use background::{publish_background_request, spawn_background};
+
 use crate::{
     error::AppError,
     routes::{generic, rollouts},
@@ -49,6 +52,12 @@ pub struct Executions {
     coordinator: OnceCell<Coordinator>,
     etcd: Option<lance_context_merge::EtcdConfig>,
     pub(crate) rollout: lance_context_merge::rollout::MergeRollout,
+    endpoint: Option<String>,
+    fencing_mode: OnceCell<bool>,
+    mode_pending: std::sync::atomic::AtomicBool,
+    background_targets: std::sync::Mutex<std::collections::HashSet<String>>,
+    request_targets: std::sync::Mutex<std::collections::HashSet<String>>,
+    request_slots: Arc<tokio::sync::Semaphore>,
     queue_timeout_secs: u64,
     idle_timeout_secs: u64,
     instance: String,
@@ -64,6 +73,12 @@ impl Executions {
             coordinator: OnceCell::new_with(coordinator),
             etcd: None,
             rollout: Default::default(),
+            endpoint: None,
+            fencing_mode: OnceCell::new(),
+            mode_pending: std::sync::atomic::AtomicBool::new(false),
+            background_targets: Default::default(),
+            request_targets: Default::default(),
+            request_slots: Arc::new(tokio::sync::Semaphore::new(4)),
             queue_timeout_secs: 600,
             idle_timeout_secs: 600,
             instance: uuid::Uuid::new_v4().to_string(),
@@ -74,6 +89,7 @@ impl Executions {
     pub fn configured(
         etcd: lance_context_merge::EtcdConfig,
         rollout: lance_context_merge::rollout::MergeRollout,
+        endpoint: Option<String>,
         timeout_secs: u64,
         queue_timeout_secs: u64,
         idle_timeout_secs: u64,
@@ -83,6 +99,12 @@ impl Executions {
             coordinator: OnceCell::new(),
             etcd: Some(etcd),
             rollout,
+            endpoint,
+            fencing_mode: OnceCell::new(),
+            mode_pending: std::sync::atomic::AtomicBool::new(false),
+            background_targets: Default::default(),
+            request_targets: Default::default(),
+            request_slots: Arc::new(tokio::sync::Semaphore::new(4)),
             timeout_secs,
             queue_timeout_secs,
             idle_timeout_secs,
@@ -91,6 +113,59 @@ impl Executions {
         }
     }
 
+    /// Unknown mode defers maintenance, never ingestion. An activated protocol
+    /// is persisted so a configuration rollback cannot reopen untracked writes.
+    pub(crate) fn fencing_mode_known(&self) -> Option<bool> {
+        if !self.rollout.fence_legacy
+            && self.coordinator.get().is_none()
+            && self
+                .etcd
+                .as_ref()
+                .is_none_or(|c| c.etcd_endpoints.is_empty())
+        {
+            return Some(false);
+        }
+        self.fencing_mode.get().copied()
+    }
+
+    pub(crate) async fn fenced_mode(&self) -> Result<bool, AppError> {
+        if let Some(mode) = self.fencing_mode_known() {
+            return Ok(mode);
+        }
+        self.fencing_mode.get_or_try_init(|| async {
+            let mode = self.coordinator().await?.legacy_fencing_mode(self.rollout.fence_legacy)
+                .await.map_err(AppError::Internal)?;
+            if mode && self.endpoint.is_none() {
+                return Err(AppError::InvalidRequest("persisted merge protocol requires MERGE_EXECUTOR_ENDPOINT; refusing legacy rollback".into()));
+            }
+            Ok(mode)
+        }).await.copied()
+    }
+
+    pub(crate) fn resolve_mode_in_background(state: Arc<AppState>) {
+        use std::sync::atomic::Ordering;
+        if state.merge_executions.fencing_mode_known().is_some()
+            || state
+                .merge_executions
+                .mode_pending
+                .swap(true, Ordering::SeqCst)
+        {
+            return;
+        }
+        tokio::spawn(async move {
+            if let Err(error) = state.merge_executions.fenced_mode().await {
+                tracing::warn!(
+                    ?error,
+                    "merge protocol unresolved; ingestion continues, untracked merges deferred"
+                );
+            }
+            state
+                .merge_executions
+                .mode_pending
+                .store(false, Ordering::SeqCst);
+        });
+    }
+
     pub(crate) fn instance(&self) -> &str {
         &self.instance
     }
@@ -155,6 +230,7 @@ impl Executions {
 pub async fn capabilities(
     State(state): State<Arc<AppState>>,
 ) -> Result<Json<serde_json::Value>, AppError> {
+    Executions::resolve_mode_in_background(state.clone());
     Ok(Json(
         serde_json::json!({"protocol": 2, "instance": state.merge_executions.instance,
         "timeout_secs": state.merge_executions.timeout_secs,
@@ -162,6 +238,7 @@ pub async fn capabilities(
         "idle_timeout_secs": state.merge_executions.idle_timeout_secs,
         "progress_protocol": 1,
         "legacy_progress_protocol": 1,
+        "fenced_legacy_merges": state.merge_executions.fencing_mode_known() == Some(true),
         "rollout_append_protocol": 1,
         "shard_name": state.instance_id,
         "owned_targets": state.merge_executions.rollout.owned_targets,
@@ -174,7 +251,10 @@ pub async fn start(
     Json(execution): Json<Execution>,
 ) -> Result<StatusCode, AppError> {
     let coordinator = state.merge_executions.coordinator().await?;
-    if !state.merge_executions.owned(&execution.target)
+    if (!state.merge_executions.owned(&execution.target)
+        && !state.merge_executions.fenced_mode().await?)
+        || state.merge_executions.rollout.draining(&execution.target)
+        || execution.background
         || execution.maintenance.is_some()
         || execution.protocol != 2
         || execution.instance != state.merge_executions.instance
@@ -205,7 +285,14 @@ pub async fn start(
     // never leave a Running fence without an executor (or cancel a live write).
     let owned = state.clone();
     tokio::spawn(async move {
-        let result = run(owned.clone(), coordinator, execution.clone(), cancelled).await;
+        let result = run(
+            owned.clone(),
+            coordinator,
+            execution.clone(),
+            cancelled,
+            None,
+        )
+        .await;
         if let Err(error) = result {
             tracing::error!(id = %execution.id, target = %execution.target, %error, "merge execution ownership unresolved");
         }
@@ -224,6 +311,7 @@ async fn run(
     coordinator: Coordinator,
     execution: Execution,
     cancelled: watch::Receiver<bool>,
+    mut slot: Option<tokio::sync::OwnedSemaphorePermit>,
 ) -> Result<(), String> {
     // CAS errors are ambiguous: never retry storage work. Reconciliation sees
     // the durable fence, so an uncertain admission cannot release ownership.
@@ -246,7 +334,6 @@ async fn run(
         tokio::time::sleep(Duration::from_secs(1)).await;
     };
     let target = execution.target.clone();
-    let mut slot = None;
     let uri = match target.strip_prefix("generic:") {
         Some(name) => state.generic_uri(name),
         None => state.rollout_uri(&target),
@@ -293,13 +380,15 @@ async fn run(
     // not exceed the compiler's async layout recursion limit.
     let execute: futures::future::BoxFuture<'_, Result<usize, String>> = Box::pin(async {
         // No payload work starts before admission to the process-wide slot.
-        slot = execute_scoped(
-            async { Ok(state.acquire_merge_slot().await) },
-            Duration::from_secs(execution.queue_timeout_secs),
-            cancelled.clone(),
-        )
-        .await
-        .map_err(|e| format!("merge queue wait: {e}"))?;
+        if slot.is_none() {
+            slot = execute_scoped(
+                async { Ok(state.acquire_merge_slot().await) },
+                Duration::from_secs(execution.queue_timeout_secs),
+                cancelled.clone(),
+            )
+            .await
+            .map_err(|e| format!("merge queue wait: {e}"))?;
+        }
         if !coordinator.publish_progress(&running, 0).await? {
             return Err("merge ownership revoked before execution".into());
         }
@@ -479,7 +568,7 @@ mod tests {
     use tokio::io::AsyncWriteExt;
 
     #[tokio::test]
-    async fn unavailable_etcd_does_not_block_startup_or_disable_legacy_self_merge() {
+    async fn unavailable_etcd_does_not_block_startup_or_flush_configuration() {
         use clap::Parser;
         let dir = tempfile::tempdir().unwrap();
         let config = crate::config::ServerConfig::parse_from([
diff --git a/crates/lance-context-server/src/merge_execution/background.rs b/crates/lance-context-server/src/merge_execution/background.rs
new file mode 100644
index 0000000..b71f112
--- /dev/null
+++ b/crates/lance-context-server/src/merge_execution/background.rs
@@ -0,0 +1,575 @@
+//! Count/timer and direct HTTP merges retain their executor independently of
+//! callers. Admission uses the same durable owner and commit guard as fanout.
+use super::*;
+use tokio::sync::{oneshot, OwnedSemaphorePermit};
+
+type Reply = oneshot::Receiver<Result<usize, String>>;
+
+struct TargetGuard {
+    state: Arc<AppState>,
+    target: String,
+    request: bool,
+}
+impl Drop for TargetGuard {
+    fn drop(&mut self) {
+        let targets = if self.request {
+            &self.state.merge_executions.request_targets
+        } else {
+            &self.state.merge_executions.background_targets
+        };
+        targets.lock().unwrap().remove(&self.target);
+    }
+}
+
+/// Demand must reach the master even when every worker execution slot is
+/// occupied. Metadata publication has its own small bound and per-table
+/// coalescing; no payload slot or flush-pass wait is needed.
+pub(crate) fn publish_background_request(state: Arc<AppState>, target: String) {
+    let Ok(permit) = state
+        .merge_executions
+        .request_slots
+        .clone()
+        .try_acquire_owned()
+    else {
+        return;
+    };
+    if !state
+        .merge_executions
+        .request_targets
+        .lock()
+        .unwrap()
+        .insert(target.clone())
+    {
+        return;
+    }
+    let guard = TargetGuard {
+        state: state.clone(),
+        target: target.clone(),
+        request: true,
+    };
+    tokio::spawn(async move {
+        let _permit = permit;
+        let _guard = guard;
+        let result = tokio::time::timeout(
+            Duration::from_secs(2),
+            state.merge_executions.request_merge(&target),
+        )
+        .await;
+        let label = match result {
+            Ok(Ok(())) => "ok",
+            outcome => {
+                tracing::warn!(target, ?outcome, "master merge demand publication deferred");
+                "failed"
+            }
+        };
+        metrics::counter!("rollout_wal_background_requests_total", "result" => label).increment(1);
+    });
+}
+
+/// Never wait for a slot or coordinator IO on the flush sweep. Coalescing and
+/// the pre-acquired process slot bound both admission requests and executors.
+/// Dropping the returned receiver does not drop an admitted storage write.
+pub(crate) async fn spawn_background(
+    state: Arc<AppState>,
+    target: String,
+) -> Result<Reply, AppError> {
+    state.merge_executions.legacy_allowed(&target)?;
+    if !state.merge_executions.rollout.fence_legacy
+        && state.merge_executions.fencing_mode_known() != Some(true)
+    {
+        return Err(AppError::InvalidRequest(
+            "background fencing is disabled".into(),
+        ));
+    }
+    let endpoint = state.merge_executions.endpoint.clone().ok_or_else(|| {
+        AppError::InvalidRequest("background merge executor endpoint is missing".into())
+    })?;
+    let slot = state
+        .merge_slots
+        .as_ref()
+        .ok_or_else(|| AppError::InvalidRequest("background merges require bounded slots".into()))?
+        .clone()
+        .try_acquire_owned()
+        .map_err(|_| AppError::Overloaded("merge slots are occupied".into()))?;
+    if !state
+        .merge_executions
+        .background_targets
+        .lock()
+        .unwrap()
+        .insert(target.clone())
+    {
+        return Err(AppError::Overloaded(
+            "table already has a local merge admission".into(),
+        ));
+    }
+    let guard = TargetGuard {
+        state: state.clone(),
+        target: target.clone(),
+        request: false,
+    };
+    let mut execution = Execution::new(
+        &target,
+        &endpoint,
+        state.merge_executions.instance(),
+        state.merge_executions.timeout_secs,
+    );
+    execution.background = true;
+    execution.queue_timeout_secs = state.merge_executions.queue_timeout_secs;
+    execution.idle_timeout_secs = state.merge_executions.idle_timeout_secs;
+    let (cancel, cancelled) = watch::channel(false);
+    state
+        .merge_executions
+        .running
+        .lock()
+        .await
+        .insert(execution.id.clone(), cancel);
+    let (reply, result) = oneshot::channel();
+    // Retain reconciliation BEFORE the first fallible coordinator operation.
+    tokio::spawn(async move {
+        let _guard = guard;
+        let outcome = std::panic::AssertUnwindSafe(execute_background(
+            state.clone(),
+            &execution,
+            cancelled,
+            slot,
+        ))
+        .catch_unwind()
+        .await
+        .unwrap_or_else(|_| Err("background executor panicked; durable fence retained".into()));
+        if let Err(error) = &outcome {
+            tracing::warn!(target = %target, id = %execution.id, %error, "background merge did not complete");
+        }
+        state
+            .merge_executions
+            .running
+            .lock()
+            .await
+            .remove(&execution.id);
+        let _ = reply.send(outcome);
+    });
+    Ok(result)
+}
+
+async fn execute_background(
+    state: Arc<AppState>,
+    execution: &Execution,
+    cancelled: watch::Receiver<bool>,
+    slot: OwnedSemaphorePermit,
+) -> Result<usize, String> {
+    if !state
+        .merge_executions
+        .fenced_mode()
+        .await
+        .map_err(|e| format!("{e:?}"))?
+    {
+        return Err("fenced merge protocol is not active".into());
+    }
+    let coordinator = state
+        .merge_executions
+        .coordinator()
+        .await
+        .map_err(|e| format!("{e:?}"))?;
+    if *cancelled.borrow() {
+        return Err("background admission cancelled before reservation".into());
+    }
+    loop {
+        match coordinator.reserve_background(execution).await {
+            Ok(true) => break,
+            outcome => {
+                match coordinator.get(&execution.target).await {
+                    Ok(Some(current)) if current.id == execution.id => break,
+                    Ok(_) if matches!(outcome, Ok(false)) => {
+                        return Err("background merge deferred by ownership, admission gate or retry budget".into());
+                    }
+                    _ => {
+                        // An uncertain transaction may have admitted this exact
+                        // execution. Never start an untracked fallback or forget it.
+                        tokio::time::sleep(Duration::from_secs(1)).await;
+                    }
+                }
+            }
+        }
+    }
+    run(
+        state,
+        coordinator.clone(),
+        execution.clone(),
+        cancelled,
+        Some(slot),
+    )
+    .await?;
+    loop {
+        match coordinator.get(&execution.target).await {
+            Ok(Some(current)) if current.id == execution.id => {
+                if !matches!(current.phase, Phase::Finished | Phase::Recovered) {
+                    return Err(
+                        "background merge needs master storage recovery; fence retained".into(),
+                    );
+                }
+                match coordinator.release_background(&current).await {
+                    // False preserves a terminal record for an existing master
+                    // claim. Only that master may transfer the target lock.
+                    Ok(_) => return current.error.map_or(Ok(current.reclaimed), Err),
+                    Err(error) => {
+                        tracing::warn!(id = %execution.id, %error, "retrying background terminal release")
+                    }
+                }
+            }
+            Ok(_) => return Err("background execution reconciled by another owner".into()),
+            Err(error) => {
+                tracing::warn!(id = %execution.id, %error, "reconciling background terminal outcome")
+            }
+        }
+        tokio::time::sleep(Duration::from_secs(1)).await;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::sweeper;
+    use lance_context_core::{rollout_record_from_add_request, RolloutStore};
+    use tokio::sync::RwLock;
+
+    async fn fixture(slots: usize, idle: u64) -> (Arc<AppState>, Coordinator, tempfile::TempDir) {
+        let client =
+            etcd_client::Client::connect([std::env::var("ETCD_TEST_ENDPOINTS").unwrap()], None)
+                .await
+                .unwrap();
+        let coordinator = Coordinator::new(
+            client,
+            format!("/background-runtime/{}", uuid::Uuid::new_v4()),
+        );
+        let dir = tempfile::tempdir().unwrap();
+        let mut state = AppState::new_for_test(dir.path().to_path_buf()).await;
+        state.merge_slots = Some(Arc::new(tokio::sync::Semaphore::new(slots)));
+        state.rollout_merge_after_generations = 1;
+        state.merge_executions = Executions::new(Some(coordinator.clone()), 3600);
+        state.merge_executions.rollout.fence_legacy = true;
+        state.merge_executions.endpoint = Some("http://127.0.0.1:1".into());
+        state.merge_executions.idle_timeout_secs = idle;
+        (Arc::new(state), coordinator, dir)
+    }
+
+    async fn pending(state: &Arc<AppState>, name: &str) -> Arc<RwLock<RolloutStore>> {
+        let _ = rollouts::create_rollout_store(
+            State(state.clone()),
+            Json(lance_context_api::CreateRolloutStoreRequest {
+                name: name.into(),
+                storage_options: None,
+            }),
+        )
+        .await
+        .unwrap();
+        let store = state.get_or_open_rollout_store(name).await.unwrap();
+        let row = rollout_record_from_add_request(&lance_context_api::AddRolloutRequest {
+            id: "row".into(),
+            rollout_id: "trajectory".into(),
+            ..Default::default()
+        });
+        store.read().await.add(&[row]).await.unwrap();
+        store.read().await.flush().await.unwrap();
+        store
+    }
+
+    async fn running(coordinator: &Coordinator, target: &str) -> Execution {
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if let Some(e) = coordinator.get(target).await.unwrap() {
+                    if e.phase == Phase::Running {
+                        return e;
+                    }
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .unwrap()
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn disconnected_http_keeps_admitted_executor_and_reclaims_real_wal() {
+        let (state, coordinator, _dir) = fixture(1, 60).await;
+        let store = pending(&state, "hot").await;
+        let held = store.write().await;
+        let caller = tokio::spawn(rollouts::merge_wal(
+            State(state.clone()),
+            axum::extract::Path("hot".into()),
+        ));
+        let e = running(&coordinator, "hot").await;
+        assert!(e.background);
+        caller.abort();
+        assert!(caller.await.unwrap_err().is_cancelled());
+        assert_eq!(state.merge_slots.as_ref().unwrap().available_permits(), 0);
+        assert_eq!(coordinator.get("hot").await.unwrap().unwrap().id, e.id);
+        drop(held);
+        tokio::time::timeout(Duration::from_secs(20), async {
+            loop {
+                if !state
+                    .merge_executions
+                    .background_targets
+                    .lock()
+                    .unwrap()
+                    .contains("hot")
+                {
+                    break;
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .unwrap();
+        assert!(coordinator.get("hot").await.unwrap().is_none());
+        assert_eq!(
+            store.read().await.pending_wal_generations().await.unwrap(),
+            0
+        );
+        assert!(store.read().await.get_by_id("row").await.unwrap().is_some());
+        assert_eq!(state.merge_slots.as_ref().unwrap().available_permits(), 1);
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn count_and_timer_triggers_execute_without_holding_flush_sweep() {
+        let (state, coordinator, _dir) = fixture(1, 60).await;
+        let store = pending(&state, "hot").await;
+        // A reader permits the count predicate and preparation, but blocks
+        // commit. The sweeper must return while the owned executor waits.
+        let held = store.read().await;
+        tokio::time::timeout(
+            Duration::from_secs(2),
+            sweeper::flush_pass_coordinated(
+                vec![("hot".into(), store.clone())],
+                Duration::from_secs(5),
+                state.merge_slots.clone(),
+                Some(state.clone()),
+            ),
+        )
+        .await
+        .unwrap();
+        let e = running(&coordinator, "hot").await;
+        tokio::time::timeout(
+            Duration::from_secs(2),
+            sweeper::merge_pass_coordinated(
+                vec![("hot".into(), store.clone())],
+                Duration::from_secs(5),
+                Some(state.clone()),
+            ),
+        )
+        .await
+        .unwrap();
+        assert_eq!(coordinator.get("hot").await.unwrap().unwrap().id, e.id);
+        drop(held);
+        tokio::time::timeout(Duration::from_secs(20), async {
+            while store.read().await.pending_wal_generations().await.unwrap() != 0 {
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .unwrap();
+        state.merge_executions.shutdown().await;
+        assert!(coordinator.get("hot").await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn busy_slots_do_not_create_unbounded_admissions_or_hide_other_flushes() {
+        let (state, coordinator, _dir) = fixture(1, 60).await;
+        let store = pending(&state, "hot").await;
+        let held = state.acquire_merge_slot().await;
+        for _ in 0..32 {
+            assert!(matches!(
+                spawn_background(state.clone(), "hot".into()).await,
+                Err(AppError::Overloaded(_))
+            ));
+        }
+        tokio::time::timeout(
+            Duration::from_secs(2),
+            sweeper::flush_pass_coordinated(
+                vec![("hot".into(), store.clone())],
+                Duration::from_secs(5),
+                state.merge_slots.clone(),
+                Some(state.clone()),
+            ),
+        )
+        .await
+        .unwrap();
+        assert!(state.merge_executions.running.lock().await.is_empty());
+        tokio::time::timeout(Duration::from_secs(20), async {
+            loop {
+                if coordinator
+                    .request_page(None)
+                    .await
+                    .unwrap()
+                    .0
+                    .iter()
+                    .any(|r| r.target == "hot")
+                {
+                    break;
+                }
+                // Publication has a bounded metadata timeout. Model the next
+                // periodic flush after a deferred publish while retaining the
+                // occupied execution slot; retry must not need merge capacity.
+                tokio::time::timeout(
+                    Duration::from_secs(2),
+                    sweeper::flush_pass_coordinated(
+                        vec![("hot".into(), store.clone())],
+                        Duration::from_secs(5),
+                        state.merge_slots.clone(),
+                        Some(state.clone()),
+                    ),
+                )
+                .await
+                .expect("metadata retry must not hold the flush sweep");
+                assert!(state.merge_executions.running.lock().await.is_empty());
+                tokio::time::sleep(Duration::from_millis(100)).await;
+            }
+        })
+        .await
+        .expect("saturated worker must still request master catchup");
+        assert!(coordinator.get("hot").await.unwrap().is_none());
+        assert_eq!(
+            store.read().await.pending_wal_generations().await.unwrap(),
+            1
+        );
+        drop(held);
+        assert_eq!(
+            spawn_background(state.clone(), "hot".into())
+                .await
+                .unwrap()
+                .await
+                .unwrap()
+                .unwrap(),
+            1
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn another_tables_success_does_not_renew_a_stalled_executor() {
+        let (state, coordinator, _dir) = fixture(2, 10).await;
+        let slow = pending(&state, "slow").await;
+        let fast = pending(&state, "fast").await;
+        let held = slow.write().await;
+        let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let commits = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let healthy = tokio::spawn({
+            let state = state.clone();
+            let stop = stop.clone();
+            let commits = commits.clone();
+            async move {
+                while !stop.load(std::sync::atomic::Ordering::SeqCst) {
+                    let reclaimed = spawn_background(state.clone(), "fast".into())
+                        .await
+                        .unwrap()
+                        .await
+                        .unwrap()
+                        .unwrap();
+                    assert!(reclaimed > 0);
+                    let id = commits.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                    let row =
+                        rollout_record_from_add_request(&lance_context_api::AddRolloutRequest {
+                            id: format!("row-{id}"),
+                            rollout_id: "trajectory".into(),
+                            ..Default::default()
+                        });
+                    fast.read().await.add(&[row]).await.unwrap();
+                    fast.read().await.flush().await.unwrap();
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                }
+            }
+        });
+        // Establish actual commit traffic before the idle window starts. Cold
+        // index/storage setup can otherwise consume the entire short window on
+        // a busy test host without ever exercising concurrent progress.
+        tokio::time::timeout(Duration::from_secs(30), async {
+            while commits.load(std::sync::atomic::Ordering::SeqCst) < 2 {
+                assert!(!healthy.is_finished());
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .unwrap();
+        let stalled = spawn_background(state.clone(), "slow".into())
+            .await
+            .unwrap();
+        running(&coordinator, "slow").await;
+        let before = commits.load(std::sync::atomic::Ordering::SeqCst);
+        let error = tokio::time::timeout(Duration::from_secs(30), stalled)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap_err();
+        // Real commits continue throughout the stalled table's idle window.
+        assert!(!healthy.is_finished());
+        stop.store(true, std::sync::atomic::Ordering::SeqCst);
+        healthy.await.unwrap();
+        assert!(commits.load(std::sync::atomic::Ordering::SeqCst) - before >= 2);
+        assert!(error.contains("no-progress"), "{error}");
+        assert!(coordinator.get("slow").await.unwrap().is_none());
+        drop(held);
+        assert_eq!(
+            slow.read().await.pending_wal_generations().await.unwrap(),
+            1
+        );
+        assert_eq!(state.merge_slots.as_ref().unwrap().available_permits(), 2);
+    }
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn restarted_worker_keeps_fencing_after_flag_rollback() {
+        let (state, coordinator, dir) = fixture(1, 60).await;
+        let store = pending(&state, "hot").await;
+        assert!(state.merge_executions.fenced_mode().await.unwrap());
+        let mut restarted = AppState::new_for_test(dir.path().to_path_buf()).await;
+        restarted.merge_slots = Some(Arc::new(tokio::sync::Semaphore::new(1)));
+        restarted.merge_executions = Executions::new(Some(coordinator.clone()), 3600);
+        restarted.merge_executions.endpoint = Some("http://127.0.0.1:1".into());
+        assert!(!restarted.merge_executions.rollout.fence_legacy);
+        assert!(restarted.merge_executions.fenced_mode().await.unwrap());
+        let restarted = Arc::new(restarted);
+        let held = store.write().await;
+        // The reopened worker has its own handle. Hold its lock to inspect the
+        // actual HTTP admission before allowing storage work to run.
+        let reopened = restarted.get_or_open_rollout_store("hot").await.unwrap();
+        let blocked = reopened.write().await;
+        let task = tokio::spawn(rollouts::merge_wal(
+            State(restarted.clone()),
+            axum::extract::Path("hot".into()),
+        ));
+        assert!(running(&coordinator, "hot").await.background);
+        drop(blocked);
+        drop(held);
+        let Json(result) = task.await.unwrap().unwrap();
+        assert_eq!(result.reclaimed, 1);
+        assert!(coordinator.get("hot").await.unwrap().is_none());
+        let missing_endpoint = Executions::new(Some(coordinator), 3600);
+        assert!(
+            missing_endpoint.fenced_mode().await.is_err(),
+            "rollback must not silently restore legacy writes"
+        );
+    }
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn fenced_mode_blocks_untracked_worker_compaction_before_store_lock() {
+        let (state, _coordinator, _dir) = fixture(1, 60).await;
+        let store = pending(&state, "hot").await;
+        let held = store.read().await;
+        let result = tokio::time::timeout(
+            Duration::from_secs(2),
+            rollouts::compact_rollout(
+                State(state.clone()),
+                axum::extract::Path("hot".into()),
+                Json(serde_json::from_value(serde_json::json!({})).unwrap()),
+            ),
+        )
+        .await
+        .unwrap();
+        assert!(matches!(result, Err(AppError::Overloaded(_))));
+        drop(held);
+        assert_eq!(
+            store.read().await.pending_wal_generations().await.unwrap(),
+            1
+        );
+    }
+}
diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs
index 07f3b43..287ba02 100644
--- a/crates/lance-context-server/src/routes/generic.rs
+++ b/crates/lance-context-server/src/routes/generic.rs
@@ -297,6 +297,19 @@ pub async fn merge_generic_wal(
     state
         .merge_executions
         .legacy_allowed(&format!("generic:{name}"))?;
+    if state.merge_executions.fenced_mode().await? {
+        crate::merge_execution::publish_background_request(
+            state.clone(),
+            format!("generic:{name}"),
+        );
+        let result =
+            crate::merge_execution::spawn_background(state, format!("generic:{name}")).await?;
+        let reclaimed = result
+            .await
+            .map_err(|e| AppError::Internal(e.to_string()))?
+            .map_err(AppError::Internal)?;
+        return Ok(Json(serde_json::json!({"reclaimed": reclaimed})));
+    }
     let execution_state = state.clone();
     let execution_name = name.clone();
     state
diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs
index 8df9514..1b00fc1 100644
--- a/crates/lance-context-server/src/routes/rollouts.rs
+++ b/crates/lance-context-server/src/routes/rollouts.rs
@@ -641,6 +641,11 @@ pub async fn compact_rollout(
     // Once a table is draining or owned, only the master's fenced maintenance
     // path may publish a rewrite. Check before opening or locking its handle.
     state.merge_executions.legacy_allowed(&name)?;
+    if state.merge_executions.fenced_mode().await? {
+        return Err(AppError::Overloaded(
+            "fenced table maintenance must run through the master".into(),
+        ));
+    }
     let store_lock = state.get_or_open_rollout_store(&name).await?;
 
     let config = if req.target_rows_per_fragment.is_some() || req.materialize_deletions.is_some() {
@@ -715,6 +720,15 @@ pub async fn merge_wal(
     Path(name): Path<String>,
 ) -> Result<Json<MergeWalResponse>, AppError> {
     state.merge_executions.legacy_allowed(&name)?;
+    if state.merge_executions.fenced_mode().await? {
+        crate::merge_execution::publish_background_request(state.clone(), name.clone());
+        let result = crate::merge_execution::spawn_background(state, name).await?;
+        let reclaimed = result
+            .await
+            .map_err(|e| AppError::Internal(e.to_string()))?
+            .map_err(AppError::Internal)?;
+        return Ok(Json(MergeWalResponse { reclaimed }));
+    }
     let execution_state = state.clone();
     let execution_name = name.clone();
     state
diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs
index cebd07c..122080c 100644
--- a/crates/lance-context-server/src/state.rs
+++ b/crates/lance-context-server/src/state.rs
@@ -364,10 +364,31 @@ impl AppState {
                 "merge execution, queue and idle timeouts must be positive".into(),
             ));
         }
+        if config.merge_rollout.fence_legacy || config.merge_executor_endpoint.is_some() {
+            let endpoint = config.merge_executor_endpoint.as_deref().unwrap_or("");
+            let valid_endpoint = endpoint.parse::<axum::http::Uri>().is_ok_and(|url| {
+                matches!(url.scheme_str(), Some("http" | "https"))
+                    && url
+                        .authority()
+                        .is_some_and(|a| !a.host().is_empty() && !a.as_str().contains('@'))
+                    && url.query().is_none()
+                    && matches!(url.path(), "" | "/")
+                    && !endpoint.contains('#')
+            });
+            if !valid_endpoint
+                || config.etcd.etcd_endpoints.is_empty()
+                || config.rollout_merge_concurrency == 0
+            {
+                return Err(AppError::InvalidRequest(
+                    "fenced legacy merges require ETCD_ENDPOINTS, a stable MERGE_EXECUTOR_ENDPOINT and positive ROLLOUT_MERGE_CONCURRENCY".into(),
+                ));
+            }
+        }
         Ok(Self {
             merge_executions: crate::merge_execution::Executions::configured(
                 config.merge_etcd_config(),
                 config.merge_rollout.clone(),
+                config.merge_executor_endpoint.clone(),
                 config.merge_execution_timeout_secs,
                 config.merge_queue_timeout_secs,
                 config.merge_idle_timeout_secs,
diff --git a/crates/lance-context-server/src/sweeper.rs b/crates/lance-context-server/src/sweeper.rs
index 66449ee..783980e 100644
--- a/crates/lance-context-server/src/sweeper.rs
+++ b/crates/lance-context-server/src/sweeper.rs
@@ -481,7 +481,26 @@ async fn route_merge<S: Sweepable>(
     // unchanged.
     spawn_publish_demand(state.clone(), target.clone(), S::clone(store));
     if !state.merge_executions.owned(&target) {
-        return Ok(false);
+        match state.merge_executions.fencing_mode_known() {
+            Some(false) => return Ok(false),
+            None if !state.merge_executions.rollout.fence_legacy => {
+                crate::merge_execution::Executions::resolve_mode_in_background(state.clone());
+                return Ok(true);
+            }
+            _ => {}
+        }
+        if !count_trigger || store.count_merge_due().await? {
+            crate::merge_execution::publish_background_request(state.clone(), target.clone());
+            match crate::merge_execution::spawn_background(state.clone(), target).await {
+                Ok(_reply) => {} // Executor is retained independently of this sweep.
+                Err(crate::error::AppError::Overloaded(_)) => {
+                    metrics::counter!("rollout_wal_self_merge_skipped_total", "kind" => S::kind())
+                        .increment(1);
+                }
+                Err(error) => return Err(format!("{error:?}")),
+            }
+        }
+        return Ok(true);
     }
     if !count_trigger || store.count_merge_due().await? {
         state.merge_executions.request_merge(&target).await?;
diff --git a/docs/worker-merge-admission.md b/docs/worker-merge-admission.md
new file mode 100644
index 0000000..cccf471
--- /dev/null
+++ b/docs/worker-merge-admission.md
@@ -0,0 +1,103 @@
+# Durable worker merge admission (integration in progress)
+
+Worker count-triggered and timer-triggered merges previously wrote without a
+persistent execution record. A master could observe a hung request but could
+not establish what that writer might still commit. This path records those
+executions before storage work and reuses the owned executor's useful-progress
+watchdog, conditional commit permits, cancellation and storage fencing.
+
+## Behavior
+
+`MERGE_FENCE_LEGACY=true` enables the protocol for new rollout and generic worker
+merges, independently of the static native-ownership list. It requires a stable
+`MERGE_EXECUTOR_ENDPOINT`, private etcd configuration, and positive
+`ROLLOUT_MERGE_CONCURRENCY`. Master fanout remains serial. Ordinary HTTP merges,
+count triggers and cleanup timers use the same process slots and executor.
+
+A background trigger tries a slot before spawning; it never waits for capacity
+or etcd on the flush sweep. Due demand is also published through a separate
+four-request metadata bound, so occupied execution slots do not hide a hot table
+from the master. Per-table admission and demand are coalesced. The detached
+executor exists before admission IO, so a disconnected HTTP caller cannot
+abandon a writer. Its slot remains occupied until its storage leaves drain or
+are fenced. An ambiguous terminal record is retained for master recovery.
+
+Admission atomically persists the execution, exclusive owner and coalesced
+request. It respects canonical claims, catchup ownership, prepared-commit
+priority, admission closure and persisted failure budgets. A terminal worker
+cannot delete an owner being transferred to a waiting master claim.
+
+The master inventories worker records as well as maintenance records, including
+records whose original worker is no longer configured. It observes actual
+execution progress before cancellation, reconciles ambiguous commits through
+the existing storage barrier, and never treats an empty endpoint list as a
+successful catchup. New-table demand can enqueue fenced worker fanout without
+adding a static owned target. Until automatic native routing is enabled, an
+existing append wildcard applies only to explicitly owned tables; new targets
+continue through fenced worker fanout instead of entering an ineligible local
+execution path.
+
+Worker HTTP compaction is refused in this mode; tracked master maintenance must
+handle it. This prevents an untracked worker rewrite from bypassing merge
+ownership.
+
+Concurrent merge progress can invalidate several read snapshots in well under
+a second. Consistent reads now distinguish that progress from repeated missing
+data at the same base version: up to 16 snapshot attempts are allowed, while
+three missing-file failures without a new base commit still stop the request.
+No additional attempt starts after the two-second retry budget is consumed;
+that budget does not cancel an otherwise valid long scan. Reads never return
+partial results or discard required WAL generations to satisfy this bound.
+Rollout point reads apply that recovery to their WAL fallback while preserving
+the cheap cached-base hit and WAL-only semantics. Generic filtered/point reads
+keep their cached-view behavior on success or ordinary miss, but recover an
+actual missing-file race through a fresh consistent snapshot. Pinned reads do
+not advance; invalid filters are not retried as missing storage.
+
+## Persisted protocol and startup
+
+The private-etcd `worker-merge-admission-mode` key is persistent and unleased.
+Once activated, disabling the initiation flag on a restarted participant does
+not restore legacy writes. An unknown newer marker is rejected, including when
+activation is requested. The master resolves this mode at startup. Workers
+resolve it independently of startup/flush; unresolved metadata defers maintenance
+and cannot authorize a legacy fallback. Participants without etcd and without
+the opt-in keep standalone legacy behavior.
+
+A worker that has resolved legacy mode retains that mode for its incarnation.
+Consequently, initial activation requires a coordinated transition covering
+all old worker/master instances and external helpers. A marker, a Pod restart,
+or an empty execution inventory does not prove that historical writers are
+quiescent. Removing the shared etcd namespace/configuration or deploying an old
+binary is not a supported rollback after activation.
+
+## Automatic native routing (validation in progress)
+
+After the initial legacy-writer transition is verified, explicitly enabling
+`MASTER_LOCAL_MERGE_AUTO=true` with `MASTER_LOCAL_MERGE_TARGETS=*` routes new
+rollout/generic demand through resident masters without changing a table
+allowlist. Worker protocol activation also enables fenced master maintenance.
+Automatic routing requires that persisted protocol, stable shard IDs, positive
+merge capacity and the existing shared memory budget. It creates no Pods or
+second queue. Draining targets and datagen tables remain excluded.
+
+The canonical task claim excludes new background admission. A progressing
+background executor is reconciled to completion before native catchup starts;
+uncertain storage writes retain their fence until recovery succeeds. Native
+catchup uses bounded passes and publishes continuation demand before releasing
+the claim, so another master can resume it. Existing explicit append-protocol
+targets retain that optimization; newly selected targets use native sealed-WAL
+catchup, including generic updates.
+
+Neither automatic routing nor the activation marker proves old writers are
+quiescent. Before enabling this in production, finish and validate:
+
+- Complete old-writer coverage, including helpers and pre-upgrade master work.
+- Combined staging of worker admission, tracked maintenance and automatic
+  resident execution with ongoing ingestion, progress stalls and restart.
+- Measured production handoff under the actual traffic and memory limits.
+
+The source paths and regression tests are implemented but final validation is
+still running. Keep automatic routing disabled until the initial transition
+is proven. Existing production helpers remain necessary until the handoff is
+demonstrated.