fix(merge): align composite @unique key separator with intake (U+001F)

The branch-merge path (update_unique_constraints) joined composite key
columns with '|', while intake joins with U+001F. The same @unique(a, b)
was keyed two different ways, and '|'-join can raise phantom merge
conflicts for values containing '|' (e.g. ('x|y','z') vs ('x','y|z')).

Factor the tuple-join into one shared helper (loader::composite_unique_key)
so the intake and merge paths cannot drift again. Add branching regression
tests for edge @unique(src, dst) on the merge path.

Refs MR-983.

crates/omnigraph/src/exec/merge.rs

@@ -697,7 +697,7 @@ fn update_unique_constraints(
             if any_null {
                 continue;
             }
-            let value = parts.join("|");
+            let value = crate::loader::composite_unique_key(&parts);
             let row_id = row_id_at(batch, row)?;
             if let Some(first_row_id) = seen.insert(value.clone(), row_id.clone()) {
                 conflicts.push(MergeConflict {

crates/omnigraph/src/loader/mod.rs

@@ -1466,11 +1466,7 @@ pub(crate) fn enforce_unique_constraints_intra_batch(
             if any_null {
                 continue;
             }
-            // Join on the unit separator (U+001F) — a control char highly
-            // unlikely to occur in real data, keeping composite keys
-            // effectively unambiguous. Matches `exec/merge.rs::row_signature`,
-            // which uses the same separator.
-            let value = parts.join("\u{1f}");
+            let value = composite_unique_key(&parts);
             if let Some(prev_row) = seen.insert(value.clone(), row) {
                 return Err(OmniError::manifest(format!(
                     "@unique violation on {}.{}: value '{}' appears in rows {} and {}",
@@ -1486,6 +1482,18 @@ pub(crate) fn enforce_unique_constraints_intra_batch(
     Ok(())
 }
 
+/// Join one row's rendered, non-null column values into a single composite
+/// uniqueness key. The separator is the unit separator (U+001F) — a control
+/// char highly unlikely to occur in real data, so distinct tuples like
+/// `("a|b", "c")` and `("a", "b|c")` stay distinct rather than colliding.
+///
+/// Shared by the intake path (`enforce_unique_constraints_intra_batch`) and
+/// the branch-merge path (`exec/merge.rs::update_unique_constraints`) so the
+/// two cannot silently drift to incompatible keyings.
+pub(crate) fn composite_unique_key(parts: &[String]) -> String {
+    parts.join("\u{1f}")
+}
+
 /// Render a unique constraint's columns for error messages: a single column
 /// as `col`, a composite as `(a, b)`.
 fn format_unique_columns(columns: &[String]) -> String {

crates/omnigraph/tests/branching.rs

@@ -39,6 +39,26 @@ query insert_user($name: String, $email: String) {
 }
 "#;
 
+const EDGE_UNIQUE_SCHEMA: &str = r#"
+node Person {
+    name: String @key
+}
+
+edge Knows: Person -> Person {
+    @unique(src, dst)
+}
+"#;
+
+const EDGE_UNIQUE_DATA: &str = r#"{"type":"Person","data":{"name":"Alice"}}
+{"type":"Person","data":{"name":"Bob"}}
+{"type":"Person","data":{"name":"Carol"}}"#;
+
+const EDGE_UNIQUE_MUTATIONS: &str = r#"
+query add_knows($from: String, $to: String) {
+    insert Knows { from: $from, to: $to }
+}
+"#;
+
 const CARDINALITY_SCHEMA: &str = r#"
 node Person {
     name: String @key
@@ -1119,6 +1139,87 @@ async fn branch_merge_reports_unique_violation_conflict() {
     }
 }
 
+/// Regression for the MR-983 follow-up: the branch-merge path must enforce an
+/// edge composite `@unique(src, dst)` as a true composite key, consistent with
+/// the intake path. Two branches inserting the *same* (src, dst) pair must
+/// conflict on merge.
+#[tokio::test]
+async fn branch_merge_reports_composite_unique_violation_conflict() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let mut main = init_db_from_schema_and_data(&dir, EDGE_UNIQUE_SCHEMA, EDGE_UNIQUE_DATA).await;
+    main.branch_create("feature").await.unwrap();
+
+    let mut feature = Omnigraph::open(uri).await.unwrap();
+
+    mutate_main(
+        &mut main,
+        EDGE_UNIQUE_MUTATIONS,
+        "add_knows",
+        &params(&[("$from", "Alice"), ("$to", "Bob")]),
+    )
+    .await
+    .unwrap();
+
+    mutate_branch(
+        &mut feature,
+        "feature",
+        EDGE_UNIQUE_MUTATIONS,
+        "add_knows",
+        &params(&[("$from", "Alice"), ("$to", "Bob")]),
+    )
+    .await
+    .unwrap();
+
+    let err = main.branch_merge("feature", "main").await.unwrap_err();
+    match err {
+        OmniError::MergeConflicts(conflicts) => {
+            assert!(conflicts.iter().any(|conflict| {
+                conflict.table_key == "edge:Knows"
+                    && conflict.kind == MergeConflictKind::UniqueViolation
+            }));
+        }
+        other => panic!("expected merge conflicts, got {other:?}"),
+    }
+}
+
+/// Sibling to the above: pairs sharing `src` but differing on `dst` are unique
+/// on the (src, dst) tuple and must merge cleanly. Guards against the composite
+/// degrading back into a single-field `@unique(src)` on the merge path.
+#[tokio::test]
+async fn branch_merge_allows_distinct_composite_unique_pairs() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let mut main = init_db_from_schema_and_data(&dir, EDGE_UNIQUE_SCHEMA, EDGE_UNIQUE_DATA).await;
+    main.branch_create("feature").await.unwrap();
+
+    let mut feature = Omnigraph::open(uri).await.unwrap();
+
+    mutate_main(
+        &mut main,
+        EDGE_UNIQUE_MUTATIONS,
+        "add_knows",
+        &params(&[("$from", "Alice"), ("$to", "Bob")]),
+    )
+    .await
+    .unwrap();
+
+    mutate_branch(
+        &mut feature,
+        "feature",
+        EDGE_UNIQUE_MUTATIONS,
+        "add_knows",
+        &params(&[("$from", "Alice"), ("$to", "Carol")]),
+    )
+    .await
+    .unwrap();
+
+    main.branch_merge("feature", "main")
+        .await
+        .expect("distinct (src, dst) pairs are unique on the composite and must merge cleanly");
+    assert_eq!(count_rows(&main, "edge:Knows").await, 2);
+}
+
 #[tokio::test]
 async fn branch_merge_reports_cardinality_violation_conflict() {
     let dir = tempfile::tempdir().unwrap();