diff --git a/crates/lance-context-core/src/merge_write_scope.rs b/crates/lance-context-core/src/merge_write_scope.rs
index 248c462..7c5e3cb 100644
--- a/crates/lance-context-core/src/merge_write_scope.rs
+++ b/crates/lance-context-core/src/merge_write_scope.rs
@@ -35,6 +35,57 @@ struct Progress {
     uncertain: bool,
 }
 
+/// Completed work for one execution. Decoded bytes are Arrow buffer sizes,
+/// not network bytes. File encoding is not evidence of a committed merge.
+#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct MergeWork {
+    /// Parallel staging can occupy several phases at once. Entering/leaving
+    /// a phase is diagnostic only and never advances completed_steps.
+    pub active_stages: std::collections::BTreeMap<String, u32>,
+    pub wal_batches_read: u64,
+    pub wal_rows_read: u64,
+    pub wal_decoded_bytes: u64,
+    pub encoded_bytes: u64,
+    pub encoded_rows: u64,
+    pub encoded_files: u64,
+    pub base_commits: u64,
+    pub generations_reclaimed: u64,
+    pub last_completed_stage: Option<String>,
+}
+
+pub(crate) struct WorkStage {
+    scope: Option<Arc<MergeWriteScope>>,
+    name: &'static str,
+}
+
+impl Drop for WorkStage {
+    fn drop(&mut self) {
+        if let Some(scope) = &self.scope {
+            let mut work = scope.work.lock().unwrap();
+            if let Some(count) = work.active_stages.get_mut(self.name) {
+                *count -= 1;
+                if *count == 0 {
+                    work.active_stages.remove(self.name);
+                }
+            }
+        }
+    }
+}
+
+pub(crate) fn work_stage(name: &'static str) -> WorkStage {
+    let scope = CURRENT.try_with(Arc::clone).ok();
+    if let Some(scope) = &scope {
+        *scope
+            .work
+            .lock()
+            .unwrap()
+            .active_stages
+            .entry(name.into())
+            .or_default() += 1;
+    }
+    WorkStage { scope, name }
+}
+
 pub trait CommitAuthorizer: std::fmt::Debug + Send + Sync {
     fn authorize<'a>(
         &'a self,
@@ -46,6 +97,7 @@ pub trait CommitAuthorizer: std::fmt::Debug + Send + Sync {
 #[derive(Debug, Default)]
 pub struct MergeWriteScope {
     completed_steps: Arc<AtomicU64>,
+    work: Mutex<MergeWork>,
     observe_file_io: bool,
     local_io: Mutex<Vec<(lance_io::utils::tracking_store::IOTracker, u64)>>,
     progress: Mutex<Progress>,
@@ -56,6 +108,9 @@ pub struct MergeWriteScope {
 }
 
 impl MergeWriteScope {
+    pub fn work(&self) -> MergeWork {
+        self.work.lock().unwrap().clone()
+    }
     pub fn completed_steps(&self) -> u64 {
         self.local_io.lock().unwrap().iter().fold(
             self.completed_steps.load(Ordering::Relaxed),
@@ -148,6 +203,38 @@ pub fn checkpoint() {
     let _ = CURRENT.try_with(|scope| scope.completed_steps.fetch_add(1, Ordering::Relaxed));
 }
 
+pub(crate) fn wal_batch_read(rows: usize, bytes: usize) {
+    let _ = CURRENT.try_with(|scope| {
+        let mut work = scope.work.lock().unwrap();
+        work.wal_batches_read += 1;
+        work.wal_rows_read = work.wal_rows_read.saturating_add(rows as u64);
+        work.wal_decoded_bytes = work.wal_decoded_bytes.saturating_add(bytes as u64);
+        work.last_completed_stage = Some("wal_read".into());
+    });
+    checkpoint();
+}
+
+pub(crate) fn base_committed() {
+    let _ = CURRENT.try_with(|scope| {
+        let mut work = scope.work.lock().unwrap();
+        work.base_commits += 1;
+        work.last_completed_stage = Some("base_commit".into());
+    });
+    checkpoint();
+}
+
+pub(crate) fn generations_reclaimed(count: usize) {
+    if count == 0 {
+        return;
+    }
+    let _ = CURRENT.try_with(|scope| {
+        let mut work = scope.work.lock().unwrap();
+        work.generations_reclaimed = work.generations_reclaimed.saturating_add(count as u64);
+        work.last_completed_stage = Some("wal_drain".into());
+    });
+    checkpoint();
+}
+
 /// Capture progress explicitly for Lance callbacks which can run on another task.
 pub(crate) fn write_progress() -> lance::dataset::write::WriteProgressFn {
     let scope = CURRENT.try_with(Arc::clone).ok();
@@ -156,10 +243,25 @@ pub(crate) fn write_progress() -> lance::dataset::write::WriteProgressFn {
         let current = (stats.bytes_written, stats.rows_written, stats.files_written);
         let mut old = previous.lock().unwrap();
         if current.0 > old.0 || current.1 > old.1 || current.2 > old.2 {
-            *old = current;
             if let Some(scope) = &scope {
+                let mut work = scope.work.lock().unwrap();
+                work.encoded_bytes = work
+                    .encoded_bytes
+                    .saturating_add(current.0.saturating_sub(old.0));
+                work.encoded_rows = work
+                    .encoded_rows
+                    .saturating_add(current.1.saturating_sub(old.1));
+                work.encoded_files = work
+                    .encoded_files
+                    .saturating_add(current.2.saturating_sub(old.2) as u64);
+                work.last_completed_stage = Some("file_encoding".into());
                 scope.completed_steps.fetch_add(1, Ordering::Relaxed);
             }
+            *old = (
+                old.0.max(current.0),
+                old.1.max(current.1),
+                old.2.max(current.2),
+            );
         }
     })
 }
@@ -592,6 +694,27 @@ mod tests {
     use super::*;
     use std::sync::atomic::{AtomicBool, Ordering};
 
+    #[tokio::test]
+    async fn phase_changes_and_observation_do_not_count_as_completed_work() {
+        let scope = MergeWriteScope::new();
+        scope
+            .run(async {
+                let first = work_stage("memory_wait");
+                let second = work_stage("memory_wait");
+                assert_eq!(scope.work().active_stages["memory_wait"], 2);
+                assert_eq!(scope.completed_steps(), 0);
+                drop(first);
+                assert_eq!(scope.work().active_stages["memory_wait"], 1);
+                drop(second);
+                assert!(scope.work().active_stages.is_empty());
+                assert_eq!(scope.completed_steps(), 0);
+                wal_batch_read(3, 1024);
+                assert_eq!(scope.completed_steps(), 1);
+                assert_eq!(scope.work().wal_rows_read, 3);
+            })
+            .await;
+    }
+
     #[test]
     fn recovery_rejects_unsafe_and_external_commit_handlers() {
         for uri in [
diff --git a/crates/lance-context-core/src/rollout_append.rs b/crates/lance-context-core/src/rollout_append.rs
index 26e912b..235e122 100644
--- a/crates/lance-context-core/src/rollout_append.rs
+++ b/crates/lance-context-core/src/rollout_append.rs
@@ -323,8 +323,11 @@ pub(crate) async fn reconcile_shard(dataset: &Dataset, shard: Uuid) -> Result<us
             .filter(|g| generations.contains(&g.generation))
             .map(|g| g.path.clone())
             .collect();
+        let phase = crate::merge_write_scope::work_stage("wal_drain");
         crate::merge_write_scope::drain_generations(store, manifest.writer_epoch, generations)
             .await?;
+        crate::merge_write_scope::generations_reclaimed(count);
+        drop(phase);
         let object_store = dataset.object_store(None).await?;
         let root = dataset
             .branch_location()
@@ -406,6 +409,7 @@ impl AppendCoordinator {
         max_generations: usize,
         max_bytes: usize,
     ) -> Result<(Vec<AppendPlan>, usize)> {
+        let _phase = crate::merge_write_scope::work_stage("planning");
         if max_generations == 0
             || max_generations > MAX_PLAN_GENERATIONS
             || max_bytes == 0
@@ -683,6 +687,7 @@ pub(crate) async fn commit_files(
     fragments: Vec<Fragment>,
     merged: Vec<MergedGeneration>,
 ) -> Result<()> {
+    let _phase = crate::merge_write_scope::work_stage("base_commit");
     let operation = Operation::Update {
         removed_fragment_ids: Vec::new(),
         updated_fragments: Vec::new(),
@@ -703,6 +708,7 @@ pub(crate) async fn commit_files(
             .execute(Transaction::new(version, operation, None)),
     )
     .await?;
+    crate::merge_write_scope::base_committed();
     dataset.checkout_latest().await?;
     Ok(())
 }
@@ -737,11 +743,14 @@ pub async fn stage(
         ));
     }
     let replay_ranges = ReplayRanges::load(&dataset, plan.shard)?;
+    let phase = crate::merge_write_scope::work_stage("memory_wait");
     let mut reservation = budget.reserve(plan.max_bytes.min(budget.limit())).await;
+    drop(phase);
     let mut batches = Vec::new();
     let mut bytes = 0usize;
     let mut completed = 0;
     'generations: for generation in &plan.generations {
+        let _phase = crate::merge_write_scope::work_stage("wal_read");
         let path = format!(
             "{}/_mem_wal/{}/{}",
             uri.trim_end_matches('/'),
@@ -753,6 +762,10 @@ pub async fn stage(
         let mut current = Vec::new();
         while let Some(batch) = stream.try_next().await? {
             let batch = align_batch_to_schema(batch, schema.clone())?;
+            crate::merge_write_scope::wal_batch_read(
+                batch.num_rows(),
+                batch.get_array_memory_size(),
+            );
             bytes = bytes.saturating_add(batch.get_array_memory_size());
             if !reservation.try_grow_to(bytes.saturating_mul(2)) {
                 if completed == 0 {
@@ -763,7 +776,6 @@ pub async fn stage(
                 break 'generations;
             }
             current.push(batch);
-            crate::merge_write_scope::checkpoint();
         }
         let needs_lookup = replay_ranges
             .as_ref()
@@ -784,6 +796,7 @@ pub async fn stage(
     if batches.iter().any(|(_, needs_lookup)| *needs_lookup)
         && !dataset.manifest().fragments.is_empty()
     {
+        let _phase = crate::merge_write_scope::work_stage("legacy_id_lookup");
         let mut ids = HashSet::new();
         for (batch, _) in batches.iter().filter(|(_, needs_lookup)| *needs_lookup) {
             let column = batch
@@ -847,6 +860,7 @@ pub async fn stage(
     let fragments = if output.is_empty() {
         Vec::new()
     } else {
+        let _phase = crate::merge_write_scope::work_stage("file_encoding");
         let params = WriteParams {
             mode: WriteMode::Append,
             max_bytes_per_file: plan.max_bytes,
@@ -957,6 +971,44 @@ mod tests {
         MergeMemoryBudget::new(8 * 1024 * 1024)
     }
 
+    #[tokio::test]
+    async fn work_report_distinguishes_encoding_from_durable_reclamation() {
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let store = writer(uri, "a").await;
+        put(&store, "first", 4096).await;
+        let scope = crate::merge_write_scope::MergeWriteScope::new();
+        scope
+            .run(async {
+                let mut coordinator = AppendCoordinator::open(uri, None).await.unwrap();
+                let (mut plans, _) = coordinator
+                    .plan(&["a".into()], 64, 1024 * 1024)
+                    .await
+                    .unwrap();
+                let part = stage(uri, plans.remove(0), budget(), None).await.unwrap();
+                let encoded = scope.work();
+                assert_eq!(encoded.wal_rows_read, 1);
+                assert!(encoded.wal_decoded_bytes >= 4096);
+                assert_eq!(encoded.encoded_rows, 1);
+                assert!(encoded.encoded_bytes > 0);
+                assert!(encoded.encoded_files > 0);
+                assert_eq!(encoded.base_commits, 0);
+                assert_eq!(encoded.generations_reclaimed, 0);
+                assert_eq!(coordinator.commit(vec![part]).await.unwrap(), 1);
+                let committed = scope.work();
+                assert_eq!(committed.base_commits, 1);
+                assert_eq!(committed.generations_reclaimed, 1);
+                assert_eq!(committed.last_completed_stage.as_deref(), Some("wal_drain"));
+                let sequence = scope.completed_steps();
+                tokio::task::yield_now().await;
+                assert_eq!(scope.completed_steps(), sequence);
+                assert_eq!(scope.work(), committed);
+            })
+            .await;
+        scope.drain().await;
+        assert_eq!(rows(uri).await, 1);
+    }
+
     #[derive(Debug)]
     struct NoPreparationCommit;
     impl crate::merge_write_scope::CommitAuthorizer for NoPreparationCommit {
diff --git a/crates/lance-context-master/src/maintenance_execution.rs b/crates/lance-context-master/src/maintenance_execution.rs
index 976e045..fc92960 100644
--- a/crates/lance-context-master/src/maintenance_execution.rs
+++ b/crates/lance-context-master/src/maintenance_execution.rs
@@ -156,22 +156,42 @@ async fn watch(
     coordinator: &Coordinator,
     execution: &Execution,
     scope: &MergeWriteScope,
+    executor_id: &str,
+    task_id: &str,
 ) -> String {
     let mut sequence = scope.completed_steps();
     let mut changed = tokio::time::Instant::now();
+    let mut last_progress_at_ms = (sequence > 0).then(lance_context_merge::failure::now_ms);
     loop {
         tokio::time::sleep(Duration::from_secs(1)).await;
         let current = scope.completed_steps();
         if current != sequence {
             changed = tokio::time::Instant::now();
             sequence = current;
+            last_progress_at_ms = Some(lance_context_merge::failure::now_ms());
         }
         if changed.elapsed() >= Duration::from_secs(execution.idle_timeout_secs) {
+            tracing::warn!(target = %execution.target, execution = %execution.id, task_id, executor_id,
+                sequence, work = ?scope.work(), idle_seconds = changed.elapsed().as_secs(),
+                "maintenance stopped making progress; cancelling preparation before drain/fence recovery");
             return "maintenance no-progress deadline exceeded".into();
         }
         // Publishing an unchanged sequence checks ownership but does not count
         // as progress. A healthy lease never resets the idle deadline.
-        match coordinator.publish_progress(execution, sequence).await {
+        let report = serde_json::json!({
+            "executor_id": executor_id,
+            "task_id": task_id,
+            "sequence": sequence,
+            "observed_at_ms": lance_context_merge::failure::now_ms(),
+            "last_progress_at_ms": last_progress_at_ms,
+            "idle_for_ms": changed.elapsed().as_millis() as u64,
+            "idle_timeout_secs": execution.idle_timeout_secs,
+            "work": scope.work(),
+        });
+        match coordinator
+            .publish_work_progress(execution, sequence, report)
+            .await
+        {
             Ok(true) => {}
             Ok(false) => return "maintenance ownership revoked".into(),
             Err(error) => return format!("maintenance progress publication failed: {error}"),
@@ -243,11 +263,12 @@ where
         execution: running.clone(),
         uri,
     }));
+    let executor_id = state.admission.status().executor_id;
     let outcome = std::panic::AssertUnwindSafe(async {
         let watched = async {
             tokio::select! {
                 result = scope.run(work) => result,
-                error = watch(&coordinator, &running, &scope) => Err(error),
+                error = watch(&coordinator, &running, &scope, &executor_id, &claim.task.id) => Err(error),
             }
         };
         watched.await
@@ -255,6 +276,9 @@ where
     .catch_unwind()
     .await
     .unwrap_or_else(|_| Err("maintenance executor panicked".into()));
+    tracing::info!(target = %running.target, execution = %running.id,
+        task_id = %claim.task.id, %executor_id, work = ?scope.work(),
+        success = outcome.is_ok(), "maintenance work outcome before commit drain");
     // The work future has been dropped, so no preparation may issue another
     // commit. The shielded manifest leaves still need acknowledgement or fencing.
     let drained = tokio::time::timeout(
@@ -371,6 +395,71 @@ mod tests {
         state.task_store.claim_next().await.unwrap().unwrap()
     }
 
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn work_report_heartbeats_do_not_extend_actual_progress_deadline() {
+        let (_dir, state) = fixture().await;
+        let claim = claim(&state, TaskKind::MergeWal).await;
+        let coordinator = state.task_store.merge_coordinator();
+        let mut execution = Execution::new("table", "master:catchup", "test", 60);
+        execution.idle_timeout_secs = 5;
+        assert!(coordinator
+            .reserve(&state.task_store.merge_claim(&claim), &execution)
+            .await
+            .unwrap());
+        let running = coordinator.start(&execution).await.unwrap().unwrap();
+        let scope = MergeWriteScope::new();
+        let watchdog = watch(
+            &coordinator,
+            &running,
+            &scope,
+            "executor-test",
+            &claim.task.id,
+        );
+        let producer = async {
+            // Make real checkpoint progress beyond the entire idle allowance.
+            for _ in 0..12 {
+                scope
+                    .run(async {
+                        lance_context_core::merge_write_scope::checkpoint();
+                    })
+                    .await;
+                tokio::time::sleep(Duration::from_millis(500)).await;
+            }
+            let moving = coordinator.work_progress(&running).await.unwrap().unwrap();
+            assert_eq!(moving["executor_id"], "executor-test");
+            assert_eq!(moving["task_id"], claim.task.id);
+            // Wait for actual publications instead of assuming an etcd RPC
+            // plus a one-second timer always completes in 1.1 wall seconds.
+            let first = loop {
+                let report = coordinator.work_progress(&running).await.unwrap().unwrap();
+                if report["sequence"].as_u64() == Some(scope.completed_steps()) {
+                    break report;
+                }
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            };
+            let second = loop {
+                let report = coordinator.work_progress(&running).await.unwrap().unwrap();
+                if report["observed_at_ms"].as_u64() > first["observed_at_ms"].as_u64() {
+                    break report;
+                }
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            };
+            assert_eq!(first["sequence"], second["sequence"]);
+            assert_eq!(first["last_progress_at_ms"], second["last_progress_at_ms"]);
+            assert!(
+                second["observed_at_ms"].as_u64().unwrap()
+                    > first["observed_at_ms"].as_u64().unwrap()
+            );
+        };
+        let (error, ()) = tokio::time::timeout(Duration::from_secs(30), async {
+            tokio::join!(watchdog, producer)
+        })
+        .await
+        .unwrap();
+        assert!(error.contains("no-progress"));
+    }
+
     #[tokio::test]
     #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn stalled_local_kinds_cancel_and_release_without_stopping_other_tables() {
diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs
index aa4ac1b..bb3f7a2 100644
--- a/crates/lance-context-master/src/routes.rs
+++ b/crates/lance-context-master/src/routes.rs
@@ -706,6 +706,70 @@ async fn executor_status(State(state): State<Arc<MasterState>>) -> Json<crate::a
     Json(state.admission.status())
 }
 
+#[derive(Deserialize)]
+struct MergeProgressQuery {
+    target: String,
+}
+
+/// Exact target reads only. A live task/lease without execution telemetry is
+/// explicitly unknown, not evidence that a legacy merge is making progress.
+async fn merge_progress(
+    State(state): State<Arc<MasterState>>,
+    Query(query): Query<MergeProgressQuery>,
+) -> Result<Json<serde_json::Value>, MasterError> {
+    if query.target.is_empty() || query.target.len() > 512 {
+        return Err(MasterError::InvalidRequest("invalid merge target".into()));
+    }
+    let coordinator = state.task_store.merge_coordinator();
+    let task_id = state
+        .task_store
+        .get_active_id(TaskKind::MergeWal, &query.target)
+        .await
+        .map_err(MasterError::from_lance)?;
+    let Some(execution) = coordinator
+        .get(&query.target)
+        .await
+        .map_err(MasterError::Internal)?
+    else {
+        return Ok(Json(
+            serde_json::json!({"target": query.target, "task_id": task_id, "status": if task_id.is_some() { "progress_unknown" } else { "no_active_record" }}),
+        ));
+    };
+    let progress = coordinator
+        .progress(&execution)
+        .await
+        .map_err(MasterError::Internal)?;
+    let work = coordinator
+        .work_progress(&execution)
+        .await
+        .map_err(MasterError::Internal)?;
+    if work
+        .as_ref()
+        .is_some_and(|report| report["sequence"].as_u64() != progress.as_ref().map(|p| p.sequence))
+    {
+        return Ok(Json(
+            serde_json::json!({"target":query.target,"status":"progress_changed_retry"}),
+        ));
+    }
+    if coordinator
+        .get(&query.target)
+        .await
+        .map_err(MasterError::Internal)?
+        .as_ref()
+        != Some(&execution)
+    {
+        return Ok(Json(
+            serde_json::json!({"target":query.target,"status":"execution_changed_retry"}),
+        ));
+    }
+    Ok(Json(serde_json::json!({
+        "target": query.target, "task_id": task_id, "execution": execution,
+        "progress": progress, "work_report": work,
+        "sampled_at_ms": lance_context_merge::failure::now_ms(),
+        "status": if work.is_some() { "reported" } else { "progress_details_unavailable" },
+    })))
+}
+
 async fn drain_executor(
     State(state): State<Arc<MasterState>>,
     Json(request): Json<DrainRequest>,
@@ -720,6 +784,7 @@ async fn drain_executor(
 pub fn api_router() -> Router<Arc<MasterState>> {
     Router::new()
         .route("/executor", get(executor_status))
+        .route("/merge-progress", get(merge_progress))
         .route("/executor/drain", post(drain_executor))
         .route("/experiments", get(list_experiments))
         .route("/experiments/{name}", get(get_experiment))
@@ -853,6 +918,44 @@ mod tests {
             .unwrap_or_default()
     }
 
+    #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
+    async fn merge_progress_never_calls_a_live_legacy_task_healthy() {
+        let dir = TempDir::new().unwrap();
+        let state = MasterState::new(test_config(&dir)).await.unwrap();
+        state
+            .task_store
+            .enqueue(TaskKind::MergeWal, "legacy", vec![])
+            .await
+            .unwrap();
+        let claim = state.task_store.claim_next().await.unwrap().unwrap();
+        let Json(report) = merge_progress(
+            State(state.clone()),
+            Query(MergeProgressQuery {
+                target: "legacy".into(),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(report["status"], "progress_unknown");
+        assert_eq!(report["task_id"], claim.task.id);
+        assert!(report.get("work_report").is_none());
+        state
+            .task_store
+            .finish(claim, Ok("done".into()))
+            .await
+            .unwrap();
+        let Json(report) = merge_progress(
+            State(state),
+            Query(MergeProgressQuery {
+                target: "legacy".into(),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(report["status"], "no_active_record");
+    }
+
     fn test_record(id: &str, with_blob: bool) -> RolloutRecord {
         let bytes = with_blob.then(|| b"master-blob".to_vec());
         RolloutRecord {
diff --git a/crates/lance-context-merge/src/lib.rs b/crates/lance-context-merge/src/lib.rs
index 7d547f2..ab2cf4b 100644
--- a/crates/lance-context-merge/src/lib.rs
+++ b/crates/lance-context-merge/src/lib.rs
@@ -306,6 +306,7 @@ impl Coordinator {
         operations.extend([
             self.remove_permits(execution),
             self.remove_progress(execution),
+            self.remove_work_progress(execution),
             TxnOp::delete(key, None),
             TxnOp::put(
                 target_lock_key(&self.prefix, &execution.target),
@@ -705,6 +706,55 @@ mod tests {
         assert!(!coordinator.publish_progress(&running, 6).await.unwrap());
     }
 
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn detailed_work_keeps_legacy_progress_cas_and_cannot_resurrect_execution() {
+        let (coordinator, _, proof, _) = fixture().await;
+        let execution = Execution::new("table", "master:catchup", "boot", 3600);
+        assert!(coordinator.reserve(&proof, &execution).await.unwrap());
+        let running = coordinator.start(&execution).await.unwrap().unwrap();
+        let report = serde_json::json!({"sequence": 4, "work": {"wal_rows_read": 100}});
+        assert!(coordinator
+            .publish_work_progress(&running, 4, report.clone())
+            .await
+            .unwrap());
+        assert_eq!(
+            coordinator.work_progress(&running).await.unwrap(),
+            Some(report.clone())
+        );
+        assert_eq!(
+            coordinator
+                .progress(&running)
+                .await
+                .unwrap()
+                .unwrap()
+                .sequence,
+            4
+        );
+        // A stale observation cannot revoke completed new work. Details have
+        // their own key, so the established exact-byte sequence CAS still works.
+        assert!(!coordinator.revoke_stalled(&running, Some(3)).await.unwrap());
+        assert!(coordinator.revoke_stalled(&running, Some(4)).await.unwrap());
+        assert!(!coordinator
+            .publish_work_progress(&running, 5, report.clone())
+            .await
+            .unwrap());
+        let uncertain = coordinator.get("table").await.unwrap().unwrap();
+        let frozen = coordinator
+            .freeze(&proof, &uncertain)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(coordinator.finish_recovery(&proof, &frozen).await.unwrap());
+        let recovered = coordinator.get("table").await.unwrap().unwrap();
+        assert!(coordinator.release(&proof, &recovered).await.unwrap());
+        assert!(coordinator.work_progress(&running).await.unwrap().is_none());
+        assert!(!coordinator
+            .publish_work_progress(&running, 6, report)
+            .await
+            .unwrap());
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
     async fn recovery_closes_commit_admission_and_retains_every_allowed_version() {
diff --git a/crates/lance-context-merge/src/progress.rs b/crates/lance-context-merge/src/progress.rs
index 0cd8cb4..5cbdcf6 100644
--- a/crates/lance-context-merge/src/progress.rs
+++ b/crates/lance-context-merge/src/progress.rs
@@ -9,6 +9,63 @@ pub struct ExecutionProgress {
 }
 
 impl Coordinator {
+    fn work_key(&self, execution: &Execution) -> String {
+        format!(
+            "{}/merge-work/{}",
+            self.prefix.trim_end_matches('/'),
+            execution.id
+        )
+    }
+
+    /// Publish work details atomically with the existing progress sequence.
+    /// Keep the old sequence wire format intact: mixed-version executors use
+    /// its exact bytes in the no-progress revocation compare-and-swap.
+    pub async fn publish_work_progress(
+        &self,
+        execution: &Execution,
+        sequence: u64,
+        report: serde_json::Value,
+    ) -> Result<bool> {
+        self.transact(
+            vec![Compare::value(
+                execution_key(&self.prefix, &execution.target),
+                CompareOp::Equal,
+                encode(execution),
+            )],
+            vec![
+                TxnOp::put(
+                    self.progress_key(execution),
+                    serde_json::to_vec(&ExecutionProgress { sequence }).unwrap(),
+                    None,
+                ),
+                TxnOp::put(
+                    self.work_key(execution),
+                    serde_json::to_vec(&report).map_err(|e| e.to_string())?,
+                    None,
+                ),
+            ],
+        )
+        .await
+    }
+
+    pub async fn work_progress(&self, execution: &Execution) -> Result<Option<serde_json::Value>> {
+        let response = self
+            .client
+            .clone()
+            .get(self.work_key(execution), None)
+            .await
+            .map_err(|e| e.to_string())?;
+        response
+            .kvs()
+            .first()
+            .map(|kv| serde_json::from_slice(kv.value()).map_err(|e| e.to_string()))
+            .transpose()
+    }
+
+    pub(crate) fn remove_work_progress(&self, execution: &Execution) -> TxnOp {
+        TxnOp::delete(self.work_key(execution), None)
+    }
+
     fn progress_key(&self, execution: &Execution) -> String {
         format!(
             "{}/merge-progress/{}",
diff --git a/docs/merge-work-progress.md b/docs/merge-work-progress.md
new file mode 100644
index 0000000..de1bfb8
--- /dev/null
+++ b/docs/merge-work-progress.md
@@ -0,0 +1,43 @@
+# Actual merge progress
+
+`GET /api/v1/merge-progress?target=<table>` reads the exact canonical task,
+execution and work report from etcd. It performs no task inventory or table
+payload scan. A live legacy task without execution telemetry returns
+`progress_unknown`; a lease heartbeat is not classified as merge progress.
+Older owned executors without detailed reports return
+`progress_details_unavailable` alongside their existing progress sequence.
+
+Resident maintenance reports executor process identity, canonical task ID,
+execution ID, observed time, last progress time, monotonic idle duration and
+configured idle allowance. Append merge work reports:
+
+* WAL batches/rows read and decoded Arrow bytes (not network byte throughput).
+* Encoded bytes, rows and files (not yet committed generations).
+* Successful base commits and generations reclaimed after manifest drain.
+* Active phase counts and the last completed stage. Several shard preparation
+  futures may be reading, waiting for memory or encoding at once.
+
+Phase transitions and reporting heartbeats do not advance the existing progress
+sequence. A continued WAL read, encoding callback or acknowledged commit does.
+The existing no-progress watchdog remains authoritative: continued work outlives
+the idle allowance, while unchanged completed work triggers cancellation of
+preparation followed by commit drain/storage fencing before replacement. Retry
+backoff applies after a failure; it is not evidence of activity or a substitute
+for progress detection.
+
+Detailed reports and the established sequence are published in one
+execution-qualified transaction. The sequence keeps its original wire format,
+so mixed-version no-progress compare-and-swap remains valid. The API rejects a
+sample if the execution changed or independently read sequence/report versions
+disagree. Consumers must compare samples of the same execution and check the
+report timestamp; `reported` is not a health verdict. Zero append counters on a
+different maintenance kind are not evidence that its work is stalled.
+
+The report is live execution state and is removed on normal release. A final
+structured log records completed work when the work future ends, before commit
+drain; that log alone does not certify safe writer termination. The existing
+task completion record remains the terminal result.
+
+This does not make legacy unfenced handlers cancellable, replay raw WAL or
+change ownership admission. Such handlers require a separately proved stop or
+storage fencing barrier before another publisher may take over.