Harden branch recovery and cleanup retention (#344)

* Harden branch recovery and cleanup retention

* Fail closed on blocked partial rollback

crates/omnigraph/src/branch_control.rs

@@ -0,0 +1,699 @@
+//! Crash classification for Lance's native branch controls.
+//!
+//! Lance branch creation is deliberately two-phase: it first commits a shallow
+//! clone under `tree/{branch}` and then writes authoritative `BranchContents`.
+//! Deletion removes `BranchContents` before reclaiming that tree.  These helpers
+//! keep OmniGraph on Lance's public APIs while making both operations bounded
+//! and idempotent under the documented single-writer-process branch-control
+//! boundary. Live graph names are path-prefix-disjoint because Lance cannot
+//! recursively reclaim an ancestor tree while a path-child branch remains.
+
+use std::collections::HashMap;
+
+use lance::Dataset;
+use lance::dataset::refs::{BranchContents, BranchIdentifier, check_valid_branch};
+
+use crate::error::{OmniError, Result};
+
+/// Result of a recoverable native create attempt.
+pub(crate) enum BranchCreateOutcome {
+    /// The requested branch is durably authoritative and its dataset opens.
+    Created(Dataset),
+    /// The target ref existed before this invocation. Callers classify its
+    /// ownership at their own logical layer; this helper never deletes it.
+    /// A ref that appears after target absence was captured is instead a typed
+    /// authority-change error and cannot enter an orphan-reclaim path.
+    RefAlreadyExists,
+}
+
+fn lance_error(error: lance::Error) -> OmniError {
+    OmniError::Lance(error.to_string())
+}
+
+/// Lance's local backend reports `NotFound` when the target tree is already
+/// absent, while object stores commonly treat the empty-prefix delete as a
+/// success. Normalize both to the idempotent contract every OmniGraph branch
+/// reconciler needs.
+pub(crate) async fn force_delete_branch_idempotent(
+    dataset: &mut Dataset,
+    branch: &str,
+) -> Result<()> {
+    let branches = list_branch_contents(dataset).await?;
+    if let Some(child) = path_descendant(&branches, branch) {
+        return Err(OmniError::manifest_conflict(format!(
+            "cannot reclaim branch '{branch}' while live branch '{child}' shares its physical \
+             Lance path; delete the child branch first"
+        )));
+    }
+    force_delete_branch_tree_unchecked(dataset, branch).await
+}
+
+async fn force_delete_branch_tree_unchecked(dataset: &mut Dataset, branch: &str) -> Result<()> {
+    match dataset.force_delete_branch(branch).await {
+        Ok(()) | Err(lance::Error::RefNotFound { .. }) | Err(lance::Error::NotFound { .. }) => {
+            Ok(())
+        }
+        Err(error) => Err(lance_error(error)),
+    }
+}
+
+async fn list_branch_contents(dataset: &Dataset) -> Result<HashMap<String, BranchContents>> {
+    dataset.list_branches().await.map_err(lance_error)
+}
+
+async fn branch_contents(dataset: &Dataset, branch: &str) -> Result<Option<BranchContents>> {
+    Ok(list_branch_contents(dataset).await?.get(branch).cloned())
+}
+
+pub(crate) fn path_descendant<'a>(
+    branches: &'a HashMap<String, BranchContents>,
+    branch: &str,
+) -> Option<&'a str> {
+    let prefix = format!("{branch}/");
+    branches
+        .keys()
+        .map(String::as_str)
+        .filter(|candidate| candidate.starts_with(&prefix))
+        .min()
+}
+
+fn path_collision<'a>(
+    branches: &'a HashMap<String, BranchContents>,
+    branch: &str,
+) -> Option<&'a str> {
+    let branch_prefix = format!("{branch}/");
+    branches
+        .keys()
+        .map(String::as_str)
+        .filter(|candidate| {
+            *candidate != branch
+                && *candidate != "main"
+                && (candidate.starts_with(&branch_prefix)
+                    || branch.starts_with(&format!("{candidate}/")))
+        })
+        .min()
+}
+
+fn encode_identifier(identifier: &BranchIdentifier) -> Result<String> {
+    serde_json::to_string(identifier).map_err(|error| {
+        OmniError::manifest_internal(format!("failed to encode Lance branch identifier: {error}"))
+    })
+}
+
+async fn authority_appeared_after_absence(dataset: &Dataset, branch: &str) -> Result<OmniError> {
+    match branch_contents(dataset, branch).await? {
+        Some(contents) => Ok(OmniError::manifest_read_set_changed(
+            format!("branch_identifier:{branch}"),
+            None,
+            Some(encode_identifier(&contents.identifier)?),
+        )),
+        None => Ok(OmniError::manifest_conflict(format!(
+            "branch authority for '{branch}' changed during absent-only reclaim; refusing \
+             destructive cleanup"
+        ))),
+    }
+}
+
+/// Reclaim only when the authoritative ref is freshly absent. Returns `false`
+/// if a ref exists and therefore nothing was touched. A live path-child is a
+/// conflict because Lance deliberately keeps the ancestor directory in that
+/// state; the graph namespace prevents new instances of that overlap.
+pub(crate) async fn reclaim_ref_absent_tree(dataset: &mut Dataset, branch: &str) -> Result<bool> {
+    let branches = list_branch_contents(dataset).await?;
+    if branches.contains_key(branch) {
+        return Ok(false);
+    }
+    if let Some(child) = path_descendant(&branches, branch) {
+        return Err(OmniError::manifest_conflict(format!(
+            "cannot reclaim absent branch '{branch}' while live branch '{child}' shares its \
+             physical Lance path; delete the child branch first"
+        )));
+    }
+    // This is the final authority read before the destructive call. The
+    // supported single-writer-process gate boundary prevents a local create
+    // from interleaving after it; an exact ref observed here is never passed to
+    // Lance's broad force-delete primitive.
+    force_delete_branch_tree_unchecked(dataset, branch).await?;
+    Ok(true)
+}
+
+/// A completed create receives an identifier equal to the parent's complete
+/// identifier mapping plus one newly-generated `(parent_version, uuid)` entry.
+/// The UUID itself cannot be pre-minted through Lance's public API, so this is
+/// the strongest completion proof available inside the supported
+/// single-writer-process boundary.
+fn matches_create_expectation(
+    contents: &BranchContents,
+    parent_branch: Option<&str>,
+    parent_version: u64,
+    parent_identifier: &BranchIdentifier,
+) -> bool {
+    let expected_parent = parent_branch.filter(|branch| *branch != "main");
+    let mapping = &contents.identifier.version_mapping;
+    let parent_mapping = &parent_identifier.version_mapping;
+    contents.parent_branch.as_deref() == expected_parent
+        && contents.parent_version == parent_version
+        && mapping.len() == parent_mapping.len() + 1
+        && mapping.starts_with(parent_mapping)
+        && mapping
+            .last()
+            .is_some_and(|(version, uuid)| *version == parent_version && !uuid.is_empty())
+}
+
+/// Create a branch with a bounded recovery classifier.
+///
+/// The caller must already hold OmniGraph's schema → source/target branch →
+/// table gate envelope. An absent `BranchContents` ref makes any same-name tree
+/// derived garbage, so it is reclaimed before the first attempt. An ambiguous
+/// native error is then classified from fresh authority: matching contents are
+/// accepted as a lost acknowledgement, mismatching contents are never deleted,
+/// and a ref-less clone is reclaimed before one bounded retry.
+pub(crate) async fn create_branch_recoverably(
+    source: &mut Dataset,
+    branch: &str,
+    source_version: u64,
+) -> Result<BranchCreateOutcome> {
+    // Lance validates inside phase 2 today. Validate before phase 1 so an
+    // invalid name cannot leave a clone-only zombie.
+    check_valid_branch(branch).map_err(lance_error)?;
+    if source.version().version != source_version {
+        return Err(OmniError::manifest_conflict(format!(
+            "branch source moved before native create: expected version {}, current {}",
+            source_version,
+            source.version().version
+        )));
+    }
+
+    let parent_branch = source.manifest().branch.clone();
+    let parent_identifier = source.branch_identifier().await.map_err(lance_error)?;
+
+    let initial_branches = list_branch_contents(source).await?;
+    if initial_branches.contains_key(branch) {
+        return Ok(BranchCreateOutcome::RefAlreadyExists);
+    }
+    if let Some(conflicting) = path_collision(&initial_branches, branch) {
+        return Err(OmniError::manifest_conflict(format!(
+            "cannot create branch '{branch}' while live branch '{conflicting}' shares its \
+             physical Lance path; live graph branch names may not be ancestors or descendants"
+        )));
+    }
+    if !reclaim_ref_absent_tree(source, branch).await? {
+        return Err(authority_appeared_after_absence(source, branch).await?);
+    }
+
+    for attempt in 0..2 {
+        let native_error = match source
+            .create_branch(branch, source_version, None)
+            .await
+            .map_err(lance_error)
+        {
+            Ok(created) => match crate::failpoints::maybe_fail(
+                crate::failpoints::names::BRANCH_CREATE_POST_NATIVE,
+            ) {
+                Ok(()) => return Ok(BranchCreateOutcome::Created(created)),
+                Err(error) => error,
+            },
+            Err(error) => error,
+        };
+
+        let observed = branch_contents(source, branch).await.map_err(|classifier_error| {
+            OmniError::manifest_internal(format!(
+                "native create of branch '{branch}' returned an ambiguous error ({native_error}); \
+                 reading BranchContents to classify it also failed ({classifier_error})"
+            ))
+        })?;
+        if let Some(contents) = observed {
+            if !matches_create_expectation(
+                &contents,
+                parent_branch.as_deref(),
+                source_version,
+                &parent_identifier,
+            ) {
+                return Err(OmniError::manifest_read_set_changed(
+                    format!("branch_identifier:{branch}"),
+                    None,
+                    Some(encode_identifier(&contents.identifier)?),
+                ));
+            }
+            let created = source.checkout_branch(branch).await.map_err(|error| {
+                OmniError::manifest_internal(format!(
+                    "branch '{}' has matching authoritative metadata after an ambiguous create, \
+                     but its branch dataset cannot be opened: {}; original native error: {}",
+                    branch, error, native_error
+                ))
+            })?;
+            return Ok(BranchCreateOutcome::Created(created));
+        }
+
+        match reclaim_ref_absent_tree(source, branch).await {
+            Ok(true) => {}
+            Ok(false) => {
+                return Err(authority_appeared_after_absence(source, branch).await?);
+            }
+            Err(cleanup_error) => {
+                return Err(OmniError::manifest_internal(format!(
+                    "native create of branch '{}' failed before authoritative metadata was visible \
+                     ({native_error}); clone-only cleanup also failed ({cleanup_error})",
+                    branch
+                )));
+            }
+        }
+        if attempt == 1 {
+            return Err(native_error);
+        }
+    }
+
+    unreachable!("bounded native branch-create loop returns from every attempt")
+}
+
+/// Delete a branch and classify Lance's authority-removal → tree-cleanup gap.
+///
+/// A missing ref after an error is logical success. Tree reclamation is derived
+/// state and is retried best-effort; a recreated identifier is never deleted.
+pub(crate) async fn delete_branch_recoverably(
+    dataset: &mut Dataset,
+    branch: &str,
+    expected_identifier: &BranchIdentifier,
+) -> Result<()> {
+    let initial = list_branch_contents(dataset).await?;
+    let Some(initial_contents) = initial.get(branch) else {
+        match reclaim_ref_absent_tree(dataset, branch).await {
+            Ok(true) => {}
+            Ok(false) => {
+                return Err(authority_appeared_after_absence(dataset, branch).await?);
+            }
+            Err(cleanup_error) => {
+                tracing::warn!(
+                    target: "omnigraph::branch_control",
+                    branch,
+                    error = %cleanup_error,
+                    "branch authority is already deleted; derived tree reclaim remains pending",
+                );
+            }
+        }
+        return Ok(());
+    };
+    if initial_contents.identifier != *expected_identifier {
+        return Err(OmniError::manifest_read_set_changed(
+            format!("branch_identifier:{branch}"),
+            Some(encode_identifier(expected_identifier)?),
+            Some(encode_identifier(&initial_contents.identifier)?),
+        ));
+    }
+    if let Some(child) = path_descendant(&initial, branch) {
+        return Err(OmniError::manifest_conflict(format!(
+            "cannot delete branch '{branch}' while live branch '{child}' shares its physical \
+             Lance path; delete the child branch first"
+        )));
+    }
+
+    let native_result = match dataset.delete_branch(branch).await.map_err(lance_error) {
+        Ok(()) => {
+            crate::failpoints::maybe_fail(crate::failpoints::names::BRANCH_DELETE_POST_NATIVE)
+        }
+        Err(error) => Err(error),
+    };
+    let native_error = match native_result {
+        Ok(()) => return Ok(()),
+        Err(error) => error,
+    };
+
+    let observed = branch_contents(dataset, branch)
+        .await
+        .map_err(|classifier_error| {
+            OmniError::manifest_internal(format!(
+                "native delete of branch '{branch}' returned an ambiguous error ({native_error}); \
+             reading BranchContents to classify it also failed ({classifier_error})"
+            ))
+        })?;
+    match observed {
+        None => {
+            match reclaim_ref_absent_tree(dataset, branch).await {
+                Ok(true) => {}
+                Ok(false) => {
+                    return Err(authority_appeared_after_absence(dataset, branch).await?);
+                }
+                Err(cleanup_error) => {
+                    tracing::warn!(
+                        target: "omnigraph::branch_control",
+                        branch,
+                        error = %cleanup_error,
+                        "branch authority is deleted; derived tree reclaim remains pending",
+                    );
+                }
+            }
+            Ok(())
+        }
+        Some(contents) if contents.identifier == *expected_identifier => Err(native_error),
+        Some(contents) => Err(OmniError::manifest_read_set_changed(
+            format!("branch_identifier:{branch}"),
+            Some(encode_identifier(expected_identifier)?),
+            Some(encode_identifier(&contents.identifier)?),
+        )),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+
+    use arrow_array::{Int32Array, RecordBatch, RecordBatchIterator};
+    use arrow_schema::{DataType, Field, Schema};
+    use lance::dataset::{WriteMode, WriteParams};
+
+    async fn test_dataset(dir: &tempfile::TempDir) -> Dataset {
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Int32Array::from(vec![1]))],
+        )
+        .unwrap();
+        let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+        // forbidden-api-allow: test-only raw Lance fixture for native branch-control truth cells
+        Dataset::write(
+            reader,
+            dir.path().to_str().unwrap(),
+            Some(WriteParams {
+                mode: WriteMode::Create,
+                auto_cleanup: None,
+                skip_auto_cleanup: true,
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap()
+    }
+
+    #[test]
+    fn create_match_requires_exact_parent_incarnation_prefix() {
+        let parent = BranchIdentifier {
+            version_mapping: vec![(2, "parent-a".to_string())],
+        };
+        let matching = BranchContents {
+            parent_branch: Some("source".to_string()),
+            identifier: BranchIdentifier {
+                version_mapping: vec![(2, "parent-a".to_string()), (7, "child".to_string())],
+            },
+            parent_version: 7,
+            create_at: 0,
+            manifest_size: 0,
+            metadata: Default::default(),
+        };
+        assert!(matches_create_expectation(
+            &matching,
+            Some("source"),
+            7,
+            &parent
+        ));
+
+        let foreign_parent = BranchIdentifier {
+            version_mapping: vec![(2, "parent-b".to_string())],
+        };
+        assert!(!matches_create_expectation(
+            &matching,
+            Some("source"),
+            7,
+            &foreign_parent
+        ));
+    }
+
+    #[tokio::test]
+    async fn delete_accepts_absent_authority_and_reclaims_remaining_tree() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+        let identifier = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+        std::fs::remove_file(
+            dir.path()
+                .join("_refs")
+                .join("branches")
+                .join("feature.json"),
+        )
+        .unwrap();
+
+        delete_branch_recoverably(&mut dataset, "feature", &identifier)
+            .await
+            .unwrap();
+        assert!(!dir.path().join("tree").join("feature").exists());
+    }
+
+    #[tokio::test]
+    async fn delete_preserves_native_error_while_same_identifier_remains() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        let mut feature = dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+        let identifier = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+        let feature_version = feature.version().version;
+        feature
+            .create_branch("dependent", feature_version, None)
+            .await
+            .unwrap();
+
+        let error = delete_branch_recoverably(&mut dataset, "feature", &identifier)
+            .await
+            .expect_err("a lineage-dependent ref must preserve Lance's delete refusal");
+        assert!(error.to_string().contains("referenc"));
+        assert_eq!(
+            dataset
+                .list_branches()
+                .await
+                .unwrap()
+                .get("feature")
+                .unwrap()
+                .identifier,
+            identifier
+        );
+    }
+
+    #[tokio::test]
+    async fn delete_rejects_recreated_identifier_without_removing_it() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+        let original = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+        dataset.delete_branch("feature").await.unwrap();
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+        let recreated = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+        assert_ne!(original, recreated);
+
+        let error = delete_branch_recoverably(&mut dataset, "feature", &original)
+            .await
+            .expect_err("a recreated target must be fenced by its identifier");
+        let OmniError::Manifest(error) = error else {
+            panic!("expected typed manifest conflict");
+        };
+        match error.details {
+            Some(crate::error::ManifestConflictDetails::ReadSetChanged {
+                member,
+                expected: Some(expected),
+                actual: Some(actual),
+            }) => {
+                assert_eq!(member, "branch_identifier:feature");
+                assert_eq!(expected, serde_json::to_string(&original).unwrap());
+                assert_eq!(actual, serde_json::to_string(&recreated).unwrap());
+            }
+            other => panic!("expected identifier ReadSetChanged, got {other:?}"),
+        }
+        assert_eq!(
+            dataset
+                .list_branches()
+                .await
+                .unwrap()
+                .get("feature")
+                .unwrap()
+                .identifier,
+            recreated,
+            "classifier must never delete the recreated authority"
+        );
+    }
+
+    #[tokio::test]
+    async fn lower_create_chokepoint_rejects_prefix_collision_before_clone() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+
+        let error = match create_branch_recoverably(&mut dataset, "feature/child", version).await {
+            Ok(_) => panic!("the lower branch-control surface must enforce prefix disjointness"),
+            Err(error) => error,
+        };
+        assert!(error.to_string().contains("feature"));
+        assert!(error.to_string().contains("physical Lance path"));
+        assert!(
+            !dataset
+                .list_branches()
+                .await
+                .unwrap()
+                .contains_key("feature/child"),
+            "admission refusal must precede authoritative ref creation"
+        );
+        assert!(
+            !dir.path()
+                .join("tree")
+                .join("feature")
+                .join("child")
+                .exists(),
+            "admission refusal must precede Lance's shallow-clone phase"
+        );
+    }
+
+    #[tokio::test]
+    async fn lower_create_preserves_preexisting_same_name_ref() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+        let before = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+
+        let outcome = create_branch_recoverably(&mut dataset, "feature", version)
+            .await
+            .unwrap();
+        assert!(matches!(outcome, BranchCreateOutcome::RefAlreadyExists));
+        let after = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+        assert_eq!(
+            before, after,
+            "classification must not replace the live ref"
+        );
+    }
+
+    #[tokio::test]
+    async fn absent_only_reclaim_preserves_authority_that_appeared_after_prior_absence() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        assert!(
+            !dataset
+                .list_branches()
+                .await
+                .unwrap()
+                .contains_key("feature"),
+            "precondition: an earlier classifier observed target absence"
+        );
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+        let before = dataset
+            .list_branches()
+            .await
+            .unwrap()
+            .get("feature")
+            .unwrap()
+            .identifier
+            .clone();
+
+        assert!(
+            !reclaim_ref_absent_tree(&mut dataset, "feature")
+                .await
+                .unwrap(),
+            "the final authority check must refuse destructive cleanup"
+        );
+        assert_eq!(
+            dataset
+                .list_branches()
+                .await
+                .unwrap()
+                .get("feature")
+                .unwrap()
+                .identifier,
+            before
+        );
+    }
+
+    #[tokio::test]
+    async fn force_reclaim_reports_lexically_first_live_physical_path_child() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut dataset = test_dataset(&dir).await;
+        let version = dataset.version().version;
+        dataset
+            .create_branch("feature/zeta", version, None)
+            .await
+            .unwrap();
+        dataset
+            .create_branch("feature/alpha", version, None)
+            .await
+            .unwrap();
+        dataset
+            .create_branch("feature", version, None)
+            .await
+            .unwrap();
+
+        let error = force_delete_branch_idempotent(&mut dataset, "feature")
+            .await
+            .expect_err("ancestor reclaim must not report success while a path-child is live");
+        assert!(error.to_string().contains("feature/alpha"));
+        assert!(
+            dataset
+                .list_branches()
+                .await
+                .unwrap()
+                .contains_key("feature"),
+            "preflight refusal must preserve ancestor authority"
+        );
+    }
+}

crates/omnigraph/src/db/graph_coordinator.rs

@@ -238,10 +238,10 @@ impl GraphCoordinator {
         let branch = normalize_branch_name(name)?
             .ok_or_else(|| OmniError::manifest("cannot create branch 'main'".to_string()))?;
 
-        // Manifest is the single branch authority (it forks `__manifest` first).
-        // The commit graph is a pure `__manifest` projection (Phase B), so a
-        // branch create is one atomic manifest op — no derived commit-graph
-        // branch to fork, and nothing to roll back.
+        // Manifest BranchContents is the single branch authority. Lance creates
+        // it in two physical phases (shallow clone, then BranchContents); the
+        // manifest coordinator classifies/reclaims a clone-only zombie before
+        // a bounded retry. No graph-lineage branch is created or rolled back.
         self.manifest.create_branch(&branch).await
     }
 
@@ -255,10 +255,11 @@ impl GraphCoordinator {
             )));
         }
 
-        // Manifest is the single branch authority (Phase B): one atomic op makes
-        // the branch cease to exist. The commit graph is a pure `__manifest`
-        // projection with no derived branch to reclaim; the per-table data forks
-        // are reclaimed by `cleanup`, not here.
+        // Removing manifest BranchContents is the logical visibility point.
+        // Lance reclaims the branch tree afterward, so an error may still mean
+        // logical deletion succeeded; the manifest coordinator reclassifies
+        // that outcome from fresh authority. Per-table data forks remain
+        // derived state and are reclaimed by the engine afterward.
         self.manifest.delete_branch(&branch).await
     }
 

crates/omnigraph/src/db/manifest.rs

@@ -634,10 +634,14 @@ impl ManifestCoordinator {
 
     pub async fn create_branch(&mut self, name: &str) -> Result<()> {
         let mut ds = self.dataset.clone();
-        ds.create_branch(name, self.version(), None)
-            .await
-            .map_err(|e| OmniError::Lance(e.to_string()))?;
-        Ok(())
+        match crate::branch_control::create_branch_recoverably(&mut ds, name, self.version())
+            .await?
+        {
+            crate::branch_control::BranchCreateOutcome::Created(_) => Ok(()),
+            crate::branch_control::BranchCreateOutcome::RefAlreadyExists => Err(
+                OmniError::manifest_conflict(format!("branch '{}' already exists", name)),
+            ),
+        }
     }
 
     pub async fn delete_branch(&mut self, name: &str) -> Result<()> {
@@ -649,9 +653,17 @@ impl ManifestCoordinator {
             crate::instrumentation::manifest_wrapper(),
         )
         .await?;
-        ds.delete_branch(name)
+        let branches = ds
+            .list_branches()
             .await
-            .map_err(|e| OmniError::Lance(e.to_string()))?;
+            .map_err(|error| OmniError::Lance(error.to_string()))?;
+        let expected_identifier = branches
+            .get(name)
+            .ok_or_else(|| OmniError::manifest_not_found(format!("branch '{}' not found", name)))?
+            .identifier
+            .clone();
+        crate::branch_control::delete_branch_recoverably(&mut ds, name, &expected_identifier)
+            .await?;
         self.dataset = open_manifest_dataset(&self.root_uri, self.active_branch.as_deref()).await?;
         self.known_state = read_manifest_state(&self.dataset).await?;
         Ok(())

crates/omnigraph/src/db/manifest/recovery.rs

@@ -1817,8 +1817,28 @@ async fn process_sidecar(
                 return Ok(false);
             }
             if matches!(mode, RecoveryMode::Full) {
-                cleanup_unpublished_no_effect_forks(root_uri, storage.as_ref(), sidecar, &states)
-                    .await?;
+                if let NoEffectForkCleanup::DeferredPathChild {
+                    table_path,
+                    target_branch,
+                    path_child,
+                } = cleanup_unpublished_no_effect_forks(
+                    root_uri,
+                    storage.as_ref(),
+                    sidecar,
+                    &states,
+                )
+                .await?
+                {
+                    warn!(
+                        operation_id = sidecar.operation_id.as_str(),
+                        table_path,
+                        branch = target_branch,
+                        path_child,
+                        "recovery: deferring no-effect fork cleanup until legacy path-child \
+                         branches are deleted leaf-first"
+                    );
+                    return Ok(false);
+                }
             }
             warn!(
                 operation_id = sidecar.operation_id.as_str(),
@@ -2369,6 +2389,16 @@ enum EffectOwnership {
     Unverifiable,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum NoEffectForkCleanup {
+    Complete,
+    DeferredPathChild {
+        table_path: String,
+        target_branch: String,
+        path_child: String,
+    },
+}
+
 /// Remove first-touch named-branch refs created by an Armed v3 attempt that
 /// never landed this table's planned transaction.
 ///
@@ -2386,9 +2416,9 @@ async fn cleanup_unpublished_no_effect_forks(
     storage: &dyn StorageAdapter,
     sidecar: &RecoverySidecar,
     states: &[ClassifiedTable],
-) -> Result<()> {
+) -> Result<NoEffectForkCleanup> {
     if sidecar.protocol_v3.is_none() {
-        return Ok(());
+        return Ok(NoEffectForkCleanup::Complete);
     }
 
     let all_sidecars = list_sidecars(root_uri, storage).await?;
@@ -2432,9 +2462,38 @@ async fn cleanup_unpublished_no_effect_forks(
             .list_branches()
             .await
             .map_err(|error| OmniError::Lance(error.to_string()))?;
+        if let Some(child) = crate::branch_control::path_descendant(&branches, target_branch) {
+            // Lance cannot reclaim an ancestor tree while a slash-separated
+            // path-child remains. Old stores could admit that namespace shape.
+            // Keep the ownership sidecar and let open complete so the operator
+            // can delete the child branch first; the next Full sweep then
+            // rechecks authority and reclaims the ancestor fork.
+            warn!(
+                operation_id = sidecar.operation_id.as_str(),
+                table_path = pin.table_path.as_str(),
+                branch = target_branch,
+                path_child = child,
+                "recovery: deferring unpublished fork cleanup for legacy path overlap"
+            );
+            return Ok(NoEffectForkCleanup::DeferredPathChild {
+                table_path: pin.table_path.clone(),
+                target_branch: target_branch.to_string(),
+                path_child: child.to_string(),
+            });
+        }
         let Some(contents) = branches.get(target_branch) else {
-            // Crash-before-fork, or a previous cleanup pass that crashed before
-            // its rollback publish. Both are already converged physically.
+            // BranchContents is authoritative, but Lance create writes the
+            // shallow-cloned target dataset first. The absent-ref state is
+            // therefore either crash-before-fork (idempotent no-op) or an
+            // exact clone-only zombie owned by this already-armed sidecar.
+            // Reclaim both through Lance's force API before retiring intent;
+            // merely skipping here leaves the zombie blocking every retry.
+            if !crate::branch_control::reclaim_ref_absent_tree(&mut dataset, target_branch).await? {
+                return Err(OmniError::manifest_conflict(format!(
+                    "target ref '{target_branch}' appeared during no-effect recovery; refusing \
+                     to retire its ownership intent"
+                )));
+            }
             continue;
         };
         if contents.parent_version != pin.expected_version {
@@ -2468,7 +2527,7 @@ async fn cleanup_unpublished_no_effect_forks(
             .await
             .map_err(|error| OmniError::Lance(error.to_string()))?;
     }
-    Ok(())
+    Ok(NoEffectForkCleanup::Complete)
 }
 
 fn table_requires_rollback_effect(state: &ClassifiedTable) -> bool {
@@ -2546,9 +2605,33 @@ async fn roll_back_sidecar(
     sidecar: &RecoverySidecar,
     states: &[ClassifiedTable],
 ) -> Result<()> {
+    // An Armed multi-table attempt can create every first-touch ref and then
+    // land effects on only a subset. No-effect refs are not selected by the
+    // rollback manifest publish, so they must be removed BEFORE that publish.
+    // If recovery crashed after publishing first, the fixed rollback outcome
+    // would make the next pass finalize/delete the sidecar without ever seeing
+    // the still-orphaned refs. A rollback that owns any physical effect may not
+    // defer and let read-write open succeed: legacy writers are not all enrolled
+    // in the v3 preparation barrier. Fail closed until the path child is removed.
+    if let NoEffectForkCleanup::DeferredPathChild {
+        table_path,
+        target_branch,
+        path_child,
+    } = cleanup_unpublished_no_effect_forks(root_uri, storage, sidecar, states).await?
+    {
+        return Err(OmniError::manifest_internal(format!(
+            "OCC recovery sidecar '{}' owns physical effects but cannot clean unpublished fork \
+             '{}:{}' while legacy path-child '{}' is live; refusing read-write open; delete the \
+             child branch leaf-first using an existing handle or an offline Lance-level branch \
+             tool, then reopen",
+            sidecar.operation_id, table_path, target_branch, path_child
+        )));
+    }
+
     // Once the fixed rollback commit is visible, early recovery finalization no
     // longer has pre-restore table observations. Persist the exact audit plan
-    // first so that path can replay it without fabricating outcomes from pins.
+    // after fork cleanup and before the first restore so that path can replay it
+    // without fabricating outcomes from pins.
     let prepared_v3 = if sidecar.protocol_v3.is_some() {
         Some(prepare_v3_rollback_audit_plan(root_uri, storage, sidecar, states).await?)
     } else {
@@ -2556,14 +2639,6 @@ async fn roll_back_sidecar(
     };
     let sidecar = prepared_v3.as_ref().unwrap_or(sidecar);
 
-    // An Armed multi-table attempt can create every first-touch ref and then
-    // land effects on only a subset. No-effect refs are not selected by the
-    // rollback manifest publish, so they must be removed BEFORE that publish.
-    // If recovery crashed after publishing first, the fixed rollback outcome
-    // would make the next pass finalize/delete the sidecar without ever seeing
-    // the still-orphaned refs.
-    cleanup_unpublished_no_effect_forks(root_uri, storage, sidecar, states).await?;
-
     // Restore every drifted table (RolledPastExpected / UnexpectedAtP1 /
     // UnexpectedMultistep) to its manifest-pinned content, then PUBLISH so
     // `manifest == Lance HEAD` for each — symmetric with roll-forward. The

crates/omnigraph/src/db/omnigraph.rs

@@ -1825,7 +1825,41 @@ impl Omnigraph {
         queue_keys
     }
 
+    fn ensure_branch_create_namespace_safe(target: &str, branches: &[String]) -> Result<()> {
+        if branches.iter().any(|candidate| candidate == target) {
+            return Err(OmniError::manifest_conflict(format!(
+                "branch '{}' already exists",
+                target
+            )));
+        }
+
+        let target_prefix = format!("{target}/");
+        if let Some(conflicting) = branches.iter().find(|candidate| {
+            candidate.as_str() != "main"
+                && (candidate.starts_with(&target_prefix)
+                    || target.starts_with(&format!("{candidate}/")))
+        }) {
+            return Err(OmniError::manifest_conflict(format!(
+                "cannot create branch '{target}' while live branch '{conflicting}' shares its \
+                 physical Lance path; live graph branch names may not be ancestors or descendants"
+            )));
+        }
+
+        Ok(())
+    }
+
     async fn ensure_branch_delete_safe(&self, branch: &str, branches: &[String]) -> Result<()> {
+        let path_prefix = format!("{branch}/");
+        if let Some(child) = branches
+            .iter()
+            .find(|candidate| candidate.starts_with(&path_prefix))
+        {
+            return Err(OmniError::manifest_conflict(format!(
+                "cannot delete branch '{branch}' while live branch '{child}' shares its physical \
+                 Lance path; delete the child branch first"
+            )));
+        }
+
         let descendants = self
             .coordinator
             .read()
@@ -1921,7 +1955,9 @@ impl Omnigraph {
             .map(|entry| (entry.table_key.clone(), entry.table_path.clone()))
             .collect::<Vec<_>>();
 
-        // Authority flip (+ best-effort commit-graph reclaim) — must succeed.
+        // Authority removal is the logical branch deletion. Lance tree cleanup
+        // follows that ref removal; the coordinator classifies an absent ref as
+        // success even if the physical cleanup acknowledgement failed.
         self.coordinator.write().await.branch_delete(branch).await?;
         // Best-effort per-table fork reclaim; cleanup reconciles any leftover.
         self.cleanup_deleted_branch_tables(branch, &owned_tables)
@@ -1998,20 +2034,8 @@ impl Omnigraph {
         self.refresh_coordinator_only().await?;
         self.ensure_schema_apply_not_locked("branch_create").await?;
         self.ensure_schema_state_valid().await?;
-        if self
-            .coordinator
-            .read()
-            .await
-            .all_branches()
-            .await?
-            .iter()
-            .any(|branch| branch == &target)
-        {
-            return Err(OmniError::manifest_conflict(format!(
-                "branch '{}' already exists",
-                target
-            )));
-        }
+        let branches = self.coordinator.read().await.all_branches().await?;
+        Self::ensure_branch_create_namespace_safe(&target, &branches)?;
         self.coordinator.write().await.branch_create(name).await
     }
 
@@ -2107,20 +2131,8 @@ impl Omnigraph {
         self.ensure_schema_apply_not_locked("branch_create_from")
             .await?;
         self.ensure_schema_state_valid().await?;
-        if self
-            .coordinator
-            .read()
-            .await
-            .all_branches()
-            .await?
-            .iter()
-            .any(|candidate| candidate == &target_branch)
-        {
-            return Err(OmniError::manifest_conflict(format!(
-                "branch '{}' already exists",
-                target_branch
-            )));
-        }
+        let branches = self.coordinator.read().await.all_branches().await?;
+        Self::ensure_branch_create_namespace_safe(&target_branch, &branches)?;
         // Operate on a freshly-opened source coordinator that's owned locally
         // — never touch `self.coordinator`. The pre-fix implementation used
         // `swap_coordinator_for_branch` + operate + `restore_coordinator` as

crates/omnigraph/src/db/omnigraph/optimize.rs

@@ -18,12 +18,13 @@
 //!   older manifest versions until `cleanup` runs.
 //! * `cleanup_all_tables` — Lance `cleanup_old_versions` on every table.
 //!   Removes manifests (and their unique fragments) older than the configured
-//!   retention. Destructive to version history — callers should gate this
-//!   behind an explicit confirm flag at the CLI layer.
+//!   retention, capped at the oldest main-table version inherited by any live
+//!   lazy graph branch. Destructive to unreferenced version history — callers
+//!   should gate this behind an explicit confirm flag at the CLI layer.
 //!
 //! Both orchestrate the graph's node + edge datasets from main authority;
-//! cleanup preserves Lance-referenced named-branch history according to its
-//! retention policy.
+//! cleanup preserves both Lance-referenced native branch history and the
+//! graph-level lazy-branch references Lance cannot observe.
 
 use std::time::Duration;
 
@@ -786,7 +787,8 @@ async fn compact_internal_table(
 
 /// Run Lance `cleanup_old_versions` on every node + edge table on `main`,
 /// using [`CleanupPolicyOptions`]. The latest manifest is always preserved
-/// regardless (Lance invariant).
+/// regardless (Lance invariant), and the requested cutoff is capped at the
+/// oldest main-table version inherited by a live lazy graph branch.
 pub async fn cleanup_all_tables(
     db: &mut Omnigraph,
     options: CleanupPolicyOptions,
@@ -896,8 +898,77 @@ pub async fn cleanup_all_tables(
         ));
     }
 
+    // Lance protects versions referenced by its own per-dataset branches, but
+    // an OmniGraph branch is lazy: until a table is first written on that
+    // branch its manifest entry points directly at an older MAIN version and
+    // no Lance branch ref exists on the data table. Resolve every live graph
+    // branch from fresh authority while schema + all branch/table gates are
+    // held, then cap each main dataset's GC cutoff at its oldest such pin.
+    // Main itself participates: its manifest-visible version must open and
+    // equal Lance HEAD, so uncovered drift is repaired before cleanup rather
+    // than letting HEAD-based GC collect graph-visible authority.
+    // Any branch snapshot read failure aborts before the first table GC: an
+    // unknown live reference is never evidence that a version is disposable.
+    let mut oldest_live_main_version_by_path = std::collections::HashMap::<String, u64>::new();
+    for branch_target in &graph_branches {
+        if branch_target
+            .as_deref()
+            .is_some_and(crate::db::is_internal_system_branch)
+        {
+            continue;
+        }
+        let branch_label = branch_target.as_deref().unwrap_or("main");
+        let branch_snapshot = db
+            .fresh_snapshot_for_branch(branch_target.as_deref())
+            .await
+            .map_err(|err| {
+                OmniError::manifest_conflict(format!(
+                    "cleanup could not classify live branch '{branch_label}'; refusing version GC: {err}"
+                ))
+            })?;
+        for entry in branch_snapshot
+            .entries()
+            .filter(|entry| entry.table_branch.is_none())
+        {
+            // Validate that the exact protected version is still openable
+            // before GC starts. This catches pre-existing damage from an older
+            // cleanup implementation and keeps the sweep fail-closed instead
+            // of deleting unrelated history around an already-broken branch.
+            entry.open(db.root_uri(), None).await.map_err(|err| {
+                OmniError::manifest_conflict(format!(
+                    "cleanup could not classify live branch '{branch_label}' table '{}' at main version {}; refusing version GC: {err}",
+                    entry.table_key, entry.table_version
+                ))
+            })?;
+            let full_path = format!("{}/{}", db.root_uri, entry.table_path);
+            if branch_target.is_none() {
+                let head = db.storage().open_dataset_head(&full_path, None).await?;
+                if head.version() != entry.table_version {
+                    return Err(OmniError::manifest_conflict(format!(
+                        "cleanup found uncovered HEAD drift for table '{}': manifest version {}, \
+                         Lance HEAD {}; run `omnigraph repair` before version GC",
+                        entry.table_key,
+                        entry.table_version,
+                        head.version()
+                    )));
+                }
+            }
+            oldest_live_main_version_by_path
+                .entry(full_path)
+                .and_modify(|oldest| *oldest = (*oldest).min(entry.table_version))
+                .or_insert(entry.table_version);
+        }
+    }
+
     let before_timestamp = options.older_than.map(|d| Utc::now() - d);
     let keep_versions = options.keep_versions;
+    let table_tasks = table_tasks
+        .into_iter()
+        .map(|(table_key, full_path)| {
+            let live_main_floor = oldest_live_main_version_by_path.get(&full_path).copied();
+            (table_key, full_path, live_main_floor)
+        })
+        .collect::<Vec<_>>();
 
     if table_tasks.is_empty() {
         return Ok(Vec::new());
@@ -911,7 +982,7 @@ pub async fn cleanup_all_tables(
     // cleanup is the convergence backstop, so it must do as much as it can and
     // converge on re-run rather than fail wholesale (invariant 13).
     let results: Vec<TableCleanupStats> = futures::stream::iter(table_tasks.into_iter())
-        .map(|(table_key, full_path)| async move {
+        .map(|(table_key, full_path, live_main_floor)| async move {
             let outcome: Result<RemovalStats> = async {
                 crate::failpoints::maybe_fail(crate::failpoints::names::CLEANUP_TABLE_GC)?;
                 // `cleanup_old_versions` is a Lance-only maintenance API not
@@ -919,9 +990,36 @@ pub async fn cleanup_all_tables(
                 // above for the same rationale. Unwrap via `into_dataset()`.
                 let handle = storage.open_dataset_head(&full_path, None).await?;
                 let ds = handle.into_dataset();
-                let before_version = keep_versions
-                    .map(|n| ds.version().version.saturating_sub(n as u64))
-                    .filter(|v| *v > 0);
+                let requested_before_version = if let Some(keep) = keep_versions {
+                    // Lance versions are not safely derivable from HEAD
+                    // arithmetic after prior GC. Use the actual ordered
+                    // version list so `keep=N` retains exactly the newest N
+                    // available versions (with HEAD as the unavoidable floor
+                    // when N=0).
+                    let versions = ds
+                        .versions()
+                        .await
+                        .map_err(|error| OmniError::Lance(error.to_string()))?;
+                    let retain = (keep as usize).max(1);
+                    let cutoff = if versions.len() <= retain {
+                        versions.first()
+                    } else {
+                        versions.get(versions.len() - retain)
+                    }
+                    .ok_or_else(|| {
+                        OmniError::manifest_internal(format!(
+                            "cleanup found no versions for open table '{table_key}'"
+                        ))
+                    })?;
+                    Some(cutoff.version)
+                } else {
+                    None
+                };
+                let before_version = match (requested_before_version, live_main_floor) {
+                    (Some(requested), Some(floor)) => Some(requested.min(floor)),
+                    (None, Some(floor)) => Some(floor),
+                    (requested, None) => requested,
+                };
                 let policy = CleanupPolicy {
                     before_timestamp,
                     before_version,

crates/omnigraph/src/exec/staging.rs

@@ -1155,6 +1155,10 @@ impl StagedMutation {
                 row_count: state.row_count,
                 version_metadata: state.version_metadata,
             });
+            crate::failpoints::maybe_fail(crate::failpoints::names::MUTATION_POST_TABLE_COMMIT)
+                .map_err(|error| {
+                    OmniError::recovery_required(operation_id.clone(), error.to_string())
+                })?;
         }
 
         if let Err(error) = confirm_occ_sidecar_phase_b(

crates/omnigraph/src/failpoints.rs

@@ -38,7 +38,15 @@ pub(crate) fn maybe_fail_retryable_contention(name: &str) -> Result<()> {
 /// reference these constants instead of bare string literals, so a typo is a
 /// compile error rather than a silently-never-firing failpoint.
 pub mod names {
+    /// After Lance returns success from its two-phase native create, before
+    /// OmniGraph acknowledges it. Recovery must classify the matching
+    /// BranchContents as a completed create (lost acknowledgement).
+    pub const BRANCH_CREATE_POST_NATIVE: &str = "branch_create.post_native";
     pub const BRANCH_DELETE_BEFORE_TABLE_CLEANUP: &str = "branch_delete.before_table_cleanup";
+    /// After Lance returns success from native delete, before OmniGraph
+    /// acknowledges it. Recovery must classify the absent BranchContents as a
+    /// completed logical deletion.
+    pub const BRANCH_DELETE_POST_NATIVE: &str = "branch_delete.post_native";
     /// Branch delete holds the schema, target-branch, and fresh-catalog table
     /// envelope and has completed its final recovery check, before the native
     /// manifest-ref mutation.
@@ -94,6 +102,11 @@ pub mod names {
     /// After every deferred first-touch table ref is created under a durable
     /// v3 sidecar, before any staged data transaction advances target HEAD.
     pub const MUTATION_POST_FORK_PRE_COMMIT: &str = "mutation.post_fork_pre_commit";
+    /// After each exact staged table transaction advances HEAD, before the next
+    /// table effect or Phase-B confirmation. Used to leave a real partial
+    /// multi-table v3 attempt whose remaining first-touch fork still needs
+    /// recovery cleanup.
+    pub const MUTATION_POST_TABLE_COMMIT: &str = "mutation.post_table_commit";
     /// After the v3 ownership sidecar is durable but before the first deferred
     /// named-table ref is created. Recovery must accept the absent target ref.
     pub const MUTATION_POST_SIDECAR_PRE_FORK: &str = "mutation.post_sidecar_pre_fork";

crates/omnigraph/src/lib.rs

@@ -7,6 +7,7 @@
 // future Lance bumps stop needing this.
 #![recursion_limit = "256"]
 
+mod branch_control;
 pub mod changes;
 pub mod db;
 pub mod embedding;

crates/omnigraph/src/storage_layer.rs

@@ -295,14 +295,12 @@ pub trait TableStorage: sealed::Sealed + Send + Sync + Debug {
         target_branch: &str,
     ) -> Result<ForkOutcome<SnapshotHandle>>;
 
-    async fn delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()>;
-
-    /// Idempotent variant of `delete_branch` used by the best-effort fork
-    /// reclaim under branch delete (`db/omnigraph.rs::cleanup_deleted_branch_tables`)
+    /// Idempotent branch-tree reclaim used by the best-effort fork cleanup
+    /// under branch delete (`db/omnigraph.rs::cleanup_deleted_branch_tables`)
     /// and by the orphan-fork reconciler in `optimize`. Tolerates an
     /// already-absent branch (both Lance's `RefNotFound` and the local-store
     /// `NotFound` quirk on a missing `tree/{branch}/` dir). A still-referenced
-    /// branch (`RefConflict`) still surfaces as `OmniError::Lance`.
+    /// branch (`RefConflict`) or live physical path-child remains an error.
     async fn force_delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()>;
 
     /// List the named Lance branches present on the dataset at `dataset_uri`.
@@ -591,10 +589,6 @@ impl TableStorage for TableStore {
         )
     }
 
-    async fn delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()> {
-        TableStore::delete_branch(self, dataset_uri, branch).await
-    }
-
     async fn force_delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()> {
         TableStore::force_delete_branch(self, dataset_uri, branch).await
     }

crates/omnigraph/src/table_store.rs

@@ -277,19 +277,6 @@ impl TableStore {
         }
     }
 
-    pub async fn delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()> {
-        let mut ds = crate::instrumentation::open_dataset(
-            dataset_uri,
-            crate::instrumentation::VersionResolution::Latest,
-            Some(&self.session),
-            crate::instrumentation::table_wrapper(),
-        )
-        .await?;
-        ds.delete_branch(branch)
-            .await
-            .map_err(|e| OmniError::Lance(e.to_string()))
-    }
-
     /// List the named Lance branches present on the dataset at `dataset_uri`.
     /// The `cleanup` orphan reconciler diffs this against the manifest branch
     /// set to find orphaned per-table forks. `main`/default is not a named
@@ -311,15 +298,16 @@ impl TableStore {
 
     /// Idempotently drop `branch` from the dataset at `dataset_uri`.
     ///
-    /// Unlike [`delete_branch`](Self::delete_branch), this tolerates an
-    /// already-absent branch — both a missing contents ref (Lance's
+    /// This tolerates an already-absent branch — both a missing contents ref (Lance's
     /// `force_delete_branch` handles that) and a missing `tree/{branch}/`
     /// directory (the local-store `NotFound` quirk pinned by
     /// `lance_surface_guards::force_delete_branch_semantics`). Safe to call on a
     /// possibly-orphaned or already-reclaimed fork.
     ///
-    /// A branch that still has referencing descendants (`RefConflict`) is NOT
-    /// tolerated: that is a real ordering error and surfaces as `OmniError::Lance`.
+    /// A branch that still has referencing descendants (`RefConflict`) or a
+    /// live physical path-child is NOT tolerated: those are real ordering
+    /// errors. The graph namespace prevents new path-prefix overlaps; surfacing
+    /// legacy ones keeps cleanup from falsely reporting a reclaim Lance skipped.
     /// Used by the eager best-effort reclaim in `cleanup_deleted_branch_tables`
     /// and the `cleanup` orphan reconciler.
     pub async fn force_delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()> {
@@ -330,11 +318,7 @@ impl TableStore {
             crate::instrumentation::table_wrapper(),
         )
         .await?;
-        match ds.force_delete_branch(branch).await {
-            Ok(()) => Ok(()),
-            Err(lance::Error::RefNotFound { .. }) | Err(lance::Error::NotFound { .. }) => Ok(()),
-            Err(e) => Err(OmniError::Lance(e.to_string())),
-        }
+        crate::branch_control::force_delete_branch_idempotent(&mut ds, branch).await
     }
 
     pub fn ensure_expected_version(
@@ -387,49 +371,28 @@ impl TableStore {
             .map_err(|e| OmniError::Lance(e.to_string()))?;
         self.ensure_expected_version(&source_ds, table_key, source_version)?;
 
-        if let Err(create_err) = source_ds
-            .create_branch(target_branch, source_version, None)
-            .await
+        let created = match crate::branch_control::create_branch_recoverably(
+            &mut source_ds,
+            target_branch,
+            source_version,
+        )
+        .await?
         {
-            // Disambiguate the failure: only a genuinely pre-existing ref is a
-            // reclaim candidate. Mapping EVERY create_branch failure to
-            // `RefAlreadyExists` would route a transient I/O / version / Lance
-            // internal error into the destructive reclaim path. So check whether
-            // the ref actually exists; if not, the failure is real — propagate
-            // it (preserving error fidelity) rather than force-deleting.
-            //
-            // `list_branches` reads `_refs/branches/` from the store, so it sees
-            // a fully-formed manifest-unreferenced fork (our common case — a
-            // create_branch that completed but whose manifest publish did not).
-            // It does NOT see a phase-1-only Lance "zombie" (tree dir written,
-            // no BranchContents) — but neither does `cleanup`'s reconciler, also
-            // list_branches-based. A zombie only forms if create_branch is
-            // interrupted *between its two internal phases* (a far narrower
-            // window than the manifest-publish gap), and it surfaces here as the
-            // propagated create error requiring manual reclaim. We deliberately
-            // do NOT force-delete on a not-found-ref failure: it is
-            // indistinguishable from a transient error on a fresh create, and
-            // force-deleting there is the destructive overreach this guard
-            // removes. The caller holds the per-(table, branch) write queue, so
-            // no in-process writer races this fork; a cross-process create
-            // between our check and now is the documented one-winner-CAS gap and
-            // propagates as a retryable error.
-            let ref_exists = source_ds
-                .list_branches()
-                .await
-                .map(|b| b.contains_key(target_branch))
-                .unwrap_or(false);
-            if ref_exists {
+            crate::branch_control::BranchCreateOutcome::Created(dataset) => dataset,
+            crate::branch_control::BranchCreateOutcome::RefAlreadyExists => {
                 return Ok(ForkOutcome::RefAlreadyExists);
             }
-            return Err(OmniError::Lance(create_err.to_string()));
-        }
+        };
 
         // The ref is now independently durable. Any error from this point is an
         // ambiguous/post-effect outcome to the caller and must retain an armed
         // recovery intent rather than being treated as a safe pre-effect retry.
         crate::failpoints::maybe_fail(crate::failpoints::names::FORK_POST_CREATE_PRE_OPEN)?;
 
+        // Re-open through the shared session for normal cache behavior. The
+        // returned handle above is used only as proof that the matching branch
+        // dataset was openable during classification.
+        drop(created);
         let ds = self
             .open_dataset_head(dataset_uri, Some(target_branch))
             .await?;

crates/omnigraph/tests/branching.rs

@@ -128,6 +128,19 @@ async fn branch_create_open_list_and_lazy_branching_work() {
     let uri = dir.path().to_str().unwrap();
     let mut main = init_and_load(&dir).await;
 
+    main.branch_create("feature").await.unwrap();
+    // Reproduce Lance's phase-1-only crash state: keep the shallow-cloned
+    // `tree/feature` dataset but remove BranchContents, its sole logical
+    // authority. A same-name graph create must reclaim the zombie and retry,
+    // rather than surfacing DatasetAlreadyExists forever.
+    std::fs::remove_file(
+        dir.path()
+            .join("__manifest")
+            .join("_refs")
+            .join("branches")
+            .join("feature.json"),
+    )
+    .unwrap();
     main.branch_create("feature").await.unwrap();
     assert_eq!(main.branch_list().await.unwrap(), vec!["main", "feature"]);
 
@@ -1044,6 +1057,18 @@ async fn branch_created_from_non_main_inherits_branch_state() {
         .branch_create_from(ReadTarget::branch("feature"), "experiment")
         .await
         .unwrap();
+    std::fs::remove_file(
+        dir.path()
+            .join("__manifest")
+            .join("_refs")
+            .join("branches")
+            .join("experiment.json"),
+    )
+    .unwrap();
+    feature
+        .branch_create_from(ReadTarget::branch("feature"), "experiment")
+        .await
+        .expect("non-main create must also reclaim a clone-only target");
 
     assert_eq!(
         feature.branch_list().await.unwrap(),
@@ -1530,6 +1555,17 @@ async fn branch_api_rejects_reserved_main_and_same_source_target_merge() {
     let err = db.branch_delete("main").await.unwrap_err();
     assert!(err.to_string().contains("cannot delete branch 'main'"));
 
+    let err = db.branch_create("bad branch").await.unwrap_err();
+    assert!(err.to_string().contains("invalid") || err.to_string().contains("allowed"));
+    assert!(
+        !dir.path()
+            .join("__manifest")
+            .join("tree")
+            .join("bad branch")
+            .exists(),
+        "branch names must be validated before Lance's shallow-clone phase"
+    );
+
     let err = db.branch_merge("main", "main").await.unwrap_err();
     assert!(err.to_string().contains("distinct source and target"));
 
@@ -1574,6 +1610,91 @@ async fn branch_delete_removes_owned_table_branches_and_allows_recreate() {
     assert_eq!(count_rows_branch(&main, "feature", "node:Person").await, 5);
 }
 
+#[tokio::test]
+async fn branch_namespace_rejects_live_physical_path_prefix_collisions() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = init_and_load(&dir).await;
+
+    db.branch_create("feature").await.unwrap();
+    let err = db.branch_create("feature/child").await.unwrap_err();
+    assert!(
+        err.to_string().contains("physical Lance path")
+            && err.to_string().contains("ancestors or descendants"),
+        "prefix collision must be actionable; got: {err}"
+    );
+    assert!(
+        !dir.path()
+            .join("__manifest")
+            .join("tree")
+            .join("feature")
+            .join("child")
+            .exists(),
+        "prefix admission must reject before Lance creates the target clone"
+    );
+
+    db.branch_delete("feature").await.unwrap();
+    db.branch_create("feature/child").await.unwrap();
+    let err = db.branch_create("feature").await.unwrap_err();
+    assert!(
+        err.to_string().contains("physical Lance path")
+            && err.to_string().contains("feature/child"),
+        "ancestor creation must reject the inverse prefix collision; got: {err}"
+    );
+    assert!(
+        !lance::Dataset::open(&format!("{}/__manifest", dir.path().display()))
+            .await
+            .unwrap()
+            .list_branches()
+            .await
+            .unwrap()
+            .contains_key("feature"),
+        "inverse admission refusal must not create an ancestor ref"
+    );
+}
+
+#[tokio::test]
+async fn branch_delete_refuses_legacy_physical_path_children() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let mut db = init_and_load(&dir).await;
+    db.branch_create("feature/child").await.unwrap();
+
+    // Forge a graph written before the prefix-disjoint namespace invariant.
+    // Lance permits both refs but intentionally cannot reclaim the ancestor's
+    // dataset directory while the child path is live.
+    let mut manifest = lance::Dataset::open(&format!("{uri}/__manifest"))
+        .await
+        .unwrap();
+    let version = manifest.version().version;
+    manifest
+        .create_branch("feature", version, None)
+        .await
+        .unwrap();
+
+    let err = db.branch_delete("feature").await.unwrap_err();
+    assert!(
+        err.to_string().contains("feature/child")
+            && err.to_string().contains("delete the child branch first"),
+        "legacy prefix collisions must be deleted leaf-first; got: {err}"
+    );
+    assert!(
+        manifest
+            .list_branches()
+            .await
+            .unwrap()
+            .contains_key("feature"),
+        "refusal must not remove ancestor authority"
+    );
+
+    db.branch_delete("feature/child").await.unwrap();
+    db.branch_delete("feature").await.unwrap();
+    assert_eq!(
+        db.branch_list().await.unwrap(),
+        vec!["main"],
+        "legacy overlap must converge when deleted leaf-first"
+    );
+}
+
 #[tokio::test]
 async fn branch_delete_rejects_branches_still_referenced_by_descendants() {
     let dir = tempfile::tempdir().unwrap();

crates/omnigraph/tests/failpoints.rs

@@ -56,6 +56,50 @@ fn node_table_uri(root: &str, type_name: &str) -> String {
     format!("{}/nodes/{hash:016x}", root.trim_end_matches('/'))
 }
 
+// Lance can durably complete a native ref mutation while the caller observes
+// an error (for example, a lost object-store acknowledgement). BranchContents
+// is the logical authority in both directions: matching create metadata means
+// success, and an absent ref means delete succeeded even if tree cleanup or the
+// acknowledgement failed. Neither control emits graph lineage or a main-table
+// version.
+#[tokio::test]
+#[serial]
+async fn native_branch_controls_reclassify_lost_acknowledgements() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    let before_version = version_main(&db).await.unwrap();
+    let before_commits = db.list_commits(Some("main")).await.unwrap().len();
+
+    {
+        let _fp = ScopedFailPoint::new(names::BRANCH_CREATE_POST_NATIVE, "return");
+        db.branch_create("feature")
+            .await
+            .expect("matching BranchContents must classify a lost create acknowledgement");
+    }
+    assert!(
+        db.branch_list()
+            .await
+            .unwrap()
+            .iter()
+            .any(|branch| branch == "feature")
+    );
+
+    {
+        let _fp = ScopedFailPoint::new(names::BRANCH_DELETE_POST_NATIVE, "return");
+        db.branch_delete("feature")
+            .await
+            .expect("absent BranchContents must classify a lost delete acknowledgement");
+    }
+    assert_eq!(db.branch_list().await.unwrap(), vec!["main".to_string()]);
+    assert_eq!(version_main(&db).await.unwrap(), before_version);
+    assert_eq!(
+        db.list_commits(Some("main")).await.unwrap().len(),
+        before_commits,
+        "native branch controls must not manufacture graph lineage"
+    );
+}
+
 // Branch delete flips the manifest authority first, then reclaims the per-table
 // forks best-effort. A failure during that reclaim (here, the
 // `branch_delete.before_table_cleanup` failpoint, standing in for a transient
@@ -647,6 +691,31 @@ async fn armed_first_touch_recovery_accepts_missing_target_ref() {
         "precondition: crash happened before the target ref was created"
     );
 
+    // Materialize the narrower Lance crash window underneath the already-
+    // durable writer intent: shallow clone present, BranchContents absent.
+    // Full recovery must not equate "not listed" with "no physical fork".
+    let mut person = lance::Dataset::open(&person_uri).await.unwrap();
+    let person_version = person.version().version;
+    person
+        .create_branch("feature", person_version, None)
+        .await
+        .unwrap();
+    std::fs::remove_file(
+        std::path::Path::new(&person_uri)
+            .join("_refs")
+            .join("branches")
+            .join("feature.json"),
+    )
+    .unwrap();
+    assert!(
+        std::path::Path::new(&person_uri)
+            .join("tree")
+            .join("feature")
+            .exists(),
+        "precondition: clone-only target tree exists"
+    );
+    drop(person);
+
     // Stage A is a synchronous, branch-aware barrier. The Armed sidecar has no
     // table effect for a HEAD-drift check to discover, but it is still ownership:
     // another mutation on the same branch must name it and stop before preparing
@@ -727,6 +796,314 @@ async fn armed_first_touch_recovery_accepts_missing_target_ref() {
             .exists(),
         "full recovery must remove the empty armed intent"
     );
+    assert!(
+        !std::path::Path::new(&person_uri)
+            .join("tree")
+            .join("feature")
+            .exists(),
+        "full recovery must reclaim an unlisted clone-only table fork"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn armed_first_touch_recovery_defers_legacy_path_overlap_until_leaf_delete() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap().to_string();
+    let db = helpers::init_and_load(&dir).await;
+    let main_rows = helpers::count_rows(&db, "node:Person").await;
+
+    // Materialize the leaf first so its graph snapshot owns a matching
+    // per-table Lance branch. New OmniGraph versions reject the inverse live
+    // namespace, but old stores could contain it.
+    db.branch_create("feature/child").await.unwrap();
+    db.mutate(
+        "feature/child",
+        MUTATION_QUERIES,
+        "insert_person",
+        &mixed_params(&[("$name", "Leaf")], &[("$age", 22)]),
+    )
+    .await
+    .unwrap();
+
+    // Forge only the legacy graph-level ancestor admission. It inherits main,
+    // so an interrupted first write will own an unpublished table fork.
+    let mut manifest = lance::Dataset::open(&format!("{uri}/__manifest"))
+        .await
+        .unwrap();
+    let manifest_version = manifest.version().version;
+    manifest
+        .create_branch("feature", manifest_version, None)
+        .await
+        .unwrap();
+    drop(manifest);
+
+    {
+        let _failpoint = ScopedFailPoint::new(names::MUTATION_POST_SIDECAR_PRE_FORK, "return");
+        let error = db
+            .mutate(
+                "feature",
+                MUTATION_QUERIES,
+                "insert_person",
+                &mixed_params(&[("$name", "Ancestor")], &[("$age", 23)]),
+            )
+            .await
+            .expect_err("failure after arming must leave the ancestor recovery intent");
+        assert!(matches!(error, OmniError::RecoveryRequired { .. }));
+    }
+    let operation_id = single_sidecar_operation_id(dir.path());
+
+    // Reproduce Lance's narrower clone-only crash window under the already
+    // durable intent. The live leaf shares the ancestor's physical path, so a
+    // force-delete of the ancestor cannot safely complete yet.
+    let person_uri = node_table_uri(&uri, "Person");
+    let mut person = lance::Dataset::open(&person_uri).await.unwrap();
+    let person_version = person.version().version;
+    person
+        .create_branch("feature", person_version, None)
+        .await
+        .unwrap();
+    std::fs::remove_file(
+        std::path::Path::new(&person_uri)
+            .join("_refs")
+            .join("branches")
+            .join("feature.json"),
+    )
+    .unwrap();
+    assert!(
+        person
+            .list_branches()
+            .await
+            .unwrap()
+            .contains_key("feature/child"),
+        "precondition: live leaf table branch exists"
+    );
+    assert!(
+        !person
+            .list_branches()
+            .await
+            .unwrap()
+            .contains_key("feature"),
+        "precondition: ancestor is clone-only"
+    );
+    drop(person);
+    drop(db);
+
+    let recovered = Omnigraph::open(&uri)
+        .await
+        .expect("legacy path overlap must defer cleanup instead of wedging read-write open");
+    assert!(
+        dir.path()
+            .join("__recovery")
+            .join(format!("{operation_id}.json"))
+            .exists(),
+        "deferred cleanup must retain its ownership sidecar"
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&recovered, "feature", "node:Person").await,
+        main_rows,
+        "the empty ancestor intent must remain unpublished"
+    );
+
+    recovered
+        .branch_delete("feature/child")
+        .await
+        .expect("open handle must permit the documented leaf-first remediation");
+    drop(recovered);
+
+    let recovered = Omnigraph::open(&uri)
+        .await
+        .expect("the next Full sweep must finish ancestor cleanup");
+    assert!(
+        !dir.path()
+            .join("__recovery")
+            .join(format!("{operation_id}.json"))
+            .exists(),
+        "recovery must retire the intent after the path child is gone"
+    );
+    assert!(
+        !std::path::Path::new(&person_uri)
+            .join("tree")
+            .join("feature")
+            .exists(),
+        "the clone-only ancestor tree must be reclaimed after leaf deletion"
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&recovered, "feature", "node:Person").await,
+        main_rows
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn partial_first_touch_recovery_fails_closed_on_legacy_path_overlap() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap().to_string();
+    let db = helpers::init_and_load(&dir).await;
+    let main_person_rows = helpers::count_rows(&db, "node:Person").await;
+    let main_edge_rows = helpers::count_rows(&db, "edge:Knows").await;
+
+    let main_snapshot = db.snapshot_of("main").await.unwrap();
+    let table_pins = ["node:Person", "edge:Knows"]
+        .into_iter()
+        .map(|table_key| {
+            let entry = main_snapshot.entry(table_key).unwrap();
+            (
+                table_key.to_string(),
+                format!("{uri}/{}", entry.table_path),
+                entry.table_version,
+            )
+        })
+        .collect::<Vec<_>>();
+
+    db.branch_create("feature").await.unwrap();
+    let operation_id = {
+        let _failpoint = ScopedFailPoint::new(names::MUTATION_POST_TABLE_COMMIT, "return");
+        let error = db
+            .mutate(
+                "feature",
+                MUTATION_QUERIES,
+                "insert_person_and_friend",
+                &mixed_params(
+                    &[("$name", "Ancestor"), ("$friend", "Alice")],
+                    &[("$age", 23)],
+                ),
+            )
+            .await
+            .expect_err("the first exact table effect must leave a partial v3 intent");
+        match error {
+            OmniError::RecoveryRequired { operation_id, .. } => operation_id,
+            other => panic!("expected RecoveryRequired, got {other}"),
+        }
+    };
+    assert_eq!(single_sidecar_operation_id(dir.path()), operation_id);
+
+    // All deferred first-touch refs are created before the table-commit loop.
+    // Staging order is intentionally non-semantic, so either table may own the
+    // one durable effect while its sibling remains at the exact fork point.
+    let mut head_deltas = Vec::new();
+    let mut heads_before_open = Vec::new();
+    for (table_key, table_uri, expected_version) in &table_pins {
+        let mut root = lance::Dataset::open(table_uri).await.unwrap();
+        let branches = root.list_branches().await.unwrap();
+        assert!(
+            branches.contains_key("feature"),
+            "precondition: {table_key} has its armed first-touch ref"
+        );
+        let head = root
+            .checkout_branch("feature")
+            .await
+            .unwrap()
+            .version()
+            .version;
+        heads_before_open.push((table_key.clone(), head));
+        head_deltas.push(head.checked_sub(*expected_version).unwrap());
+
+        // Forge the legacy namespace only after the partial effect exists. New
+        // OmniGraph versions reject this path-prefix overlap at branch create,
+        // but old stores can contain it. Clone from the exact feature HEAD so
+        // the child itself introduces no extra ancestor movement.
+        root.create_branch("feature/child", ("feature", head), None)
+            .await
+            .unwrap();
+    }
+    head_deltas.sort_unstable();
+    assert_eq!(
+        head_deltas,
+        vec![0, 1],
+        "exactly one table effect must be durable while its sibling remains an untouched fork"
+    );
+
+    let mut manifest = lance::Dataset::open(&format!("{uri}/__manifest"))
+        .await
+        .unwrap();
+    let feature_manifest_version = manifest
+        .checkout_branch("feature")
+        .await
+        .unwrap()
+        .version()
+        .version;
+    manifest
+        .create_branch("feature/child", ("feature", feature_manifest_version), None)
+        .await
+        .unwrap();
+    drop(manifest);
+
+    let open_error = match Omnigraph::open(&uri).await {
+        Ok(_) => panic!(
+            "Full recovery must not return a writable handle while owned effects remain unrolled"
+        ),
+        Err(error) => error,
+    };
+    assert!(
+        open_error.to_string().contains("owns physical effects")
+            && open_error.to_string().contains("feature/child"),
+        "failure must explain the safe leaf-first remediation boundary: {open_error}"
+    );
+    assert!(
+        dir.path()
+            .join("__recovery")
+            .join(format!("{operation_id}.json"))
+            .exists(),
+        "failed Full recovery must retain exact ownership"
+    );
+    let manifest_after_failed_open = lance::Dataset::open(&format!("{uri}/__manifest"))
+        .await
+        .unwrap()
+        .checkout_branch("feature")
+        .await
+        .unwrap()
+        .version()
+        .version;
+    assert_eq!(
+        manifest_after_failed_open, feature_manifest_version,
+        "failed rollback preflight must not publish the manifest"
+    );
+    for ((table_key, table_uri, _), (_, before)) in table_pins.iter().zip(heads_before_open.iter())
+    {
+        let root = lance::Dataset::open(table_uri).await.unwrap();
+        let after = root
+            .checkout_branch("feature")
+            .await
+            .unwrap()
+            .version()
+            .version;
+        assert_eq!(
+            after, *before,
+            "failed preflight must not restore or publish {table_key}"
+        );
+    }
+
+    // A handle that predates the interrupted attempt can remove the graph leaf
+    // under the normal branch-control gates. This fixture forged its table refs
+    // outside the manifest, so finish the documented offline leaf cleanup at
+    // Lance level. Once the physical overlap is gone, the next quiesced Full
+    // sweep can compensate the owned effect atomically.
+    db.branch_delete("feature/child").await.unwrap();
+    for (_, table_uri, _) in &table_pins {
+        lance::Dataset::open(table_uri)
+            .await
+            .unwrap()
+            .force_delete_branch("feature/child")
+            .await
+            .unwrap();
+    }
+    drop(db);
+
+    let recovered = Omnigraph::open(&uri)
+        .await
+        .expect("rollback must converge after leaf-first remediation");
+    assert!(helpers::recovery::sidecar_operation_ids(dir.path()).is_empty());
+    assert_eq!(
+        helpers::count_rows_branch(&recovered, "feature", "node:Person").await,
+        main_person_rows
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&recovered, "feature", "edge:Knows").await,
+        main_edge_rows
+    );
 }
 
 #[tokio::test]

crates/omnigraph/tests/lance_surface_guards.rs

@@ -338,7 +338,7 @@ async fn _compile_uncommitted_merge_insert_field_shape() -> lance::Result<()> {
 // The branch-delete reconciler (`db/omnigraph/optimize.rs::reconcile_orphaned_branches`)
 // and the eager best-effort reclaim in `cleanup_deleted_branch_tables` call
 // `force_delete_branch` to drop orphaned branch refs. The single-authority
-// design relies on three facts pinned here:
+// design relies on five facts pinned here:
 //   1. plain `delete_branch` errors on a missing ref (so the design uses the
 //      force variant instead);
 //   2. `force_delete_branch` removes an existing (forked) branch — the orphan
@@ -347,7 +347,14 @@ async fn _compile_uncommitted_merge_insert_field_shape() -> lance::Result<()> {
 //      errors on the local store, because `remove_dir_all`'s NotFound is not
 //      caught for Lance's native error variant. `TableStore::force_delete_branch`
 //      wraps this to be fully idempotent. Pin the raw quirk so a future Lance
-//      fix (which would let us simplify the wrapper) is noticed.
+//      fix (which would let us simplify the wrapper) is noticed;
+//   4. a clone-only zombie (branch dataset present, BranchContents absent)
+//      blocks raw create and is reclaimed by `force_delete_branch`. Lance's
+//      create is explicitly two-phase, so this is the crash state OmniGraph's
+//      native branch-control wrapper must heal before retrying;
+//   5. a live slash-name path-child makes force delete remove an ancestor's
+//      BranchContents but intentionally retain its dataset files. OmniGraph's
+//      prefix-disjoint live-name invariant prevents this false-success shape.
 
 #[tokio::test]
 async fn force_delete_branch_semantics() {
@@ -379,6 +386,56 @@ async fn force_delete_branch_semantics() {
         "force_delete_branch on a fully-absent branch no longer errors — \
          TableStore::force_delete_branch's NotFound tolerance can be simplified."
     );
+
+    // (4) Exact phase-1-only create state: create the shallow-cloned branch
+    // dataset, then remove only its authoritative BranchContents ref. This is
+    // the same fixture Lance's own dataset-versioning test uses for a zombie.
+    ds.create_branch("zombie", base, None).await.unwrap();
+    std::fs::remove_file(
+        std::path::Path::new(uri)
+            .join("_refs")
+            .join("branches")
+            .join("zombie.json"),
+    )
+    .unwrap();
+    assert!(
+        !ds.list_branches().await.unwrap().contains_key("zombie"),
+        "BranchContents is the authority; the clone-only tree must not list as a branch"
+    );
+    assert!(
+        ds.create_branch("zombie", base, None).await.is_err(),
+        "the clone-only tree should block an unclassified raw create"
+    );
+    ds.force_delete_branch("zombie").await.unwrap();
+    assert!(
+        !std::path::Path::new(uri)
+            .join("tree")
+            .join("zombie")
+            .exists(),
+        "force_delete_branch must reclaim the clone-only tree"
+    );
+
+    // (5) Slash-separated names overlap physically. A path-child created from
+    // main is not a lineage descendant of its lexical ancestor, so raw force
+    // delete removes the ancestor ref but deliberately leaves its dataset
+    // files to avoid recursively deleting the child.
+    ds.create_branch("ancestor/child", base, None)
+        .await
+        .unwrap();
+    ds.create_branch("ancestor", base, None).await.unwrap();
+    ds.force_delete_branch("ancestor").await.unwrap();
+    assert!(
+        !ds.list_branches().await.unwrap().contains_key("ancestor"),
+        "raw force delete still removes authoritative ancestor metadata"
+    );
+    assert!(
+        std::path::Path::new(uri)
+            .join("tree")
+            .join("ancestor")
+            .join("_versions")
+            .exists(),
+        "Lance must retain ancestor dataset files while a physical path-child is live"
+    );
 }
 
 // --- Guard 10: blob-column compaction works in this Lance ------------------

crates/omnigraph/tests/maintenance.rs

@@ -17,8 +17,8 @@ use omnigraph::loader::{LoadMode, load_jsonl};
 use omnigraph::table_store::{IndexCoverage, TableStore};
 
 use helpers::{
-    MUTATION_QUERIES, TEST_DATA, TEST_SCHEMA, count_rows, init_and_load, mixed_params, mutate_main,
-    snapshot_main,
+    MUTATION_QUERIES, TEST_DATA, TEST_SCHEMA, count_rows, count_rows_branch, init_and_load,
+    mixed_params, mutate_main, snapshot_main,
 };
 
 /// Filesystem URI of a node sub-table, mirroring the engine's layout
@@ -925,7 +925,9 @@ async fn cleanup_without_any_policy_option_errors() {
 #[tokio::test]
 async fn cleanup_keep_one_preserves_head_and_table_remains_readable() {
     let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap().to_string();
     let mut db = init_and_load(&dir).await;
+    add_person_fragments(&mut db).await;
 
     let people_before = count_rows(&db, "node:Person").await;
     assert!(
@@ -933,9 +935,21 @@ async fn cleanup_keep_one_preserves_head_and_table_remains_readable() {
         "fixture should seed Person rows for this test to be meaningful"
     );
 
-    // Most aggressive version-based cleanup short of forcing keep=0. Lance's
-    // contract is that head is always preserved regardless, so the table
-    // must remain openable and rows must still be visible.
+    let person_uri = node_table_uri(&uri, "Person");
+    assert!(
+        Dataset::open(&person_uri)
+            .await
+            .unwrap()
+            .versions()
+            .await
+            .unwrap()
+            .len()
+            > 1,
+        "precondition: Person must have history to collect"
+    );
+
+    // Most aggressive version-based cleanup short of forcing keep=0. `keep`
+    // is exact over the available version list, not HEAD arithmetic.
     let _stats = db
         .cleanup(CleanupPolicyOptions {
             keep_versions: Some(1),
@@ -945,6 +959,52 @@ async fn cleanup_keep_one_preserves_head_and_table_remains_readable() {
         .unwrap();
 
     assert_eq!(count_rows(&db, "node:Person").await, people_before);
+    assert_eq!(
+        Dataset::open(&person_uri)
+            .await
+            .unwrap()
+            .versions()
+            .await
+            .unwrap()
+            .len(),
+        1,
+        "keep=1 must retain exactly the latest available version"
+    );
+}
+
+#[tokio::test]
+async fn cleanup_keep_exceeding_history_preserves_every_available_version() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap().to_string();
+    let mut db = init_and_load(&dir).await;
+    let person_uri = node_table_uri(&uri, "Person");
+    let before = Dataset::open(&person_uri)
+        .await
+        .unwrap()
+        .versions()
+        .await
+        .unwrap()
+        .len();
+    assert!(before > 1, "fixture must contain version history");
+
+    db.cleanup(CleanupPolicyOptions {
+        keep_versions: Some(10),
+        older_than: None,
+    })
+    .await
+    .unwrap();
+
+    let after = Dataset::open(&person_uri)
+        .await
+        .unwrap()
+        .versions()
+        .await
+        .unwrap()
+        .len();
+    assert_eq!(
+        after, before,
+        "keep greater than available history must not become an unbounded cleanup"
+    );
 }
 
 #[tokio::test]
@@ -970,6 +1030,224 @@ async fn cleanup_older_than_zero_preserves_head() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn cleanup_preserves_main_version_pinned_by_live_lazy_branch() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = init_and_load(&dir).await;
+
+    db.branch_create("feature").await.unwrap();
+    let feature_before = db.snapshot_of(ReadTarget::branch("feature")).await.unwrap();
+    let feature_person = feature_before.entry("node:Person").unwrap();
+    assert_eq!(
+        feature_person.table_branch, None,
+        "precondition: Person must still be inherited lazily from main"
+    );
+    let pinned_main_version = feature_person.table_version;
+    let feature_people_before = count_rows_branch(&db, "feature", "node:Person").await;
+
+    // Move main far enough that keep=1 would collect the version inherited by
+    // the lazy branch unless cleanup accounts for graph-level branch pins.
+    add_person_fragments(&mut db).await;
+    let main_person_version = db
+        .snapshot_of(ReadTarget::branch("main"))
+        .await
+        .unwrap()
+        .entry("node:Person")
+        .unwrap()
+        .table_version;
+    assert!(
+        pinned_main_version < main_person_version.saturating_sub(1),
+        "precondition: lazy-branch pin must fall outside keep=1 retention"
+    );
+
+    db.cleanup(CleanupPolicyOptions {
+        keep_versions: Some(1),
+        older_than: None,
+    })
+    .await
+    .unwrap();
+
+    assert_eq!(
+        count_rows_branch(&db, "feature", "node:Person").await,
+        feature_people_before,
+        "cleanup must preserve the exact main-table version inherited by a live lazy branch"
+    );
+
+    // The same floor must constrain a time-only policy. Lance combines the
+    // timestamp and version predicates with AND, so injecting the branch pin
+    // as `before_version` keeps the inherited version even when every old
+    // manifest satisfies the timestamp cutoff.
+    db.cleanup(CleanupPolicyOptions {
+        keep_versions: None,
+        older_than: Some(Duration::from_secs(0)),
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        count_rows_branch(&db, "feature", "node:Person").await,
+        feature_people_before,
+        "time-only cleanup must honor the same live lazy-branch floor"
+    );
+}
+
+#[tokio::test]
+async fn cleanup_uses_oldest_pin_across_multiple_live_lazy_branches() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = init_and_load(&dir).await;
+
+    db.branch_create("a-old").await.unwrap();
+    let old_rows = count_rows_branch(&db, "a-old", "node:Person").await;
+    add_person_fragments(&mut db).await;
+    db.branch_create("z-new").await.unwrap();
+    let new_rows = count_rows_branch(&db, "z-new", "node:Person").await;
+    mutate_main(
+        &mut db,
+        MUTATION_QUERIES,
+        "insert_person",
+        &mixed_params(&[("$name", "Ivan")], &[("$age", 44)]),
+    )
+    .await
+    .unwrap();
+
+    db.cleanup(CleanupPolicyOptions {
+        keep_versions: Some(1),
+        older_than: None,
+    })
+    .await
+    .unwrap();
+
+    assert_eq!(
+        count_rows_branch(&db, "a-old", "node:Person").await,
+        old_rows,
+        "the oldest live pin must win over later branch pins"
+    );
+    assert_eq!(
+        count_rows_branch(&db, "z-new", "node:Person").await,
+        new_rows,
+        "newer lazy pins must remain readable too"
+    );
+}
+
+#[tokio::test]
+async fn cleanup_fails_closed_when_live_lazy_branch_pin_is_unopenable() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap().to_string();
+    let mut db = init_and_load(&dir).await;
+
+    db.branch_create("feature").await.unwrap();
+    let pinned_main_version = db
+        .snapshot_of(ReadTarget::branch("feature"))
+        .await
+        .unwrap()
+        .entry("node:Person")
+        .unwrap()
+        .table_version;
+    add_person_fragments(&mut db).await;
+
+    // Simulate damage created by an older cleanup implementation: raw Lance
+    // sees no native Person branch for the lazy graph branch and removes its
+    // inherited main manifest.
+    let person_uri = node_table_uri(&uri, "Person");
+    let ds = Dataset::open(&person_uri).await.unwrap();
+    let head = ds.version().version;
+    assert!(pinned_main_version < head, "precondition: main advanced");
+    let removed = lance::dataset::cleanup::cleanup_old_versions(
+        &ds,
+        lance::dataset::cleanup::CleanupPolicy {
+            before_timestamp: None,
+            before_version: Some(head),
+            delete_unverified: false,
+            error_if_tagged_old_versions: false,
+            clean_referenced_branches: false,
+            delete_rate_limit: None,
+        },
+    )
+    .await
+    .unwrap();
+    assert!(
+        removed.old_versions > 0,
+        "precondition: raw Lance cleanup removed old main versions"
+    );
+    let company_uri = node_table_uri(&uri, "Company");
+    let company_versions_before = Dataset::open(&company_uri)
+        .await
+        .unwrap()
+        .versions()
+        .await
+        .unwrap()
+        .len();
+
+    let err = db
+        .cleanup(CleanupPolicyOptions {
+            keep_versions: Some(1),
+            older_than: None,
+        })
+        .await
+        .expect_err("cleanup must fail closed when a live lazy pin cannot be opened");
+    let message = err.to_string();
+    assert!(
+        message.contains("could not classify live branch 'feature'")
+            && message.contains("node:Person"),
+        "error must identify the unclassifiable live reference; got: {message}"
+    );
+    assert_eq!(
+        Dataset::open(&company_uri)
+            .await
+            .unwrap()
+            .versions()
+            .await
+            .unwrap()
+            .len(),
+        company_versions_before,
+        "the graph-wide preflight must fail before any unrelated table GC"
+    );
+}
+
+#[tokio::test]
+async fn cleanup_refuses_uncovered_main_head_drift_before_any_version_gc() {
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().to_str().unwrap().to_string();
+    let mut db = init_and_load(&dir).await;
+    let (manifest_version, head_version, _) = forge_person_compaction_drift(&mut db, &root).await;
+    let company_uri = node_table_uri(&root, "Company");
+    let company_versions_before = Dataset::open(&company_uri)
+        .await
+        .unwrap()
+        .versions()
+        .await
+        .unwrap()
+        .len();
+
+    let err = db
+        .cleanup(CleanupPolicyOptions {
+            keep_versions: Some(1),
+            older_than: None,
+        })
+        .await
+        .expect_err("cleanup must not garbage-collect around uncovered HEAD drift");
+    let message = err.to_string();
+    assert!(
+        message.contains("uncovered HEAD drift")
+            && message.contains("node:Person")
+            && message.contains("repair"),
+        "drift refusal must be actionable; got: {message}"
+    );
+    let (manifest_after, head_after, _) = person_manifest_and_head(&db, &root).await;
+    assert_eq!(manifest_after, manifest_version);
+    assert_eq!(head_after, head_version);
+    assert_eq!(
+        Dataset::open(&company_uri)
+            .await
+            .unwrap()
+            .versions()
+            .await
+            .unwrap()
+            .len(),
+        company_versions_before,
+        "drift preflight must abort before unrelated table history is collected"
+    );
+}
+
 #[tokio::test]
 async fn cleanup_then_optimize_preserves_rows_and_table_remains_writable() {
     // Cleanup destroys version history; the concern is that subsequent