Report store version conflicts as typed values, not parsed messages (closes #3032)

packages/core/src/db/mod.rs

@@ -16,5 +16,5 @@ pub(crate) use sqlite_store::collection_not_root;
 pub(crate) use sqlite_store::tx::Tx;
 pub use sqlite_store::{
     ensure_sqlite_vec_registered, BulkNodeRow, RelationshipRecord, ResolvedEntity, SqliteStore,
-    StoreChange, StoreOperation,
+    StoreChange, StoreOperation, VersionConflict,
 };

packages/core/src/db/sqlite_store/mod.rs

@@ -87,6 +87,33 @@ impl RelationshipRecord {
     }
 }
 
+/// An optimistic-concurrency mismatch observed by a store write: `node_id`
+/// was expected at `expected` but is persisted at `actual`.
+///
+/// Store methods return it as `Ok(Err(VersionConflict))` rather than as an
+/// `anyhow` error (ADR-069 §2a: a version mismatch is an expected outcome, not
+/// a transaction failure), so the service maps it to
+/// `NodeServiceError::VersionConflict` by type — never by parsing message
+/// text — and carries the real versions through to the caller.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct VersionConflict {
+    pub node_id: String,
+    pub expected: i64,
+    pub actual: i64,
+}
+
+impl std::fmt::Display for VersionConflict {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "version conflict on node '{}': expected {}, actual {}",
+            self.node_id, self.expected, self.actual
+        )
+    }
+}
+
+impl std::error::Error for VersionConflict {}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum StoreOperation {
     Created,
@@ -663,7 +690,11 @@ mod tests {
                         .iter()
                         .map(|(id, ver)| (id.as_str(), *ver))
                         .collect();
-                    SqliteStore::move_children_to_parent_in_tx(tx, &new_parent_id, &refs).await
+                    // A conflict must fail the closure, not return `Ok`, or
+                    // `with_transaction` would commit the edges already swapped.
+                    SqliteStore::move_children_to_parent_in_tx(tx, &new_parent_id, &refs)
+                        .await?
+                        .map_err(anyhow::Error::new)
                 })
             })
             .await
@@ -2225,6 +2256,40 @@ mod tests {
         Ok(())
     }
 
+    /// A gated DELETE that removes nothing is only a version conflict when the
+    /// persisted version actually differs. A child at its expected version with
+    /// no `has_child` edge (a root) must fail as a plain error, not surface as a
+    /// conflict carrying equal expected and actual versions.
+    #[tokio::test]
+    async fn test_move_children_to_parent_store_root_child_is_not_a_conflict() -> Result<()> {
+        let (store, _temp) = create_test_store().await?;
+
+        let new_parent = Node::new("text".to_string(), "New Parent".to_string(), json!({}));
+        let new_parent_id = new_parent.id.clone();
+        store.create_node(new_parent, None, None).await?;
+
+        let root = Node::new("text".to_string(), "Root".to_string(), json!({}));
+        let root_id = root.id.clone();
+        store.create_node(root, None, None).await?;
+        let root_version = store.get_node(&root_id).await?.unwrap().version;
+
+        let err =
+            move_children_to_parent(&store, &new_parent_id, &[(root_id.as_str(), root_version)])
+                .await
+                .expect_err("a root child has no edge to replace");
+        assert!(
+            err.downcast_ref::<VersionConflict>().is_none(),
+            "a missing parent edge is not a version conflict, got: {err:#}"
+        );
+        assert!(
+            err.to_string().contains("has no parent edge to replace"),
+            "expected the missing-parent-edge error, got: {err:#}"
+        );
+        assert_eq!(store.get_parent_id(&root_id).await?, None);
+
+        Ok(())
+    }
+
     // C3c: store-layer atomicity — exercises the in-transaction version check
     // that the service-level pre-validation cannot catch (concurrent version bump).
     #[tokio::test]
@@ -2264,15 +2329,18 @@ mod tests {
         )
         .await;
 
-        assert!(
-            result.is_err(),
-            "stale version should cause store-level failure"
-        );
-        let err_msg = result.unwrap_err().to_string();
-        assert!(
-            err_msg.contains("VERSION_CONFLICT"),
-            "error should contain VERSION_CONFLICT, got: {}",
-            err_msg
+        let err = result.expect_err("stale version should cause store-level failure");
+        let conflict = err
+            .downcast_ref::<VersionConflict>()
+            .unwrap_or_else(|| panic!("expected a typed VersionConflict, got: {err:#}"));
+        let child2_actual = store.get_node(&child2_id).await?.unwrap().version;
+        assert_eq!(
+            conflict,
+            &VersionConflict {
+                node_id: child2_id.clone(),
+                expected: stale_version,
+                actual: child2_actual,
+            }
         );
 
         // ALL-OR-NOTHING: child1 must NOT have moved (transaction was rolled back).

packages/core/src/db/sqlite_store/nodes.rs

@@ -1335,18 +1335,20 @@ impl SqliteStore {
     /// drifting from — a second walk.
     ///
     /// Returns `(existed, deleted_nodes)` where `deleted_nodes` contains the target and all
-    /// descendants that were deleted (empty vec when target didn't exist).
+    /// descendants that were deleted (empty vec when target didn't exist), or
+    /// `Ok(Err(VersionConflict))` when the target's persisted version is not
+    /// `expected_version` — nothing is deleted in that case.
     pub async fn delete_subtree_atomic(
         &self,
         node_id: &str,
         expected_version: i64,
         subtree_ids: &[String],
         source: Option<String>,
-    ) -> Result<(bool, Vec<Node>)> {
+    ) -> Result<std::result::Result<(bool, Vec<Node>), VersionConflict>> {
         // Collect the target + all descendants before mutating.
         let target = match self.get_node(node_id).await? {
             Some(n) => n,
-            None => return Ok((false, vec![])),
+            None => return Ok(Ok((false, vec![]))),
         };
 
         // Only the target can be a schema node (descendants are its description
@@ -1356,12 +1358,11 @@ impl SqliteStore {
 
         // OCC check on target before entering the transaction.
         if target.version != expected_version {
-            return Err(anyhow::anyhow!(
-                "version_conflict:{}:{}:{}",
-                node_id,
-                expected_version,
-                target.version
-            ));
+            return Ok(Err(VersionConflict {
+                node_id: node_id.to_string(),
+                expected: expected_version,
+                actual: target.version,
+            }));
         }
 
         let all_ids: Vec<String> = subtree_ids.to_vec();
@@ -1412,16 +1413,16 @@ impl SqliteStore {
             Some(row) => row.get(0)?,
             None => {
                 // Node disappeared between pre-check and transaction — idempotent.
-                return Ok((false, vec![]));
+                return Ok(Ok((false, vec![])));
             }
         };
         if actual_version != expected_version {
-            return Err(anyhow::anyhow!(
-                "version_conflict:{}:{}:{}",
-                node_id,
-                expected_version,
-                actual_version
-            ));
+            // Dropping `tx` uncommitted rolls it back; nothing was written yet.
+            return Ok(Err(VersionConflict {
+                node_id: node_id.to_string(),
+                expected: expected_version,
+                actual: actual_version,
+            }));
         }
 
         // Chunked, but every chunk executes against the same `tx` opened above —
@@ -1460,7 +1461,7 @@ impl SqliteStore {
             });
         }
 
-        Ok((true, nodes_to_delete))
+        Ok(Ok((true, nodes_to_delete)))
     }
 
     /// Delete all descendants of `parent_id` (the full child subtree) without touching the parent.
@@ -2833,11 +2834,15 @@ impl SqliteStore {
 
     /// Re-parent an ordered set of existing children to `new_parent_id` inside
     /// `tx`. Validates each child's version using `SELECT changes()` after a
-    /// version-gated DELETE — any mismatch returns an error, so the caller's
-    /// transaction rolls back every edge moved so far (all-or-nothing OCC).
+    /// version-gated DELETE.
     ///
     /// Returns the assigned fractional order for each child, preserving input
-    /// array order as sibling order under the new parent.
+    /// array order as sibling order under the new parent, or
+    /// `Ok(Err(VersionConflict))` carrying the first mismatched child's
+    /// expected and persisted versions. On a conflict, edges for earlier
+    /// children have already been swapped in `tx`: the caller must turn the
+    /// conflict into an `Err` from its transaction closure so every one of
+    /// them rolls back (all-or-nothing OCC).
     ///
     /// `_in_tx` only (ADR-069 §1a): the service composes the version bumps and
     /// event emission into the same transaction, so edges and bumps commit
@@ -2846,9 +2851,9 @@ impl SqliteStore {
         tx: &Tx<'_>,
         new_parent_id: &str,
         children: &[(&str, i64)],
-    ) -> Result<Vec<f64>> {
+    ) -> Result<std::result::Result<Vec<f64>, VersionConflict>> {
         if children.is_empty() {
-            return Ok(Vec::new());
+            return Ok(Ok(Vec::new()));
         }
 
         let now = Utc::now().to_rfc3339();
@@ -2902,15 +2907,41 @@ impl SqliteStore {
                 .query("SELECT changes()", libsql::params![])
                 .await
                 .context("Failed to query changes()")?;
-            if let Some(row) = changes_rows.next().await? {
-                let affected: i64 = row.get(0)?;
-                if affected == 0 {
-                    return Err(anyhow::anyhow!(
-                        "VERSION_CONFLICT: node '{}' version mismatch (expected {})",
-                        child_id,
-                        expected_version
-                    ));
+            let affected: i64 = match changes_rows.next().await? {
+                Some(row) => row.get(0)?,
+                None => return Err(anyhow::anyhow!("SELECT changes() returned no row")),
+            };
+            if affected == 0 {
+                // Nothing was deleted: either the version moved on, or the
+                // child has no has_child edge to replace. Read the persisted
+                // version in this same transaction to tell them apart.
+                let mut version_rows = db
+                    .query(
+                        "SELECT version FROM node WHERE id = ?1",
+                        libsql::params![child_id.clone()],
+                    )
+                    .await
+                    .context("Failed to read child version after a gated DELETE")?;
+                let actual: i64 = match version_rows.next().await? {
+                    Some(row) => row.get(0)?,
+                    None => {
+                        return Err(anyhow::anyhow!(
+                            "Node not found while moving children: {}",
+                            child_id
+                        ))
+                    }
+                };
+                if actual != *expected_version {
+                    return Ok(Err(VersionConflict {
+                        node_id: child_id,
+                        expected: *expected_version,
+                        actual,
+                    }));
                 }
+                return Err(anyhow::anyhow!(
+                    "Node '{}' has no parent edge to replace",
+                    child_id
+                ));
             }
 
             // Insert new has_child edge under new_parent_id.
@@ -2931,7 +2962,7 @@ impl SqliteStore {
             .context("Failed to insert new has_child edge")?;
         }
 
-        Ok(orders)
+        Ok(Ok(orders))
     }
 
     pub async fn get_schema(&self, node_type: &str) -> Result<Option<Value>> {
@@ -5341,7 +5372,8 @@ mod large_subtree_chunking_tests {
         // subtree nodes" without deleting anything.
         let (existed, deleted_nodes) = store
             .delete_subtree_atomic(&root.id, root.version, &subtree_ids, None)
-            .await?;
+            .await?
+            .expect("target is at root.version");
 
         assert!(existed);
         assert_eq!(deleted_nodes.len(), DESCENDANT_COUNT + 1);
@@ -5368,7 +5400,8 @@ mod large_subtree_chunking_tests {
         // call's chunks committed together as one unit.
         let (existed_again, _) = store
             .delete_subtree_atomic(&root.id, root.version, &subtree_ids, None)
-            .await?;
+            .await?
+            .expect("target is at root.version");
         assert!(!existed_again);
 
         Ok(())
@@ -5442,7 +5475,8 @@ mod large_subtree_chunking_tests {
 
         let (existed, deleted_nodes) = store
             .delete_subtree_atomic(&root.id, root.version, &subtree_ids, None)
-            .await?;
+            .await?
+            .expect("target is at root.version");
 
         assert!(existed);
         assert_eq!(deleted_nodes.len(), MULTI_CHUNK_COUNT + 1);

packages/core/src/services/error.rs

@@ -173,6 +173,12 @@ pub enum NodeServiceError {
     },
 }
 
+impl From<crate::db::VersionConflict> for NodeServiceError {
+    fn from(c: crate::db::VersionConflict) -> Self {
+        Self::version_conflict(c.node_id, c.expected, c.actual)
+    }
+}
+
 impl NodeServiceError {
     /// Create a node not found error
     pub fn node_not_found(id: impl Into<String>) -> Self {

packages/core/src/services/node_service/crud.rs

@@ -1935,19 +1935,7 @@ impl NodeService {
                 self.client_id.clone(),
             )
             .await
-            .map_err(|e| {
-                let msg = e.to_string();
-                // Parse version_conflict sentinel: "version_conflict:<id>:<expected>:<actual>"
-                if let Some(rest) = msg.strip_prefix("version_conflict:") {
-                    let parts: Vec<&str> = rest.splitn(3, ':').collect();
-                    if parts.len() == 3 {
-                        let exp: i64 = parts[1].parse().unwrap_or(expected_version);
-                        let act: i64 = parts[2].parse().unwrap_or(0);
-                        return NodeServiceError::version_conflict(node_id, exp, act);
-                    }
-                }
-                NodeServiceError::query_failed(msg)
-            })?;
+            .map_err(|e| NodeServiceError::query_failed(e.to_string()))??;
 
         if !existed {
             return Ok(crate::models::DeleteResult {

packages/core/src/services/node_service/hierarchy.rs

@@ -713,9 +713,9 @@ impl NodeService {
                 )));
             }
 
-            // Root nodes have no has_child edge — the in-transaction DELETE would
-            // return 0 changes and be misidentified as a version conflict. Reject
-            // root nodes explicitly so callers get a clear InvalidParent error.
+            // Root nodes have no has_child edge to replace, so the in-transaction
+            // swap would fail with a generic store error. Reject them here so
+            // callers get a clear hierarchy-violation error before any write.
             let Some(former_parent) = self.get_parent(node_id).await?.map(|p| p.id) else {
                 return Err(NodeServiceError::hierarchy_violation(format!(
                     "Root node '{}' cannot be batch-moved (no parent edge to replace)",
@@ -752,24 +752,16 @@ impl NodeService {
                         .iter()
                         .map(|(id, ver)| (id.as_str(), *ver))
                         .collect();
+                    // The second `?` turns a `VersionConflict` into this
+                    // closure's `Err`, which is what rolls back the edges
+                    // already swapped for earlier children.
                     let orders = crate::db::SqliteStore::move_children_to_parent_in_tx(
                         tx.store_tx(),
                         &new_parent_id,
                         &children_with_versions,
                     )
                     .await
-                    .map_err(|e| {
-                        let msg = e.to_string();
-                        // The store embeds the node ID in the error string:
-                        // "VERSION_CONFLICT: node '<id>' ...". Parse it out so
-                        // the caller gets an actionable conflict message.
-                        if let Some(rest) = msg.strip_prefix("VERSION_CONFLICT: node '") {
-                            let node_id = rest.split('\'').next().unwrap_or("unknown");
-                            NodeServiceError::version_conflict(node_id, 0, 0)
-                        } else {
-                            NodeServiceError::query_failed(msg)
-                        }
-                    })?;
+                    .map_err(|e| NodeServiceError::query_failed(e.to_string()))??;
 
                     let mut updated = Vec::with_capacity(nodes.len());
                     for ((node, order), former_parent) in