diff --git a/crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt b/crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt
new file mode 100644
index 000000000..f2c08ae5c
--- /dev/null
+++ b/crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt
@@ -0,0 +1,74 @@
+# issue: none
+# red_on: 2026-09-07, main @ 5f94a741: `branch merge main into feature` failed with `storage: Concurrent modification: table version 4 already exists for identity … (edge:Knows) with different state`
+# notes: sibling of merge_adopt_lower_source_version_keeps_edges: two edge writes
+# notes: on `feature` instead of three make the two edge tables reach the same
+# notes: version number on different native branches. Adopting `main` into
+# notes: `feature` must fast-forward and show all four edges.
+
+--- schema
+node Person {
+    name: String @key
+}
+
+edge Knows: Person -> Person
+
+--- seed
+{"type":"Person","data":{"name":"a"}}
+{"type":"Person","data":{"name":"b"}}
+{"type":"Person","data":{"name":"c"}}
+{"edge":"Knows","from":"a","to":"b"}
+
+--- mutate
+branch create feature
+
+--- expect ok
+
+--- mutate branch: feature
+query add_b_c() {
+    insert Knows { from: "b", to: "c" }
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate branch: feature
+query add_c_a() {
+    insert Knows { from: "c", to: "a" }
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate
+branch merge feature
+
+--- expect outcome: fast_forward
+
+--- mutate
+query add_a_c_on_main() {
+    insert Knows { from: "a", to: "c" }
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate
+branch merge main into feature
+
+--- expect outcome: fast_forward
+
+--- query branch: feature
+query edges_on_feature() {
+    match {
+        $x: Person
+        $x knows $y
+    }
+    return { $x.name, $y.name }
+}
+
+--- expect unordered
+{"x.name": "a", "y.name": "b"}
+{"x.name": "b", "y.name": "c"}
+{"x.name": "c", "y.name": "a"}
+{"x.name": "a", "y.name": "c"}
+
+--- expect shape
+x.name: String
+y.name: String
diff --git a/crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt b/crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt
new file mode 100644
index 000000000..e54f87756
--- /dev/null
+++ b/crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt
@@ -0,0 +1,92 @@
+# issue: none
+# notes: from PR 630, the lazy-target arm. `main` writes its node table four
+# notes: times; `feature` writes once, takes `main` by a three-way merge, then
+# notes: sets a's age; `child` forks from `main` and never writes, so its node
+# notes: table is `main`'s at a version above `feature`'s. Merging `feature`
+# notes: into `child` must show feature's value on child.
+
+--- schema
+node Person {
+    name: String @key
+    age: I32
+}
+
+--- seed
+{"type":"Person","data":{"name":"a","age":1}}
+{"type":"Person","data":{"name":"b","age":1}}
+
+--- mutate
+branch create feature
+
+--- expect ok
+
+--- mutate
+query a_2_on_main() {
+    update Person set { age: 2 } where name = "a"
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate
+query a_3_on_main() {
+    update Person set { age: 3 } where name = "a"
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate
+query a_4_on_main() {
+    update Person set { age: 4 } where name = "a"
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate
+query a_5_on_main() {
+    update Person set { age: 5 } where name = "a"
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate branch: feature
+query b_9_on_feature() {
+    update Person set { age: 9 } where name = "b"
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate
+branch merge main into feature
+
+--- expect outcome: merged
+
+--- mutate branch: feature
+query a_50_on_feature() {
+    update Person set { age: 50 } where name = "a"
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate
+branch create child
+
+--- expect ok
+
+--- mutate
+branch merge feature into child
+
+--- expect outcome: fast_forward
+
+--- query branch: child
+query ages_on_child() {
+    match { $p: Person }
+    return { $p.name, $p.age }
+}
+
+--- expect unordered
+{"p.name": "a", "p.age": 50}
+{"p.name": "b", "p.age": 9}
+
+--- expect shape
+p.name: String
+p.age: I32
diff --git a/crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt b/crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt
new file mode 100644
index 000000000..1caff7a50
--- /dev/null
+++ b/crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt
@@ -0,0 +1,113 @@
+# issue: none
+# red_on: 2026-09-07, main @ 5f94a741: after `branch merge main into feature` reported fast_forward, feature's edges lacked a->c; merging feature back then dropped a->c from main
+# notes: from PR 630. `feature` writes its edge table three times, so its table
+# notes: version passes `main`'s. Merge `feature` into `main`, add an edge on
+# notes: `main`, merge `main` into `feature`: the adopted edge must be visible
+# notes: on `feature`. Then edit only a node on `feature` and merge back: `main`
+# notes: must keep the edge.
+
+--- schema
+node Person {
+    name: String @key
+}
+
+edge Knows: Person -> Person
+
+--- seed
+{"type":"Person","data":{"name":"a"}}
+{"type":"Person","data":{"name":"b"}}
+{"type":"Person","data":{"name":"c"}}
+{"edge":"Knows","from":"a","to":"b"}
+
+--- mutate
+branch create feature
+
+--- expect ok
+
+--- mutate branch: feature
+query add_b_c() {
+    insert Knows { from: "b", to: "c" }
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate branch: feature
+query add_c_a() {
+    insert Knows { from: "c", to: "a" }
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate branch: feature
+query drop_b_c() {
+    delete Knows where from = "b"
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate
+branch merge feature
+
+--- expect outcome: fast_forward
+
+--- mutate
+query add_a_c_on_main() {
+    insert Knows { from: "a", to: "c" }
+}
+
+--- expect affected: nodes=0 edges=1
+
+--- mutate
+branch merge main into feature
+
+--- expect outcome: fast_forward
+
+--- query branch: feature
+query edges_on_feature() {
+    match {
+        $x: Person
+        $x knows $y
+    }
+    return { $x.name, $y.name }
+}
+
+--- expect unordered
+{"x.name": "a", "y.name": "b"}
+{"x.name": "a", "y.name": "c"}
+{"x.name": "c", "y.name": "a"}
+
+--- expect shape
+x.name: String
+y.name: String
+
+--- mutate branch: feature
+query add_d_on_feature() {
+    insert Person { name: "d" }
+}
+
+--- expect affected: nodes=1 edges=0
+
+--- mutate
+branch merge feature
+
+--- expect outcome: fast_forward
+
+--- restart
+
+--- query
+query edges_on_main() {
+    match {
+        $x: Person
+        $x knows $y
+    }
+    return { $x.name, $y.name }
+}
+
+--- expect unordered
+{"x.name": "a", "y.name": "b"}
+{"x.name": "a", "y.name": "c"}
+{"x.name": "c", "y.name": "a"}
+
+--- expect shape
+x.name: String
+y.name: String
diff --git a/crates/omnigraph/src/db/manifest/state.rs b/crates/omnigraph/src/db/manifest/state.rs
index 18b5c3600..25449c191 100644
--- a/crates/omnigraph/src/db/manifest/state.rs
+++ b/crates/omnigraph/src/db/manifest/state.rs
@@ -26,6 +26,19 @@ pub struct DatasetEntry {
     pub(crate) version_metadata: TableVersionMetadata,
 }
 
+impl DatasetEntry {
+    /// Field-for-field equal registration, the Lance manifest metadata included.
+    pub fn same_registration(&self, other: &DatasetEntry) -> bool {
+        self.identity == other.identity
+            && self.type_key == other.type_key
+            && self.dataset_path == other.dataset_path
+            && self.published_dataset_version == other.published_dataset_version
+            && self.native_dataset_branch == other.native_dataset_branch
+            && self.entity_count == other.entity_count
+            && self.version_metadata == other.version_metadata
+    }
+}
+
 #[derive(Debug, Clone)]
 pub(super) struct ManifestState {
     pub(super) version: u64,
diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs
index 45b1fca2b..3cfe7e0ed 100644
--- a/crates/omnigraph/src/exec/merge.rs
+++ b/crates/omnigraph/src/exec/merge.rs
@@ -3499,11 +3499,25 @@ fn row_id_at(batch: &RecordBatch, row: usize) -> Result<String> {
     Ok(ids.value(row).to_string())
 }
 
+/// A source version at or below the target's must be written onto the target's
+/// lineage (`docs/dev/merge.md` §Table classification); rows still decide the delta.
+fn adopt_requires_target_lineage(
+    source_entry: &crate::db::DatasetEntry,
+    target_entry: Option<&crate::db::DatasetEntry>,
+) -> bool {
+    target_entry.is_some_and(|target| {
+        source_entry.published_dataset_version <= target.published_dataset_version
+    })
+}
+
 fn adopt_advances_head(
     target_active: Option<&str>,
     source_entry: &crate::db::DatasetEntry,
     target_entry: Option<&crate::db::DatasetEntry>,
 ) -> bool {
+    if adopt_requires_target_lineage(source_entry, target_entry) {
+        return true;
+    }
     match (target_active, source_entry.native_dataset_branch.as_deref()) {
         // Source on a branch, target on main — delta applied onto main's lineage.
         (None, Some(_)) => true,
@@ -3512,23 +3526,13 @@ fn adopt_advances_head(
             target_entry.and_then(|entry| entry.native_dataset_branch.as_deref())
                 == Some(target_branch)
         }
-        // Source on main (pointer switch) or target doesn't own (fork): no advance.
         _ => false,
     }
 }
 
-/// Classify a table whose target state equals base (the adopt / fast-forward
-/// case). A proven insertion-only descendant becomes
-/// [`CandidateTableState::AdoptPureInserts`]; every other non-empty delta that
-/// advances target HEAD becomes [`CandidateTableState::AdoptWithDelta`] with
-/// its write payload pre-computed for recovery planning. Pointer switches and
-/// forks become [`CandidateTableState::AdoptSourceState`] and do not advance
-/// data HEAD.
-///
-/// The HEAD-advancing subcases mirror [`publish_adopted_source_state`]: source
-/// on a branch with the target either on main or owning the table. Computing the
-/// delta here (rather than inside the publish) is what closes the recovery gap —
-/// the classifier knows whether the publish will move Lance HEAD.
+/// Classify a table whose target equals base: a proven insertion-only descendant
+/// is `AdoptPureInserts`, any other HEAD-advancing delta (a source at or below the
+/// target's version included) is `AdoptWithDelta`, a newer pointer or fork is `AdoptSourceState`.
 async fn classify_adopt(
     target_db: &Omnigraph,
     catalog: &Catalog,
@@ -3622,12 +3626,9 @@ async fn classify_general_adopt(
     }
 }
 
-/// What publishing a table's adopted source state does to `__manifest`.
-///
-/// An empty delta does not imply an empty publish: source and target can hold
-/// the same content at different Lance versions. Planning purely lets
-/// classification drop a table whose registration is already stored, which the
-/// registry guard would otherwise reject (#473).
+/// What publishing a table's adopted source state does to `__manifest`; planning
+/// lets classification drop a table whose registration is already stored (#473)
+/// or whose source version is at or below the target's.
 #[must_use = "the adopt plan decides whether this table is a merge candidate"]
 enum AdoptPublish {
     /// The planned registration is field-for-field the stored entry.
@@ -3642,11 +3643,8 @@ enum AdoptPublish {
     },
 }
 
-/// Plan what adopting the source's table state publishes, without an effect.
-///
-/// Reaching a branch-bearing arm means the delta was empty: the HEAD-advancing
-/// case is classified [`CandidateTableState::AdoptWithDelta`] and published by
-/// [`publish_adopted_delta`].
+/// Plan what adopting the source's table state publishes, without an effect;
+/// only a source newer than the target reaches a branch-bearing arm.
 fn plan_adopted_source_state(
     target_active: Option<&str>,
     source_entry: &crate::db::DatasetEntry,
@@ -3740,10 +3738,11 @@ fn keep_publishing_candidate(
         CandidateTableState::AdoptSourceState {
             validation_delta: None
         }
-    ) && matches!(
-        plan_adopted_source_state(target_active, source_entry, target_entry, table_key),
-        AdoptPublish::Nothing
-    );
+    ) && (adopt_requires_target_lineage(source_entry, target_entry)
+        || matches!(
+            plan_adopted_source_state(target_active, source_entry, target_entry, table_key),
+            AdoptPublish::Nothing
+        ));
     if publishes_nothing {
         return None;
     }
@@ -3784,6 +3783,21 @@ mod adopt_plan_tests {
         }
     }
 
+    /// `adopt_requires_target_lineage` mirrors the projection fold's `>=`
+    /// (`db/manifest/state.rs`): a source at or below the target's version
+    /// advances HEAD; a newer source on main does not; no target entry, no rule.
+    #[test]
+    fn source_at_or_below_target_version_advances_head() {
+        let target = entry(4, Some("feature"), 3, "manifest-v4");
+        let below = entry(3, None, 3, "manifest-v3");
+        let equal = entry(4, None, 3, "manifest-v4-main");
+        let above = entry(5, None, 3, "manifest-v5");
+        assert!(adopt_advances_head(Some("feature"), &below, Some(&target)));
+        assert!(adopt_advances_head(Some("feature"), &equal, Some(&target)));
+        assert!(!adopt_advances_head(Some("feature"), &above, Some(&target)));
+        assert!(!adopt_advances_head(Some("feature"), &below, None));
+    }
+
     /// The #473 shape: the source advanced two Lance versions on a branch and
     /// came back to the target's content. The plan must be `Nothing`.
     ///
@@ -3914,6 +3928,16 @@ async fn publish_adopted_source_state(
     )?;
 
     match plan_adopted_source_state(target_active, source_entry, target_entry, table_key) {
+        AdoptPublish::Pointer(update)
+            if target_entry.is_some_and(|current| {
+                update.published_dataset_version <= current.published_dataset_version
+            }) =>
+        {
+            Err(OmniError::manifest_internal(format!(
+                "branch merge table '{table_key}' plans a pointer at version {} at or below the target's registration; classification must route that onto the target lineage",
+                update.published_dataset_version
+            )))
+        }
         AdoptPublish::Pointer(update) => Ok(update),
         AdoptPublish::Fork {
             source_branch,
@@ -5058,6 +5082,7 @@ impl Omnigraph {
             }),
         };
         let mut candidates: HashMap<String, CandidateTableState> = HashMap::new();
+        let mut fenced_but_unpublished_table_keys: Vec<String> = Vec::new();
         let empty_external_preflight = crate::table_store::ExternalBlobPreflight::default();
         let mut blob_table_keys = HashSet::new();
         let mut blob_selection = crate::table_store::PersistedBlobSelection::default();
@@ -5088,7 +5113,7 @@ impl Omnigraph {
             let has_blob = schema_has_blob(&schema_for_table_key(catalog, table_key)?)?;
             if !has_blob {
                 if same_manifest_state(base_entry, target_entry) {
-                    if let Some(candidate) = classify_adopt(
+                    match classify_adopt(
                         self,
                         catalog,
                         base_snapshot,
@@ -5100,7 +5125,10 @@ impl Omnigraph {
                     )
                     .await?
                     {
-                        candidates.insert(table_key.clone(), candidate);
+                        Some(candidate) => {
+                            candidates.insert(table_key.clone(), candidate);
+                        }
+                        None => fenced_but_unpublished_table_keys.push(table_key.clone()),
                     }
                 } else {
                     let table_walk_timing = crate::instrumentation::start_merge_timing(
@@ -5301,8 +5329,11 @@ impl Omnigraph {
                     )
                     .await?
                 };
-                if let Some(candidate) = candidate {
-                    candidates.insert(table_key.clone(), candidate);
+                match candidate {
+                    Some(candidate) => {
+                        candidates.insert(table_key.clone(), candidate);
+                    }
+                    None => fenced_but_unpublished_table_keys.push(table_key.clone()),
                 }
                 continue;
             }
@@ -5475,6 +5506,7 @@ impl Omnigraph {
 
         let expected_versions = candidates
             .keys()
+            .chain(fenced_but_unpublished_table_keys.iter())
             .filter_map(|table_key| {
                 let identity = target_snapshot
                     .dataset(table_key)
diff --git a/crates/omnigraph/tests/branching.rs b/crates/omnigraph/tests/branching.rs
index ac2fcccfd..86aeffcf6 100644
--- a/crates/omnigraph/tests/branching.rs
+++ b/crates/omnigraph/tests/branching.rs
@@ -1856,6 +1856,227 @@ async fn branch_merge_applies_node_insert_to_main() {
     assert_eq!(qr.num_rows(), 1);
 }
 
+/// Rust because the pins are native table versions, the target ref's physical
+/// HEAD, and the entry retained on an empty delta; the row-visible half is
+/// `merge_adopt_*.gqt`.
+#[tokio::test]
+async fn branch_merge_preserves_state_when_native_versions_differ() {
+    for branch_updates in [8, 2] {
+        for lazy_target in [false, true] {
+            let dir = tempfile::tempdir().unwrap();
+            let uri = dir.path().to_str().unwrap();
+            let main = init_and_load(&dir).await;
+            main.branch_create("feature").await.unwrap();
+            let history_branch = if lazy_target { "main" } else { "feature" };
+            let history_updates = branch_updates + i64::from(lazy_target);
+            for age in 40..40 + history_updates {
+                main.mutate(
+                    history_branch,
+                    MUTATION_QUERIES,
+                    "set_age",
+                    &mixed_params(&[("$name", "Alice")], &[("$age", age)]),
+                )
+                .await
+                .unwrap();
+            }
+            let (source, target) = if lazy_target {
+                main.mutate(
+                    "feature",
+                    MUTATION_QUERIES,
+                    "set_age",
+                    &mixed_params(&[("$name", "Bob")], &[("$age", 26)]),
+                )
+                .await
+                .unwrap();
+                assert_eq!(
+                    main.branch_merge("main", "feature").await.unwrap(),
+                    MergeOutcome::Merged
+                );
+                main.mutate(
+                    "feature",
+                    MUTATION_QUERIES,
+                    "set_age",
+                    &mixed_params(&[("$name", "Alice")], &[("$age", 50)]),
+                )
+                .await
+                .unwrap();
+                main.branch_create_from(ReadTarget::branch("main"), "child")
+                    .await
+                    .unwrap();
+                ("feature", "child")
+            } else {
+                assert_eq!(
+                    main.branch_merge("feature", "main").await.unwrap(),
+                    MergeOutcome::FastForward
+                );
+                main.mutate(
+                    "main",
+                    MUTATION_QUERIES,
+                    "set_age",
+                    &mixed_params(&[("$name", "Alice")], &[("$age", 50)]),
+                )
+                .await
+                .unwrap();
+                ("main", "feature")
+            };
+            let target_native = graph_native_ref(uri, target).await;
+            let source_entry = snapshot_branch(&main, source)
+                .await
+                .unwrap()
+                .dataset("node:Person")
+                .unwrap()
+                .clone();
+            let target_entry = snapshot_branch(&main, target)
+                .await
+                .unwrap()
+                .dataset("node:Person")
+                .unwrap()
+                .clone();
+            if branch_updates == 2 {
+                assert_eq!(
+                    source_entry.published_dataset_version, target_entry.published_dataset_version,
+                    "fixture must exercise equal numeric versions on different refs"
+                );
+            } else {
+                assert!(
+                    source_entry.published_dataset_version < target_entry.published_dataset_version,
+                    "fixture must exercise a lower source version"
+                );
+            }
+            assert_ne!(
+                source_entry.native_dataset_branch,
+                target_entry.native_dataset_branch
+            );
+            assert_eq!(
+                target_entry.native_dataset_branch.as_deref() == Some(target_native.as_str()),
+                !lazy_target
+            );
+            assert_eq!(
+                main.branch_merge(source, target).await.unwrap(),
+                MergeOutcome::FastForward
+            );
+            let merged_entry = snapshot_branch(&main, target)
+                .await
+                .unwrap()
+                .dataset("node:Person")
+                .unwrap()
+                .clone();
+            assert!(
+                merged_entry.published_dataset_version > target_entry.published_dataset_version,
+                "{target}, {branch_updates} updates: changed rows must advance target's own version"
+            );
+            assert_eq!(
+                merged_entry.native_dataset_branch.as_deref(),
+                Some(target_native.as_str())
+            );
+            let reopened = Omnigraph::open(uri).await.unwrap();
+            for handle in [&main, &reopened] {
+                let result = handle
+                    .query(
+                        ReadTarget::branch(target),
+                        TEST_QUERIES,
+                        "get_person",
+                        &params(&[("$name", "Alice")]),
+                    )
+                    .await
+                    .unwrap();
+                let batch = result.concat_batches().unwrap();
+                assert_eq!(
+                    batch
+                        .column(1)
+                        .as_any()
+                        .downcast_ref::<Int32Array>()
+                        .unwrap()
+                        .value(0),
+                    50,
+                    "{target}, {branch_updates} updates: source value must survive adoption"
+                );
+            }
+            main.mutate(
+                target,
+                MUTATION_QUERIES,
+                "add_friend",
+                &params(&[("$from", "Alice"), ("$to", "Diana")]),
+            )
+            .await
+            .unwrap();
+            assert_eq!(
+                main.branch_merge(target, "main").await.unwrap(),
+                MergeOutcome::FastForward
+            );
+            let result = main
+                .query(
+                    ReadTarget::branch("main"),
+                    TEST_QUERIES,
+                    "get_person",
+                    &params(&[("$name", "Alice")]),
+                )
+                .await
+                .unwrap();
+            let batch = result.concat_batches().unwrap();
+            assert_eq!(
+                batch
+                    .column(1)
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .unwrap()
+                    .value(0),
+                50,
+                "{target}, {branch_updates} updates: an unrelated edit must not roll back main"
+            );
+
+            let before_empty = snapshot_branch(&main, target)
+                .await
+                .unwrap()
+                .dataset("node:Person")
+                .unwrap()
+                .clone();
+            let table_uri = format!("{uri}/{}", before_empty.dataset_path);
+            let head_before =
+                open_dataset_head(&table_uri, before_empty.native_dataset_branch.as_deref())
+                    .await
+                    .version()
+                    .version;
+            assert_eq!(
+                main.branch_merge("main", target).await.unwrap(),
+                MergeOutcome::FastForward
+            );
+            let reopened = Omnigraph::open(uri).await.unwrap();
+            for handle in [&main, &reopened] {
+                let after_empty = snapshot_branch(handle, target)
+                    .await
+                    .unwrap()
+                    .dataset("node:Person")
+                    .unwrap()
+                    .clone();
+                assert_eq!(after_empty.type_key, before_empty.type_key);
+                assert_eq!(after_empty.dataset_path, before_empty.dataset_path);
+                assert_eq!(
+                    after_empty.native_dataset_branch,
+                    before_empty.native_dataset_branch
+                );
+                assert_eq!(
+                    after_empty.published_dataset_version,
+                    before_empty.published_dataset_version
+                );
+                assert_eq!(after_empty.entity_count, before_empty.entity_count);
+                assert!(
+                    after_empty.same_registration(&before_empty),
+                    "empty adoption must retain the target's Lance manifest metadata"
+                );
+            }
+            assert_eq!(
+                open_dataset_head(&table_uri, before_empty.native_dataset_branch.as_deref())
+                    .await
+                    .version()
+                    .version,
+                head_before,
+                "empty adoption must not advance the physical target HEAD"
+            );
+        }
+    }
+}
+
 #[tokio::test]
 async fn branch_merge_records_single_latest_commit_with_two_parents() {
     let dir = tempfile::tempdir().unwrap();
diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index b7c0ea5ab..ee1c7ca4c 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -10604,23 +10604,26 @@ async fn pre_upgrade_v1_branch_merge_sidecar_rolls_forward_not_back() {
 /// would record the global head as the merge parent on a non-main
 /// target, and future merges between the same pair would lose
 /// already-up-to-date detection.
-#[tokio::test]
+#[test]
 #[serial]
 #[serial(branch_merge_phase_b)]
-async fn branch_merge_phase_b_failure_recovered_on_non_main_target() {
+fn branch_merge_phase_b_failure_recovered_on_non_main_target() {
+    on_big_stack(branch_merge_phase_b_failure_recovered_on_non_main_target_inner);
+}
+
+async fn branch_merge_phase_b_failure_recovered_on_non_main_target_inner() {
     use omnigraph::loader::{LoadMode, load_jsonl};
 
     let _scenario = FailScenario::setup();
-    let dir = tempfile::tempdir().unwrap();
-    let uri = dir.path().to_str().unwrap().to_string();
-    let operation_id;
-
-    // Setup:
-    //   main: alice
-    //   target_branch (off main): + bob (target moved past base)
-    //   source_branch (off main): + carol (source moved past base)
-    // Merge: source_branch → target_branch
-    {
+    for (case, target_updates, source_branch) in [
+        ("three-way", 0, "source_branch"),
+        ("equal native versions", 2, "main"),
+        ("lower source native version", 8, "main"),
+        ("lower source native version, lazy target", 8, "feature"),
+    ] {
+        let lazy_target = source_branch == "feature";
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap().to_string();
         let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap();
         load_jsonl(
             &db,
@@ -10630,86 +10633,234 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() {
         )
         .await
         .unwrap();
-        db.branch_create("target_branch").await.unwrap();
-        db.mutate(
-            "target_branch",
-            MUTATION_QUERIES,
-            "insert_person",
-            &mixed_params(&[("$name", "Bob")], &[("$age", 40)]),
-        )
-        .await
-        .unwrap();
-        db.branch_create("source_branch").await.unwrap();
-        db.mutate(
-            "source_branch",
-            MUTATION_QUERIES,
-            "insert_person",
-            &mixed_params(&[("$name", "Carol")], &[("$age", 50)]),
-        )
-        .await
-        .unwrap();
-    }
+        if lazy_target {
+            db.branch_create(source_branch).await.unwrap();
+            for age in 40..40 + target_updates + 1 {
+                db.mutate(
+                    "main",
+                    MUTATION_QUERIES,
+                    "set_age",
+                    &mixed_params(&[("$name", "alice")], &[("$age", age)]),
+                )
+                .await
+                .unwrap();
+            }
+            db.mutate(
+                source_branch,
+                MUTATION_QUERIES,
+                "insert_person",
+                &mixed_params(&[("$name", "Bob")], &[("$age", 26)]),
+            )
+            .await
+            .unwrap();
+            assert_eq!(
+                db.branch_merge("main", source_branch).await.unwrap(),
+                omnigraph::db::MergeOutcome::Merged,
+                "{case}"
+            );
+            db.mutate(
+                source_branch,
+                MUTATION_QUERIES,
+                "set_age",
+                &mixed_params(&[("$name", "alice")], &[("$age", 50)]),
+            )
+            .await
+            .unwrap();
+            db.branch_create("target_branch").await.unwrap();
+        } else {
+            db.branch_create("target_branch").await.unwrap();
+        }
+        if !lazy_target && target_updates == 0 {
+            db.mutate(
+                "target_branch",
+                MUTATION_QUERIES,
+                "insert_person",
+                &mixed_params(&[("$name", "Bob")], &[("$age", 40)]),
+            )
+            .await
+            .unwrap();
+            db.branch_create(source_branch).await.unwrap();
+            db.mutate(
+                source_branch,
+                MUTATION_QUERIES,
+                "insert_person",
+                &mixed_params(&[("$name", "Carol")], &[("$age", 50)]),
+            )
+            .await
+            .unwrap();
+        } else if !lazy_target {
+            for age in 40..40 + target_updates {
+                db.mutate(
+                    "target_branch",
+                    MUTATION_QUERIES,
+                    "set_age",
+                    &mixed_params(&[("$name", "alice")], &[("$age", age)]),
+                )
+                .await
+                .unwrap();
+            }
+            assert_eq!(
+                db.branch_merge("target_branch", "main").await.unwrap(),
+                omnigraph::db::MergeOutcome::FastForward,
+                "{case}"
+            );
+            db.mutate(
+                "main",
+                MUTATION_QUERIES,
+                "set_age",
+                &mixed_params(&[("$name", "alice")], &[("$age", 50)]),
+            )
+            .await
+            .unwrap();
+        }
 
-    let main_person_pin = {
-        let db = Omnigraph::open(&uri).await.unwrap();
-        db.snapshot_of(omnigraph::db::ReadTarget::branch("main"))
+        let main_person_pin = db
+            .snapshot_of(ReadTarget::branch("main"))
             .await
             .unwrap()
             .dataset("node:Person")
             .expect("main must have Person")
-            .published_dataset_version
-    };
-    let target_parent_commit_id = branch_head_commit_id(dir.path(), "target_branch")
+            .published_dataset_version;
+        let source_person = db
+            .snapshot_of(ReadTarget::branch(source_branch))
+            .await
+            .unwrap()
+            .dataset("node:Person")
+            .unwrap()
+            .clone();
+        if target_updates > 0 {
+            let target_version = db
+                .snapshot_of(ReadTarget::branch("target_branch"))
+                .await
+                .unwrap()
+                .dataset("node:Person")
+                .unwrap()
+                .published_dataset_version;
+            assert_eq!(
+                source_person.published_dataset_version.cmp(&target_version),
+                if target_updates == 2 {
+                    std::cmp::Ordering::Equal
+                } else {
+                    std::cmp::Ordering::Less
+                },
+                "fixture must exercise {case}"
+            );
+        }
+        let source_head = branch_head_commit_id(dir.path(), source_branch)
+            .await
+            .unwrap();
+        let target_parent_commit_id = branch_head_commit_id(dir.path(), "target_branch")
+            .await
+            .unwrap();
+        drop(db);
+
+        let operation_id = {
+            let db = Omnigraph::open(&uri).await.unwrap();
+            let _failpoint = ScopedFailPoint::new(
+                names::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+                "return",
+            );
+            let err = db
+                .branch_merge(source_branch, "target_branch")
+                .await
+                .unwrap_err();
+            assert!(
+                err.to_string().contains(
+                    "injected failpoint triggered: branch_merge.post_phase_b_pre_manifest_commit"
+                ),
+                "{case}: unexpected error: {err}"
+            );
+            single_sidecar_operation_id(dir.path())
+        };
+
+        let db = Omnigraph::open(&uri).await.unwrap();
+        drop(db);
+        assert_post_recovery_invariants(
+            dir.path(),
+            &operation_id,
+            RecoveryExpectation::RolledForwardOriginalLineage {
+                tables: vec![
+                    TableExpectation::branch("node:Person", "target_branch")
+                        .expected_main_manifest_pin(main_person_pin)
+                        .expected_recovery_parent_commit_id(target_parent_commit_id),
+                ],
+            },
+        )
         .await
         .unwrap();
 
-    // Setup: failpoint fires after the per-table publish loop completes
-    // but before commit_manifest_updates. Sidecar persists with
-    // branch=Some("target_branch").
-    {
         let db = Omnigraph::open(&uri).await.unwrap();
-        let _failpoint = ScopedFailPoint::new(
-            names::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
-            "return",
-        );
-        let err = db
-            .branch_merge("source_branch", "target_branch")
+        let recovered_source = db
+            .snapshot_of(ReadTarget::branch(source_branch))
             .await
-            .unwrap_err();
-        assert!(
-            err.to_string().contains(
-                "injected failpoint triggered: branch_merge.post_phase_b_pre_manifest_commit"
-            ),
-            "unexpected error: {err}"
+            .unwrap();
+        let recovered_source_person = recovered_source.dataset("node:Person").unwrap();
+        assert_eq!(
+            recovered_source_person.published_dataset_version,
+            source_person.published_dataset_version,
+            "{case}: recovery must not advance the source table"
         );
-        let recovery_dir = dir.path().join("__recovery");
-        let sidecar_count = std::fs::read_dir(&recovery_dir).unwrap().count();
         assert_eq!(
-            sidecar_count, 1,
-            "exactly one sidecar must persist after non-main branch_merge failure"
+            recovered_source_person.native_dataset_branch, source_person.native_dataset_branch,
+            "{case}: recovery must preserve the source ref"
         );
-        operation_id = single_sidecar_operation_id(dir.path());
+        assert_eq!(
+            branch_head_commit_id(dir.path(), source_branch)
+                .await
+                .unwrap(),
+            source_head,
+            "{case}: recovery must not change source lineage"
+        );
+        let recovered_commit =
+            omnigraph::db::commit_graph::CommitGraph::open_at_branch(&uri, "target_branch")
+                .await
+                .unwrap()
+                .head_commit()
+                .await
+                .unwrap()
+                .unwrap();
+        assert_eq!(
+            recovered_commit.merged_parent_commit_id.as_deref(),
+            Some(source_head.as_str()),
+            "{case}: recovery must retain the captured source parent"
+        );
+        let expected = if lazy_target {
+            vec![("Bob", 26), ("alice", 50)]
+        } else if target_updates == 0 {
+            vec![("Bob", 40), ("Carol", 50), ("alice", 30)]
+        } else {
+            vec![("alice", 50)]
+        };
+        let rows = helpers::read_table_branch(&db, "target_branch", "node:Person").await;
+        assert_eq!(
+            rows.iter().map(RecordBatch::num_rows).sum::<usize>(),
+            expected.len(),
+            "{case}: recovery must retain every merged row"
+        );
+        for (name, expected_age) in expected {
+            let result = db
+                .query(
+                    ReadTarget::branch("target_branch"),
+                    TEST_QUERIES,
+                    "get_person",
+                    &params(&[("$name", name)]),
+                )
+                .await
+                .unwrap();
+            let batch = result.concat_batches().unwrap();
+            assert_eq!(batch.num_rows(), 1, "{case}: missing {name}");
+            assert_eq!(
+                batch
+                    .column(1)
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .unwrap()
+                    .value(0),
+                expected_age,
+                "{case}: recovery must publish {name}'s captured source value"
+            );
+        }
     }
-
-    // Recovery: reopen runs full sweep. The BranchMerge sidecar's branch
-    // = Some("target_branch"); D2 fix opens a per-branch CommitGraph
-    // for the audit append so the merge-parent linkage is correct.
-    let db = Omnigraph::open(&uri).await.unwrap();
-    drop(db);
-
-    assert_post_recovery_invariants(
-        dir.path(),
-        &operation_id,
-        RecoveryExpectation::RolledForwardOriginalLineage {
-            tables: vec![
-                TableExpectation::branch("node:Person", "target_branch")
-                    .expected_main_manifest_pin(main_person_pin)
-                    .expected_recovery_parent_commit_id(target_parent_commit_id),
-            ],
-        },
-    )
-    .await
-    .unwrap();
 }
 
 /// Contract: the BranchMerge sidecar's per-table `table_branch` MUST be
diff --git a/crates/omnigraph/tests/merge_fast_forward.rs b/crates/omnigraph/tests/merge_fast_forward.rs
index cb1c38115..f34934916 100644
--- a/crates/omnigraph/tests/merge_fast_forward.rs
+++ b/crates/omnigraph/tests/merge_fast_forward.rs
@@ -1395,146 +1395,189 @@ async fn fast_forward_merge_streams_blob_columns() {
 /// siblings distinct.
 #[tokio::test]
 async fn blob_changed_only_adopt_uses_known_present_update() {
-    let dir = tempfile::tempdir().unwrap();
-    let uri = dir.path().to_str().unwrap();
-    let external_dir = tempfile::tempdir().unwrap();
-    let external_path = external_dir.path().join("changed.txt");
-    std::fs::write(&external_path, b"Changed externally").unwrap();
-    let external_uri = url::Url::from_file_path(std::fs::canonicalize(&external_path).unwrap())
-        .expect("external Blob path is absolute")
-        .to_string();
-    let empty_path = external_dir.path().join("valid-empty.txt");
-    std::fs::write(&empty_path, b"").unwrap();
-    let empty_uri = url::Url::from_file_path(std::fs::canonicalize(&empty_path).unwrap())
-        .expect("empty external Blob path is absolute")
-        .to_string();
-    let external_base = url::Url::from_directory_path(external_dir.path())
-        .expect("external Blob base is absolute")
-        .to_string();
-    let policy = ExternalBlobPolicy::allow(vec![
-        ExternalBlobBase::new(external_base, ExternalBlobExecutionScope::EmbeddedOnly).unwrap(),
-    ])
-    .unwrap();
-
-    let main = Omnigraph::init(uri, BLOB_SCHEMA)
-        .await
-        .unwrap()
-        .with_external_blob_policy(policy.clone())
-        .unwrap();
-    let base_data = [
-        serde_json::json!({
-            "type": "Document",
-            "data": {"title": "changed", "content": "base64:QmFzZQ==", "note": "base"},
-        }),
-        serde_json::json!({
-            "type": "Document",
-            "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"},
-        }),
-        serde_json::json!({
-            "type": "Document",
-            "data": {"title": "null", "content": null, "note": "null"},
-        }),
-    ]
-    .into_iter()
-    .map(|row| row.to_string())
-    .collect::<Vec<_>>()
-    .join("\n");
-    main.load("main", &base_data, LoadMode::Overwrite)
-        .await
-        .unwrap();
-    main.branch_create("feature").await.unwrap();
+    const SET_NOTE: &str = r#"
+query set_note($title: String, $note: String) {
+    update Document set { note: $note } where title = $title
+}
+"#;
 
-    let feature = Omnigraph::open(uri)
-        .await
-        .unwrap()
-        .with_external_blob_policy(policy.clone())
-        .unwrap();
-    let source_data = [
-        serde_json::json!({
-            "type": "Document",
-            "data": {"title": "changed", "content": external_uri, "note": "source"},
-        }),
-        serde_json::json!({
-            "type": "Document",
-            "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"},
-        }),
-        serde_json::json!({
-            "type": "Document",
-            "data": {"title": "null", "content": null, "note": "null"},
-        }),
-    ]
-    .into_iter()
-    .map(|row| row.to_string())
-    .collect::<Vec<_>>()
-    .join("\n");
-    feature
-        .load("feature", &source_data, LoadMode::Overwrite)
-        .await
+    for source_main in [false, true] {
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let external_dir = tempfile::tempdir().unwrap();
+        let external_path = external_dir.path().join("changed.txt");
+        std::fs::write(&external_path, b"Changed externally").unwrap();
+        let external_uri = url::Url::from_file_path(std::fs::canonicalize(&external_path).unwrap())
+            .expect("external Blob path is absolute")
+            .to_string();
+        let empty_path = external_dir.path().join("valid-empty.txt");
+        std::fs::write(&empty_path, b"").unwrap();
+        let empty_uri = url::Url::from_file_path(std::fs::canonicalize(&empty_path).unwrap())
+            .expect("empty external Blob path is absolute")
+            .to_string();
+        let external_base = url::Url::from_directory_path(external_dir.path())
+            .expect("external Blob base is absolute")
+            .to_string();
+        let policy = ExternalBlobPolicy::allow(vec![
+            ExternalBlobBase::new(external_base, ExternalBlobExecutionScope::EmbeddedOnly).unwrap(),
+        ])
         .unwrap();
 
-    let merger = Omnigraph::open(uri)
-        .await
-        .unwrap()
-        .with_external_blob_policy(policy)
-        .unwrap();
-    let probes = MergeWriteProbes::default();
-    let outcome = with_merge_write_probes(probes.clone(), merger.branch_merge("feature", "main"))
-        .await
-        .unwrap();
-    assert_eq!(outcome, MergeOutcome::FastForward);
-    assert_eq!(probes.stage_known_present_update_calls(), 1);
-    assert_eq!(probes.stage_known_present_update_rows(), 1);
-    assert_eq!(probes.stage_merge_insert_calls(), 0);
-    assert_eq!(probes.stage_fenced_insert_calls(), 0);
-    assert_eq!(probes.strict_insert_preflight_calls(), 0);
-    assert_eq!(
-        probes.stage_vector_index_calls(),
-        0,
-        "general Blob adoption must also defer derived index work"
-    );
-    assert_eq!(
-        probes.external_blob_probe_inputs(),
-        1,
-        "only the changed external descriptor belongs to the adopt delta"
-    );
-    assert_eq!(probes.external_blob_probe_calls(), 1);
-    assert_eq!(probes.external_blob_payload_read_calls(), 1);
+        let main = Omnigraph::init(uri, BLOB_SCHEMA)
+            .await
+            .unwrap()
+            .with_external_blob_policy(policy.clone())
+            .unwrap();
+        let base_data = [
+            serde_json::json!({
+                "type": "Document",
+                "data": {"title": "changed", "content": "base64:QmFzZQ==", "note": "base"},
+            }),
+            serde_json::json!({
+                "type": "Document",
+                "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"},
+            }),
+            serde_json::json!({
+                "type": "Document",
+                "data": {"title": "null", "content": null, "note": "null"},
+            }),
+        ]
+        .into_iter()
+        .map(|row| row.to_string())
+        .collect::<Vec<_>>()
+        .join("\n");
+        main.load("main", &base_data, LoadMode::Overwrite)
+            .await
+            .unwrap();
+        main.branch_create("feature").await.unwrap();
 
-    assert_eq!(count_rows(&merger, "node:Document").await, 3);
-    let changed = read_managed_blob_bytes(
-        &merger,
-        ReadTarget::branch("main"),
-        node_blob_cell("Document", "changed", "content"),
-    )
-    .await;
-    assert_eq!(&changed[..], b"Changed externally");
+        let feature = Omnigraph::open(uri)
+            .await
+            .unwrap()
+            .with_external_blob_policy(policy.clone())
+            .unwrap();
+        let source_data = [
+            serde_json::json!({
+                "type": "Document",
+                "data": {"title": "changed", "content": external_uri, "note": "source"},
+            }),
+            serde_json::json!({
+                "type": "Document",
+                "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"},
+            }),
+            serde_json::json!({
+                "type": "Document",
+                "data": {"title": "null", "content": null, "note": "null"},
+            }),
+        ]
+        .into_iter()
+        .map(|row| row.to_string())
+        .collect::<Vec<_>>()
+        .join("\n");
+        let (source, target) = if source_main {
+            for step in 0..8 {
+                feature
+                    .mutate(
+                        "feature",
+                        SET_NOTE,
+                        "set_note",
+                        &params(&[("$title", "changed"), ("$note", &format!("step-{step}"))]),
+                    )
+                    .await
+                    .unwrap();
+            }
+            assert_eq!(
+                main.branch_merge("feature", "main").await.unwrap(),
+                MergeOutcome::FastForward
+            );
+            ("main", "feature")
+        } else {
+            ("feature", "main")
+        };
+        feature
+            .load(source, &source_data, LoadMode::Overwrite)
+            .await
+            .unwrap();
+        if source_main {
+            let source_snapshot = snapshot_branch(&main, source).await.unwrap();
+            let target_snapshot = snapshot_branch(&main, target).await.unwrap();
+            assert!(
+                source_snapshot
+                    .dataset("node:Document")
+                    .unwrap()
+                    .published_dataset_version
+                    < target_snapshot
+                        .dataset("node:Document")
+                        .unwrap()
+                        .published_dataset_version,
+                "source-main fixture must require target-lineage adoption"
+            );
+        }
 
-    let empty = merger
-        .read_blob_at(
-            ReadTarget::branch("main"),
-            node_blob_cell("Document", "valid-empty", "content"),
-        )
-        .await
-        .unwrap();
-    let BlobContent::External(empty) = empty.content else {
-        panic!("an unchanged retained descriptor must stay pointer-only");
-    };
-    assert_eq!(empty.uri, empty_uri);
-    assert_eq!(empty.offset, 0);
-    assert_eq!(empty.length, None);
-
-    let null = merger
-        .read_blob_at(
-            ReadTarget::branch("main"),
-            node_blob_cell("Document", "null", "content"),
+        let merger = Omnigraph::open(uri)
+            .await
+            .unwrap()
+            .with_external_blob_policy(policy)
+            .unwrap();
+        let probes = MergeWriteProbes::default();
+        let outcome = with_merge_write_probes(probes.clone(), merger.branch_merge(source, target))
+            .await
+            .unwrap();
+        assert_eq!(outcome, MergeOutcome::FastForward);
+        assert_eq!(probes.stage_known_present_update_calls(), 1);
+        assert_eq!(probes.stage_known_present_update_rows(), 1);
+        assert_eq!(probes.stage_merge_insert_calls(), 0);
+        assert_eq!(probes.stage_fenced_insert_calls(), 0);
+        assert_eq!(probes.strict_insert_preflight_calls(), 0);
+        assert_eq!(
+            probes.stage_vector_index_calls(),
+            0,
+            "general Blob adoption must also defer derived index work"
+        );
+        assert_eq!(
+            probes.external_blob_probe_inputs(),
+            1,
+            "only the changed external descriptor belongs to the adopt delta"
+        );
+        assert_eq!(probes.external_blob_probe_calls(), 1);
+        assert_eq!(probes.external_blob_payload_read_calls(), 1);
+
+        assert_eq!(count_rows_branch(&merger, target, "node:Document").await, 3);
+        let changed = read_managed_blob_bytes(
+            &merger,
+            ReadTarget::branch(target),
+            node_blob_cell("Document", "changed", "content"),
         )
-        .await
-        .unwrap_err();
-    assert!(
-        matches!(
-            null,
-            OmniError::Manifest(ref error) if error.kind == ManifestErrorKind::NotFound
-        ),
-        "an unchanged null Blob must remain null rather than becoming valid-empty: {null:?}"
-    );
+        .await;
+        assert_eq!(&changed[..], b"Changed externally");
+
+        let empty = merger
+            .read_blob_at(
+                ReadTarget::branch(target),
+                node_blob_cell("Document", "valid-empty", "content"),
+            )
+            .await
+            .unwrap();
+        let BlobContent::External(empty) = empty.content else {
+            panic!("an unchanged retained descriptor must stay pointer-only");
+        };
+        assert_eq!(empty.uri, empty_uri);
+        assert_eq!(empty.offset, 0);
+        assert_eq!(empty.length, None);
+
+        let null = merger
+            .read_blob_at(
+                ReadTarget::branch(target),
+                node_blob_cell("Document", "null", "content"),
+            )
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(
+                null,
+                OmniError::Manifest(ref error) if error.kind == ManifestErrorKind::NotFound
+            ),
+            "an unchanged null Blob must remain null rather than becoming valid-empty: {null:?}"
+        );
+    }
 }
diff --git a/docs/dev/merge.md b/docs/dev/merge.md
index 63c4c6748..03d85d19a 100644
--- a/docs/dev/merge.md
+++ b/docs/dev/merge.md
@@ -26,7 +26,8 @@ The table classifier chooses one of four routes:
 1. **No change:** source contributes nothing.
 2. **Pointer adoption:** target still equals the base and the exact source table
    state can become the target's visible pointer without copying rows. A
-   first-touch lazy target stays on this ref-only route.
+   first-touch lazy target can use this ref-only route when the version check
+   below permits it.
 3. **Proven insertion replay:** target still permits data replay and the
    complete retained source interval proves a contiguous sequence of exact-ID,
    insertion-only transactions.
@@ -37,6 +38,29 @@ An optimization miss is not a merge failure. Missing transaction history,
 unknown certificate fields, incomplete ancestry, or an unfamiliar Lance shape
 falls back to the general route.
 
+Native Lance version numbers are local to each native branch. The manifest
+projection nevertheless selects the highest registered version for a table
+lifetime, so adopting a source version less than or equal to the target's
+version cannot safely replace its pointer. A nonempty source delta uses the
+target-lineage writer, creating a native target ref lazily when needed. A
+proved empty delta retains the target's complete table entry, including its
+native ref and version metadata. The version comparison selects the
+publication route; it never proves row equality. This rule also governs Blob
+preflight so validation and publication agree on which rows are copied, which
+means a `main`-to-branch adoption that carries Blob payloads is subject to the
+keyed-write byte cap like any other delta. The proven insertion replay never
+applies to this shape: the proof needs a source version above the base, and
+the base equals the target here, so the delta is computed by the ordered walk
+even when it turns out empty.
+
+This comparison is a fence, not the class fix: the projection and the
+registry guard order registrations by the native version number, a
+per-branch counter, and the same comparison produced #473 and the projection
+miss this rule works around. Ordering registrations by the `__manifest`
+version that published them (unique and increasing within every branch's
+lineage, which is the only scope the projection ever compares) would retire
+the rule.
+
 ## Proven insertion route
 
 The internal `omnigraph.insert_absence = "v1"` transaction property says that
diff --git a/docs/releases/v0.11.0.md b/docs/releases/v0.11.0.md
index c2740736f..94440d2e0 100644
--- a/docs/releases/v0.11.0.md
+++ b/docs/releases/v0.11.0.md
@@ -4,6 +4,18 @@ Notes accumulate here until the release is cut.
 
 ## Highlights
 
+- **Repeated branch merges preserve previously merged rows and values.**
+  After a branch accumulated more table versions than `main`, merging newer
+  data from `main` into that branch could report success while retaining stale
+  data. Merging the branch back after an unrelated edit could then delete an
+  edge or roll back a value on `main`. When the two counts were equal, the
+  merge failed instead with `Concurrent modification: table version N already
+  exists for identity … with different state`. Merge now applies these changes
+  through the target's table history and preserves its existing entry when the
+  rows are unchanged; the equal-count merge fast-forwards. A `main`-to-branch
+  merge that carries Blob payloads now checks the same keyed-write byte cap as
+  a branch-to-`main` merge.
+
 - **Managed cluster creation, config upload and retirement.** `cluster create`
   requests an empty cluster, `cluster push` conditionally prepares referenced
   config files, and `cluster delete` / `cluster undo-delete` target exact
diff --git a/docs/user/branching/merge.md b/docs/user/branching/merge.md
index 86b0dfcf4..105718ef0 100644
--- a/docs/user/branching/merge.md
+++ b/docs/user/branching/merge.md
@@ -26,6 +26,13 @@ composition: follow it with `branch delete <source>`. A merge statement takes
 no commit precondition -- `POST /mutate/if-graph-commit` and `--if-commit`
 refuse one beside it -- and no front offers a conditional merge today.
 
+A merge preserves changes already integrated into a branch when you later
+merge that branch back. For example, after merging a new edge from `main`
+into `review`, editing only a node on `review` and merging it into `main`
+preserves that edge. A merge whose source and target tables reached the same
+version count on different branches fast-forwards; earlier releases refused
+it with a `table version … already exists` error.
+
 ## Outcomes
 
 - **Already up to date**: the target already contains the source changes.