diff --git a/crates/lance-context-core/src/merge_write_scope.rs b/crates/lance-context-core/src/merge_write_scope.rs
index 3cc31ae..55dcbe1 100644
--- a/crates/lance-context-core/src/merge_write_scope.rs
+++ b/crates/lance-context-core/src/merge_write_scope.rs
@@ -106,6 +106,9 @@ pub struct MergeWriteScope {
     completed_steps: Arc<AtomicU64>,
     work: Mutex<MergeWork>,
     observe_file_io: bool,
+    /// Legacy diagnostics must not introduce detached storage tasks or alter
+    /// cancellation/slot lifetimes before durable ownership is available.
+    observation_only: bool,
     local_io: Mutex<Vec<(lance_io::utils::tracking_store::IOTracker, u64)>>,
     progress: Mutex<Progress>,
     changed: Notify,
@@ -117,7 +120,23 @@ pub struct MergeWriteScope {
     io_shim: Option<Arc<dyn lance_io::object_store::WrappingObjectStore>>,
 }
 
+/// Local write observations, not a durable fence or proof of remote quiescence.
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct MergeWriteStatus {
+    pub active_commits: usize,
+    pub admission_closed: bool,
+    pub uncertain_commit: bool,
+}
+
 impl MergeWriteScope {
+    pub fn write_status(&self) -> MergeWriteStatus {
+        let progress = self.progress.lock().unwrap();
+        MergeWriteStatus {
+            active_commits: progress.active,
+            admission_closed: progress.closed,
+            uncertain_commit: progress.uncertain,
+        }
+    }
     pub fn work(&self) -> MergeWork {
         self.work.lock().unwrap().clone()
     }
@@ -139,6 +158,13 @@ impl MergeWriteScope {
         Arc::new(Self::default())
     }
 
+    pub fn for_legacy_observation() -> Arc<Self> {
+        Arc::new(Self {
+            observation_only: true,
+            ..Self::default()
+        })
+    }
+
     pub fn with_authorizer(authorizer: Arc<dyn CommitAuthorizer>) -> Arc<Self> {
         Arc::new(Self {
             authorizer: Some(authorizer),
@@ -241,6 +267,18 @@ pub(crate) fn wal_batch_read(rows: usize, bytes: usize) {
 }
 
 pub(crate) fn base_committed() {
+    // Legacy observation counts at the commit handler, including fallback
+    // paths which also invoke the owned executor's logical checkpoints.
+    if CURRENT
+        .try_with(|scope| scope.observation_only)
+        .unwrap_or(false)
+    {
+        return;
+    }
+    record_base_commit();
+}
+
+fn record_base_commit() {
     let _ = CURRENT.try_with(|scope| {
         let mut work = scope.work.lock().unwrap();
         work.base_commits += 1;
@@ -250,6 +288,16 @@ pub(crate) fn base_committed() {
 }
 
 pub(crate) fn generations_reclaimed(count: usize) {
+    if CURRENT
+        .try_with(|scope| scope.observation_only)
+        .unwrap_or(false)
+    {
+        return;
+    }
+    record_reclaimed(count);
+}
+
+fn record_reclaimed(count: usize) {
     if count == 0 {
         return;
     }
@@ -423,6 +471,18 @@ where
         scope,
         completed: false,
     };
+    if active.scope.observation_only {
+        let mut active = active;
+        let result = write.await;
+        if progressed(&result) {
+            active.scope.completed_steps.fetch_add(1, Ordering::Relaxed);
+        }
+        if result.is_err() {
+            active.scope.mark_uncertain();
+        }
+        active.completed = true;
+        return result;
+    }
     let (send, receive) = oneshot::channel();
     let leaf = tokio::spawn(async move {
         let mut active = active;
@@ -465,8 +525,10 @@ pub(crate) async fn drain_generations(
             .version
             .checked_add(1)
             .ok_or_else(|| Error::io("manifest version overflow"))?;
+        let before = next.sstables.len();
         next.sstables
             .retain(|g| !generations.contains(&g.generation));
+        let removed = before - next.sstables.len();
         authorize(&resource, next.version).await?;
         let writer = store.clone();
         let scope = CURRENT.try_with(Arc::clone).ok();
@@ -487,7 +549,15 @@ pub(crate) async fn drain_generations(
         )
         .await?;
         match written {
-            Ok(_) => return Ok(next),
+            Ok(_) => {
+                if CURRENT
+                    .try_with(|scope| scope.observation_only)
+                    .unwrap_or(false)
+                {
+                    record_reclaimed(removed);
+                }
+                return Ok(next);
+            }
             Err(Error::RetryableCommitConflict { .. }) => continue,
             Err(error) => return Err(error),
         }
@@ -718,6 +788,13 @@ impl CommitHandler for GuardedCommit {
             )
             .await?;
             *manifest = committed;
+            if location.is_ok()
+                && CURRENT
+                    .try_with(|scope| scope.observation_only)
+                    .unwrap_or(false)
+            {
+                record_base_commit();
+            }
             location
         };
         match &self.scope {
@@ -796,6 +873,45 @@ mod tests {
     use super::*;
     use std::sync::atomic::{AtomicBool, Ordering};
 
+    #[tokio::test]
+    async fn legacy_observation_retains_uncertainty_without_detaching_storage() {
+        struct LocalWrite(Arc<AtomicBool>);
+        impl Drop for LocalWrite {
+            fn drop(&mut self) {
+                self.0.store(false, Ordering::SeqCst);
+            }
+        }
+        let scope = MergeWriteScope::for_legacy_observation();
+        let writing = Arc::new(AtomicBool::new(false));
+        let flag = writing.clone();
+        let (entered, began) = oneshot::channel();
+        let run_scope = scope.clone();
+        let caller = tokio::spawn(async move {
+            run_scope
+                .run(shield(async move {
+                    flag.store(true, Ordering::SeqCst);
+                    let _local_write = LocalWrite(flag);
+                    entered.send(()).unwrap();
+                    std::future::pending::<Result<()>>().await
+                }))
+                .await
+        });
+        began.await.unwrap();
+        assert_eq!(scope.write_status().active_commits, 1);
+        caller.abort();
+        caller.await.unwrap_err();
+        assert!(
+            !writing.load(Ordering::SeqCst),
+            "diagnostics must not detach storage work"
+        );
+        assert_eq!(scope.write_status().active_commits, 0);
+        assert!(
+            scope.write_status().uncertain_commit,
+            "remote completion is still unknown"
+        );
+        assert_eq!(scope.completed_steps(), 0);
+    }
+
     #[tokio::test]
     async fn phase_changes_and_observation_do_not_count_as_completed_work() {
         let scope = MergeWriteScope::new();
@@ -939,6 +1055,8 @@ mod tests {
         began.await.unwrap();
         merge.abort();
         assert!(merge.await.unwrap_err().is_cancelled());
+        assert_eq!(scope.write_status().active_commits, 1);
+        assert!(!scope.write_status().uncertain_commit);
         let draining_scope = scope.clone();
         let drain = tokio::spawn(async move { draining_scope.drain().await });
         tokio::task::yield_now().await;
@@ -948,6 +1066,8 @@ mod tests {
         );
         release.send(()).unwrap();
         drain.await.unwrap();
+        assert_eq!(scope.write_status().active_commits, 0);
+        assert!(scope.write_status().admission_closed);
         assert!(committed.load(Ordering::SeqCst));
         assert!(!scope.has_uncertain_commit());
         assert!(
diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs
index 2c3daa0..f3e01f9 100644
--- a/crates/lance-context-core/src/store_base.rs
+++ b/crates/lance-context-core/src/store_base.rs
@@ -1353,8 +1353,11 @@ impl StorageBase {
                             break 'generations;
                         }
                     }
+                    crate::merge_write_scope::wal_batch_read(
+                        batch.num_rows(),
+                        batch.get_array_memory_size(),
+                    );
                     current_batches.push(batch);
-                    crate::merge_write_scope::checkpoint();
                 }
             }
             generation_batches.push((flushed.generation, current_batches));
@@ -1429,6 +1432,7 @@ impl StorageBase {
         if let Some(watermark) = watermark {
             let params = WriteParams {
                 mode: WriteMode::Append,
+                write_progress: Some(crate::merge_write_scope::write_progress()),
                 ..Default::default()
             };
             let staged = lance::dataset::InsertBuilder::new(Arc::new(self.dataset.clone()))
@@ -1448,7 +1452,15 @@ impl StorageBase {
                 batches.into_iter().map(Ok::<RecordBatch, ArrowError>),
                 merge_schema,
             );
-            Box::pin(self.dataset.append(reader, None)).await?;
+            Box::pin(self.dataset.append(
+                reader,
+                Some(WriteParams {
+                    mode: WriteMode::Append,
+                    write_progress: Some(crate::merge_write_scope::write_progress()),
+                    ..Default::default()
+                }),
+            ))
+            .await?;
         }
         Ok(())
     }
diff --git a/crates/lance-context-master/src/merge_execution.rs b/crates/lance-context-master/src/merge_execution.rs
index 3ff7aa6..730dffc 100644
--- a/crates/lance-context-master/src/merge_execution.rs
+++ b/crates/lance-context-master/src/merge_execution.rs
@@ -179,8 +179,11 @@ async fn run_legacy_workers(
             Err((false, error)) => {
                 // A response error, reset, or timeout after sending may leave
                 // an unfenced legacy writer alive. Do not advance the fan-out.
+                // Fetch evidence from the same worker without opening stores or
+                // querying etcd. This never permits retry, takeover or restart.
+                let progress = legacy_progress(http, endpoint, target).await;
                 return Err(format!(
-                    "merged {reclaimed} generations across {completed}/{} workers (legacy); stopped at {endpoint}: {error}",
+                    "merged {reclaimed} generations across {completed}/{} workers (legacy); stopped at {endpoint}: {error}; legacy progress: {progress}",
                     endpoints.len()
                 ));
             }
@@ -201,6 +204,67 @@ async fn run_legacy_workers(
     }
 }
 
+async fn legacy_progress(http: &reqwest::Client, endpoint: &str, target: &str) -> String {
+    let result =
+        tokio::time::timeout(Duration::from_secs(2), async {
+            let mut response = http
+                .get(format!("{endpoint}/api/v1/internal/legacy-merges"))
+                .query(&[("target", target)])
+                .send()
+                .await
+                .map_err(|e| e.to_string())?
+                .error_for_status()
+                .map_err(|e| e.to_string())?;
+            let mut bytes = Vec::new();
+            while let Some(chunk) = response.chunk().await.map_err(|e| e.to_string())? {
+                if bytes.len().saturating_add(chunk.len()) > 64 * 1024 {
+                    return Err("snapshot exceeds 64 KiB".to_string());
+                }
+                bytes.extend_from_slice(&chunk);
+            }
+            let value: serde_json::Value =
+                serde_json::from_slice(&bytes).map_err(|e| e.to_string())?;
+            if value["instance"].as_str().is_none()
+                || value["snapshot"]["protocol"] != 1
+                || value["snapshot"]["takeover_authorized"] != false
+            {
+                return Err("unsupported progress response".to_string());
+            }
+            let snapshot = &value["snapshot"];
+            let executions = snapshot["executions"]
+                .as_array()
+                .ok_or_else(|| "missing execution observations".to_string())?;
+            // Task errors are stored in etcd. Keep the durable summary small even
+            // if the endpoint returns a large history or an unexpected field.
+            let summaries: Vec<_> = executions.iter().take(8).map(|row| serde_json::json!({
+            "id": row["id"].as_str().map(|s| s.chars().take(128).collect::<String>()),
+            "phase": row["phase"].as_str().map(|s| s.chars().take(32).collect::<String>()),
+            "elapsed_ms": row["elapsed_ms"].as_u64(),
+            "completed_steps": row["completed_steps"].as_u64(),
+            "base_commits": row["work"]["base_commits"].as_u64(),
+            "generations_reclaimed": row["work"]["generations_reclaimed"].as_u64(),
+            "wal_decoded_bytes": row["work"]["wal_decoded_bytes"].as_u64(),
+            "encoded_bytes": row["work"]["encoded_bytes"].as_u64(),
+            "active_commits": row["writes"]["active_commits"].as_u64(),
+            "uncertain_commit": row["writes"]["uncertain_commit"].as_bool(),
+        })).collect();
+            Ok(serde_json::json!({
+            "instance": value["instance"].as_str().unwrap().chars().take(128).collect::<String>(),
+            "takeover_authorized": false,
+            "untracked_admissions": snapshot["untracked_admissions"].as_u64(),
+            "evicted_history": snapshot["evicted_history"].as_u64(),
+            "truncated": executions.len() > 8 || snapshot["truncated"].as_bool() != Some(false),
+            "executions": summaries,
+        }).to_string())
+        })
+        .await;
+    match result {
+        Ok(Ok(snapshot)) => snapshot,
+        Ok(Err(error)) => format!("unavailable ({error}); ownership unresolved"),
+        Err(_) => "unavailable (observation timed out); ownership unresolved".into(),
+    }
+}
+
 struct Deadlines {
     queue: tokio::time::Instant,
     idle_timeout: Duration,
@@ -1197,6 +1261,46 @@ mod legacy_connection_tests {
         .await
     }
 
+    #[tokio::test]
+    async fn progress_evidence_is_reported_but_never_allows_unsafe_fanout() {
+        let (bad, failed_server) = serve(Router::new()
+            .route("/api/v1/internal/legacy-merges", axum::routing::get(
+                |axum::extract::Query(query): axum::extract::Query<std::collections::HashMap<String, String>>| async move {
+                    assert_eq!(query.get("target").unwrap(), "table");
+                    Json(serde_json::json!({"instance":"boot-1", "snapshot":{
+                        "protocol":1, "takeover_authorized":false,
+                        "executions":[{"id":"exec-1","completed_steps":42,"writes":{"active_commits":1}}]
+                    }}))
+                }))
+            .fallback(post(|| async { StatusCode::INTERNAL_SERVER_ERROR }))).await;
+        let calls = Arc::new(AtomicUsize::new(0));
+        let (good, server) = healthy(calls.clone()).await;
+        let http = crate::state::worker_http_client().unwrap();
+        let error = run_legacy_workers(&http, &[bad, good], "table")
+            .await
+            .unwrap_err();
+        assert!(error.contains("exec-1"), "{error}");
+        assert!(error.contains("active_commits"), "{error}");
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        failed_server.abort();
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn oversized_progress_is_bounded_and_not_accepted_as_handoff_evidence() {
+        let (endpoint, server) =
+            serve(Router::new().fallback(axum::routing::get(|| async { "x".repeat(65537) }))).await;
+        let evidence = super::legacy_progress(
+            &crate::state::worker_http_client().unwrap(),
+            &endpoint,
+            "table",
+        )
+        .await;
+        assert!(evidence.contains("exceeds 64 KiB"), "{evidence}");
+        assert!(evidence.contains("ownership unresolved"));
+        server.abort();
+    }
+
     #[tokio::test]
     async fn connection_refusal_preserves_progress_on_both_sides() {
         // Reserve a non-listening port so another test cannot take it.
diff --git a/crates/lance-context-server/src/legacy_merge.rs b/crates/lance-context-server/src/legacy_merge.rs
new file mode 100644
index 0000000..4bc096c
--- /dev/null
+++ b/crates/lance-context-server/src/legacy_merge.rs
@@ -0,0 +1,323 @@
+//! Process-local legacy merge evidence. These observations never authorize
+//! takeover: other processes and pre-upgrade writers remain untracked.
+use std::{
+    collections::VecDeque,
+    future::Future,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Mutex,
+    },
+    time::Instant,
+};
+
+use axum::{
+    extract::{Query, State},
+    Json,
+};
+use lance_context_core::merge_write_scope::{MergeWork, MergeWriteScope, MergeWriteStatus};
+use serde::{Deserialize, Serialize};
+
+const MAX_TRACKED: usize = 256;
+const MAX_RETURNED: usize = 128;
+const MAX_RESPONSE: usize = 32;
+
+#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum Phase {
+    WaitingSlot,
+    Running,
+    ReturnedOk,
+    ReturnedError,
+    Interrupted,
+}
+
+struct Entry {
+    id: String,
+    target: String,
+    origin: &'static str,
+    started: Instant,
+    phase: Phase,
+    scope: Arc<MergeWriteScope>,
+    started_work: Arc<AtomicBool>,
+}
+
+#[derive(Default)]
+struct Entries {
+    active: Vec<Entry>,
+    returned: VecDeque<Entry>,
+    untracked: u64,
+    evicted: u64,
+}
+
+#[derive(Default)]
+pub(crate) struct LegacyMerges(Mutex<Entries>);
+
+#[derive(Serialize)]
+pub(crate) struct Observation {
+    id: String,
+    target: String,
+    origin: &'static str,
+    phase: Phase,
+    elapsed_ms: u64,
+    completed_steps: u64,
+    work: MergeWork,
+    writes: MergeWriteStatus,
+}
+
+#[derive(Serialize)]
+pub(crate) struct Snapshot {
+    protocol: u32,
+    /// Explicitly false even with an empty, untruncated result.
+    takeover_authorized: bool,
+    scope: &'static str,
+    untracked_admissions: u64,
+    evicted_history: u64,
+    truncated: bool,
+    executions: Vec<Observation>,
+}
+
+struct Tracked<'a> {
+    registry: &'a LegacyMerges,
+    id: String,
+    phase: Phase,
+}
+pub(crate) struct Started(Arc<AtomicBool>);
+impl Started {
+    pub(crate) fn running(&self) {
+        self.0.store(true, Ordering::Relaxed);
+    }
+}
+impl Drop for Tracked<'_> {
+    fn drop(&mut self) {
+        let mut entries = self.registry.0.lock().unwrap();
+        if let Some(index) = entries.active.iter().position(|e| e.id == self.id) {
+            let mut entry = entries.active.swap_remove(index);
+            entry.phase = self.phase;
+            if entries.returned.len() == MAX_RETURNED {
+                entries.returned.pop_front();
+                entries.evicted = entries.evicted.saturating_add(1);
+            }
+            entries.returned.push_back(entry);
+        }
+    }
+}
+
+impl LegacyMerges {
+    /// Register before waiting for the slot. Cancellation retains the scope,
+    /// including uncertainty when a storage future was cancelled.
+    /// Overflow affects telemetry only; ingestion and merge admission keep
+    /// their existing limits and are never failed by this registry.
+    pub(crate) async fn run<T, E, F, Fut>(
+        &self,
+        target: &str,
+        origin: &'static str,
+        waiting_slot: bool,
+        work: F,
+    ) -> Result<T, E>
+    where
+        F: FnOnce(Started) -> Fut,
+        Fut: Future<Output = Result<T, E>>,
+    {
+        let scope = MergeWriteScope::for_legacy_observation();
+        let started_work = Arc::new(AtomicBool::new(false));
+        let mut tracked = Tracked {
+            registry: self,
+            id: uuid::Uuid::new_v4().to_string(),
+            phase: Phase::Interrupted,
+        };
+        {
+            let mut entries = self.0.lock().unwrap();
+            if entries.active.len() < MAX_TRACKED {
+                entries.active.push(Entry {
+                    id: tracked.id.clone(),
+                    target: target.into(),
+                    origin,
+                    started: Instant::now(),
+                    phase: if waiting_slot {
+                        Phase::WaitingSlot
+                    } else {
+                        Phase::Running
+                    },
+                    scope: scope.clone(),
+                    started_work: started_work.clone(),
+                });
+            } else {
+                entries.untracked = entries.untracked.saturating_add(1);
+            }
+        }
+        let result = scope.run(work(Started(started_work))).await;
+        tracked.phase = if result.is_ok() {
+            Phase::ReturnedOk
+        } else {
+            Phase::ReturnedError
+        };
+        result
+    }
+
+    fn snapshot(&self, target: Option<&str>) -> Snapshot {
+        let entries = self.0.lock().unwrap();
+        let mut matching = entries
+            .active
+            .iter()
+            .chain(entries.returned.iter().rev())
+            .filter(|e| target.is_none_or(|t| e.target == t));
+        let executions = matching
+            .by_ref()
+            .take(MAX_RESPONSE)
+            .map(|e| Observation {
+                id: e.id.clone(),
+                target: e.target.clone(),
+                origin: e.origin,
+                phase: if e.phase == Phase::WaitingSlot && e.started_work.load(Ordering::Relaxed) {
+                    Phase::Running
+                } else {
+                    e.phase
+                },
+                elapsed_ms: e.started.elapsed().as_millis().min(u64::MAX as u128) as u64,
+                completed_steps: e.scope.completed_steps(),
+                work: e.scope.work(),
+                writes: e.scope.write_status(),
+            })
+            .collect();
+        Snapshot {
+            protocol: 1,
+            takeover_authorized: false,
+            scope: "this_process_legacy_merge_only",
+            untracked_admissions: entries.untracked,
+            evicted_history: entries.evicted,
+            truncated: matching.next().is_some(),
+            executions,
+        }
+    }
+}
+
+#[derive(Default, Deserialize)]
+pub(crate) struct Filter {
+    target: Option<String>,
+}
+
+pub(crate) async fn status(
+    State(state): State<Arc<crate::state::AppState>>,
+    Query(filter): Query<Filter>,
+) -> Json<serde_json::Value> {
+    Json(serde_json::json!({
+        "instance": state.merge_executions.instance(),
+        "snapshot": state.merge_executions.legacy.snapshot(filter.target.as_deref()),
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use lance_context_core::merge_write_scope::checkpoint;
+    use tokio::sync::oneshot;
+
+    #[tokio::test]
+    async fn concurrent_progress_is_isolated_and_cancelled_work_is_retained() {
+        let registry = Arc::new(LegacyMerges::default());
+        let (began, entered) = oneshot::channel();
+        let first = registry.clone();
+        let frozen = tokio::spawn(async move {
+            first
+                .run("frozen", "http", true, |started| async move {
+                    started.running();
+                    checkpoint();
+                    began.send(()).unwrap();
+                    std::future::pending::<Result<(), ()>>().await
+                })
+                .await
+        });
+        entered.await.unwrap();
+        let frozen_id = registry.snapshot(Some("frozen")).executions[0].id.clone();
+        for _ in 0..20 {
+            registry
+                .run("healthy", "count_sweeper", false, |_| async {
+                    for _ in 0..10 {
+                        checkpoint();
+                        tokio::task::yield_now().await;
+                    }
+                    Ok::<_, ()>(())
+                })
+                .await
+                .unwrap();
+            let snapshot = registry.snapshot(Some("frozen"));
+            assert_eq!(snapshot.executions.len(), 1);
+            assert_eq!(snapshot.executions[0].id, frozen_id);
+            assert_eq!(snapshot.executions[0].completed_steps, 1);
+            assert_eq!(snapshot.executions[0].phase, Phase::Running);
+            assert!(!snapshot.takeover_authorized);
+        }
+        frozen.abort();
+        assert!(frozen.await.unwrap_err().is_cancelled());
+        let snapshot = registry.snapshot(Some("frozen"));
+        assert_eq!(snapshot.executions[0].phase, Phase::Interrupted);
+        assert_eq!(snapshot.executions[0].completed_steps, 1);
+        assert!(!snapshot.takeover_authorized);
+        assert!(registry.snapshot(Some("absent")).executions.is_empty());
+        assert!(!registry.snapshot(Some("absent")).takeover_authorized);
+    }
+
+    #[tokio::test]
+    async fn slot_wait_does_not_report_work_and_errors_are_distinct() {
+        let registry = Arc::new(LegacyMerges::default());
+        let (began, entered) = oneshot::channel();
+        let worker_registry = registry.clone();
+        let worker = tokio::spawn(async move {
+            worker_registry
+                .run("hot", "http", true, |_| async move {
+                    began.send(()).unwrap();
+                    std::future::pending::<Result<(), ()>>().await
+                })
+                .await
+        });
+        entered.await.unwrap();
+        let snapshot = registry.snapshot(None);
+        assert_eq!(snapshot.executions[0].phase, Phase::WaitingSlot);
+        assert_eq!(snapshot.executions[0].completed_steps, 0);
+        worker.abort();
+        worker.await.unwrap_err();
+        registry
+            .run("error", "http", false, |_| async { Err::<(), _>(()) })
+            .await
+            .unwrap_err();
+        assert_eq!(
+            registry.snapshot(Some("error")).executions[0].phase,
+            Phase::ReturnedError
+        );
+    }
+
+    #[tokio::test]
+    async fn history_and_response_are_bounded_and_loss_is_explicit() {
+        let registry = LegacyMerges::default();
+        for _ in 0..MAX_RETURNED + 5 {
+            registry
+                .run("hot", "cleanup_sweeper", false, |_| async {
+                    Ok::<_, ()>(())
+                })
+                .await
+                .unwrap();
+        }
+        let snapshot = registry.snapshot(None);
+        assert_eq!(snapshot.executions.len(), MAX_RESPONSE);
+        assert!(snapshot.truncated);
+        assert_eq!(snapshot.evicted_history, 5);
+        assert!(!snapshot.takeover_authorized);
+        let mut tasks = Vec::new();
+        for _ in 0..MAX_TRACKED + 1 {
+            let mut future = Box::pin(registry.run("hot", "http", true, |_| {
+                std::future::pending::<Result<(), ()>>()
+            }));
+            assert!(futures::poll!(&mut future).is_pending());
+            tasks.push(future);
+        }
+        let snapshot = registry.snapshot(None);
+        assert!(snapshot.truncated);
+        assert_eq!(snapshot.untracked_admissions, 1);
+        assert_eq!(snapshot.executions.len(), MAX_RESPONSE);
+        assert!(snapshot
+            .executions
+            .iter()
+            .all(|e| e.phase == Phase::WaitingSlot));
+        drop(tasks);
+    }
+}
diff --git a/crates/lance-context-server/src/main.rs b/crates/lance-context-server/src/main.rs
index 5e8a5af..d83edb0 100644
--- a/crates/lance-context-server/src/main.rs
+++ b/crates/lance-context-server/src/main.rs
@@ -4,6 +4,7 @@ mod body_timeout;
 mod config;
 mod drain;
 mod error;
+mod legacy_merge;
 mod merge_execution;
 mod routes;
 mod state;
diff --git a/crates/lance-context-server/src/merge_execution.rs b/crates/lance-context-server/src/merge_execution.rs
index 4a70abc..fb6c96d 100644
--- a/crates/lance-context-server/src/merge_execution.rs
+++ b/crates/lance-context-server/src/merge_execution.rs
@@ -45,6 +45,7 @@ impl lance_context_core::merge_write_scope::CommitAuthorizer for OwnedCommitGuar
 }
 
 pub struct Executions {
+    pub(crate) legacy: crate::legacy_merge::LegacyMerges,
     coordinator: OnceCell<Coordinator>,
     etcd: Option<lance_context_merge::EtcdConfig>,
     pub(crate) rollout: lance_context_merge::rollout::MergeRollout,
@@ -59,6 +60,7 @@ impl Executions {
     #[cfg(test)]
     pub fn new(coordinator: Option<Coordinator>, timeout_secs: u64) -> Self {
         Self {
+            legacy: Default::default(),
             coordinator: OnceCell::new_with(coordinator),
             etcd: None,
             rollout: Default::default(),
@@ -77,6 +79,7 @@ impl Executions {
         idle_timeout_secs: u64,
     ) -> Self {
         Self {
+            legacy: Default::default(),
             coordinator: OnceCell::new(),
             etcd: Some(etcd),
             rollout,
@@ -88,6 +91,10 @@ impl Executions {
         }
     }
 
+    pub(crate) fn instance(&self) -> &str {
+        &self.instance
+    }
+
     pub(crate) fn owned(&self, target: &str) -> bool {
         self.rollout.owned(target)
     }
@@ -154,6 +161,7 @@ pub async fn capabilities(
         "queue_timeout_secs": state.merge_executions.queue_timeout_secs,
         "idle_timeout_secs": state.merge_executions.idle_timeout_secs,
         "progress_protocol": 1,
+        "legacy_progress_protocol": 1,
         "rollout_append_protocol": 1,
         "shard_name": state.instance_id,
         "owned_targets": state.merge_executions.rollout.owned_targets,
@@ -495,6 +503,14 @@ mod tests {
         assert!(state.merge_executions.legacy_allowed("hot").is_err());
         let Json(reply) = capabilities(State(state.clone())).await.unwrap();
         assert_eq!(reply["owned_targets"][0], "hot");
+        assert_eq!(reply["legacy_progress_protocol"], 1);
+        let Json(progress) = crate::legacy_merge::status(
+            State(state.clone()),
+            axum::extract::Query(Default::default()),
+        )
+        .await;
+        assert_eq!(progress["instance"], reply["instance"]);
+        assert_eq!(progress["snapshot"]["takeover_authorized"], false);
         assert!(state.merge_executions.coordinator.get().is_none());
     }
 
diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs
index 2717760..07f3b43 100644
--- a/crates/lance-context-server/src/routes/generic.rs
+++ b/crates/lance-context-server/src/routes/generic.rs
@@ -297,8 +297,22 @@ pub async fn merge_generic_wal(
     state
         .merge_executions
         .legacy_allowed(&format!("generic:{name}"))?;
-    let _slot = state.acquire_merge_slot().await;
-    merge_generic_wal_owned(State(state), Path(name)).await
+    let execution_state = state.clone();
+    let execution_name = name.clone();
+    state
+        .merge_executions
+        .legacy
+        .run(
+            &format!("generic:{name}"),
+            "http",
+            true,
+            |started| async move {
+                let _slot = execution_state.acquire_merge_slot().await;
+                started.running();
+                merge_generic_wal_owned(State(execution_state.clone()), Path(execution_name)).await
+            },
+        )
+        .await
 }
 
 pub(crate) async fn merge_generic_wal_owned(
diff --git a/crates/lance-context-server/src/routes/mod.rs b/crates/lance-context-server/src/routes/mod.rs
index 38f5f52..91df6ea 100644
--- a/crates/lance-context-server/src/routes/mod.rs
+++ b/crates/lance-context-server/src/routes/mod.rs
@@ -18,6 +18,10 @@ use crate::state::AppState;
 
 pub fn router() -> Router<Arc<AppState>> {
     Router::new()
+        .route(
+            "/api/v1/internal/legacy-merges",
+            get(crate::legacy_merge::status),
+        )
         .route(
             "/api/v1/internal/merge-executor",
             get(crate::merge_execution::capabilities),
diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs
index a078607..6e07183 100644
--- a/crates/lance-context-server/src/routes/rollouts.rs
+++ b/crates/lance-context-server/src/routes/rollouts.rs
@@ -711,8 +711,17 @@ pub async fn merge_wal(
     Path(name): Path<String>,
 ) -> Result<Json<MergeWalResponse>, AppError> {
     state.merge_executions.legacy_allowed(&name)?;
-    let _slot = state.acquire_merge_slot().await;
-    merge_wal_owned(State(state), Path(name)).await
+    let execution_state = state.clone();
+    let execution_name = name.clone();
+    state
+        .merge_executions
+        .legacy
+        .run(&name, "http", true, |started| async move {
+            let _slot = execution_state.acquire_merge_slot().await;
+            started.running();
+            merge_wal_owned(State(execution_state.clone()), Path(execution_name)).await
+        })
+        .await
 }
 
 pub(crate) async fn merge_wal_owned(
diff --git a/crates/lance-context-server/src/sweeper.rs b/crates/lance-context-server/src/sweeper.rs
index a01cc8e..66449ee 100644
--- a/crates/lance-context-server/src/sweeper.rs
+++ b/crates/lance-context-server/src/sweeper.rs
@@ -291,9 +291,16 @@ pub(crate) async fn flush_pass_coordinated<S: Sweepable>(
                     },
                     None => None,
                 };
-                let merged = async move {
+                let merged = async {
                     let _slot = slot;
-                    store.merge_if_due().await
+                    observe_legacy_merge(
+                        state.as_deref(),
+                        &name,
+                        kind,
+                        "count_sweeper",
+                        store.merge_if_due(),
+                    )
+                    .await
                 };
                 report_merge(
                     &name,
@@ -345,11 +352,45 @@ pub(crate) async fn merge_pass_coordinated<S: Sweepable>(
         report_merge(
             &name,
             kind,
-            tokio::time::timeout(pass_timeout, store.merge_wal()).await,
+            tokio::time::timeout(
+                pass_timeout,
+                observe_legacy_merge(
+                    state.as_deref(),
+                    &name,
+                    kind,
+                    "cleanup_sweeper",
+                    store.merge_wal(),
+                ),
+            )
+            .await,
         );
     }
 }
 
+/// The scope covers only merge work, never the independent ingestion flush.
+async fn observe_legacy_merge<T, E>(
+    state: Option<&crate::state::AppState>,
+    name: &str,
+    kind: &str,
+    origin: &'static str,
+    work: impl std::future::Future<Output = Result<T, E>>,
+) -> Result<T, E> {
+    if let Some(state) = state {
+        let target = if kind == "rollout" {
+            name.to_string()
+        } else {
+            format!("{kind}:{name}")
+        };
+        state
+            .merge_executions
+            .legacy
+            .run(&target, origin, false, |_| work)
+            .await
+    } else {
+        work.await
+    }
+}
+
 /// Hard bound on one detached demand publish (manifest read + etcd put).
 /// Longer than any healthy path, far shorter than a flush pass timeout.
 const DEMAND_PUBLISH_TIMEOUT: Duration = Duration::from_secs(5);
@@ -760,6 +801,46 @@ mod tests {
         )])
     }
 
+    #[tokio::test]
+    async fn legacy_merge_observes_real_reclamation_without_tracking_ingest_flush() {
+        let state_dir = TempDir::new().unwrap();
+        let state =
+            Arc::new(crate::state::AppState::new_for_test(state_dir.path().to_path_buf()).await);
+        let dir = TempDir::new().unwrap();
+        let store = generic_with_pending(&dir, 3, 3).await;
+        flush_pass_coordinated(
+            vec![("s".into(), store.clone())],
+            Duration::from_secs(10),
+            None,
+            Some(state.clone()),
+        )
+        .await;
+        assert_eq!(
+            store.read().await.pending_wal_generations().await.unwrap(),
+            0
+        );
+        assert_eq!(store.read().await.count_base_rows().await.unwrap(), 3);
+        let axum::Json(report) = crate::legacy_merge::status(
+            axum::extract::State(state),
+            axum::extract::Query(Default::default()),
+        )
+        .await;
+        let executions = report["snapshot"]["executions"].as_array().unwrap();
+        assert_eq!(
+            executions.len(),
+            1,
+            "flush itself must not be a maintenance execution"
+        );
+        let execution = &executions[0];
+        assert_eq!(execution["target"], "generic:s");
+        assert_eq!(execution["origin"], "count_sweeper");
+        assert_eq!(execution["phase"], "returned_ok");
+        assert_eq!(execution["work"]["generations_reclaimed"], 3);
+        assert!(execution["work"]["base_commits"].as_u64().unwrap() > 0);
+        assert_eq!(execution["writes"]["active_commits"], 0);
+        assert_eq!(report["snapshot"]["takeover_authorized"], false);
+    }
+
     async fn generic_with_pending(
         dir: &TempDir,
         merge_after_generations: usize,
diff --git a/docs/merge-recovery.md b/docs/merge-recovery.md
index 7b2e63f..838f5d8 100644
--- a/docs/merge-recovery.md
+++ b/docs/merge-recovery.md
@@ -164,6 +164,37 @@ Other maintenance tasks retain their existing repair policy.
 
 ## Default-off, per-table rollout
 
+### Legacy worker progress evidence
+
+Workers advertising `legacy_progress_protocol: 1` expose
+`GET /api/v1/internal/legacy-merges?target=<scheduler-target>`. The response
+identifies the process incarnation and individual executions, distinguishes
+slot waits from running work, and reports actual completed steps, WAL reads,
+encoded output, base commits, reclaimed generations, and local outstanding or
+uncertain commits. Polling and changing phase do not advance completed work.
+HTTP merges and both worker self-merge triggers share this instrumentation;
+ingestion flushes are not included. Cancelling the caller retains an
+`interrupted` observation. Cancelling an admitted storage future marks its
+commit uncertain: a remote write may still finish. This observation-only scope
+does not spawn storage tasks or change legacy cancellation and slot lifetimes.
+
+The endpoint reads only process memory, without opening tables or querying
+etcd. It returns at most 32 entries, prioritizing active calls, with explicit
+truncation and history-eviction counters. Tracking is capped at 256 active and
+128 returned/interrupted entries; overflow increments an untracked-admission
+counter without rejecting work. Query an exact target when the global response
+is truncated. Master captures a bounded snapshot after an ambiguous legacy RPC
+failure (2 seconds, 64 KiB input maximum). The task error stores a small typed
+summary of at most eight executions; serial fan-out still stops.
+Older workers returning 404 simply have unavailable evidence.
+
+This is **process-local evidence, not ownership or permission to take over**.
+Even an empty response or zero active commits cannot prove storage quiescence:
+other masters/helpers, evicted history, prior incarnations, and old binaries
+are outside its coverage. `takeover_authorized` is always false. These snapshots
+help identify legacy execution progress; they do not replace the migration
+barriers below or implement automatic enrollment of new tables.
+
 `MERGE_OWNED_TARGETS` and `MERGE_DRAIN_TARGETS` are comma-separated **exact**
 scheduler targets (`name` for rollout, `generic:name` for generic). Both default
 to empty; wildcards and overlapping lists are rejected. Merely configuring