Let a sub-collection stay under its parent through a sync

The merge inserted every incoming collection with `parent_id = NULL` and
never set it on update, so the hierarchy flattened on each round trip: a
collection nested on one device came back from the server at the top
level. `r.parent_id` was selected and then not read.

The id could not be copied — row ids are local, and the remote's integer
names a different collection here, or none. So carry the parent's uuid
and resolve it locally, in a second pass: rows arrive in whatever order
the query returns, and a child can precede its parent.

Guard the resolution against cycles. Each tree is acyclic alone, but the
union need not be — we may hold A above B while the remote holds B above
A — and closing that loop would make every tree walk spin.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

core/dr-catalog/src/merge.rs

@@ -121,17 +121,24 @@ pub fn merge_collections(conn: &Connection) -> Result<MergeReport, CatalogError>
     // ---- collections ------------------------------------------------------
     {
         let mut stmt = tx.prepare(
-            "SELECT r.uuid, r.name, r.parent_id, r.kind, r.selector_json,
+            // The parent arrives as a *uuid*, not `r.parent_id`: row ids are
+            // local to a catalog, so the remote's integer means nothing here.
+            "SELECT r.uuid, r.name, rp.uuid, r.kind, r.selector_json,
                     r.created, r.revision, r.modified, r.deleted,
                     l.revision, l.modified
                FROM remote_cat.collections r
-               LEFT JOIN main.collections l ON l.uuid = r.uuid",
+               LEFT JOIN main.collections l ON l.uuid = r.uuid
+               LEFT JOIN remote_cat.collections rp ON rp.id = r.parent_id",
         )?;
 
         struct Incoming {
             uuid: String,
             name: String,
             kind: i64,
+            /// The parent's uuid, resolved to a local row id once every
+            /// incoming collection exists — a child can arrive before its
+            /// parent, so this cannot be applied inline.
+            parent_uuid: Option<String>,
             selector_json: Option<String>,
             created: i64,
             revision: i64,
@@ -152,6 +159,7 @@ pub fn merge_collections(conn: &Connection) -> Result<MergeReport, CatalogError>
                     uuid: r.get(0)?,
                     name: r.get(1)?,
                     kind: r.get(3)?,
+                    parent_uuid: r.get(2)?,
                     selector_json: r.get(4)?,
                     created: r.get(5)?,
                     revision,
@@ -161,6 +169,11 @@ pub fn merge_collections(conn: &Connection) -> Result<MergeReport, CatalogError>
             })?
             .collect::<Result<_, _>>()?;
 
+        // Parentage is applied after the loop: a child can arrive before its
+        // parent, so resolving the uuid inline would find nothing and silently
+        // flatten the tree.
+        let mut reparent: Vec<(String, Option<String>)> = Vec::new();
+
         for row in rows {
             match row.verdict {
                 MergeVerdict::KeptLocal => {
@@ -182,6 +195,7 @@ pub fn merge_collections(conn: &Connection) -> Result<MergeReport, CatalogError>
                             row.modified,
                         ],
                     )?;
+                    reparent.push((row.uuid.clone(), row.parent_uuid.clone()));
                     report.inserted += 1;
                 }
                 MergeVerdict::UpdatedFromRemote => {
@@ -199,6 +213,7 @@ pub fn merge_collections(conn: &Connection) -> Result<MergeReport, CatalogError>
                             row.modified,
                         ],
                     )?;
+                    reparent.push((row.uuid.clone(), row.parent_uuid.clone()));
                     report.updated += 1;
                 }
                 MergeVerdict::DeletedByRemote => {
@@ -228,6 +243,48 @@ pub fn merge_collections(conn: &Connection) -> Result<MergeReport, CatalogError>
                 }
             }
         }
+
+        // Second pass: every incoming collection now exists locally, so a
+        // parent uuid can be resolved to a row id. A parent we have never seen
+        // resolves to NULL, which leaves the collection at the top level —
+        // wrong, but visible and recoverable, where a dangling id would not be.
+        //
+        // Without this the tree flattened on every sync: `parent_id` is a local
+        // row id and was written as NULL rather than translated, so a nested
+        // collection came back from a round trip at the top level.
+        for (uuid, parent_uuid) in reparent {
+            // Remote's tree is acyclic and so is ours, but the union of the two
+            // need not be: if we hold A above B and the remote holds B above A,
+            // applying only the winning half closes a loop, and `descendants`
+            // would then spin. Walk up from the proposed parent first; reaching
+            // the collection itself means this edge would close a cycle, so the
+            // safe move is to leave it where it is.
+            if let Some(ref parent) = parent_uuid {
+                let closes_cycle: bool = tx.query_row(
+                    "WITH RECURSIVE up(id) AS (
+                         SELECT id FROM main.collections WHERE uuid = ?2
+                         UNION
+                         SELECT c.parent_id FROM main.collections c
+                           JOIN up ON c.id = up.id
+                          WHERE c.parent_id IS NOT NULL
+                     )
+                     SELECT EXISTS(
+                         SELECT 1 FROM up
+                          WHERE id = (SELECT id FROM main.collections WHERE uuid = ?1))",
+                    rusqlite::params![uuid, parent],
+                    |r| r.get(0),
+                )?;
+                if closes_cycle {
+                    continue;
+                }
+            }
+            tx.execute(
+                "UPDATE main.collections
+                    SET parent_id = (SELECT id FROM main.collections WHERE uuid = ?2)
+                  WHERE uuid = ?1",
+                rusqlite::params![uuid, parent_uuid],
+            )?;
+        }
     }
 
     // ---- membership -------------------------------------------------------
@@ -374,6 +431,93 @@ mod tests {
         .unwrap();
     }
 
+    /// The parent's uuid for `uuid`, or None if it sits at the top level.
+    fn parent_of(c: &Connection, uuid: &str) -> Option<String> {
+        c.query_row(
+            "SELECT p.uuid FROM main.collections ch
+               JOIN main.collections p ON p.id = ch.parent_id
+              WHERE ch.uuid = ?1",
+            [uuid],
+            |r| r.get(0),
+        )
+        .ok()
+    }
+
+    #[test]
+    fn a_nested_collection_arrives_still_nested() {
+        // The tree used to flatten on every sync: `parent_id` is a local row id,
+        // and the merge wrote NULL rather than translating it through the uuid.
+        let c = two_catalogs();
+        add_collection(&c, "remote_cat", 1, "uuid-holidays", "holidays", 1);
+        add_collection(&c, "remote_cat", 2, "uuid-arosa", "arosa", 1);
+        c.execute(
+            "UPDATE remote_cat.collections SET parent_id = 1 WHERE id = 2",
+            [],
+        )
+        .unwrap();
+
+        merge_collections(&c).unwrap();
+
+        assert_eq!(
+            parent_of(&c, "uuid-arosa").as_deref(),
+            Some("uuid-holidays"),
+            "a sub-collection must not be promoted to the top level by a merge"
+        );
+    }
+
+    #[test]
+    fn a_parent_arriving_after_its_child_still_adopts_it() {
+        // Row order is whatever the query returns, so the child can be inserted
+        // first. Parentage is applied in a second pass for exactly this reason.
+        let c = two_catalogs();
+        // Child has the lower id, so it is seen before the parent exists.
+        add_collection(&c, "remote_cat", 1, "uuid-child", "arosa", 1);
+        add_collection(&c, "remote_cat", 2, "uuid-parent", "holidays", 1);
+        c.execute(
+            "UPDATE remote_cat.collections SET parent_id = 2 WHERE id = 1",
+            [],
+        )
+        .unwrap();
+
+        merge_collections(&c).unwrap();
+
+        assert_eq!(
+            parent_of(&c, "uuid-child").as_deref(),
+            Some("uuid-parent"),
+            "resolution must not depend on the order rows happen to arrive in"
+        );
+    }
+
+    #[test]
+    fn disagreeing_devices_cannot_close_a_parent_cycle() {
+        // Local holds A above B; the remote holds B above A and wins on
+        // revision. Applying that blindly makes each the other's parent, and
+        // every tree walk then spins.
+        let c = two_catalogs();
+        add_collection(&c, "main", 1, "uuid-a", "A", 1);
+        add_collection(&c, "main", 2, "uuid-b", "B", 1);
+        c.execute("UPDATE main.collections SET parent_id = 1 WHERE id = 2", [])
+            .unwrap();
+
+        // Remote: A under B, at a higher revision so it is the winner.
+        add_collection(&c, "remote_cat", 1, "uuid-a", "A", 9);
+        add_collection(&c, "remote_cat", 2, "uuid-b", "B", 9);
+        c.execute(
+            "UPDATE remote_cat.collections SET parent_id = 2 WHERE id = 1",
+            [],
+        )
+        .unwrap();
+
+        merge_collections(&c).unwrap();
+
+        let a = parent_of(&c, "uuid-a");
+        let b = parent_of(&c, "uuid-b");
+        assert!(
+            !(a.as_deref() == Some("uuid-b") && b.as_deref() == Some("uuid-a")),
+            "merge closed a cycle: A parented under B and B under A"
+        );
+    }
+
     #[test]
     fn disjoint_collections_from_two_devices_both_survive() {
         // The property the whole design exists for: neither device loses work.