feat(apply): §6 M2 — structural-divergence detection + declarative reparent

With the declared parent on the wire (M1), `apply_tree` now compares each
EXISTING group node's server parent to the manifest's (directory-nesting)
parent and resolves a divergence per a request `StructureMode` (default
`Refuse` — a pre-M2 CLI never reshapes the tree):

- `Refuse` → `ApplyError::StructuralDivergence` (422), naming the flags.
- `ForceLocal` → reparent to the declared parent IN the apply tx via the
  extracted `group_repo::reparent_group_tx` (cycle guard + structure_version
  bump), §5.6-gated (`GroupAdmin` on node + source + destination) and
  attach-ceilinged.
- `AdoptServer` → keep the server shape, reconcile content in place.

M1's `create_missing_groups_tx` generalized to `reconcile_group_structure_tx`
(create + reparent share the coarse-lock / attach-ceiling / RBAC path; both are
recorded `structurally_changed` so the pool-based attach re-check skips their
uncommitted rows). `pic plan` previews the drift (`divergence_preview` → a
`structure` row naming server + manifest parents). CLI:
`--force-local-structure` / `--adopt-server-structure` (mutually exclusive) →
the `structure_mode` wire field; the 422 surfaces verbatim. A concurrent
`pic groups reparent` between plan and apply still trips `StateMoved` (the tree
token already folds structure_version).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/apply_api.rs

@@ -19,7 +19,7 @@ use crate::app_repo::AppRepository;
 use crate::apply_service::{
     ApplyError, ApplyOwner, ApplyReport, ApplyService, Bundle, BundleTrigger, CollectionInfo,
     ExtensionPointInfo, NodeKind, OwnershipClaim, PlanResult, ProjectDecl, RouteTemplateInfo,
-    SuppressionInfo, TreeBundle, TreePlanResult, TriggerTemplateInfo,
+    StructureMode, SuppressionInfo, TreeBundle, TreePlanResult, TriggerTemplateInfo,
 };
 use crate::authz::{require, AuthzDenied, Capability};
 use crate::group_repo::GroupRepository;
@@ -331,6 +331,10 @@ struct TreeApplyRequest {
     project: Option<ProjectDecl>,
     #[serde(default)]
     takeover: bool,
+    /// §6 M2: how to resolve a group whose server parent diverges from the
+    /// manifest (default `Refuse`; a pre-M2 CLI omits it).
+    #[serde(default)]
+    structure_mode: StructureMode,
 }
 
 #[derive(Deserialize)]
@@ -370,6 +374,7 @@ async fn tree_apply_handler(
             req.prune,
             &claim,
             req.expected_token.as_deref(),
+            req.structure_mode,
         )
         .await?;
     Ok(Json(report))
@@ -571,7 +576,7 @@ impl IntoResponse for ApplyError {
     fn into_response(self) -> Response {
         let (status, body) = match &self {
             Self::AppNotFound(_) => (StatusCode::NOT_FOUND, json!({ "error": self.to_string() })),
-            Self::Invalid(_) | Self::OutsideAttachPoint(_) => (
+            Self::Invalid(_) | Self::OutsideAttachPoint(_) | Self::StructuralDivergence(_) => (
                 StatusCode::UNPROCESSABLE_ENTITY,
                 json!({ "error": self.to_string() }),
             ),

crates/manager-core/src/apply_service.rs

@@ -614,6 +614,34 @@ fn preview_ownership(
     }
 }
 
+/// §6 M2: how a tree apply resolves a group whose SERVER parent differs from the
+/// manifest's declared (directory-nesting) parent — Terraform-style
+/// detect-and-refuse. Request-only; `Refuse` is the default (a pre-M2 CLI omits
+/// it), so a divergence never silently reshapes the tree.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum StructureMode {
+    /// Refuse the apply on any structural divergence (422).
+    #[default]
+    Refuse,
+    /// Reparent each diverged group to the manifest's declared parent.
+    ForceLocal,
+    /// Accept the server's shape; reconcile content in place, no reparent.
+    AdoptServer,
+}
+
+/// §6 M2: a group node's structural state under the manifest — `in-sync` when
+/// the server parent matches the declared (directory-nesting) parent, else
+/// `diverged` (naming both parents). Previewed by `pic plan`.
+#[derive(Debug, Clone, Serialize)]
+pub struct StructurePreview {
+    pub status: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub server_parent: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub manifest_parent: Option<String>,
+}
+
 /// One node's diff within a tree plan.
 #[derive(Debug, Clone, Serialize)]
 pub struct NodePlan {
@@ -624,6 +652,9 @@ pub struct NodePlan {
     /// §7 M3: this node's ownership outcome under the tree's `[project]`.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub ownership: Option<OwnershipPreview>,
+    /// §6 M2: for a GROUP node, its structural state (in-sync / diverged).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub structure: Option<StructurePreview>,
 }
 
 /// What `plan_tree` returns: a per-node diff plus ONE combined bound-plan token
@@ -746,6 +777,11 @@ pub enum ApplyError {
     /// attach point — you can only apply within its subtree. Maps to 422.
     #[error("outside attach point: {0}")]
     OutsideAttachPoint(String),
+    /// §6 M2: a group's SERVER parent differs from the manifest's declared
+    /// (directory-nesting) parent, and no `--force-local-structure` /
+    /// `--adopt-server-structure` was given. Maps to 422 — actionable.
+    #[error("structural divergence: {0}")]
+    StructuralDivergence(String),
     #[error("authorization repo error: {0}")]
     AuthzRepo(String),
     #[error("backend: {0}")]
@@ -1656,11 +1692,15 @@ impl ApplyService {
         let mut nodes = Vec::with_capacity(prepared.len() + to_create.len());
         for p in prepared {
             let ownership = self.ownership_preview(p.owner, declared).await?;
+            // §6 M2: for an existing group node, preview a structural divergence
+            // (server parent ≠ manifest parent) before apply.
+            let structure = self.divergence_preview(p.node).await?;
             nodes.push(NodePlan {
                 kind: p.node.kind,
                 slug: p.node.slug.clone(),
                 plan: p.plan,
                 ownership,
+                structure,
             });
         }
         // §6: a to-create group is previewed with a full-create plan; if the
@@ -1672,6 +1712,7 @@ impl ApplyService {
                 slug: c.node.slug.clone(),
                 plan: c.plan,
                 ownership,
+                structure: None,
             });
         }
         Ok(TreePlanResult {
@@ -1697,6 +1738,7 @@ impl ApplyService {
         prune: bool,
         claim: &OwnershipClaim<'_>,
         expected_token: Option<&str>,
+        structure_mode: StructureMode,
     ) -> Result<ApplyReport, ApplyError> {
         let actor = claim.principal.user_id;
         let mut tx = self
@@ -1704,15 +1746,23 @@ impl ApplyService {
             .begin()
             .await
             .map_err(|e| ApplyError::Backend(e.to_string()))?;
-        // §6 Phase 0: create every group the manifest declares (by directory
-        // nesting) but the server lacks, IN this tx (parents-first), returning
-        // the complete slug → id map. Enforces the attach-point ceiling + RBAC
-        // per create and takes the coarse structural lock only when it creates.
+        // §6 Phase 0: reconcile the group tree SHAPE — create every group the
+        // manifest declares (by directory nesting) but the server lacks, and
+        // (M2) reparent a diverged group per `structure_mode` — IN this tx
+        // (parents-first), returning the complete slug → id map and the set of
+        // structurally-changed groups. Enforces the attach ceiling + RBAC per
+        // mutation and takes the coarse structural lock only when it mutates.
         // With all groups now resolvable, `prepare_tree` sees the whole tree
         // (created groups load an empty current → a full-create plan), so its
         // to-create list is empty.
-        let (resolved, created) = self
-            .create_missing_groups_tx(&mut tx, bundle, claim.project.as_ref(), claim.principal)
+        let (resolved, structurally_changed) = self
+            .reconcile_group_structure_tx(
+                &mut tx,
+                bundle,
+                claim.project.as_ref(),
+                claim.principal,
+                structure_mode,
+            )
             .await?;
         let (prepared, to_create, token) = self.prepare_tree(bundle, &resolved).await?;
         if !to_create.is_empty() {
@@ -1737,7 +1787,7 @@ impl ApplyService {
             let pg = self.resolve_group(pg_slug).await?;
             for p in &prepared {
                 if let ApplyOwner::Group(gid) = p.owner {
-                    if created.contains(&gid) {
+                    if structurally_changed.contains(&gid) {
                         continue;
                     }
                 }
@@ -1883,26 +1933,122 @@ impl ApplyService {
         Ok(map)
     }
 
-    /// §6: create every GROUP node the manifest declares but the server lacks,
-    /// INSIDE the apply tx (parents-first, from the CLI's directory ordering).
-    /// Returns the complete slug → id map (existing + created) so `prepare_tree`
-    /// resolves the whole tree. Enforces the attach-point ceiling + RBAC on each
-    /// create; claiming happens in Phase A alongside existing groups. The coarse
-    /// structural lock is taken only once, on the first create.
+    /// §6 M2: preview an EXISTING group node's structural divergence — `Some`
+    /// only when the server's parent differs from the manifest's declared
+    /// (directory-nesting) parent. `None` for an app node, a to-create group, or
+    /// an in-sync group.
+    async fn divergence_preview(
+        &self,
+        node: &TreeNode,
+    ) -> Result<Option<StructurePreview>, ApplyError> {
+        if node.kind != NodeKind::Group {
+            return Ok(None);
+        }
+        let Some(group) = self
+            .groups
+            .get_by_slug(&node.slug)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?
+        else {
+            return Ok(None); // to-create — not a divergence
+        };
+        let server_parent = match group.parent_id {
+            Some(pid) => self
+                .groups
+                .get_by_id(pid)
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?
+                .map(|g| g.slug),
+            None => None,
+        };
+        if server_parent.as_deref() == node.parent.as_deref() {
+            return Ok(None); // in-sync
+        }
+        Ok(Some(StructurePreview {
+            status: "diverged".to_string(),
+            server_parent,
+            manifest_parent: node.parent.clone(),
+        }))
+    }
+
+    /// §6: resolve a group node's DECLARED (directory-nesting) parent slug to a
+    /// `GroupId` — an in-tree group resolved earlier this tree wins, else a
+    /// committed server group; `None` = instance root. Errors if a declared
+    /// parent slug resolves to nothing.
+    async fn resolve_declared_parent(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        map: &HashMap<String, GroupId>,
+        n: &TreeNode,
+    ) -> Result<Option<GroupId>, ApplyError> {
+        let Some(pslug) = &n.parent else {
+            return Ok(None);
+        };
+        if let Some(&g) = map.get(&pslug.to_lowercase()) {
+            return Ok(Some(g));
+        }
+        crate::group_repo::read_group_id_by_slug_tx(tx, pslug)
+            .await
+            .map_err(map_group_repo_err)?
+            .map(Some)
+            .ok_or_else(|| {
+                ApplyError::Invalid(format!(
+                    "group `{}` declares parent `{pslug}`, which does not exist",
+                    n.slug
+                ))
+            })
+    }
+
+    /// §6: is `parent` within the attach-point subtree? `attach = None` (no
+    /// ceiling) → always true. A just-structurally-changed group in `ok` (its
+    /// own placement already validated within) short-circuits the ancestor walk.
+    async fn parent_within_attach(
+        &self,
+        parent: Option<GroupId>,
+        attach: Option<GroupId>,
+        ok: &HashSet<GroupId>,
+    ) -> Result<bool, ApplyError> {
+        let Some(attach) = attach else {
+            return Ok(true);
+        };
+        Ok(match parent {
+            Some(p) if p == attach || ok.contains(&p) => true,
+            Some(p) => self
+                .groups
+                .ancestors(p)
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?
+                .iter()
+                .any(|g| g.id == attach),
+            None => false,
+        })
+    }
+
+    /// §6 M1+M2: reconcile the group tree SHAPE inside the apply tx (parents-
+    /// first). A group the server lacks is CREATED; a group whose server parent
+    /// differs from the manifest's declared parent is a structural divergence,
+    /// resolved per `mode` (refuse / reparent / adopt-server). Returns the
+    /// complete slug → id map (so `prepare_tree` resolves the whole tree) and the
+    /// set of groups STRUCTURALLY changed here (created or reparented) — the
+    /// caller skips them in the pool-based attach re-check (their uncommitted row
+    /// isn't visible). Enforces the attach ceiling + RBAC per mutation; the
+    /// coarse structural lock is taken only once, on the first mutation.
     #[allow(clippy::too_many_lines)]
-    async fn create_missing_groups_tx(
+    async fn reconcile_group_structure_tx(
         &self,
         tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
         bundle: &TreeBundle,
         project: Option<&ProjectDecl>,
         principal: &Principal,
+        mode: StructureMode,
     ) -> Result<(HashMap<String, GroupId>, HashSet<GroupId>), ApplyError> {
         use crate::group_repo::{
-            create_group_tx, read_group_id_by_slug_tx, GROUP_STRUCTURAL_LOCK_KEY,
+            create_group_tx, read_group_id_by_slug_tx, read_group_node_tx, reparent_group_tx,
+            GROUP_STRUCTURAL_LOCK_KEY,
         };
 
-        // The attach-point ceiling group (if the repo declares one).
-        let attach_gid = match project.and_then(|p| p.parent_group.as_deref()) {
+        let attach_slug = project.and_then(|p| p.parent_group.as_deref());
+        let attach_gid = match attach_slug {
             Some(slug) => Some(
                 read_group_id_by_slug_tx(tx, slug)
                     .await
@@ -1915,101 +2061,125 @@ impl ApplyService {
         };
 
         let mut map: HashMap<String, GroupId> = HashMap::new();
-        let mut created: HashSet<GroupId> = HashSet::new();
-        let mut created_within_attach: HashSet<GroupId> = HashSet::new();
+        let mut changed: HashSet<GroupId> = HashSet::new();
+        let mut within_attach: HashSet<GroupId> = HashSet::new();
         let mut locked = false;
         for n in &bundle.nodes {
             if n.kind != NodeKind::Group {
                 continue;
             }
             let key = n.slug.to_lowercase();
-            if let Some(gid) = read_group_id_by_slug_tx(tx, &n.slug)
+            let mut existing = read_group_node_tx(tx, &n.slug)
                 .await
-                .map_err(map_group_repo_err)?
-            {
-                map.insert(key, gid);
-                continue;
-            }
-            // The group must be CREATED. Serialize creation with other structural
-            // mutations under the coarse lock (acquired once, on first create),
-            // re-reading after the lock in case a racer created it first.
-            if !locked {
+                .map_err(map_group_repo_err)?;
+            let declared_parent = self.resolve_declared_parent(tx, &map, n).await?;
+
+            // A structural change (create or reparent) is needed if the group is
+            // missing, or exists with a divergent parent under ForceLocal.
+            let diverged = matches!(&existing, Some((_, sp)) if *sp != declared_parent);
+            let needs_mutation =
+                existing.is_none() || (diverged && mode == StructureMode::ForceLocal);
+            if needs_mutation && !locked {
                 sqlx::query("SELECT pg_advisory_xact_lock($1)")
                     .bind(GROUP_STRUCTURAL_LOCK_KEY)
                     .execute(&mut **tx)
                     .await
                     .map_err(|e| ApplyError::Backend(e.to_string()))?;
                 locked = true;
-                if let Some(gid) = read_group_id_by_slug_tx(tx, &n.slug)
+                // Re-read under the lock — a racer may have created/moved it.
+                existing = read_group_node_tx(tx, &n.slug)
                     .await
-                    .map_err(map_group_repo_err)?
-                {
-                    map.insert(key, gid);
-                    continue;
-                }
+                    .map_err(map_group_repo_err)?;
             }
-            crate::groups_api::validate_slug_format(&n.slug).map_err(ApplyError::Invalid)?;
-            // Resolve the declared parent: an in-tree group (created/existing
-            // earlier this tree) wins, else a committed server group; `None` =
-            // instance root.
-            let parent_id = match &n.parent {
-                Some(pslug) => Some(match map.get(&pslug.to_lowercase()) {
-                    Some(&g) => g,
-                    None => read_group_id_by_slug_tx(tx, pslug)
-                        .await
-                        .map_err(map_group_repo_err)?
-                        .ok_or_else(|| {
-                            ApplyError::Invalid(format!(
-                                "group `{}` declares parent `{pslug}`, which does not exist",
-                                n.slug
-                            ))
-                        })?,
-                }),
-                None => None,
-            };
-            // Attach-point ceiling: the new group must land within the subtree.
-            if let Some(attach) = attach_gid {
-                let within = match parent_id {
-                    Some(p) if p == attach || created_within_attach.contains(&p) => true,
-                    Some(p) => self
-                        .groups
-                        .ancestors(p)
-                        .await
-                        .map_err(|e| ApplyError::Backend(e.to_string()))?
-                        .iter()
-                        .any(|g| g.id == attach),
-                    None => false,
-                };
-                if !within {
+
+            let Some((gid, server_parent)) = existing else {
+                // Create the missing group under its declared parent.
+                crate::groups_api::validate_slug_format(&n.slug).map_err(ApplyError::Invalid)?;
+                if !self
+                    .parent_within_attach(declared_parent, attach_gid, &within_attach)
+                    .await?
+                {
                     return Err(ApplyError::OutsideAttachPoint(format!(
                         "group `{}` would be created outside this project's attach point `{}`",
                         n.slug,
-                        project
-                            .and_then(|p| p.parent_group.as_deref())
-                            .unwrap_or_default()
+                        attach_slug.unwrap_or_default()
                     )));
                 }
-            }
-            // RBAC: a root group needs InstanceCreateGroup; a subgroup needs
-            // GroupAdmin on the parent (mirrors groups_api::create_group).
-            let cap = match parent_id {
-                Some(p) => Capability::GroupAdmin(p),
-                None => Capability::InstanceCreateGroup,
-            };
-            require(self.authz.as_ref(), principal, cap)
-                .await
-                .map_err(map_authz)?;
-            let name = n.name.clone().unwrap_or_else(|| n.slug.clone());
-            let gid = create_group_tx(tx, &n.slug, &name, n.description.as_deref(), parent_id)
+                // RBAC: a root group needs InstanceCreateGroup; a subgroup needs
+                // GroupAdmin on the parent (mirrors create_group).
+                let cap = match declared_parent {
+                    Some(p) => Capability::GroupAdmin(p),
+                    None => Capability::InstanceCreateGroup,
+                };
+                require(self.authz.as_ref(), principal, cap)
+                    .await
+                    .map_err(map_authz)?;
+                let name = n.name.clone().unwrap_or_else(|| n.slug.clone());
+                let gid = create_group_tx(
+                    tx,
+                    &n.slug,
+                    &name,
+                    n.description.as_deref(),
+                    declared_parent,
+                )
                 .await
                 .map_err(map_group_repo_err)?;
-            created.insert(gid);
-            if attach_gid.is_some() {
-                created_within_attach.insert(gid);
-            }
+                changed.insert(gid);
+                if attach_gid.is_some() {
+                    within_attach.insert(gid);
+                }
+                map.insert(key, gid);
+                continue;
+            };
+
+            // The group exists. If its server parent matches the manifest, it's
+            // in-sync; otherwise resolve the divergence per `mode`.
             map.insert(key, gid);
+            if server_parent == declared_parent {
+                continue;
+            }
+            match mode {
+                StructureMode::AdoptServer => {} // keep the server shape
+                StructureMode::Refuse => {
+                    return Err(ApplyError::StructuralDivergence(format!(
+                        "group `{}` is under a different parent on the server than the manifest \
+                         declares — pass --force-local-structure to reparent it, or \
+                         --adopt-server-structure to keep the server's placement",
+                        n.slug
+                    )));
+                }
+                StructureMode::ForceLocal => {
+                    // §5.6 RBAC: admin on the node + source + destination.
+                    require(self.authz.as_ref(), principal, Capability::GroupAdmin(gid))
+                        .await
+                        .map_err(map_authz)?;
+                    for side in [server_parent, declared_parent].into_iter().flatten() {
+                        require(self.authz.as_ref(), principal, Capability::GroupAdmin(side))
+                            .await
+                            .map_err(map_authz)?;
+                    }
+                    if !self
+                        .parent_within_attach(declared_parent, attach_gid, &within_attach)
+                        .await?
+                    {
+                        return Err(ApplyError::OutsideAttachPoint(format!(
+                            "reparenting group `{}` would move it outside this project's attach \
+                             point `{}`",
+                            n.slug,
+                            attach_slug.unwrap_or_default()
+                        )));
+                    }
+                    reparent_group_tx(tx, gid, declared_parent)
+                        .await
+                        .map_err(map_group_repo_err)?;
+                    changed.insert(gid);
+                    if attach_gid.is_some() {
+                        within_attach.insert(gid);
+                    }
+                }
+            }
         }
-        Ok((map, created))
+        Ok((map, changed))
     }
 
     /// Resolve nodes to owners, validate each (apps' route/trigger targets may

crates/manager-core/src/group_repo.rs

@@ -87,6 +87,69 @@ pub(crate) async fn create_group_tx(
     }
 }
 
+/// Reparent a group INSIDE an existing tx (the caller holds
+/// [`GROUP_STRUCTURAL_LOCK_KEY`]): the ancestor-walk cycle guard + the
+/// `parent_id` write + `structure_version` bump. Used by the imperative
+/// `reparent` (its own tx) and §6 M2's declarative structural reconcile (the
+/// apply tx). A same-tx just-created ancestor is visible to the cycle walk.
+pub(crate) async fn reparent_group_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: GroupId,
+    new_parent: Option<GroupId>,
+) -> Result<Group, GroupRepositoryError> {
+    if let Some(parent) = new_parent {
+        if parent == id {
+            return Err(GroupRepositoryError::Conflict(
+                "a group cannot be its own parent".into(),
+            ));
+        }
+        // Cycle guard: walk from the destination up to the root; if we reach
+        // `id`, the move would place `id` beneath itself.
+        let mut cursor = Some(parent);
+        let mut hops = 0u32;
+        while let Some(node) = cursor {
+            if node == id {
+                return Err(GroupRepositoryError::Conflict(
+                    "cannot reparent a group beneath one of its own descendants".into(),
+                ));
+            }
+            hops += 1;
+            if hops > 64 {
+                return Err(GroupRepositoryError::Conflict(
+                    "group ancestry exceeds the maximum depth".into(),
+                ));
+            }
+            let parent_of: Option<(Option<Uuid>,)> =
+                sqlx::query_as("SELECT parent_id FROM groups WHERE id = $1")
+                    .bind(node.into_inner())
+                    .fetch_optional(&mut **tx)
+                    .await?;
+            match parent_of {
+                Some((p,)) => cursor = p.map(GroupId::from),
+                None => {
+                    return Err(GroupRepositoryError::Conflict(
+                        "destination parent group does not exist".into(),
+                    ));
+                }
+            }
+        }
+    }
+    let row = sqlx::query_as::<_, GroupRow>(&format!(
+        "UPDATE groups SET \
+             parent_id = $2, \
+             structure_version = structure_version + 1, \
+             updated_at = NOW() \
+         WHERE id = $1 \
+         RETURNING {GROUP_COLS}"
+    ))
+    .bind(id.into_inner())
+    .bind(new_parent.map(GroupId::into_inner))
+    .fetch_optional(&mut **tx)
+    .await?;
+    row.map(Into::into)
+        .ok_or(GroupRepositoryError::NotFound(id))
+}
+
 /// §6: resolve a group slug → id inside the apply tx, so a group CREATED earlier
 /// in the same transaction is visible (a pool read would miss the uncommitted
 /// row). `None` if absent.
@@ -101,6 +164,21 @@ pub(crate) async fn read_group_id_by_slug_tx(
     Ok(row.map(|(id,)| id.into()))
 }
 
+/// §6 M2: read a group's `(id, parent_id)` by slug inside the tx — the input to
+/// the structural-divergence check (server parent vs manifest parent). `None`
+/// if the group doesn't exist (it is to-create).
+pub(crate) async fn read_group_node_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    slug: &str,
+) -> Result<Option<(GroupId, Option<GroupId>)>, GroupRepositoryError> {
+    let row: Option<(Uuid, Option<Uuid>)> =
+        sqlx::query_as("SELECT id, parent_id FROM groups WHERE slug = $1")
+            .bind(slug)
+            .fetch_optional(&mut **tx)
+            .await?;
+    Ok(row.map(|(id, parent)| (id.into(), parent.map(GroupId::from))))
+}
+
 #[async_trait]
 pub trait GroupRepository: Send + Sync {
     /// Every group on the instance, ordered by name. The tree is small
@@ -324,63 +402,9 @@ impl GroupRepository for PostgresGroupRepository {
             .bind(GROUP_STRUCTURAL_LOCK_KEY)
             .execute(&mut *tx)
             .await?;
-
-        if let Some(parent) = new_parent {
-            if parent == id {
-                return Err(GroupRepositoryError::Conflict(
-                    "a group cannot be its own parent".into(),
-                ));
-            }
-            // Cycle guard: walk from the destination up to the root; if we
-            // reach `id`, the move would place `id` beneath itself.
-            let mut cursor = Some(parent);
-            let mut hops = 0u32;
-            while let Some(node) = cursor {
-                if node == id {
-                    return Err(GroupRepositoryError::Conflict(
-                        "cannot reparent a group beneath one of its own descendants".into(),
-                    ));
-                }
-                hops += 1;
-                if hops > 64 {
-                    return Err(GroupRepositoryError::Conflict(
-                        "group ancestry exceeds the maximum depth".into(),
-                    ));
-                }
-                let parent_of: Option<(Option<Uuid>,)> =
-                    sqlx::query_as("SELECT parent_id FROM groups WHERE id = $1")
-                        .bind(node.into_inner())
-                        .fetch_optional(&mut *tx)
-                        .await?;
-                match parent_of {
-                    Some((p,)) => cursor = p.map(GroupId::from),
-                    // Destination parent doesn't exist.
-                    None => {
-                        return Err(GroupRepositoryError::Conflict(
-                            "destination parent group does not exist".into(),
-                        ));
-                    }
-                }
-            }
-        }
-
-        let row = sqlx::query_as::<_, GroupRow>(&format!(
-            "UPDATE groups SET \
-                 parent_id = $2, \
-                 structure_version = structure_version + 1, \
-                 updated_at = NOW() \
-             WHERE id = $1 \
-             RETURNING {GROUP_COLS}"
-        ))
-        .bind(id.into_inner())
-        .bind(new_parent.map(GroupId::into_inner))
-        .fetch_optional(&mut *tx)
-        .await?;
-        let Some(row) = row else {
-            return Err(GroupRepositoryError::NotFound(id));
-        };
+        let group = reparent_group_tx(&mut tx, id, new_parent).await?;
         tx.commit().await?;
-        Ok(row.into())
+        Ok(group)
     }
 
     async fn delete(&self, id: GroupId) -> Result<(), GroupRepositoryError> {

crates/picloud-cli/src/client.rs

@@ -1321,6 +1321,7 @@ impl Client {
 
     /// `POST /api/v1/admin/tree/apply` — reconcile a whole project tree in one
     /// transaction (Phase 5).
+    #[allow(clippy::too_many_arguments)]
     pub async fn apply_tree(
         &self,
         bundle: &serde_json::Value,
@@ -1328,6 +1329,7 @@ impl Client {
         project: Option<&crate::manifest::ManifestProject>,
         takeover: bool,
         expected_token: Option<&str>,
+        structure_mode: &str,
     ) -> Result<ApplyReportDto> {
         let body = serde_json::json!({
             "bundle": bundle,
@@ -1335,14 +1337,20 @@ impl Client {
             "expected_token": expected_token,
             "project": project,
             "takeover": takeover,
+            "structure_mode": structure_mode,
         });
         let resp = self
             .request(Method::POST, "/api/v1/admin/tree/apply")
             .json(&body)
             .send()
             .await?;
-        // 409 = stale bound plan OR a §7 ownership conflict; surface verbatim.
-        if resp.status() == reqwest::StatusCode::CONFLICT {
+        // 409 = stale bound plan OR a §7 ownership conflict; 422 = a §6 M2
+        // structural divergence (or an invalid bundle). Surface the server
+        // message verbatim — it names the resolution flags.
+        let status = resp.status();
+        if status == reqwest::StatusCode::CONFLICT
+            || status == reqwest::StatusCode::UNPROCESSABLE_ENTITY
+        {
             let body = resp.text().await.unwrap_or_default();
             let msg = parse_error_body(&body).unwrap_or(body);
             return Err(anyhow!("{msg}"));
@@ -1627,6 +1635,19 @@ pub struct NodePlanDto {
     /// §7 M3: this node's ownership outcome under the tree's `[project]`.
     #[serde(default)]
     pub ownership: Option<OwnershipPreviewDto>,
+    /// §6 M2: for a group node, its structural state (in-sync omitted; diverged
+    /// names the server + manifest parents).
+    #[serde(default)]
+    pub structure: Option<StructurePreviewDto>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct StructurePreviewDto {
+    pub status: String,
+    #[serde(default)]
+    pub server_parent: Option<String>,
+    #[serde(default)]
+    pub manifest_parent: Option<String>,
 }
 
 /// Response of `POST .../apply`: counts of what changed.

crates/picloud-cli/src/cmds/apply.rs

@@ -124,12 +124,14 @@ pub async fn run(
 
 /// `pic apply --dir <root>` (Phase 5): reconcile a whole project tree — every
 /// `picloud.toml` under `root` — in ONE atomic server transaction.
+#[allow(clippy::too_many_arguments)]
 pub async fn run_tree(
     dir: &Path,
     prune: bool,
     yes: bool,
     force: bool,
     takeover: bool,
+    structure_mode: &str,
     env: Option<&str>,
     mode: OutputMode,
 ) -> Result<()> {
@@ -157,6 +159,7 @@ pub async fn run_tree(
             project.as_ref(),
             takeover,
             expected_token.as_deref(),
+            structure_mode,
         )
         .await?;
     crate::linkstate::clear_plan(dir, token_key);

crates/picloud-cli/src/cmds/plan.rs

@@ -66,6 +66,21 @@ fn render_tree(plan: &TreePlanDto, mode: OutputMode) {
     let mut table = Table::new(["node", "kind", "op", "resource", "detail"]);
     for n in &plan.nodes {
         let node = format!("{}:{}", n.kind, n.slug);
+        if let Some(s) = &n.structure {
+            table.row([
+                node.clone(),
+                "structure".to_string(),
+                s.status.clone(),
+                format!(
+                    "server={}",
+                    s.server_parent.clone().unwrap_or_else(|| "<root>".into())
+                ),
+                format!(
+                    "manifest={}",
+                    s.manifest_parent.clone().unwrap_or_else(|| "<root>".into())
+                ),
+            ]);
+        }
         if let Some(o) = &n.ownership {
             table.row([
                 node.clone(),

crates/picloud-cli/src/main.rs

@@ -261,6 +261,14 @@ struct ApplyArgs {
     /// top of the base manifest before applying.
     #[arg(long)]
     env: Option<String>,
+    /// §6 M2: reparent a group whose server parent diverges from the
+    /// manifest's directory nesting (default: refuse on divergence).
+    #[arg(long, conflicts_with = "adopt_server_structure")]
+    force_local_structure: bool,
+    /// §6 M2: accept the server's group placement on a structural
+    /// divergence and reconcile content in place (no reparent).
+    #[arg(long)]
+    adopt_server_structure: bool,
 }
 
 #[derive(Args)]
@@ -1433,12 +1441,20 @@ async fn main() -> ExitCode {
         Cmd::Whoami => cmds::whoami::run(mode).await,
         Cmd::Apply(args) => match &args.dir {
             Some(dir) => {
+                let structure_mode = if args.force_local_structure {
+                    "force-local"
+                } else if args.adopt_server_structure {
+                    "adopt-server"
+                } else {
+                    "refuse"
+                };
                 cmds::apply::run_tree(
                     dir,
                     args.prune,
                     args.yes,
                     args.force,
                     args.takeover,
+                    structure_mode,
                     args.env.as_deref(),
                     mode,
                 )