feat(apply): §6 M1 — declarative group create (directory-nesting parentage)

Lift "groups must pre-exist" for the tree apply: a `[group]` node now carries a
declared parent (inferred from directory nesting) and `pic apply --dir` creates
the ones the server lacks, in the same transaction.

Wire: `TreeNode` gains `parent`/`name`/`description` (`#[serde(default)]` — a
pre-§6 CLI still reconciles existing groups). `discover::build_tree` computes a
group's parent as the nearest ancestor directory holding a `[group]` (topmost →
the repo's `[project] parent_group` attach point, else instance root) and emits
nodes parents-first by directory depth.

Server: `apply_tree` runs a Phase 0 (`create_missing_groups_tx`) that
resolves-or-creates every group node IN the tx (a same-tree parent resolved
before its child via the new `group_repo::{create_group_tx,
read_group_id_by_slug_tx}`), under the coarse `GROUP_STRUCTURAL_LOCK_KEY` (taken
only when creating), enforcing the attach-point ceiling + RBAC
(`InstanceCreateGroup` / `GroupAdmin(parent)`) per create. A created group is
CLAIMED in Phase A alongside existing ones (`decide_group_claim` → `Claim`).
`prepare_tree` now takes a caller-supplied `resolved_ids` map and returns a
`to_create` list the plan path previews (empty current → full-create plan,
ownership `claim`); a to-create group's token part equals its post-create part
so plan/apply tokens match across a create. `validate_bundle_for` takes
`is_group: bool` (a to-create group has no id yet); `authz_tree` skips a
to-create group (its create authz is the service gate, not an existing-group
capability).

Reparent of a diverged existing group + the detect-and-refuse flags are M2.

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

crates/manager-core/src/apply_api.rs

@@ -401,7 +401,19 @@ async fn authz_tree(
                 }
             }
             NodeKind::Group => {
-                let group_id = resolve_group_id(svc.groups.as_ref(), &node.slug).await?;
+                // §6: a group node that doesn't exist yet is to-CREATE — its
+                // authorization (InstanceCreateGroup / GroupAdmin(parent)) is
+                // enforced in the service's create path, so skip the
+                // existing-group capability check here rather than 404.
+                let Some(group) = svc
+                    .groups
+                    .get_by_slug(&node.slug)
+                    .await
+                    .map_err(|e| ApplyError::Backend(e.to_string()))?
+                else {
+                    continue;
+                };
+                let group_id = group.id;
                 match apply_prune {
                     Some(prune) => {
                         require_group_node_writes(svc, principal, group_id, &node.bundle, prune)

crates/manager-core/src/apply_service.rs

@@ -494,13 +494,28 @@ pub enum NodeKind {
     Group,
 }
 
-/// One node of a project tree: its kind, slug (must pre-exist on the server),
+/// One node of a project tree: its kind, slug, and its desired-state bundle (a
-/// and its desired-state bundle (a group's bundle has no routes/triggers).
+/// group's bundle has no routes/triggers). Request-only (deserialized from the
-/// Request-only (deserialized from the CLI's tree bundle).
+/// CLI's tree bundle).
 #[derive(Debug, Clone, Deserialize)]
 pub struct TreeNode {
     pub kind: NodeKind,
     pub slug: String,
+    /// §6: for a GROUP node, the declared parent group slug inferred from the
+    /// repo's directory nesting (`None` = the repo's attach point / instance
+    /// root). Absent for an app node, and absent on the wire from a pre-§6 CLI
+    /// (`#[serde(default)]`) — in which case a group is resolved as before and
+    /// must pre-exist. Used to CREATE a missing group and (M2) to detect a
+    /// server/manifest structural divergence.
+    #[serde(default)]
+    pub parent: Option<String>,
+    /// §6: a GROUP node's display name / description, used only when the group
+    /// is CREATED (a missing group defaults its name to the slug). Content
+    /// reconcile never touches them.
+    #[serde(default)]
+    pub name: Option<String>,
+    #[serde(default)]
+    pub description: Option<String>,
     pub bundle: Bundle,
 }
 
@@ -806,7 +821,11 @@ impl ApplyService {
         project: Option<&ProjectDecl>,
     ) -> Result<PlanResult, ApplyError> {
         let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
-        self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
+        self.validate_bundle_for(
+            matches!(owner, ApplyOwner::Group(_)),
+            bundle,
+            &keys_set(&inherited),
+        )?;
         self.check_imports_resolve(owner, bundle).await?;
         self.check_extension_points_provided(owner, bundle).await?;
         if let ApplyOwner::App(app_id) = owner {
@@ -887,11 +906,11 @@ impl ApplyService {
     /// or triggers is a 422 — those are app concerns.
     fn validate_bundle_for(
         &self,
-        owner: ApplyOwner,
+        is_group: bool,
         bundle: &Bundle,
         inherited_endpoints: &HashSet<String>,
     ) -> Result<(), ApplyError> {
-        if matches!(owner, ApplyOwner::Group(_)) {
+        if is_group {
             // §11 tail: a group MAY declare ROUTE TEMPLATES (validated as binding
             // a group-owned endpoint by `validate_bundle` via the inherited
             // targets) and TRIGGER TEMPLATES. The stateless EVENT kinds
@@ -949,7 +968,7 @@ impl ApplyService {
         // this is defense-in-depth against a hand-rolled wire `Bundle`; without
         // it the reconcile would insert an inert `app_id`-owned marker row that
         // no resolver ever reads.
-        if matches!(owner, ApplyOwner::App(_)) && !bundle.collections.is_empty() {
+        if !is_group && !bundle.collections.is_empty() {
             return Err(ApplyError::Invalid(
                 "an app manifest cannot declare shared collections — \
                  they are owned by a group"
@@ -962,7 +981,7 @@ impl ApplyService {
         // a misleading `sealed` app row the filters never consult. Defense in
         // depth over the CLI (which authors `sealed` freely but only a group
         // node inherits down).
-        if matches!(owner, ApplyOwner::App(_)) {
+        if !is_group {
             if bundle.routes.iter().any(|r| r.sealed) {
                 return Err(ApplyError::Invalid(
                     "an app route cannot be `sealed` — sealing marks a group \
@@ -1464,7 +1483,11 @@ impl ApplyService {
     ) -> Result<ApplyReport, ApplyError> {
         let actor = claim.principal.user_id;
         let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
-        self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
+        self.validate_bundle_for(
+            matches!(owner, ApplyOwner::Group(_)),
+            bundle,
+            &keys_set(&inherited),
+        )?;
         self.check_imports_resolve(owner, bundle).await?;
         self.check_extension_points_provided(owner, bundle).await?;
         if let ApplyOwner::App(app_id) = owner {
@@ -1620,7 +1643,8 @@ impl ApplyService {
         bundle: &TreeBundle,
         project: Option<&ProjectDecl>,
     ) -> Result<TreePlanResult, ApplyError> {
-        let (prepared, token) = self.prepare_tree(bundle).await?;
+        let resolved = self.resolve_existing_group_ids(bundle).await?;
+        let (prepared, to_create, token) = self.prepare_tree(bundle, &resolved).await?;
         // §6/§7 M3: attach-point ceiling preview (consistent with apply_tree).
         if let Some(pg_slug) = project.and_then(|p| p.parent_group.as_deref()) {
             let pg = self.resolve_group(pg_slug).await?;
@@ -1629,7 +1653,7 @@ impl ApplyService {
             }
         }
         let declared = project.map(|p| p.slug.as_str());
-        let mut nodes = Vec::with_capacity(prepared.len());
+        let mut nodes = Vec::with_capacity(prepared.len() + to_create.len());
         for p in prepared {
             let ownership = self.ownership_preview(p.owner, declared).await?;
             nodes.push(NodePlan {
@@ -1639,6 +1663,17 @@ impl ApplyService {
                 ownership,
             });
         }
+        // §6: a to-create group is previewed with a full-create plan; if the
+        // repo declares a `[project]`, the create will also CLAIM it.
+        for c in to_create {
+            let ownership = declared.map(|d| preview_ownership(true, None, Some(d)));
+            nodes.push(NodePlan {
+                kind: c.node.kind,
+                slug: c.node.slug.clone(),
+                plan: c.plan,
+                ownership,
+            });
+        }
         Ok(TreePlanResult {
             nodes,
             state_token: token,
@@ -1664,14 +1699,36 @@ impl ApplyService {
         expected_token: Option<&str>,
     ) -> Result<ApplyReport, ApplyError> {
         let actor = claim.principal.user_id;
-        let (prepared, token) = self.prepare_tree(bundle).await?;
+        let mut tx = self
+            .pool
+            .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.
+        // 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)
+            .await?;
+        let (prepared, to_create, token) = self.prepare_tree(bundle, &resolved).await?;
+        if !to_create.is_empty() {
+            return Err(ApplyError::Backend(
+                "internal: a group remained unresolved after create".into(),
+            ));
+        }
         if let Some(expected) = expected_token {
             if token != expected {
                 return Err(ApplyError::StateMoved);
             }
         }
-        // §6/§7 attach point: every node in the tree must be within the
+        // §6/§7 attach point: every EXISTING node in the tree must be within the
-        // project's declared `parent_group` subtree (read-only, before the tx).
+        // project's declared `parent_group` subtree. A group CREATED in Phase 0
+        // was already checked there (and its uncommitted row isn't visible to
+        // `check_within_attach`'s pool-based ancestor walk), so skip it.
         if let Some(pg_slug) = claim
             .project
             .as_ref()
@@ -1679,15 +1736,15 @@ 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) {
+                        continue;
+                    }
+                }
                 self.check_within_attach(p.owner, pg, pg_slug).await?;
             }
         }
 
-        let mut tx = self
-            .pool
-            .begin()
-            .await
-            .map_err(|e| ApplyError::Backend(e.to_string()))?;
         // Lock every node's apply key, in sorted order, so concurrent applies
         // touching any shared node serialize and the ordering can't deadlock.
         let mut lock_keys: Vec<i64> = prepared.iter().map(|p| apply_lock_key(p.owner)).collect();
@@ -1804,19 +1861,180 @@ impl ApplyService {
         Ok(report)
     }
 
+    /// §6: resolve every GROUP node's slug → id on the pool, skipping ones that
+    /// don't exist yet (they are to-create, previewed by the plan path).
+    async fn resolve_existing_group_ids(
+        &self,
+        bundle: &TreeBundle,
+    ) -> Result<HashMap<String, GroupId>, ApplyError> {
+        let mut map = HashMap::new();
+        for n in &bundle.nodes {
+            if n.kind == NodeKind::Group {
+                if let Some(g) = self
+                    .groups
+                    .get_by_slug(&n.slug)
+                    .await
+                    .map_err(|e| ApplyError::Backend(e.to_string()))?
+                {
+                    map.insert(n.slug.to_lowercase(), g.id);
+                }
+            }
+        }
+        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.
+    #[allow(clippy::too_many_lines)]
+    async fn create_missing_groups_tx(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        bundle: &TreeBundle,
+        project: Option<&ProjectDecl>,
+        principal: &Principal,
+    ) -> Result<(HashMap<String, GroupId>, HashSet<GroupId>), ApplyError> {
+        use crate::group_repo::{
+            create_group_tx, read_group_id_by_slug_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()) {
+            Some(slug) => Some(
+                read_group_id_by_slug_tx(tx, slug)
+                    .await
+                    .map_err(map_group_repo_err)?
+                    .ok_or_else(|| {
+                        ApplyError::Invalid(format!("attach point group `{slug}` does not exist"))
+                    })?,
+            ),
+            None => None,
+        };
+
+        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 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)
+                .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 {
+                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)
+                    .await
+                    .map_err(map_group_repo_err)?
+                {
+                    map.insert(key, gid);
+                    continue;
+                }
+            }
+            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 {
+                    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()
+                    )));
+                }
+            }
+            // 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)
+                .await
+                .map_err(map_group_repo_err)?;
+            created.insert(gid);
+            if attach_gid.is_some() {
+                created_within_attach.insert(gid);
+            }
+            map.insert(key, gid);
+        }
+        Ok((map, created))
+    }
+
     /// 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`.
+    ///
+    /// `resolved_ids` maps every group node's lowercased slug → its `GroupId`.
+    /// The caller pre-populates it: the plan path with the groups that already
+    /// exist (a missing one becomes a `ToCreateGroup` preview), the apply path
+    /// with existing groups PLUS the ones it just created in-tx (so the returned
+    /// to-create list is empty). A to-create group's token part is identical to
+    /// what it will be once created (empty current), so the plan and apply
+    /// tokens match across a create.
     #[allow(clippy::too_many_lines)]
     async fn prepare_tree<'a>(
         &self,
         bundle: &'a TreeBundle,
-    ) -> Result<(Vec<PreparedNode<'a>>, String), ApplyError> {
+        resolved_ids: &HashMap<String, GroupId>,
+    ) -> Result<(Vec<PreparedNode<'a>>, Vec<ToCreateGroup<'a>>, String), 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
         // script names, which an app node's binding validation may reference.
+        // A group node whose slug is not in `resolved_ids` is to-create.
         let mut group_id_by_slug: HashMap<String, GroupId> = HashMap::new();
         let mut group_script_names: HashMap<GroupId, HashSet<String>> = HashMap::new();
         let mut seen_slugs: HashSet<String> = HashSet::new();
@@ -1829,25 +2047,45 @@ impl ApplyService {
                 )));
             }
             if n.kind == NodeKind::Group {
-                let gid = self.resolve_group(&n.slug).await?;
+                if let Some(&gid) = resolved_ids.get(&n.slug.to_lowercase()) {
-                group_id_by_slug.insert(n.slug.to_lowercase(), gid);
+                    group_id_by_slug.insert(n.slug.to_lowercase(), gid);
-                group_script_names.insert(
+                    group_script_names.insert(
-                    gid,
+                        gid,
-                    n.bundle
+                        n.bundle
-                        .scripts
+                            .scripts
-                        .iter()
+                            .iter()
-                        .filter(|s| s.kind == ScriptKind::Endpoint)
+                            .filter(|s| s.kind == ScriptKind::Endpoint)
-                        .map(|s| s.name.to_lowercase())
+                            .map(|s| s.name.to_lowercase())
-                        .collect(),
+                            .collect(),
-                );
+                    );
+                }
             }
         }
 
         let mut prepared = Vec::with_capacity(bundle.nodes.len());
+        let mut to_create: Vec<ToCreateGroup<'a>> = Vec::new();
         let mut token_parts: Vec<String> = Vec::new();
         let mut versioned_groups: BTreeSet<GroupId> = BTreeSet::new();
 
         for n in &bundle.nodes {
+            // §6: a group node not in `resolved_ids` is to-create — validate its
+            // bundle, diff against an empty current (a full create), and preview
+            // it. The apply path creates it before this content reconcile.
+            if n.kind == NodeKind::Group && !group_id_by_slug.contains_key(&n.slug.to_lowercase()) {
+                self.validate_bundle_for(true, &n.bundle, &HashSet::new())?;
+                let current = CurrentState::default();
+                let current_names = self.resolve_current_target_names(&current).await?;
+                validate_email_secrets_present(&n.bundle, &current.secret_names)?;
+                let plan = compute_diff_with_names(&current, &n.bundle, &current_names);
+                token_parts.push(format!(
+                    "n|{}|{}",
+                    n.slug.to_lowercase(),
+                    state_token_with_names(&current, &current_names)
+                ));
+                to_create.push(ToCreateGroup { node: n, plan });
+                continue;
+            }
+
             let (owner, app_chain) = match n.kind {
                 NodeKind::Group => {
                     let gid = group_id_by_slug[&n.slug.to_lowercase()];
@@ -1886,7 +2124,11 @@ impl ApplyService {
                 }
                 ApplyOwner::Group(_) => HashSet::new(),
             };
-            self.validate_bundle_for(owner, &n.bundle, &inherited_names)?;
+            self.validate_bundle_for(
+                matches!(owner, ApplyOwner::Group(_)),
+                &n.bundle,
+                &inherited_names,
+            )?;
             if let ApplyOwner::App(app_id) = owner {
                 self.validate_route_hosts(app_id, &n.bundle).await?;
             }
@@ -1939,7 +2181,7 @@ impl ApplyService {
             }
         }
         token_parts.sort_unstable();
-        Ok((prepared, combine_tokens(&token_parts)))
+        Ok((prepared, to_create, combine_tokens(&token_parts)))
     }
 
     /// Resolve an app node's inherited route/trigger targets to script ids,
@@ -3537,6 +3779,16 @@ fn map_authz(d: AuthzDenied) -> ApplyError {
     }
 }
 
+/// §6: map a group-repo failure from the declarative create path — a real DB
+/// error is a 500, everything else (slug/parent conflict) is a 422 with the
+/// actionable message.
+fn map_group_repo_err(e: crate::group_repo::GroupRepositoryError) -> ApplyError {
+    match e {
+        crate::group_repo::GroupRepositoryError::Db(e) => ApplyError::Backend(e.to_string()),
+        other => ApplyError::Invalid(other.to_string()),
+    }
+}
+
 /// Owner of an apply node (Phase 5): an app or a group. The apply engine is
 /// owner-generic — only script-create ownership, var writes, the current-state
 /// load, and the advisory lock differ; the diff and all other CRUD are shared.
@@ -4325,6 +4577,16 @@ struct PreparedNode<'a> {
     app_chain: Vec<GroupId>,
 }
 
+/// §6: a GROUP node whose slug does NOT yet exist on the server — it will be
+/// CREATED (its declared parent inferred from directory nesting). Its `current`
+/// is empty and `plan` is a full create; the plan path previews it, and the
+/// apply path creates it in-tx before reconciling this content. Kept separate
+/// from `PreparedNode` because it has no `GroupId` until created.
+struct ToCreateGroup<'a> {
+    node: &'a TreeNode,
+    plan: Plan,
+}
+
 /// 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`.

crates/manager-core/src/group_repo.rs

@@ -23,7 +23,7 @@ use uuid::Uuid;
 /// cycle guard and the `parent_id` write run serialized — two concurrent
 /// reparents can't race into a cycle. Distinct from the per-app
 /// `apply_lock_key` space (a fixed sentinel, hashed-namespace-free).
-const GROUP_STRUCTURAL_LOCK_KEY: i64 = 0x6701_0047_0000_0001;
+pub(crate) const GROUP_STRUCTURAL_LOCK_KEY: i64 = 0x6701_0047_0000_0001;
 
 /// Well-known slug of the instance root group seeded by migration 0047.
 pub const ROOT_GROUP_SLUG: &str = "root";
@@ -53,6 +53,54 @@ impl GroupChildCounts {
     }
 }
 
+/// §6: create a group INSIDE an existing apply transaction, returning its id.
+/// Mirrors [`PostgresGroupRepository::create`]'s error mapping. Used by the
+/// declarative tree apply to create a group the manifest declares (by directory
+/// nesting) but the server doesn't have yet. `structure_version` starts at the
+/// column default (a fresh node has no structural history to bump).
+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<GroupId, GroupRepositoryError> {
+    let res = sqlx::query_as::<_, (Uuid,)>(
+        "INSERT INTO groups (slug, name, description, parent_id) \
+         VALUES ($1, $2, $3, $4) RETURNING id",
+    )
+    .bind(slug)
+    .bind(name)
+    .bind(description)
+    .bind(parent_id.map(GroupId::into_inner))
+    .fetch_one(&mut **tx)
+    .await;
+    match res {
+        Ok((id,)) => Ok(id.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()),
+    }
+}
+
+/// §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.
+pub(crate) async fn read_group_id_by_slug_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    slug: &str,
+) -> Result<Option<GroupId>, GroupRepositoryError> {
+    let row: Option<(Uuid,)> = sqlx::query_as("SELECT id FROM groups WHERE slug = $1")
+        .bind(slug)
+        .fetch_optional(&mut **tx)
+        .await?;
+    Ok(row.map(|(id,)| id.into()))
+}
+
 #[async_trait]
 pub trait GroupRepository: Send + Sync {
     /// Every group on the instance, ordered by name. The tree is small

crates/picloud-cli/src/discover.rs

@@ -1,10 +1,15 @@
 //! Phase 5: discover a project *tree* — every `picloud.toml` under a root
 //! directory — and assemble the wire tree bundle the server's `/tree/{plan,
-//! apply}` endpoints consume. Each manifest becomes one node (app or group);
+//! apply}` endpoints consume. Each manifest becomes one node (app or group).
-//! the server resolves ancestry from its own group tree, so the on-disk
+//!
-//! nesting is organizational, not authoritative here.
+//! §6: a GROUP node's PARENT is inferred from directory nesting — its parent is
+//! the nearest ancestor directory that also holds a `[group]` manifest; the
+//! topmost group binds to the repo's `[project] parent_group` (attach point) or
+//! the instance root. The server uses this declared parent to create a missing
+//! group and to detect structural divergence (M2). An app node declares no
+//! parent (an app pre-exists with a fixed group).
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 
 use anyhow::{bail, Context, Result};
@@ -62,9 +67,10 @@ pub fn build_tree(
         );
     }
     let root_manifest = root.join(MANIFEST_FILE);
+    // Load every manifest once, remembering its directory so we can infer group
+    // parentage from the nesting.
+    let mut loaded: Vec<(PathBuf, Manifest)> = Vec::with_capacity(paths.len());
     let mut project: Option<ManifestProject> = None;
-    let mut nodes = Vec::with_capacity(paths.len());
-    let mut seen: HashSet<(String, String)> = HashSet::new();
     for path in &paths {
         let manifest = Manifest::load_with_env(path, env)
             .with_context(|| format!("loading {}", path.display()))?;
@@ -79,15 +85,65 @@ pub fn build_tree(
                 );
             }
         }
+        loaded.push((path.clone(), manifest));
+    }
+
+    // Map each GROUP manifest's DIRECTORY → its slug, so a group node can find
+    // its nearest ancestor-directory group (its declared parent).
+    let group_dir_to_slug: HashMap<PathBuf, String> = loaded
+        .iter()
+        .filter(|(_, m)| m.is_group())
+        .map(|(path, m)| {
+            let dir = path
+                .parent()
+                .unwrap_or_else(|| Path::new("."))
+                .to_path_buf();
+            (dir, m.slug().to_string())
+        })
+        .collect();
+    let attach_point = project.as_ref().and_then(|p| p.parent_group.clone());
+
+    // Build (sort_key, node) pairs. Emit groups before apps, and shallower
+    // groups before deeper ones (a parent's directory has fewer components than
+    // its child's), so the server creates a parent group before the child that
+    // nests under it.
+    let mut keyed: Vec<((bool, usize), Value)> = Vec::with_capacity(loaded.len());
+    let mut seen: HashSet<(String, String)> = HashSet::new();
+    for (path, manifest) in &loaded {
         let base_dir = path.parent().unwrap_or_else(|| Path::new("."));
-        let bundle = build_bundle(&manifest, base_dir)?;
+        let bundle = build_bundle(manifest, base_dir)?;
-        let kind = if manifest.is_group() { "group" } else { "app" };
+        let is_group = manifest.is_group();
+        let kind = if 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 depth = base_dir.components().count();
+        let node = if is_group {
+            let parent = nearest_group_ancestor(base_dir, &group_dir_to_slug)
+                .or_else(|| attach_point.clone());
+            let g = manifest.group.as_ref();
+            let name = g.map(|g| g.name.clone());
+            let description = g.and_then(|g| g.description.clone());
+            json!({
+                "kind": kind, "slug": slug, "parent": parent,
+                "name": name, "description": description, "bundle": bundle,
+            })
+        } else {
+            json!({ "kind": kind, "slug": slug, "bundle": bundle })
+        };
+        keyed.push(((!is_group, depth), node));
     }
+    keyed.sort_by(|a, b| a.0.cmp(&b.0));
+    let nodes: Vec<Value> = keyed.into_iter().map(|(_, n)| n).collect();
     let count = nodes.len();
     Ok((json!({ "nodes": nodes }), count, project))
 }
+
+/// The slug of the nearest STRICT ancestor directory of `dir` that holds a
+/// `[group]` manifest, or `None` if there is no group above it in the tree.
+fn nearest_group_ancestor(dir: &Path, group_dirs: &HashMap<PathBuf, String>) -> Option<String> {
+    dir.ancestors()
+        .skip(1) // skip `dir` itself — we want a strict ancestor
+        .find_map(|anc| group_dirs.get(anc).cloned())
+}