feat(hierarchies): declarative group create/reparent — tree shape (M2)

M2 of the remaining-hierarchies work. `pic apply --dir` now owns the org-tree
SHAPE, not just node content: a `[group]` manifest for a group that doesn't
exist is CREATED under its declared parent, and an existing group whose declared
parent changed is REPARENTED — all inside the single tree-apply transaction
(create + reparent; structural prune is deferred to M3 with the ownership layer).

group_repo: extract transaction-aware structural mutations — `create_group_tx`,
`reparent_group_tx` (the ancestor-walk cycle guard, now reading through the tx so
it sees in-tx writes), `delete_group_tx`, and `acquire_structural_lock_tx`. The
trait `create`/`reparent`/`delete` delegate to them (one SQL definition,
behavior-preserving: the coarse structural advisory lock + structure_version
bump + delete=RESTRICT all preserved).

apply_service: `TreeNode` gains `parent_slug` + `name`. `prepare_tree` classifies
group nodes into existing (resolved id, maybe reparent) vs to-create (absent →
deferred), returning a `PreparedTree { prepared, creates, reparents, token }`.
`apply_tree` adds Phase 0 (`reconcile_tree_structure_tx`): create absent groups
parent-first (topo loop with cycle/unresolved-parent detection) and reparent
existing ones, then Phase A/B reconcile content as before. A to-create group
reconciles against an empty CurrentState (all-Create) — so it needs no DB read
and sidesteps the pool-vs-tx coupling. The bound-plan token folds a
declared-absent marker, so a group created out-of-band between plan and apply
trips StateMoved.

authz (apply_api `authz_tree`): a to-create group mirrors the interactive create
gate — root-level needs `InstanceCreateGroup`, a subgroup under an existing
parent needs only `GroupAdmin(parent)`; a reparent needs `GroupAdmin` at the
group, the SOURCE parent, and the DESTINATION parent (§5.6, parity with
`reparent_group`). No 404 on a to-create node.

CLI: `[group] parent` manifest key (else the parent is inferred from the nearest
ancestor directory's group); `build_tree` emits `parent_slug` + `name` per group
node; the apply report shows a `groups +N reparented M` line.

Reviewed by subagent (core mechanism verified sound: topo termination, empty-
CurrentState reconcile, in-tx cycle guard, atomicity, StateMoved, idempotency);
fixed the two authz-parity gaps it found (missing source-parent check on
reparent; over-gated subgroup create). Tests: a new `tree_shape` journey
(create + reparent + no-op re-plan); 393 manager-core lib + the Phase-5 tree/
group/ext journeys all green; clippy -D warnings clean.

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

crates/manager-core/src/apply_api.rs

@@ -216,6 +216,7 @@ async fn tree_apply_handler(
 /// Per-node capability check for a tree plan/apply. `prune` widens an apply's
 /// write requirement to every kind. A plan (`prune` ignored, no writes) needs
 /// only the per-node read cap.
+#[allow(clippy::too_many_lines)]
 async fn authz_tree(
     svc: &ApplyService,
     principal: &Principal,
@@ -239,11 +240,87 @@ async fn authz_tree(
                 }
             }
             NodeKind::Group => {
-                let group_id = resolve_group_id(svc.groups.as_ref(), &node.slug).await?;
+                let existing = svc
+                    .groups
+                    .get_by_slug(&node.slug)
+                    .await
+                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
+                let Some(g) = existing else {
+                    // To-create group (M2): mirror the interactive create gate
+                    // (`groups_api::create_group`). A root-level group (no
+                    // resolvable parent) needs `InstanceCreateGroup`; a subgroup
+                    // under an EXISTING parent needs only `GroupAdmin(parent)`.
+                    // A parent that is itself to-create has no id yet → fall
+                    // back to `InstanceCreateGroup`.
+                    let parent = match &node.parent_slug {
+                        Some(p) => svc
+                            .groups
+                            .get_by_slug(p)
+                            .await
+                            .map_err(|e| ApplyError::Backend(e.to_string()))?,
+                        None => None,
+                    };
+                    match parent {
+                        Some(pg) => {
+                            require(svc.authz.as_ref(), principal, Capability::GroupAdmin(pg.id))
+                                .await
+                                .map_err(map_authz)?;
+                        }
+                        None => {
+                            require(
+                                svc.authz.as_ref(),
+                                principal,
+                                Capability::InstanceCreateGroup,
+                            )
+                            .await
+                            .map_err(map_authz)?;
+                        }
+                    }
+                    continue;
+                };
+                let group_id = g.id;
                 match apply_prune {
                     Some(prune) => {
                         require_group_node_writes(svc, principal, group_id, &node.bundle, prune)
                             .await?;
+                        // Reparent: an explicit parent differing from the live
+                        // one needs GroupAdmin at the group, the SOURCE parent,
+                        // and the DESTINATION parent — mirroring the interactive
+                        // `reparent_group` (§5.6 triply-gated).
+                        if let Some(parent) = &node.parent_slug {
+                            if let Some(pg) = svc
+                                .groups
+                                .get_by_slug(parent)
+                                .await
+                                .map_err(|e| ApplyError::Backend(e.to_string()))?
+                            {
+                                if Some(pg.id) != g.parent_id {
+                                    require(
+                                        svc.authz.as_ref(),
+                                        principal,
+                                        Capability::GroupAdmin(group_id),
+                                    )
+                                    .await
+                                    .map_err(map_authz)?;
+                                    if let Some(src) = g.parent_id {
+                                        require(
+                                            svc.authz.as_ref(),
+                                            principal,
+                                            Capability::GroupAdmin(src),
+                                        )
+                                        .await
+                                        .map_err(map_authz)?;
+                                    }
+                                    require(
+                                        svc.authz.as_ref(),
+                                        principal,
+                                        Capability::GroupAdmin(pg.id),
+                                    )
+                                    .await
+                                    .map_err(map_authz)?;
+                                }
+                            }
+                        }
                     }
                     None => {
                         require(

crates/manager-core/src/apply_service.rs

@@ -390,6 +390,17 @@ pub struct TreeNode {
     pub kind: NodeKind,
     pub slug: String,
     pub bundle: Bundle,
+    /// Declared parent group slug (M2 tree shape). For a GROUP node: when the
+    /// group doesn't exist it is created under this parent (None ⇒ the instance
+    /// root); when it exists and this differs from its live parent it is
+    /// reparented. `None` on an existing group means "leave parentage alone".
+    /// Ignored for app nodes (an app's group is set via `apps create --group`).
+    #[serde(default)]
+    pub parent_slug: Option<String>,
+    /// Display name for a GROUP node that must be created (M2). Falls back to
+    /// the slug when omitted. Ignored once the node exists.
+    #[serde(default)]
+    pub name: Option<String>,
 }
 
 /// A whole project subtree submitted to `apply_tree`.
@@ -1067,8 +1078,9 @@ impl ApplyService {
     /// # Errors
     /// `Invalid` for a malformed/unknown bundle; `Backend` for repo failures.
     pub async fn plan_tree(&self, bundle: &TreeBundle) -> Result<TreePlanResult, ApplyError> {
-        let (prepared, token) = self.prepare_tree(bundle).await?;
+        let pt = self.prepare_tree(bundle).await?;
-        let nodes = prepared
+        let mut nodes: Vec<NodePlan> = pt
+            .prepared
             .into_iter()
             .map(|p| NodePlan {
                 kind: p.node.kind,
@@ -1076,9 +1088,18 @@ impl ApplyService {
                 plan: p.plan,
             })
             .collect();
+        // To-create group nodes show all-Create content — their structural
+        // creation is implied by the node not yet existing on the server.
+        for c in &pt.creates {
+            nodes.push(NodePlan {
+                kind: NodeKind::Group,
+                slug: c.slug.clone(),
+                plan: compute_diff(&CurrentState::default(), &c.node.bundle),
+            });
+        }
         Ok(TreePlanResult {
             nodes,
-            state_token: token,
+            state_token: pt.token,
         })
     }
 
@@ -1091,6 +1112,7 @@ impl ApplyService {
     /// # Errors
     /// `Invalid` for a bad bundle; `StateMoved` if the bound token is stale;
     /// `Backend` for repo/transaction failures.
+    #[allow(clippy::too_many_lines)]
     pub async fn apply_tree(
         &self,
         bundle: &TreeBundle,
@@ -1098,12 +1120,18 @@ impl ApplyService {
         actor: AdminUserId,
         expected_token: Option<&str>,
     ) -> Result<ApplyReport, ApplyError> {
-        let (prepared, token) = self.prepare_tree(bundle).await?;
+        let pt = self.prepare_tree(bundle).await?;
         if let Some(expected) = expected_token {
-            if token != expected {
+            if pt.token != expected {
                 return Err(ApplyError::StateMoved);
             }
         }
+        let PreparedTree {
+            prepared,
+            creates,
+            reparents,
+            ..
+        } = pt;
 
         let mut tx = self
             .pool
@@ -1127,6 +1155,21 @@ impl ApplyService {
         let mut group_script_index: HashMap<GroupId, HashMap<String, ScriptId>> = HashMap::new();
         let mut any_app = false;
 
+        // Phase 0 — structure reconcile (M2): create absent group nodes and
+        // reparent moved ones, in-tx so the whole apply stays all-or-nothing.
+        if !creates.is_empty() || !reparents.is_empty() {
+            self.reconcile_tree_structure_tx(
+                &mut tx,
+                &creates,
+                &reparents,
+                prune,
+                actor,
+                &mut group_script_index,
+                &mut report,
+            )
+            .await?;
+        }
+
         // Phase A — groups first: reconcile each and record its post-create
         // `name -> id` so descendant app nodes can resolve inherited bindings.
         for p in &prepared {
@@ -1197,6 +1240,107 @@ impl ApplyService {
         Ok(report)
     }
 
+    /// Phase 0 of a tree apply (M2): create absent group nodes (parent-first)
+    /// and reparent moved ones, inside the apply transaction. A created group's
+    /// content reconciles against an empty current state (all-Create); its
+    /// `name -> id` is recorded so a descendant app can bind to it later in the
+    /// same tx. Runs under the coarse structural lock so the cycle guard is sound.
+    ///
+    /// Scope: create + reparent only. Deleting a server group that's absent
+    /// from the manifest is NOT done here even under `--prune` (group delete is
+    /// RESTRICT + destructive, and one repo shouldn't reap another's nodes) —
+    /// structural prune lands with the ownership layer (M3). Use
+    /// `pic groups rm` to delete a group explicitly.
+    #[allow(clippy::too_many_arguments)]
+    async fn reconcile_tree_structure_tx(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        creates: &[GroupCreate<'_>],
+        reparents: &[(GroupId, Option<GroupId>)],
+        prune: bool,
+        actor: AdminUserId,
+        group_script_index: &mut HashMap<GroupId, HashMap<String, ScriptId>>,
+        report: &mut ApplyReport,
+    ) -> Result<(), ApplyError> {
+        crate::group_repo::acquire_structural_lock_tx(tx)
+            .await
+            .map_err(map_group_err)?;
+
+        // The default parent for a create with no declared parent.
+        let root_id = self
+            .groups
+            .get_by_slug(crate::group_repo::ROOT_GROUP_SLUG)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?
+            .map(|g| g.id);
+
+        // Topo-order creates parent-first: a create whose parent is another
+        // to-create group must run after that parent is created.
+        let create_slugs: HashSet<String> = creates.iter().map(|c| c.slug.to_lowercase()).collect();
+        let mut created_ids: HashMap<String, GroupId> = HashMap::new();
+        let mut pending: Vec<&GroupCreate<'_>> = creates.iter().collect();
+        while !pending.is_empty() {
+            let before = pending.len();
+            let mut next: Vec<&GroupCreate<'_>> = Vec::new();
+            for c in std::mem::take(&mut pending) {
+                let parent_id = match &c.parent_slug {
+                    None => root_id,
+                    Some(p) => {
+                        let pl = p.to_lowercase();
+                        if create_slugs.contains(&pl) {
+                            // Parent is also to-create; defer until it exists.
+                            let Some(id) = created_ids.get(&pl) else {
+                                next.push(c);
+                                continue;
+                            };
+                            Some(*id)
+                        } else {
+                            Some(self.resolve_existing_parent(p).await?)
+                        }
+                    }
+                };
+                let g = crate::group_repo::create_group_tx(tx, &c.slug, &c.name, None, parent_id)
+                    .await
+                    .map_err(map_group_err)?;
+                created_ids.insert(c.slug.to_lowercase(), g.id);
+                let empty = CurrentState::default();
+                let plan = compute_diff(&empty, &c.node.bundle);
+                let name_to_id = self
+                    .reconcile_node_tx(
+                        tx,
+                        ApplyOwner::Group(g.id),
+                        &c.node.bundle,
+                        &plan,
+                        &empty,
+                        &HashMap::new(),
+                        &HashMap::new(),
+                        prune,
+                        actor,
+                        report,
+                    )
+                    .await?;
+                group_script_index.insert(g.id, name_to_id);
+                report.groups_created += 1;
+            }
+            if next.len() == before {
+                return Err(ApplyError::Invalid(
+                    "project tree has a cycle (or unresolved parent) among groups being created"
+                        .into(),
+                ));
+            }
+            pending = next;
+        }
+
+        // Reparent existing groups. The in-tx cycle guard sees the creates above.
+        for (id, new_parent) in reparents {
+            crate::group_repo::reparent_group_tx(tx, *id, *new_parent)
+                .await
+                .map_err(map_group_err)?;
+            report.groups_reparented += 1;
+        }
+        Ok(())
+    }
+
     /// Resolve nodes to owners, validate each (apps' route/trigger targets may
     /// bind to in-tree ancestor group scripts), compute each node's plan, and
     /// fold a combined bound-plan token. Shared by `plan_tree` / `apply_tree`.
@@ -1204,14 +1348,16 @@ impl ApplyService {
     async fn prepare_tree<'a>(
         &self,
         bundle: &'a TreeBundle,
-    ) -> Result<(Vec<PreparedNode<'a>>, String), ApplyError> {
+    ) -> Result<PreparedTree<'a>, ApplyError> {
         if bundle.nodes.is_empty() {
             return Err(ApplyError::Invalid("project tree has no nodes".into()));
         }
-        // Register in-tree group nodes first: their (resolved) ids and endpoint
+        // Classify group nodes: existing (resolved id + live parent) vs to-create
-        // script names, which an app node's binding validation may reference.
+        // (absent → created in apply Phase 0). App nodes always pre-exist.
         let mut group_id_by_slug: HashMap<String, GroupId> = HashMap::new();
+        let mut group_parent: HashMap<String, Option<GroupId>> = HashMap::new();
         let mut group_script_names: HashMap<GroupId, HashSet<String>> = HashMap::new();
+        let mut to_create_slugs: HashSet<String> = HashSet::new();
         let mut seen_slugs: HashSet<String> = HashSet::new();
         for n in &bundle.nodes {
             let key = format!("{:?}:{}", n.kind, n.slug.to_lowercase());
@@ -1222,29 +1368,78 @@ impl ApplyService {
                 )));
             }
             if n.kind == NodeKind::Group {
-                let gid = self.resolve_group(&n.slug).await?;
+                match self
-                group_id_by_slug.insert(n.slug.to_lowercase(), gid);
+                    .groups
-                group_script_names.insert(
+                    .get_by_slug(&n.slug)
-                    gid,
+                    .await
-                    n.bundle
+                    .map_err(|e| ApplyError::Backend(e.to_string()))?
-                        .scripts
+                {
-                        .iter()
+                    Some(g) => {
-                        .filter(|s| s.kind == ScriptKind::Endpoint)
+                        group_id_by_slug.insert(n.slug.to_lowercase(), g.id);
-                        .map(|s| s.name.to_lowercase())
+                        group_parent.insert(n.slug.to_lowercase(), g.parent_id);
-                        .collect(),
+                        group_script_names.insert(
-                );
+                            g.id,
+                            n.bundle
+                                .scripts
+                                .iter()
+                                .filter(|s| s.kind == ScriptKind::Endpoint)
+                                .map(|s| s.name.to_lowercase())
+                                .collect(),
+                        );
+                    }
+                    None => {
+                        to_create_slugs.insert(n.slug.to_lowercase());
+                    }
+                }
             }
         }
 
         let mut prepared = Vec::with_capacity(bundle.nodes.len());
+        let mut creates: Vec<GroupCreate<'a>> = Vec::new();
+        let mut reparents: Vec<(GroupId, Option<GroupId>)> = Vec::new();
         let mut token_parts: Vec<String> = Vec::new();
         let mut versioned_groups: BTreeSet<GroupId> = BTreeSet::new();
 
         for n in &bundle.nodes {
+            // To-create group: no id/current. Validate its bundle (group rules)
+            // and fold a "declared-absent" token part, then defer creation to
+            // apply Phase 0. Its content is all-Create against an empty state.
+            if n.kind == NodeKind::Group && to_create_slugs.contains(&n.slug.to_lowercase()) {
+                let placeholder = ApplyOwner::Group(uuid::Uuid::nil().into());
+                self.validate_bundle_for(placeholder, &n.bundle, &HashSet::new())?;
+                token_parts.push(format!("nx|{}|absent", n.slug.to_lowercase()));
+                creates.push(GroupCreate {
+                    slug: n.slug.clone(),
+                    name: n.name.clone().unwrap_or_else(|| n.slug.clone()),
+                    parent_slug: n.parent_slug.clone(),
+                    node: n,
+                });
+                continue;
+            }
+
             let (owner, app_chain) = match n.kind {
                 NodeKind::Group => {
                     let gid = group_id_by_slug[&n.slug.to_lowercase()];
                     versioned_groups.insert(gid);
+                    // Reparent an EXISTING group when an explicit parent_slug
+                    // differs from its live parent. The target must already
+                    // exist (reparenting into a freshly-created group in the
+                    // same apply needs a second apply — a known limitation).
+                    if let Some(parent_slug) = &n.parent_slug {
+                        if to_create_slugs.contains(&parent_slug.to_lowercase()) {
+                            return Err(ApplyError::Invalid(format!(
+                                "group `{}` reparents under `{parent_slug}`, which this \
+                                 apply is also creating; apply the new parent first",
+                                n.slug
+                            )));
+                        }
+                        let target = self.resolve_existing_parent(parent_slug).await?;
+                        let current_parent =
+                            group_parent.get(&n.slug.to_lowercase()).copied().flatten();
+                        if Some(target) != current_parent {
+                            reparents.push((gid, Some(target)));
+                        }
+                    }
                     (ApplyOwner::Group(gid), Vec::new())
                 }
                 NodeKind::App => {
@@ -1316,7 +1511,26 @@ impl ApplyService {
             }
         }
         token_parts.sort_unstable();
-        Ok((prepared, combine_tokens(&token_parts)))
+        Ok(PreparedTree {
+            prepared,
+            creates,
+            reparents,
+            token: combine_tokens(&token_parts),
+        })
+    }
+
+    /// Resolve a declared parent slug to an EXISTING group id (reparent target).
+    async fn resolve_existing_parent(&self, slug: &str) -> Result<GroupId, ApplyError> {
+        self.groups
+            .get_by_slug(slug)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?
+            .map(|g| g.id)
+            .ok_or_else(|| {
+                ApplyError::Invalid(format!(
+                    "parent group `{slug}` does not exist (a reparent target must already exist)"
+                ))
+            })
     }
 
     /// Resolve an app node's inherited route/trigger targets to script ids,
@@ -1383,18 +1597,6 @@ impl ApplyService {
             .ok_or_else(|| ApplyError::AppNotFound(ident.to_string()))
     }
 
-    async fn resolve_group(&self, ident: &str) -> Result<GroupId, ApplyError> {
-        let found = if let Ok(uuid) = ident.parse::<uuid::Uuid>() {
-            self.groups.get_by_id(uuid.into()).await
-        } else {
-            self.groups.get_by_slug(ident).await
-        };
-        found
-            .map_err(|e| ApplyError::Backend(e.to_string()))?
-            .map(|g| g.id)
-            .ok_or_else(|| ApplyError::AppNotFound(format!("group {ident}")))
-    }
-
     fn script_imports(&self, bs: &BundleScript) -> Result<Vec<String>, ApplyError> {
         let res = if bs.kind == ScriptKind::Module {
             self.validator.validate_module(&bs.source)
@@ -2899,6 +3101,10 @@ pub struct ApplyReport {
     pub extension_points_updated: u32,
     #[serde(default)]
     pub extension_points_deleted: u32,
+    #[serde(default)]
+    pub groups_created: u32,
+    #[serde(default)]
+    pub groups_reparented: u32,
     #[serde(skip_serializing_if = "Vec::is_empty")]
     pub warnings: Vec<String>,
 }
@@ -2921,6 +3127,26 @@ struct PreparedNode<'a> {
     app_chain: Vec<GroupId>,
 }
 
+/// A group node declared in the tree but absent on the server — created in
+/// `apply_tree` Phase 0. Its content reconciles against an empty current state.
+struct GroupCreate<'a> {
+    slug: String,
+    name: String,
+    /// Declared parent slug; `None` ⇒ the instance root.
+    parent_slug: Option<String>,
+    node: &'a TreeNode,
+}
+
+/// Output of `prepare_tree`: the per-node content plans plus the structural
+/// changes (group creates / reparents) and the combined bound-plan token.
+struct PreparedTree<'a> {
+    prepared: Vec<PreparedNode<'a>>,
+    creates: Vec<GroupCreate<'a>>,
+    /// Existing groups to reparent: `(group id, new parent id)`.
+    reparents: Vec<(GroupId, Option<GroupId>)>,
+    token: String,
+}
+
 /// FNV-1a over a set of already-sorted per-resource token parts — the same
 /// hashing `state_token` uses, lifted to combine many nodes into one tree
 /// token. Order-independent because the caller sorts `parts`.
@@ -3028,6 +3254,16 @@ fn map_repo(e: ScriptRepositoryError) -> ApplyError {
     }
 }
 
+fn map_group_err(e: crate::group_repo::GroupRepositoryError) -> ApplyError {
+    use crate::group_repo::GroupRepositoryError as E;
+    // Conflict (cycle, slug clash, non-empty delete) and not-found are caller
+    // errors → 422; a raw DB failure → 500.
+    match &e {
+        E::Db(_) => ApplyError::Backend(e.to_string()),
+        E::Conflict(_) | E::NotFound(_) => ApplyError::Invalid(e.to_string()),
+    }
+}
+
 fn map_trig(e: TriggerRepoError) -> ApplyError {
     match e {
         TriggerRepoError::Invalid(m) => ApplyError::Invalid(m),

crates/manager-core/src/group_repo.rs

@@ -191,27 +191,10 @@ impl GroupRepository for PostgresGroupRepository {
         description: Option<&str>,
         parent_id: Option<GroupId>,
     ) -> Result<Group, GroupRepositoryError> {
-        let res = sqlx::query_as::<_, GroupRow>(&format!(
+        let mut tx = self.pool.begin().await?;
-            "INSERT INTO groups (slug, name, description, parent_id) \
+        let g = create_group_tx(&mut tx, slug, name, description, parent_id).await?;
-             VALUES ($1, $2, $3, $4) \
+        tx.commit().await?;
-             RETURNING {GROUP_COLS}"
+        Ok(g)
-        ))
-        .bind(slug)
-        .bind(name)
-        .bind(description)
-        .bind(parent_id.map(GroupId::into_inner))
-        .fetch_one(&self.pool)
-        .await;
-        match res {
-            Ok(row) => Ok(row.into()),
-            Err(sqlx::Error::Database(e)) if e.is_unique_violation() => Err(
-                GroupRepositoryError::Conflict(format!("slug {slug:?} is already in use")),
-            ),
-            Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => Err(
-                GroupRepositoryError::Conflict("parent group does not exist".into()),
-            ),
-            Err(e) => Err(e.into()),
-        }
     }
 
     async fn rename(
@@ -247,96 +230,165 @@ impl GroupRepository for PostgresGroupRepository {
     ) -> Result<Group, GroupRepositoryError> {
         let mut tx = self.pool.begin().await?;
         // Coarse structural lock: serialize all structural mutations so the
-        // cycle guard + parent write can't interleave with a concurrent
+        // cycle guard + parent write can't race a concurrent reparent.
-        // reparent and race into a cycle.
+        acquire_structural_lock_tx(&mut tx).await?;
-        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        let g = reparent_group_tx(&mut tx, id, new_parent).await?;
-            .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));
-        };
         tx.commit().await?;
-        Ok(row.into())
+        Ok(g)
     }
 
     async fn delete(&self, id: GroupId) -> Result<(), GroupRepositoryError> {
-        // Pre-check for a clean message; the FK RESTRICT is the real guard.
+        let mut tx = self.pool.begin().await?;
-        let counts = self.child_counts(id).await?;
+        delete_group_tx(&mut tx, id).await?;
-        if !counts.is_empty() {
+        tx.commit().await?;
-            return Err(GroupRepositoryError::Conflict(format!(
+        Ok(())
-                "group still has {} subgroup(s) and {} app(s); move or delete them first",
+    }
-                counts.subgroups, counts.apps
+}
-            )));
+
+// ----------------------------------------------------------------------------
+// Transaction-aware structural mutations (Phase 5+ declarative tree apply).
+//
+// The declarative `apply_tree` reconciles a whole subtree in ONE transaction,
+// so group create/reparent/delete must run inside the caller's tx (not on the
+// pool) to stay all-or-nothing. These free functions hold the SQL; the trait
+// methods above delegate to them (begin → fn → commit) so there is one SQL
+// definition each. The caller takes [`acquire_structural_lock_tx`] ONCE before
+// the structure phase, so the cycle guard + parent writes serialize against
+// concurrent reparents exactly as the single-shot `reparent` does.
+// ----------------------------------------------------------------------------
+
+/// Take the coarse structural lock inside the caller's transaction. Held until
+/// the tx commits/rolls back. Call once before any `*_group_tx` mutation.
+pub(crate) async fn acquire_structural_lock_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+) -> Result<(), GroupRepositoryError> {
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(GROUP_STRUCTURAL_LOCK_KEY)
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
+/// Insert a group inside the caller's tx. Maps unique/FK violations to a clean
+/// conflict. (The structural lock is not required for a pure insert, but the
+/// apply path holds it for the whole phase anyway.)
+pub(crate) async fn create_group_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    slug: &str,
+    name: &str,
+    description: Option<&str>,
+    parent_id: Option<GroupId>,
+) -> Result<Group, GroupRepositoryError> {
+    let res = sqlx::query_as::<_, GroupRow>(&format!(
+        "INSERT INTO groups (slug, name, description, parent_id) \
+         VALUES ($1, $2, $3, $4) RETURNING {GROUP_COLS}"
+    ))
+    .bind(slug)
+    .bind(name)
+    .bind(description)
+    .bind(parent_id.map(GroupId::into_inner))
+    .fetch_one(&mut **tx)
+    .await;
+    match res {
+        Ok(row) => Ok(row.into()),
+        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => Err(
+            GroupRepositoryError::Conflict(format!("slug {slug:?} is already in use")),
+        ),
+        Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => Err(
+            GroupRepositoryError::Conflict("parent group does not exist".into()),
+        ),
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Reparent a group inside the caller's tx. Runs the ancestor-walk cycle guard
+/// against the IN-TX tree (so it sees parent writes made earlier in the same
+/// apply). The caller must already hold [`acquire_structural_lock_tx`].
+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(),
+            ));
         }
-        let res = sqlx::query("DELETE FROM groups WHERE id = $1")
+        let mut cursor = Some(parent);
-            .bind(id.into_inner())
+        let mut hops = 0u32;
-            .execute(&self.pool)
+        while let Some(node) = cursor {
-            .await;
+            if node == id {
-        match res {
+                return Err(GroupRepositoryError::Conflict(
-            Ok(r) if r.rows_affected() == 0 => Err(GroupRepositoryError::NotFound(id)),
+                    "cannot reparent a group beneath one of its own descendants".into(),
-            Ok(_) => Ok(()),
+                ));
-            Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => {
+            }
-                // Lost a race with a concurrent child insert.
+            hops += 1;
-                Err(GroupRepositoryError::Conflict(
+            if hops > 64 {
-                    "group still has descendants; move or delete them first".into(),
+                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(),
+                    ));
+                }
             }
-            Err(e) => Err(e.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))
+}
+
+/// Delete an empty group inside the caller's tx (delete = RESTRICT). Refused
+/// with a clean conflict if it still has child groups or apps.
+pub(crate) async fn delete_group_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: GroupId,
+) -> Result<(), GroupRepositoryError> {
+    let counts: (i64, i64) = sqlx::query_as(
+        "SELECT (SELECT COUNT(*) FROM groups WHERE parent_id = $1), \
+                (SELECT COUNT(*) FROM apps WHERE group_id = $1)",
+    )
+    .bind(id.into_inner())
+    .fetch_one(&mut **tx)
+    .await?;
+    if counts.0 != 0 || counts.1 != 0 {
+        return Err(GroupRepositoryError::Conflict(format!(
+            "group still has {} subgroup(s) and {} app(s); move or delete them first",
+            counts.0, counts.1
+        )));
+    }
+    let res = sqlx::query("DELETE FROM groups WHERE id = $1")
+        .bind(id.into_inner())
+        .execute(&mut **tx)
+        .await;
+    match res {
+        Ok(r) if r.rows_affected() == 0 => Err(GroupRepositoryError::NotFound(id)),
+        Ok(_) => Ok(()),
+        Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => {
+            Err(GroupRepositoryError::Conflict(
+                "group still has descendants; move or delete them first".into(),
+            ))
+        }
+        Err(e) => Err(e.into()),
+    }
 }
 
 #[derive(sqlx::FromRow)]

crates/picloud-cli/src/client.rs

@@ -1417,6 +1417,10 @@ pub struct ApplyReportDto {
     #[serde(default)]
     pub extension_points_deleted: u32,
     #[serde(default)]
+    pub groups_created: u32,
+    #[serde(default)]
+    pub groups_reparented: u32,
+    #[serde(default)]
     pub warnings: Vec<String>,
 }
 

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

@@ -94,6 +94,16 @@ pub async fn run(
                 report.extension_points_deleted
             ),
         );
+    // Structural changes only happen on a tree apply; omit the line otherwise.
+    if report.groups_created > 0 || report.groups_reparented > 0 {
+        block.field(
+            "groups",
+            format!(
+                "+{} reparented {}",
+                report.groups_created, report.groups_reparented
+            ),
+        );
+    }
     for w in &report.warnings {
         block.field("warning", w.clone());
     }
@@ -170,6 +180,16 @@ pub async fn run_tree(
                 report.extension_points_deleted
             ),
         );
+    // Structural changes only happen on a tree apply; omit the line otherwise.
+    if report.groups_created > 0 || report.groups_reparented > 0 {
+        block.field(
+            "groups",
+            format!(
+                "+{} reparented {}",
+                report.groups_created, report.groups_reparented
+            ),
+        );
+    }
     for w in &report.warnings {
         block.field("warning", w.clone());
     }

crates/picloud-cli/src/discover.rs

@@ -4,7 +4,7 @@
 //! the server resolves ancestry from its own group tree, so the on-disk
 //! nesting is organizational, not authoritative here.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 
 use anyhow::{bail, Context, Result};
@@ -56,20 +56,62 @@ pub fn build_tree(root: &Path, env: Option<&str>) -> Result<(Value, usize)> {
             root.display()
         );
     }
-    let mut nodes = Vec::with_capacity(paths.len());
+    // First pass: load every manifest and index its DIRECTORY → (slug, is_group)
-    let mut seen: HashSet<(String, String)> = HashSet::new();
+    // so a group node's parent can be inferred from the enclosing directory.
+    let mut loaded: Vec<(PathBuf, Manifest)> = Vec::with_capacity(paths.len());
+    let mut group_dirs: HashMap<PathBuf, String> = HashMap::new();
     for path in &paths {
         let manifest = Manifest::load_with_env(path, env)
             .with_context(|| format!("loading {}", path.display()))?;
-        let base_dir = path.parent().unwrap_or_else(|| Path::new("."));
+        let dir = path
-        let bundle = build_bundle(&manifest, base_dir)?;
+            .parent()
+            .unwrap_or_else(|| Path::new("."))
+            .to_path_buf();
+        if manifest.is_group() {
+            group_dirs.insert(dir.clone(), manifest.slug().to_string());
+        }
+        loaded.push((dir, manifest));
+    }
+
+    let mut nodes = Vec::with_capacity(loaded.len());
+    let mut seen: HashSet<(String, String)> = HashSet::new();
+    for (dir, manifest) in &loaded {
+        let bundle = build_bundle(manifest, dir)?;
         let kind = if manifest.is_group() { "group" } else { "app" };
         let slug = manifest.slug().to_string();
         if !seen.insert((kind.to_string(), slug.clone())) {
             bail!("project tree declares {kind} `{slug}` more than once");
         }
-        nodes.push(json!({ "kind": kind, "slug": slug, "bundle": bundle }));
+        let mut node = json!({ "kind": kind, "slug": slug, "bundle": bundle });
+        // Group nodes carry their parent (explicit `[group] parent`, else the
+        // nearest ancestor directory's group) + display name, so the server can
+        // create/reparent them (M2). App nodes ignore both server-side.
+        if let Some(g) = &manifest.group {
+            let parent = g
+                .parent
+                .clone()
+                .or_else(|| nearest_ancestor_group(dir, &group_dirs));
+            let obj = node.as_object_mut().expect("json object");
+            obj.insert("name".into(), json!(g.name));
+            if let Some(p) = parent {
+                obj.insert("parent_slug".into(), json!(p));
+            }
+        }
+        nodes.push(node);
     }
     let count = nodes.len();
     Ok((json!({ "nodes": nodes }), count))
 }
+
+/// The slug of the nearest ancestor directory (above `dir`) that holds a group
+/// manifest — the directory-derived parent for a nested group node.
+fn nearest_ancestor_group(dir: &Path, group_dirs: &HashMap<PathBuf, String>) -> Option<String> {
+    let mut cur = dir.parent();
+    while let Some(d) = cur {
+        if let Some(slug) = group_dirs.get(d) {
+            return Some(slug.clone());
+        }
+        cur = d.parent();
+    }
+    None
+}

crates/picloud-cli/src/manifest.rs

@@ -209,16 +209,20 @@ pub struct ManifestApp {
     pub description: Option<String>,
 }
 
-/// A `[group]` node (Phase 5): a group's own declarative content — its scripts
+/// A `[group]` node (Phase 5/M2): a group's own declarative content — its
-/// and `[vars]`. The group must already exist on the server (created with
+/// scripts and `[vars]`. With M2, `pic apply --dir` also creates the group if
-/// `pic groups create`); the manifest reconciles its content, not the tree
+/// it doesn't exist and reparents it under `parent`. When `parent` is omitted,
-/// shape. In a nested project the parent is inferred from the directory tree.
+/// the parent is inferred from the enclosing directory's group manifest (or the
+/// instance root for a top-level group).
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ManifestGroup {
     pub slug: String,
     pub name: String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub description: Option<String>,
+    /// Explicit parent group slug (M2). Overrides the directory-derived parent.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub parent: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

crates/picloud-cli/tests/cli.rs

@@ -41,5 +41,6 @@ mod scripts;
 mod secrets;
 mod staleness;
 mod tree;
+mod tree_shape;
 mod triggers;
 mod vars;

crates/picloud-cli/tests/tree_shape.rs

@@ -0,0 +1,113 @@
+//! M2 declarative tree SHAPE via `pic apply --dir`:
+//!   * a group manifest for a group that does NOT exist yet is CREATED under
+//!     its declared parent, with its content, in one atomic apply,
+//!   * re-applying the same tree is a no-op (the group now exists),
+//!   * changing the declared parent REPARENTS the group on the next apply.
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::{GroupGuard, ScriptGuard};
+
+/// Write a single-group project dir declaring group `slug` under `parent` with
+/// one endpoint script.
+fn group_dir(slug: &str, parent: &str) -> TempDir {
+    let dir = TempDir::new().expect("tempdir");
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/hello.rhai"), r#""hi""#).unwrap();
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!(
+            "[group]\nslug = \"{slug}\"\nname = \"Child Group\"\nparent = \"{parent}\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n"
+        ),
+    )
+    .unwrap();
+    dir
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn tree_apply_creates_then_reparents_a_group() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let parent = common::unique_slug("ts-par");
+    let child = common::unique_slug("ts-child");
+
+    // Guards drop in reverse order: parent (last) ← child (mid) ← script (first).
+    let _gp = GroupGuard::new(&env.url, &env.token, &parent);
+    let _gc = GroupGuard::new(&env.url, &env.token, &child);
+    common::pic_as(&env)
+        .args(["groups", "create", &parent])
+        .assert()
+        .success();
+
+    // --- Create: the child group does NOT exist; apply --dir creates it. ---
+    let dir = group_dir(&child, &parent);
+    let out = common::pic_as(&env)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .output()
+        .expect("apply --dir");
+    assert!(
+        out.status.success(),
+        "tree apply (create) failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+
+    // The group now exists with its script.
+    let groups = ls(&env, &["groups", "ls"]);
+    assert!(groups.contains(&child), "created group missing:\n{groups}");
+    let scripts = ls(&env, &["scripts", "ls", "--group", &child]);
+    let script_id = scripts
+        .lines()
+        .map(common::cells)
+        .find(|c| c.get(2) == Some(&"hello"))
+        .and_then(|c| c.first().map(|s| (*s).to_string()))
+        .unwrap_or_else(|| panic!("group script `hello` should exist:\n{scripts}"));
+    let _gs = ScriptGuard::new(&env.url, &env.token, &script_id);
+
+    // Re-applying the same tree is a no-op (group + content unchanged).
+    let plan_out = common::pic_as(&env)
+        .args(["plan", "--dir"])
+        .arg(dir.path())
+        .output()
+        .expect("plan --dir");
+    let plan_txt = String::from_utf8_lossy(&plan_out.stdout).to_lowercase();
+    assert!(
+        !plan_txt.contains("create") && !plan_txt.contains("delete"),
+        "re-plan of an applied tree must be a no-op:\n{plan_txt}"
+    );
+
+    // --- Reparent: move the child under the instance root. ---
+    let dir2 = group_dir(&child, "root");
+    let out = common::pic_as(&env)
+        .args(["apply", "--dir"])
+        .arg(dir2.path())
+        .output()
+        .expect("apply --dir reparent");
+    assert!(
+        out.status.success(),
+        "tree apply (reparent) failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    // Re-plan is a no-op now that the parent matches.
+    let plan_out = common::pic_as(&env)
+        .args(["plan", "--dir"])
+        .arg(dir2.path())
+        .output()
+        .expect("plan --dir after reparent");
+    let plan_txt = String::from_utf8_lossy(&plan_out.stdout).to_lowercase();
+    assert!(
+        !plan_txt.contains("create") && !plan_txt.contains("delete"),
+        "re-plan after reparent must be a no-op:\n{plan_txt}"
+    );
+}
+
+fn ls(env: &common::TestEnv, args: &[&str]) -> String {
+    String::from_utf8(common::pic_as(env).args(args).output().expect("ls").stdout).unwrap()
+}