refactor(apply): split reconcile_group_structure_tx create/reparent branches

Review #4 (LOW, cleanup). `reconcile_group_structure_tx` carried a
`#[allow(clippy::too_many_lines)]` over a ~145-line body doing create +
reparent + attach/RBAC/lock in one loop. Extract the create branch
(`create_group_node_tx`) and the divergence-resolution branch
(`reparent_diverged_group_tx`) into helper methods, threading the shared
attach-ceiling / principal / mode context through a small `ReconcileCtx` — the
loop is now thin enough to drop the allow. Also improve the child-before-parent
error in `resolve_declared_parent` to hint that group nodes must be ordered
parents-first (only a hand-rolled bundle hits it; the CLI already sorts).

No behavior change. Documents the Tier-1 review follow-ups in the design doc.

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

crates/manager-core/migrations/0066_projects.sql

@@ -0,0 +1,41 @@
+-- §7 multi-repo ownership: the `projects` registry + wiring the inert
+-- `owner_project` seam.
+--
+-- A "project" is a repo-root that declaratively MANAGES a slice of the shared
+-- group tree (contains its manifest as something it applies, not merely
+-- references). §7's model is SINGLE-OWNER-PER-NODE: each group node is owned by
+-- exactly one project; the first `pic apply` that declares a `[project]` claims
+-- the node, and a second repo applying to an owned node is refused unless it
+-- passes a group-admin-gated `--takeover`. Ownership ⟂ RBAC — it records which
+-- manifest is authoritative, not whether a principal may act.
+--
+-- Identity is a committed `[project] slug` (stable across clones); the server
+-- assigns the UUID (server-internal). The `owner_project UUID` column already
+-- exists on `groups` (0047, inert) — this migration creates the table it was
+-- always meant to reference and wires the FK. Apps carry NO owner column: an
+-- app inherits ownership from its NEAREST claimed ancestor group (the ancestor
+-- walk is the isolation boundary), faithful to §7's "don't co-own a node —
+-- split config downward" corollary.
+
+CREATE TABLE projects (
+    id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    -- Instance-global identity, declared in the repo's committed root manifest
+    -- and frozen. First apply with a new slug registers the project.
+    slug       TEXT NOT NULL UNIQUE,
+    name       TEXT NOT NULL,
+    -- The admin who first registered it. SET NULL (not CASCADE) — losing the
+    -- creator's account must not orphan-delete the project or un-claim its tree.
+    created_by UUID REFERENCES admin_users(id) ON DELETE SET NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Wire the inert 0047 seam. ON DELETE SET NULL un-claims the owned nodes (they
+-- fall back to UI/API-owned) rather than cascading a destructive tree delete —
+-- deleting a project must never destroy the groups/apps it happened to manage.
+ALTER TABLE groups
+    ADD CONSTRAINT groups_owner_project_fk
+    FOREIGN KEY (owner_project) REFERENCES projects(id) ON DELETE SET NULL;
+
+-- Ownership lookups: "what does project X own?" (pic projects ls) and the
+-- LEFT JOIN behind pic groups ls' owner column.
+CREATE INDEX groups_owner_project_idx ON groups (owner_project);

crates/manager-core/src/apply_api.rs

@@ -18,8 +18,8 @@ use serde_json::json;
 use crate::app_repo::AppRepository;
 use crate::apply_service::{
     ApplyError, ApplyOwner, ApplyReport, ApplyService, Bundle, BundleTrigger, CollectionInfo,
-    ExtensionPointInfo, NodeKind, PlanResult, RouteTemplateInfo, SuppressionInfo, TreeBundle,
-    TreePlanResult, TriggerTemplateInfo,
+    ExtensionPointInfo, NodeKind, OwnershipClaim, PlanResult, ProjectDecl, RouteTemplateInfo,
+    StructureMode, SuppressionInfo, TreeBundle, TreePlanResult, TriggerTemplateInfo,
 };
 use crate::authz::{require, AuthzDenied, Capability};
 use crate::group_repo::GroupRepository;
@@ -179,6 +179,23 @@ async fn group_extension_points_handler(
     Ok(Json(report))
 }
 
+/// A `plan` request (§7 M3): the desired bundle plus the optional declaring
+/// `[project]`, so the plan can preview the ownership outcome + attach ceiling.
+#[derive(Deserialize)]
+pub struct PlanRequest {
+    // Flattened so the bundle fields sit at the JSON top level — restoring the
+    // pre-§7 plan wire, which took a BARE `Json<Bundle>`. An older `pic` that
+    // POSTs a bare bundle still deserializes (`project` defaults to `None`), and
+    // a newer client sends the same bundle fields plus a top-level `project`
+    // key. (Apply is deliberately NOT flattened — its pre-§7 wire was already
+    // `{bundle, prune, …}`, so flattening it would instead break old apply
+    // clients; the two endpoints are asymmetric because their histories are.)
+    #[serde(flatten)]
+    pub bundle: Bundle,
+    #[serde(default)]
+    pub project: Option<ProjectDecl>,
+}
+
 #[derive(Deserialize)]
 pub struct ApplyRequest {
     pub bundle: Bundle,
@@ -188,6 +205,12 @@ pub struct ApplyRequest {
     /// refuses (409) if the app's live state has changed since.
     #[serde(default)]
     pub expected_token: Option<String>,
+    /// §7 ownership: the declaring repo's `[project]` (absent = no claim).
+    #[serde(default)]
+    pub project: Option<ProjectDecl>,
+    /// §7 ownership: reassign a node owned by another project (group-admin).
+    #[serde(default)]
+    pub takeover: bool,
 }
 
 async fn apply_handler(
@@ -198,12 +221,17 @@ async fn apply_handler(
 ) -> Result<Json<ApplyReport>, ApplyError> {
     let app_id = resolve_app_id(svc.apps.as_ref(), &id_or_slug).await?;
     require_app_node_writes(&svc, &principal, app_id, &req.bundle, req.prune).await?;
+    let claim = OwnershipClaim {
+        project: req.project,
+        takeover: req.takeover,
+        principal: &principal,
+    };
     let report = svc
         .apply(
             app_id,
             &req.bundle,
             req.prune,
-            principal.user_id,
+            &claim,
             req.expected_token.as_deref(),
         )
         .await?;
@@ -214,7 +242,7 @@ async fn plan_handler(
     State(svc): State<ApplyService>,
     Extension(principal): Extension<Principal>,
     Path(id_or_slug): Path<String>,
-    Json(bundle): Json<Bundle>,
+    Json(req): Json<PlanRequest>,
 ) -> Result<Json<PlanResult>, ApplyError> {
     let app_id = resolve_app_id(svc.apps.as_ref(), &id_or_slug).await?;
     // NOTE: the returned `Plan` discloses live secret NAMES (not values). That
@@ -226,7 +254,7 @@ async fn plan_handler(
     require(svc.authz.as_ref(), &principal, Capability::AppRead(app_id))
         .await
         .map_err(map_authz)?;
-    let plan = svc.plan(app_id, &bundle).await?;
+    let plan = svc.plan(app_id, &req.bundle, req.project.as_ref()).await?;
     Ok(Json(plan))
 }
 
@@ -240,7 +268,7 @@ async fn group_plan_handler(
     State(svc): State<ApplyService>,
     Extension(principal): Extension<Principal>,
     Path(id_or_slug): Path<String>,
-    Json(bundle): Json<Bundle>,
+    Json(req): Json<PlanRequest>,
 ) -> Result<Json<PlanResult>, ApplyError> {
     let group_id = resolve_group_id(svc.groups.as_ref(), &id_or_slug).await?;
     // Plan discloses the group's script + var + secret names; viewer-tier reads.
@@ -251,7 +279,13 @@ async fn group_plan_handler(
     )
     .await
     .map_err(map_authz)?;
-    let plan = svc.plan_owner(ApplyOwner::Group(group_id), &bundle).await?;
+    let plan = svc
+        .plan_owner(
+            ApplyOwner::Group(group_id),
+            &req.bundle,
+            req.project.as_ref(),
+        )
+        .await?;
     Ok(Json(plan))
 }
 
@@ -262,13 +296,19 @@ async fn group_apply_handler(
     Json(req): Json<ApplyRequest>,
 ) -> Result<Json<ApplyReport>, ApplyError> {
     let group_id = resolve_group_id(svc.groups.as_ref(), &id_or_slug).await?;
-    require_group_node_writes(&svc, &principal, group_id, &req.bundle, req.prune).await?;
+    svc.require_group_node_writes(&principal, group_id, &req.bundle, req.prune)
+        .await?;
+    let claim = OwnershipClaim {
+        project: req.project,
+        takeover: req.takeover,
+        principal: &principal,
+    };
     let report = svc
         .apply_owner(
             ApplyOwner::Group(group_id),
             &req.bundle,
             req.prune,
-            principal.user_id,
+            &claim,
             req.expected_token.as_deref(),
         )
         .await?;
@@ -287,15 +327,35 @@ struct TreeApplyRequest {
     prune: bool,
     #[serde(default)]
     expected_token: Option<String>,
+    /// §7 ownership: one `[project]` for the whole tree (the root manifest's).
+    #[serde(default)]
+    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)]
+struct TreePlanRequest {
+    // Flattened for the same bare-bundle wire compatibility as `PlanRequest`.
+    #[serde(flatten)]
+    bundle: TreeBundle,
+    #[serde(default)]
+    project: Option<ProjectDecl>,
 }
 
 async fn tree_plan_handler(
     State(svc): State<ApplyService>,
     Extension(principal): Extension<Principal>,
-    Json(bundle): Json<TreeBundle>,
+    Json(req): Json<TreePlanRequest>,
 ) -> Result<Json<TreePlanResult>, ApplyError> {
-    authz_tree(&svc, &principal, &bundle, None).await?;
-    Ok(Json(svc.plan_tree(&bundle).await?))
+    authz_tree(&svc, &principal, &req.bundle, None).await?;
+    Ok(Json(
+        svc.plan_tree(&req.bundle, req.project.as_ref()).await?,
+    ))
 }
 
 async fn tree_apply_handler(
@@ -304,12 +364,18 @@ async fn tree_apply_handler(
     Json(req): Json<TreeApplyRequest>,
 ) -> Result<Json<ApplyReport>, ApplyError> {
     authz_tree(&svc, &principal, &req.bundle, Some(req.prune)).await?;
+    let claim = OwnershipClaim {
+        project: req.project,
+        takeover: req.takeover,
+        principal: &principal,
+    };
     let report = svc
         .apply_tree(
             &req.bundle,
             req.prune,
-            principal.user_id,
+            &claim,
             req.expected_token.as_deref(),
+            req.structure_mode,
         )
         .await?;
     Ok(Json(report))
@@ -341,10 +407,22 @@ 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)
+                        svc.require_group_node_writes(principal, group_id, &node.bundle, prune)
                             .await?;
                     }
                     None => {
@@ -427,36 +505,6 @@ async fn require_app_node_writes(
     Ok(())
 }
 
-/// Group-node write caps (editor-tier): scripts → `GroupScriptsWrite`, vars →
-/// `GroupVarsWrite`, both on prune.
-async fn require_group_node_writes(
-    svc: &ApplyService,
-    principal: &Principal,
-    group_id: GroupId,
-    bundle: &Bundle,
-    prune: bool,
-) -> Result<(), ApplyError> {
-    if prune || !bundle.scripts.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            principal,
-            Capability::GroupScriptsWrite(group_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    if prune || !bundle.vars.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            principal,
-            Capability::GroupVarsWrite(group_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    Ok(())
-}
-
 /// Resolve a slug-or-id path param to a `GroupId`, mapping miss → 404.
 async fn resolve_group_id(
     groups: &dyn GroupRepository,
@@ -499,11 +547,15 @@ 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::Invalid(_) | Self::OutsideAttachPoint(_) | Self::StructuralDivergence(_) => (
                 StatusCode::UNPROCESSABLE_ENTITY,
                 json!({ "error": self.to_string() }),
             ),
             Self::StateMoved => (StatusCode::CONFLICT, json!({ "error": self.to_string() })),
+            // §7 ownership conflict — actionable (declare [project], --takeover).
+            Self::OwnershipConflict(_) => {
+                (StatusCode::CONFLICT, json!({ "error": self.to_string() }))
+            }
             Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })),
             Self::AuthzRepo(e) => {
                 tracing::error!(error = %e, "apply authz repo error");
@@ -523,3 +575,41 @@ impl IntoResponse for ApplyError {
         (status, Json(body)).into_response()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // §7 wire compat: the plan endpoints `#[serde(flatten)]` the bundle, so a
+    // pre-§7 client that POSTs a BARE bundle (no `{bundle: ...}` wrapper, no
+    // `project`) must still deserialize — `project` defaults to `None`.
+    #[test]
+    fn plan_request_accepts_a_bare_bundle() {
+        let bare = serde_json::json!({
+            "scripts": [],
+            "secrets": ["TOKEN"],
+        });
+        let req: PlanRequest = serde_json::from_value(bare).expect("bare bundle deserializes");
+        assert!(req.project.is_none());
+        assert_eq!(req.bundle.secrets, vec!["TOKEN".to_string()]);
+    }
+
+    // The newer wire shape — the same bundle fields at the top level plus a
+    // `project` key (what the current CLI sends) — also deserializes.
+    #[test]
+    fn plan_request_accepts_a_flattened_project() {
+        let with_proj = serde_json::json!({
+            "scripts": [],
+            "project": { "slug": "acme" },
+        });
+        let req: PlanRequest = serde_json::from_value(with_proj).expect("flattened project");
+        assert_eq!(req.project.expect("project").slug, "acme");
+    }
+
+    #[test]
+    fn tree_plan_request_accepts_a_bare_bundle() {
+        let bare = serde_json::json!({ "nodes": [] });
+        let req: TreePlanRequest = serde_json::from_value(bare).expect("bare tree bundle");
+        assert!(req.project.is_none());
+    }
+}

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,11 +53,141 @@ 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()),
+    }
+}
+
+/// 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.
+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()))
+}
+
+/// §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
     /// (org structure), so callers assemble the hierarchy in memory.
     async fn list(&self) -> Result<Vec<Group>, GroupRepositoryError>;
+    /// Every group paired with its owning project's slug (`None` when the node
+    /// is unclaimed — no manifest owns it). One LEFT JOIN instead of an N+1
+    /// per-group project lookup; backs `pic groups ls`' owner column (§7).
+    async fn list_with_owner(&self) -> Result<Vec<(Group, Option<String>)>, GroupRepositoryError>;
     async fn get_by_id(&self, id: GroupId) -> Result<Option<Group>, GroupRepositoryError>;
     async fn get_by_slug(&self, slug: &str) -> Result<Option<Group>, GroupRepositoryError>;
     /// Direct children groups of `parent`.
@@ -105,8 +235,8 @@ impl PostgresGroupRepository {
     }
 }
 
-const GROUP_COLS: &str =
-    "id, parent_id, slug, name, description, structure_version, created_at, updated_at";
+const GROUP_COLS: &str = "id, parent_id, slug, name, description, structure_version, created_at, \
+     updated_at, owner_project";
 
 #[async_trait]
 impl GroupRepository for PostgresGroupRepository {
@@ -119,6 +249,24 @@ impl GroupRepository for PostgresGroupRepository {
         Ok(rows.into_iter().map(Into::into).collect())
     }
 
+    async fn list_with_owner(&self) -> Result<Vec<(Group, Option<String>)>, GroupRepositoryError> {
+        // Columns are qualified (`g.`) because `id`/`slug`/`name` exist on both
+        // groups and projects — a bare GROUP_COLS would be ambiguous here.
+        let rows = sqlx::query_as::<_, GroupWithOwnerRow>(
+            "SELECT g.id, g.parent_id, g.slug, g.name, g.description, \
+                    g.structure_version, g.created_at, g.updated_at, g.owner_project, \
+                    p.slug AS owner_slug \
+             FROM groups g LEFT JOIN projects p ON p.id = g.owner_project \
+             ORDER BY g.name",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| (r.group.into(), r.owner_slug))
+            .collect())
+    }
+
     async fn get_by_id(&self, id: GroupId) -> Result<Option<Group>, GroupRepositoryError> {
         let row = sqlx::query_as::<_, GroupRow>(&format!(
             "SELECT {GROUP_COLS} FROM groups WHERE id = $1"
@@ -157,7 +305,8 @@ impl GroupRepository for PostgresGroupRepository {
                  SELECT {GROUP_COLS}, 0 AS depth FROM groups WHERE id = $1
                UNION ALL
                  SELECT g.id, g.parent_id, g.slug, g.name, g.description, \
-                        g.structure_version, g.created_at, g.updated_at, c.depth + 1 \
+                        g.structure_version, g.created_at, g.updated_at, g.owner_project, \
+                        c.depth + 1 \
                  FROM groups g JOIN chain c ON g.id = c.parent_id \
                  WHERE c.depth < 64
              )
@@ -253,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> {
@@ -349,6 +444,16 @@ struct GroupRow {
     structure_version: i64,
     created_at: chrono::DateTime<chrono::Utc>,
     updated_at: chrono::DateTime<chrono::Utc>,
+    owner_project: Option<Uuid>,
+}
+
+/// `list_with_owner` row: a `GroupRow` flattened with its owning project's
+/// slug (`NULL` when the group is unclaimed).
+#[derive(sqlx::FromRow)]
+struct GroupWithOwnerRow {
+    #[sqlx(flatten)]
+    group: GroupRow,
+    owner_slug: Option<String>,
 }
 
 impl From<GroupRow> for Group {
@@ -360,6 +465,7 @@ impl From<GroupRow> for Group {
             name: r.name,
             description: r.description,
             structure_version: r.structure_version,
+            owner_project: r.owner_project.map(Into::into),
             created_at: r.created_at,
             updated_at: r.updated_at,
         }

crates/manager-core/src/groups_api.rs

@@ -158,11 +158,28 @@ pub struct PatchMemberRequest {
 /// any authenticated admin sees the full tree; per-action authz still
 /// gates every mutation and all app access. Tighten in Phase 3 when groups
 /// carry inheritable config.
+/// §7: a group plus its owning project's slug (`None` = unclaimed / UI-owned).
+/// The group fields are flattened, so existing consumers (dashboard, `pic
+/// groups tree`) keep working; `owner` is a new, optional field.
+#[derive(Serialize)]
+struct GroupListItem {
+    #[serde(flatten)]
+    group: Group,
+    owner: Option<String>,
+}
+
 async fn list_groups(
     State(s): State<GroupsState>,
     Extension(_principal): Extension<Principal>,
-) -> Result<Json<Vec<Group>>, GroupsApiError> {
-    Ok(Json(s.groups.list().await?))
+) -> Result<Json<Vec<GroupListItem>>, GroupsApiError> {
+    let items = s
+        .groups
+        .list_with_owner()
+        .await?
+        .into_iter()
+        .map(|(group, owner)| GroupListItem { group, owner })
+        .collect();
+    Ok(Json(items))
 }
 
 async fn create_group(
@@ -434,25 +451,32 @@ async fn resolve_group(groups: &dyn GroupRepository, ident: &str) -> Result<Grou
     found.ok_or_else(|| GroupsApiError::GroupNotFound(ident.to_string()))
 }
 
-/// Same rule as app slugs: `^[a-z0-9][a-z0-9-]{0,62}$`, no reserved words.
-fn validate_slug(slug: &str) -> Result<(), GroupsApiError> {
-    let invalid = |reason: &str| GroupsApiError::InvalidSlug(format!("{slug:?}: {reason}"));
+/// The shared slug FORMAT rule (`^[a-z0-9][a-z0-9-]{0,62}$`), returning the
+/// human reason on failure so each caller can wrap it in its own error type.
+/// Shared by group slugs and §7 project slugs; the reserved-word check is
+/// layered on separately by callers that route slugs (groups), not by projects
+/// (never routed). See [`crate::apply_service`]'s `validate_project_slug`.
+pub(crate) fn validate_slug_format(slug: &str) -> Result<(), String> {
     if slug.is_empty() || slug.len() > SLUG_MAX {
-        return Err(invalid("must be 1–63 characters"));
+        return Err("must be 1–63 characters".to_string());
     }
-    let mut chars = slug.chars();
-    let first = chars.next().unwrap();
+    let first = slug.chars().next().expect("non-empty checked above");
     if !(first.is_ascii_lowercase() || first.is_ascii_digit()) {
-        return Err(invalid("must start with a lowercase letter or digit"));
+        return Err("must start with a lowercase letter or digit".to_string());
     }
     if !slug
         .chars()
         .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
     {
-        return Err(invalid(
-            "may contain only lowercase letters, digits, and hyphens",
-        ));
+        return Err("may contain only lowercase letters, digits, and hyphens".to_string());
     }
+    Ok(())
+}
+
+/// Same rule as app slugs: `^[a-z0-9][a-z0-9-]{0,62}$`, no reserved words.
+fn validate_slug(slug: &str) -> Result<(), GroupsApiError> {
+    let invalid = |reason: &str| GroupsApiError::InvalidSlug(format!("{slug:?}: {reason}"));
+    validate_slug_format(slug).map_err(|reason| invalid(&reason))?;
     if RESERVED_SLUGS.contains(&slug) {
         return Err(invalid("is a reserved word"));
     }

crates/manager-core/src/lib.rs

@@ -79,6 +79,8 @@ pub mod module_source;
 pub mod outbox_event_emitter;
 pub mod outbox_repo;
 pub mod principal_resolver;
+pub mod project_repo;
+pub mod projects_api;
 pub mod pubsub_repo;
 pub mod pubsub_service;
 pub mod queue_repo;
@@ -223,6 +225,8 @@ pub use outbox_repo::{
     NewOutboxRow, OutboxRepo, OutboxRepoError, OutboxRow, OutboxSourceKind, PostgresOutboxRepo,
 };
 pub use principal_resolver::{AdminPrincipalResolver, PrincipalResolver, PrincipalResolverError};
+pub use project_repo::{PostgresProjectRepository, ProjectRepository, ProjectRepositoryError};
+pub use projects_api::{projects_router, ProjectsApiError, ProjectsState};
 pub use pubsub_repo::{PostgresPubsubRepo, PublishCtx, PubsubRepo, PubsubRepoError};
 pub use pubsub_service::{PubsubServiceImpl, SubscriberTokenConfig};
 pub use realtime_authority::RealtimeAuthorityImpl;

crates/manager-core/src/project_repo.rs

@@ -0,0 +1,118 @@
+//! Read access over the `projects` registry (§7 multi-repo ownership).
+//!
+//! A project is registered by the APPLY path — an upsert-by-slug inside the
+//! reconcile transaction (see `apply_service`), so a claim registers the
+//! project atomically with the write. This repo is the READ side only, used
+//! for ownership visibility (`pic projects ls`, the `pic groups ls` owner
+//! column) and tests. Writes deliberately do not live here.
+
+use async_trait::async_trait;
+use picloud_shared::{Project, ProjectId};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ProjectRepositoryError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[async_trait]
+pub trait ProjectRepository: Send + Sync {
+    /// Every project on the instance, ordered by slug.
+    async fn list(&self) -> Result<Vec<Project>, ProjectRepositoryError>;
+    /// Every project paired with how many group nodes it owns — backs
+    /// `pic projects ls`. One GROUP BY instead of N per-project counts.
+    async fn list_with_counts(&self) -> Result<Vec<(Project, i64)>, ProjectRepositoryError>;
+    async fn get_by_id(&self, id: ProjectId) -> Result<Option<Project>, ProjectRepositoryError>;
+    async fn get_by_slug(&self, slug: &str) -> Result<Option<Project>, ProjectRepositoryError>;
+}
+
+pub struct PostgresProjectRepository {
+    pool: PgPool,
+}
+
+impl PostgresProjectRepository {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+const PROJECT_COLS: &str = "id, slug, name, created_by, created_at";
+
+#[async_trait]
+impl ProjectRepository for PostgresProjectRepository {
+    async fn list_with_counts(&self) -> Result<Vec<(Project, i64)>, ProjectRepositoryError> {
+        let rows = sqlx::query_as::<_, ProjectCountRow>(
+            "SELECT p.id, p.slug, p.name, p.created_by, p.created_at, \
+                    COUNT(g.id) AS owned_groups \
+             FROM projects p LEFT JOIN groups g ON g.owner_project = p.id \
+             GROUP BY p.id ORDER BY p.slug",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| (r.project.into(), r.owned_groups))
+            .collect())
+    }
+
+    async fn list(&self) -> Result<Vec<Project>, ProjectRepositoryError> {
+        let rows = sqlx::query_as::<_, ProjectRow>(&format!(
+            "SELECT {PROJECT_COLS} FROM projects ORDER BY slug"
+        ))
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows.into_iter().map(Into::into).collect())
+    }
+
+    async fn get_by_id(&self, id: ProjectId) -> Result<Option<Project>, ProjectRepositoryError> {
+        let row = sqlx::query_as::<_, ProjectRow>(&format!(
+            "SELECT {PROJECT_COLS} FROM projects WHERE id = $1"
+        ))
+        .bind(id.into_inner())
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(Into::into))
+    }
+
+    async fn get_by_slug(&self, slug: &str) -> Result<Option<Project>, ProjectRepositoryError> {
+        let row = sqlx::query_as::<_, ProjectRow>(&format!(
+            "SELECT {PROJECT_COLS} FROM projects WHERE slug = $1"
+        ))
+        .bind(slug)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(Into::into))
+    }
+}
+
+#[derive(sqlx::FromRow)]
+struct ProjectRow {
+    id: Uuid,
+    slug: String,
+    name: String,
+    created_by: Option<Uuid>,
+    created_at: chrono::DateTime<chrono::Utc>,
+}
+
+/// `list_with_counts` row: a project flattened with its owned-group count.
+#[derive(sqlx::FromRow)]
+struct ProjectCountRow {
+    #[sqlx(flatten)]
+    project: ProjectRow,
+    owned_groups: i64,
+}
+
+impl From<ProjectRow> for Project {
+    fn from(r: ProjectRow) -> Self {
+        Self {
+            id: r.id.into(),
+            slug: r.slug,
+            name: r.name,
+            created_by: r.created_by.map(Into::into),
+            created_at: r.created_at,
+        }
+    }
+}

crates/manager-core/src/projects_api.rs

@@ -0,0 +1,81 @@
+//! §7 M3 read-only projects surface: `GET /api/v1/admin/projects` — every
+//! project (repo-root) with the count of group nodes it owns. Backs
+//! `pic projects ls`. Behind the `/admin` middleware, so any authenticated
+//! admin can list them (a project reveals only org structure, like the groups
+//! list); ownership WRITES stay in the apply path.
+
+use std::sync::Arc;
+
+use axum::extract::State;
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Json, Response};
+use axum::routing::get;
+use axum::{Extension, Router};
+use chrono::{DateTime, Utc};
+use picloud_shared::Principal;
+use serde::Serialize;
+use serde_json::json;
+
+use crate::project_repo::{ProjectRepository, ProjectRepositoryError};
+
+#[derive(Clone)]
+pub struct ProjectsState {
+    pub projects: Arc<dyn ProjectRepository>,
+}
+
+/// Mounted under `/api/v1/admin`.
+pub fn projects_router(state: ProjectsState) -> Router {
+    Router::new()
+        .route("/projects", get(list_projects))
+        .with_state(state)
+}
+
+#[derive(Serialize)]
+struct ProjectListItem {
+    slug: String,
+    name: String,
+    owned_groups: i64,
+    created_at: DateTime<Utc>,
+}
+
+async fn list_projects(
+    State(s): State<ProjectsState>,
+    Extension(_principal): Extension<Principal>,
+) -> Result<Json<Vec<ProjectListItem>>, ProjectsApiError> {
+    let items = s
+        .projects
+        .list_with_counts()
+        .await?
+        .into_iter()
+        .map(|(p, owned_groups)| ProjectListItem {
+            slug: p.slug,
+            name: p.name,
+            owned_groups,
+            created_at: p.created_at,
+        })
+        .collect();
+    Ok(Json(items))
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ProjectsApiError {
+    #[error("backend: {0}")]
+    Backend(String),
+}
+
+impl From<ProjectRepositoryError> for ProjectsApiError {
+    fn from(e: ProjectRepositoryError) -> Self {
+        Self::Backend(e.to_string())
+    }
+}
+
+impl IntoResponse for ProjectsApiError {
+    fn into_response(self) -> Response {
+        tracing::error!(error = %self, "projects api error");
+        (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({ "error": "internal error" })),
+        )
+            .into_response()
+    }
+}

crates/manager-core/tests/expected_schema.txt

@@ -318,6 +318,13 @@ table: outbox
   claimed_by: text NULL
   created_at: timestamp with time zone NOT NULL default=now()
 
+table: projects
+  id: uuid NOT NULL default=gen_random_uuid()
+  slug: text NOT NULL
+  name: text NOT NULL
+  created_by: uuid NULL
+  created_at: timestamp with time zone NOT NULL default=now()
+
 table: pubsub_trigger_details
   trigger_id: uuid NOT NULL
   topic_pattern: text NOT NULL
@@ -579,6 +586,7 @@ indexes on group_queue_messages:
   idx_group_queue_messages_group_collection: public.group_queue_messages USING btree (group_id, collection)
 
 indexes on groups:
+  groups_owner_project_idx: public.groups USING btree (owner_project)
   groups_parent_id_idx: public.groups USING btree (parent_id)
   groups_pkey: public.groups USING btree (id)
   groups_slug_key: public.groups USING btree (slug)
@@ -595,6 +603,10 @@ indexes on outbox:
   idx_outbox_due: public.outbox USING btree (next_attempt_at) WHERE (claimed_at IS NULL)
   outbox_pkey: public.outbox USING btree (id)
 
+indexes on projects:
+  projects_pkey: public.projects USING btree (id)
+  projects_slug_key: public.projects USING btree (slug)
+
 indexes on pubsub_trigger_details:
   pubsub_trigger_details_pkey: public.pubsub_trigger_details USING btree (trigger_id)
 
@@ -817,6 +829,7 @@ constraints on group_queue_messages:
   [PRIMARY KEY] group_queue_messages_pkey: PRIMARY KEY (id)
 
 constraints on groups:
+  [FOREIGN KEY] groups_owner_project_fk: FOREIGN KEY (owner_project) REFERENCES projects(id) ON DELETE SET NULL
   [FOREIGN KEY] groups_parent_id_fkey: FOREIGN KEY (parent_id) REFERENCES groups(id) ON DELETE RESTRICT
   [PRIMARY KEY] groups_pkey: PRIMARY KEY (id)
   [UNIQUE] groups_slug_key: UNIQUE (slug)
@@ -834,6 +847,11 @@ constraints on outbox:
   [FOREIGN KEY] outbox_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
   [PRIMARY KEY] outbox_pkey: PRIMARY KEY (id)
 
+constraints on projects:
+  [FOREIGN KEY] projects_created_by_fkey: FOREIGN KEY (created_by) REFERENCES admin_users(id) ON DELETE SET NULL
+  [PRIMARY KEY] projects_pkey: PRIMARY KEY (id)
+  [UNIQUE] projects_slug_key: UNIQUE (slug)
+
 constraints on pubsub_trigger_details:
   [FOREIGN KEY] pubsub_trigger_details_trigger_id_fkey: FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE
   [PRIMARY KEY] pubsub_trigger_details_pkey: PRIMARY KEY (trigger_id)
@@ -973,3 +991,4 @@ constraints on vars:
   0063: materialized unique
   0064: shared topic queue kinds
   0065: group queues
+  0066: projects

crates/manager-core/tests/projects_repo.rs

@@ -0,0 +1,106 @@
+//! §7 repo round-trip: `list_with_owner` pairs each group with its owning
+//! project's slug (`None` when unclaimed), and `ancestors()` surfaces
+//! `owner_project` for every node — the nearest-claimed fold the ownership
+//! claim path (M1.3) relies on.
+//!
+//! Deterministic; skips cleanly when `DATABASE_URL` is unset.
+
+use picloud_manager_core::group_repo::{GroupRepository, PostgresGroupRepository};
+use picloud_manager_core::project_repo::{PostgresProjectRepository, ProjectRepository};
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("projects_repo: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&url)
+        .await
+        .expect("connect to DATABASE_URL");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("apply migrations");
+    Some(pool)
+}
+
+fn uniq(prefix: &str) -> String {
+    format!("{prefix}-{}", &Uuid::new_v4().to_string()[..8])
+}
+
+#[tokio::test]
+async fn list_with_owner_and_ancestors_surface_ownership() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let groups = PostgresGroupRepository::new(pool.clone());
+    let projects = PostgresProjectRepository::new(pool.clone());
+
+    // A project + a parent/child group pair; claim the CHILD only (a parent is
+    // deliberately left unclaimed to pin the `None` path).
+    let proj_slug = uniq("proj");
+    let (proj_id,): (Uuid,) =
+        sqlx::query_as("INSERT INTO projects (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(&proj_slug)
+            .fetch_one(&pool)
+            .await
+            .expect("insert project");
+
+    let parent = groups
+        .create(&uniq("parent"), "Parent", None, None)
+        .await
+        .expect("create parent");
+    let child_slug = uniq("child");
+    let child = groups
+        .create(&child_slug, "Child", None, Some(parent.id))
+        .await
+        .expect("create child");
+    sqlx::query("UPDATE groups SET owner_project = $1 WHERE id = $2")
+        .bind(proj_id)
+        .bind(child.id.into_inner())
+        .execute(&pool)
+        .await
+        .expect("claim child");
+
+    // list_with_owner: the child pairs with the project slug; the parent is
+    // unclaimed (None). (The list spans the whole instance — filter by slug.)
+    let listed = groups.list_with_owner().await.expect("list_with_owner");
+    let child_row = listed
+        .iter()
+        .find(|(g, _)| g.slug == child_slug)
+        .expect("child present in list_with_owner");
+    assert_eq!(child_row.1.as_deref(), Some(proj_slug.as_str()));
+    let parent_row = listed
+        .iter()
+        .find(|(g, _)| g.id == parent.id)
+        .expect("parent present");
+    assert_eq!(parent_row.1, None, "unclaimed group has no owner slug");
+
+    // ancestors(child) is nearest-first and carries owner_project on each node —
+    // the child holds its claim, the parent is None. This is the exact fold
+    // input the nearest-claimed resolution walks.
+    let chain = groups.ancestors(child.id).await.expect("ancestors");
+    assert_eq!(chain[0].id, child.id, "ancestors is nearest-first");
+    assert!(
+        chain[0].owner_project.is_some(),
+        "child ancestor row carries its claim"
+    );
+    let parent_in_chain = chain
+        .iter()
+        .find(|g| g.id == parent.id)
+        .expect("parent in chain");
+    assert!(parent_in_chain.owner_project.is_none());
+
+    // Project reads round-trip by slug and by id.
+    let by_slug = projects.get_by_slug(&proj_slug).await.expect("get_by_slug");
+    assert_eq!(by_slug.as_ref().map(|p| p.id.into_inner()), Some(proj_id));
+    let by_id = projects
+        .get_by_id(by_slug.unwrap().id)
+        .await
+        .expect("get_by_id");
+    assert_eq!(by_id.map(|p| p.slug), Some(proj_slug));
+}

crates/picloud-cli/src/client.rs

@@ -39,6 +39,25 @@ fn seg(s: &str) -> String {
     utf8_percent_encode(s, PATH_SEGMENT).to_string()
 }
 
+/// Build a `plan` request body: the bundle fields at the JSON top level plus an
+/// optional `project` key (§7). The plan endpoints `#[serde(flatten)]` the
+/// bundle, so a bare bundle (no `[project]`) is the pre-§7 wire shape — this
+/// keeps `pic plan` compatible across a CLI/server version skew in either
+/// direction (an old server ignores the extra `project`; a new server accepts
+/// the bare bundle).
+fn plan_body(
+    bundle: &serde_json::Value,
+    project: Option<&crate::manifest::ManifestProject>,
+) -> Result<serde_json::Value> {
+    let mut body = bundle.clone();
+    if let Some(p) = project {
+        if let Some(obj) = body.as_object_mut() {
+            obj.insert("project".to_string(), serde_json::to_value(p)?);
+        }
+    }
+    Ok(body)
+}
+
 pub struct Client {
     http: reqwest::Client,
     url: String,
@@ -153,7 +172,8 @@ impl Client {
     // --- Groups (Phase 2) -------------------------------------------------
 
     /// `GET /api/v1/admin/groups` — the full flat list (assemble the tree
-    /// client-side from `parent_id`).
+    /// client-side from `parent_id`). Ignores the §7 `owner` field the server
+    /// also returns (see [`Self::groups_list_with_owner`]).
     pub async fn groups_list(&self) -> Result<Vec<Group>> {
         let resp = self
             .request(Method::GET, "/api/v1/admin/groups")
@@ -162,6 +182,27 @@ impl Client {
         decode(resp).await
     }
 
+    /// Same endpoint as [`Self::groups_list`], but captures the §7 owning
+    /// project's slug alongside each group — backs `pic groups ls`' owner
+    /// column.
+    pub async fn groups_list_with_owner(&self) -> Result<Vec<GroupListItem>> {
+        let resp = self
+            .request(Method::GET, "/api/v1/admin/groups")
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `GET /api/v1/admin/projects` — every §7 project + its owned-group count.
+    /// Backs `pic projects ls`.
+    pub async fn projects_list(&self) -> Result<Vec<ProjectDto>> {
+        let resp = self
+            .request(Method::GET, "/api/v1/admin/projects")
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
     /// `GET /api/v1/admin/groups/{id_or_slug}` — group + path + children.
     pub async fn groups_get(&self, ident: &str) -> Result<GroupDetailDto> {
         let ident = seg(ident);
@@ -1208,14 +1249,16 @@ impl Client {
         kind: NodeKind,
         slug: &str,
         bundle: &serde_json::Value,
+        project: Option<&crate::manifest::ManifestProject>,
     ) -> Result<PlanDto> {
         let slug = seg(slug);
+        let body = plan_body(bundle, project)?;
         let resp = self
             .request(
                 Method::POST,
                 &format!("/api/v1/admin/{}/{slug}/plan", kind.path()),
             )
-            .json(bundle)
+            .json(&body)
             .send()
             .await?;
         decode(resp).await
@@ -1223,12 +1266,15 @@ impl Client {
 
     /// `POST /api/v1/admin/{apps|groups}/{id_or_slug}/apply` — reconcile an
     /// app OR group node in one transaction (Phase 5).
+    #[allow(clippy::too_many_arguments)]
     pub async fn apply_node(
         &self,
         kind: NodeKind,
         slug: &str,
         bundle: &serde_json::Value,
         prune: bool,
+        project: Option<&crate::manifest::ManifestProject>,
+        takeover: bool,
         expected_token: Option<&str>,
     ) -> Result<ApplyReportDto> {
         let slug = seg(slug);
@@ -1236,6 +1282,8 @@ impl Client {
             "bundle": bundle,
             "prune": prune,
             "expected_token": expected_token,
+            "project": project,
+            "takeover": takeover,
         });
         let resp = self
             .request(
@@ -1245,27 +1293,27 @@ impl Client {
             .json(&body)
             .send()
             .await?;
-        // The apply endpoint returns 409 only for a stale bound plan
-        // (`StateMoved`): the app changed since `pic plan` recorded its
-        // token. Surface an actionable next step instead of a bare
-        // `HTTP 409`.
+        // The apply endpoint returns 409 for a stale bound plan (`StateMoved`)
+        // OR a §7 ownership conflict. Both server messages are self-contained
+        // and actionable, so surface the message verbatim.
         if resp.status() == reqwest::StatusCode::CONFLICT {
             let body = resp.text().await.unwrap_or_default();
             let msg = parse_error_body(&body).unwrap_or(body);
-            return Err(anyhow!(
-                "{msg}\nThe live state changed since your last `pic plan`. Re-run \
-                 `pic plan` to review the new diff, then `pic apply` — or \
-                 `pic apply --force` to apply without re-reviewing."
-            ));
+            return Err(anyhow!("{msg}"));
         }
         decode(resp).await
     }
 
     /// `POST /api/v1/admin/tree/plan` — diff a whole project tree (Phase 5).
-    pub async fn plan_tree(&self, bundle: &serde_json::Value) -> Result<TreePlanDto> {
+    pub async fn plan_tree(
+        &self,
+        bundle: &serde_json::Value,
+        project: Option<&crate::manifest::ManifestProject>,
+    ) -> Result<TreePlanDto> {
+        let body = plan_body(bundle, project)?;
         let resp = self
             .request(Method::POST, "/api/v1/admin/tree/plan")
-            .json(bundle)
+            .json(&body)
             .send()
             .await?;
         decode(resp).await
@@ -1273,29 +1321,39 @@ 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,
         prune: bool,
+        project: Option<&crate::manifest::ManifestProject>,
+        takeover: bool,
         expected_token: Option<&str>,
+        structure_mode: &str,
     ) -> Result<ApplyReportDto> {
         let body = serde_json::json!({
             "bundle": bundle,
             "prune": prune,
             "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?;
-        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}\nThe project tree changed since your last `pic plan`. Re-run \
-                 `pic plan --dir` to review, then `pic apply --dir` — or add `--force`."
-            ));
+            return Err(anyhow!("{msg}"));
         }
         decode(resp).await
     }
@@ -1513,6 +1571,9 @@ pub struct PlanDto {
     /// in `.picloud/` and replayed to `apply` for the bound-plan check.
     #[serde(default)]
     pub state_token: String,
+    /// §7 M3: the ownership outcome this apply would produce.
+    #[serde(default)]
+    pub ownership: Option<OwnershipPreviewDto>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -1523,6 +1584,24 @@ pub struct ChangeDto {
     pub detail: Option<String>,
 }
 
+/// §7 M3 plan preview: `action` ∈ claim/owned/conflict/unclaimed; `owner` is
+/// the current owner's slug (for owned/conflict); `blast_radius` (M3.2) lists
+/// other projects' apps a group change fans out to.
+#[derive(Debug, Deserialize)]
+pub struct OwnershipPreviewDto {
+    pub action: String,
+    #[serde(default)]
+    pub owner: Option<String>,
+    #[serde(default)]
+    pub blast_radius: Vec<ProjectImpactDto>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ProjectImpactDto {
+    pub project: String,
+    pub apps: u32,
+}
+
 /// Response of `POST .../tree/plan` (Phase 5): a per-node diff + one combined
 /// bound-plan token for the whole subtree.
 #[derive(Debug, Deserialize)]
@@ -1553,6 +1632,22 @@ pub struct NodePlanDto {
     pub collections: Vec<ChangeDto>,
     #[serde(default)]
     pub suppressions: Vec<ChangeDto>,
+    /// §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.
@@ -1652,6 +1747,28 @@ pub struct GroupDetailDto {
     pub apps: Vec<App>,
 }
 
+/// One row of `pic groups ls`: a group plus its §7 owning project's slug
+/// (`None` = unclaimed / UI-owned). The group fields flatten in, so `.group.*`
+/// reaches them.
+#[derive(Debug, Deserialize)]
+pub struct GroupListItem {
+    #[serde(flatten)]
+    pub group: Group,
+    #[serde(default)]
+    pub owner: Option<String>,
+}
+
+/// One row of `pic projects ls` (§7 M3): a project + how many group nodes it
+/// owns.
+#[derive(Debug, Deserialize)]
+pub struct ProjectDto {
+    pub slug: String,
+    pub name: String,
+    #[serde(default)]
+    pub owned_groups: i64,
+    pub created_at: DateTime<Utc>,
+}
+
 #[derive(Debug, Serialize)]
 pub struct CreateRouteBody<'a> {
     pub host_kind: HostKind,

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

@@ -11,21 +11,32 @@ use anyhow::{Context, Result};
 use crate::client::{Client, NodeKind};
 use crate::cmds::plan::build_bundle;
 use crate::config;
-use crate::manifest::Manifest;
+use crate::manifest::{Manifest, ManifestProject};
 use crate::output::{KvBlock, OutputMode};
 
+#[allow(clippy::too_many_arguments)]
 pub async fn run(
     manifest_path: &Path,
     env: Option<&str>,
     prune: bool,
     yes: bool,
     force: bool,
+    takeover: bool,
+    approve: &[String],
     mode: OutputMode,
 ) -> Result<()> {
     let creds = config::resolve()?;
     let client = Client::from_creds(&creds)?;
 
     let manifest = Manifest::load_with_env(manifest_path, env)?;
+    // §3 M3: the env-approval gate is a property of the PROJECT, which lives at
+    // the repo root. A leaf manifest carries no `[project]`, so find the
+    // governing one (own block, else the nearest ancestor's) before gating.
+    let governing = manifest
+        .project
+        .clone()
+        .or_else(|| find_governing_project(manifest_path));
+    require_env_approval(governing.as_ref(), env, approve)?;
     let base_dir = manifest_path.parent().unwrap_or_else(|| Path::new("."));
     let bundle = build_bundle(&manifest, base_dir)?;
     let kind = if manifest.is_group() {
@@ -52,7 +63,15 @@ pub async fn run(
     };
 
     let report = client
-        .apply_node(kind, &slug, &bundle, prune, expected_token.as_deref())
+        .apply_node(
+            kind,
+            &slug,
+            &bundle,
+            prune,
+            manifest.project.as_ref(),
+            takeover,
+            expected_token.as_deref(),
+        )
         .await?;
     crate::linkstate::clear_plan(base_dir, &slug);
 
@@ -115,17 +134,22 @@ 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,
+    approve: &[String],
     env: Option<&str>,
     mode: OutputMode,
 ) -> Result<()> {
     let creds = config::resolve()?;
     let client = Client::from_creds(&creds)?;
-    let (bundle, node_count) = crate::discover::build_tree(dir, env)?;
+    let (bundle, node_count, project) = crate::discover::build_tree(dir, env)?;
+    require_env_approval(project.as_ref(), env, approve)?;
 
     if prune && !yes {
         confirm_prune(&format!("{node_count} node(s) under {}", dir.display()))?;
@@ -141,7 +165,14 @@ pub async fn run_tree(
     };
 
     let report = client
-        .apply_tree(&bundle, prune, expected_token.as_deref())
+        .apply_tree(
+            &bundle,
+            prune,
+            project.as_ref(),
+            takeover,
+            expected_token.as_deref(),
+            structure_mode,
+        )
         .await?;
     crate::linkstate::clear_plan(dir, token_key);
 
@@ -205,6 +236,46 @@ pub async fn run_tree(
 /// irreversible. Require an explicit go-ahead: an interactive `y`, or `--yes`
 /// for non-interactive/CI use. Refuse a non-interactive prune without `--yes`
 /// rather than silently deleting (review the deletions first with `pic plan`).
+/// §3 M3: refuse `pic apply --env <e>` to a confirm-required env
+/// (`[project.environments]` with `confirm = true`) unless it was explicitly
+/// `--approve <e>`d. A blanket `--yes` does NOT cover a gated env (§4.2 "CI must
+/// opt in per environment"). An unlisted or `confirm = false` env applies freely.
+/// §3 M3: find the `[project]` that governs a single-file apply. The applied
+/// manifest's own block wins (the caller checks that first); otherwise walk up
+/// from its directory to the nearest ancestor `picloud.toml` that declares one
+/// (the repo root), so `pic apply --file services/api/picloud.toml` still
+/// honors the repo's env-approval gate. Loaded WITHOUT the env overlay — only
+/// the `[project]` block matters here, and `load_with_env` would require an
+/// overlay next to every ancestor and spuriously skip the governing manifest.
+fn find_governing_project(manifest_path: &Path) -> Option<ManifestProject> {
+    let dir = manifest_path.parent()?;
+    dir.ancestors().skip(1).find_map(|anc| {
+        let candidate = anc.join(crate::manifest::MANIFEST_FILE);
+        Manifest::load(&candidate).ok().and_then(|m| m.project)
+    })
+}
+
+fn require_env_approval(
+    project: Option<&ManifestProject>,
+    env: Option<&str>,
+    approve: &[String],
+) -> Result<()> {
+    let (Some(env), Some(project)) = (env, project) else {
+        return Ok(());
+    };
+    let Some(policy) = project.environments.get(env) else {
+        return Ok(());
+    };
+    if policy.confirm && !approve.iter().any(|a| a == env) {
+        anyhow::bail!(
+            "environment `{env}` is confirm-required (`[project.environments]`); \
+             re-run with `--approve {env}` to apply to it — a blanket `--yes` does \
+             not cover a gated environment"
+        );
+    }
+    Ok(())
+}
+
 fn confirm_prune(slug: &str) -> Result<()> {
     if !std::io::stdin().is_terminal() {
         anyhow::bail!(

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

@@ -16,19 +16,24 @@ use crate::output::{KvBlock, OutputMode, Table};
 pub async fn ls(mode: OutputMode) -> Result<()> {
     let creds = config::resolve()?;
     let client = Client::from_creds(&creds)?;
-    let groups = client.groups_list().await?;
-    let mut table = Table::new(["slug", "name", "parent", "created_at"]);
-    let by_id: BTreeMap<_, _> = groups.iter().map(|g| (g.id, g.slug.clone())).collect();
+    let groups = client.groups_list_with_owner().await?;
+    let mut table = Table::new(["slug", "name", "parent", "owner", "created_at"]);
+    let by_id: BTreeMap<_, _> = groups
+        .iter()
+        .map(|g| (g.group.id, g.group.slug.clone()))
+        .collect();
     for g in &groups {
         let parent = g
+            .group
             .parent_id
             .and_then(|p| by_id.get(&p).cloned())
             .unwrap_or_else(|| "-".into());
         table.row([
-            g.slug.clone(),
-            g.name.clone(),
+            g.group.slug.clone(),
+            g.group.name.clone(),
             parent,
-            g.created_at.to_rfc3339(),
+            g.owner.clone().unwrap_or_else(|| "—".into()),
+            g.group.created_at.to_rfc3339(),
         ]);
     }
     table.print(mode);

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

@@ -116,6 +116,7 @@ fn scaffold_manifest(slug: &str, name: &str) -> Manifest {
             extension_points: Vec::new(),
         }),
         group: None,
+        project: None,
         scripts: vec![ManifestScript {
             name: "hello".into(),
             file: "scripts/hello.rhai".into(),

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

@@ -16,6 +16,7 @@ pub mod logout;
 pub mod logs;
 pub mod members;
 pub mod plan;
+pub mod projects;
 pub mod pull;
 pub mod queues;
 pub mod routes;

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

@@ -31,7 +31,9 @@ pub async fn run(manifest_path: &Path, env: Option<&str>, mode: OutputMode) -> R
         NodeKind::App
     };
 
-    let plan = client.plan_node(kind, manifest.slug(), &bundle).await?;
+    let plan = client
+        .plan_node(kind, manifest.slug(), &bundle, manifest.project.as_ref())
+        .await?;
     // Record the bound-plan token so a subsequent `pic apply` can detect the
     // node changing underneath the reviewed plan (best-effort — a read-only
     // plan still succeeds if the project dir isn't writable).
@@ -49,8 +51,8 @@ pub async fn run(manifest_path: &Path, env: Option<&str>, mode: OutputMode) -> R
 pub async fn run_tree(dir: &Path, env: Option<&str>, mode: OutputMode) -> Result<()> {
     let creds = config::resolve()?;
     let client = Client::from_creds(&creds)?;
-    let (bundle, _count) = crate::discover::build_tree(dir, env)?;
-    let plan = client.plan_tree(&bundle).await?;
+    let (bundle, _count, project) = crate::discover::build_tree(dir, env)?;
+    let plan = client.plan_tree(&bundle, project.as_ref()).await?;
     if !plan.state_token.is_empty() {
         if let Err(e) = crate::linkstate::write_plan(dir, TREE_TOKEN_KEY, &plan.state_token) {
             eprintln!("warning: could not record plan state for `pic apply`: {e}");
@@ -64,6 +66,39 @@ 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(),
+                "ownership".to_string(),
+                o.action.clone(),
+                o.owner.clone().unwrap_or_default(),
+                String::new(),
+            ]);
+            for b in &o.blast_radius {
+                table.row([
+                    node.clone(),
+                    "blast_radius".to_string(),
+                    b.project.clone(),
+                    format!("{} app(s)", b.apps),
+                    String::new(),
+                ]);
+            }
+        }
         let groups: [(&str, &Vec<ChangeDto>); 8] = [
             ("script", &n.scripts),
             ("route", &n.routes),
@@ -185,6 +220,22 @@ fn tagged(kind: &str, spec: impl Serialize) -> Result<Value> {
 
 fn render(plan: &PlanDto, mode: OutputMode) {
     let mut table = Table::new(["kind", "op", "resource", "detail"]);
+    if let Some(o) = &plan.ownership {
+        table.row([
+            "ownership".to_string(),
+            o.action.clone(),
+            o.owner.clone().unwrap_or_default(),
+            String::new(),
+        ]);
+        for b in &o.blast_radius {
+            table.row([
+                "blast_radius".to_string(),
+                b.project.clone(),
+                format!("{} app(s)", b.apps),
+                String::new(),
+            ]);
+        }
+    }
     let groups: [(&str, &Vec<ChangeDto>); 8] = [
         ("script", &plan.scripts),
         ("route", &plan.routes),

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

@@ -0,0 +1,29 @@
+//! `pic projects` — read-only view of §7 multi-repo ownership projects.
+//!
+//! A project is the repo-root that declaratively owns a slice of the group
+//! tree. Ownership is CLAIMED by `pic apply` (first apply with a `[project]`
+//! slug registers it); this command only lists what exists.
+
+use anyhow::Result;
+
+use crate::client::Client;
+use crate::config;
+use crate::output::{OutputMode, Table};
+
+/// `pic projects ls` — every project + how many group nodes it owns.
+pub async fn ls(mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let projects = client.projects_list().await?;
+    let mut table = Table::new(["slug", "name", "owned_groups", "created_at"]);
+    for p in &projects {
+        table.row([
+            p.slug.clone(),
+            p.name.clone(),
+            p.owned_groups.to_string(),
+            p.created_at.to_rfc3339(),
+        ]);
+    }
+    table.print(mode);
+    Ok(())
+}

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

@@ -242,6 +242,7 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) ->
             extension_points,
         }),
         group: None,
+        project: None,
         scripts: manifest_scripts,
         routes,
         triggers: manifest_triggers,

crates/picloud-cli/src/discover.rs

@@ -1,17 +1,22 @@
 //! 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);
-//! the server resolves ancestry from its own group tree, so the on-disk
-//! nesting is organizational, not authoritative here.
+//! apply}` endpoints consume. Each manifest becomes one node (app or group).
+//!
+//! §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};
 use serde_json::{json, Value};
 
 use crate::cmds::plan::build_bundle;
-use crate::manifest::{Manifest, MANIFEST_FILE};
+use crate::manifest::{Manifest, ManifestProject, MANIFEST_FILE};
 
 /// Find every `picloud.toml` under `root` (recursively), skipping the
 /// `.picloud/` link-state dir, `.git`, and `scripts/` source dirs. Per-env
@@ -46,9 +51,14 @@ fn walk(dir: &Path, out: &mut Vec<PathBuf>) -> Result<()> {
 }
 
 /// Build the wire tree bundle (`{ nodes: [{kind, slug, bundle}] }`) from every
-/// manifest under `root`. Returns the JSON plus the node count. Rejects a tree
-/// that names the same (kind, slug) twice.
-pub fn build_tree(root: &Path, env: Option<&str>) -> Result<(Value, usize)> {
+/// manifest under `root`. Returns the JSON, the node count, and the owning
+/// `[project]` (§7) declared in the tree ROOT's `picloud.toml` (if any) — one
+/// project owns the whole tree; a `[project]` in any other node is ignored with
+/// a note. Rejects a tree that names the same (kind, slug) twice.
+pub fn build_tree(
+    root: &Path,
+    env: Option<&str>,
+) -> Result<(Value, usize, Option<ManifestProject>)> {
     let paths = find_manifests(root)?;
     if paths.is_empty() {
         bail!(
@@ -56,20 +66,93 @@ pub fn build_tree(root: &Path, env: Option<&str>) -> Result<(Value, usize)> {
             root.display()
         );
     }
-    let mut nodes = Vec::with_capacity(paths.len());
-    let mut seen: HashSet<(String, String)> = HashSet::new();
+    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;
     for path in &paths {
         let manifest = Manifest::load_with_env(path, env)
             .with_context(|| format!("loading {}", path.display()))?;
+        if let Some(p) = &manifest.project {
+            if path.as_path() == root_manifest {
+                project = Some(p.clone());
+            } else {
+                eprintln!(
+                    "note: [project] in {} is ignored — declare it in the tree root {}",
+                    path.display(),
+                    root_manifest.display()
+                );
+                // §3 M3: a non-root [project] that carries env-approval policy is
+                // a silent-safety trap — the gate is dropped with the block. Call
+                // it out explicitly so a misplaced gate isn't mistaken for active.
+                if !p.environments.is_empty() {
+                    eprintln!(
+                        "warning: its [project.environments] approval gating is NOT enforced — \
+                         only the tree root's [project] governs env approval"
+                    );
+                }
+            }
+        }
+        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 kind = if manifest.is_group() { "group" } else { "app" };
+        let bundle = build_bundle(manifest, base_dir)?;
+        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))
+    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())
 }

crates/picloud-cli/src/main.rs

@@ -58,6 +58,12 @@ enum Cmd {
         cmd: GroupsCmd,
     },
 
+    /// §7 multi-repo ownership projects (read-only).
+    Projects {
+        #[command(subcommand)]
+        cmd: ProjectsCmd,
+    },
+
     /// Script management.
     Scripts {
         #[command(subcommand)]
@@ -246,10 +252,27 @@ struct ApplyArgs {
     /// since the last `pic plan`).
     #[arg(long)]
     force: bool,
+    /// §7 multi-repo ownership: reassign a group node owned by another
+    /// `[project]` to this manifest's project. Requires group-admin on the
+    /// node; without it an apply to an owned node is refused (409).
+    #[arg(long)]
+    takeover: bool,
     /// Merge the `picloud.<env>.toml` overlay (per-env slug + secrets) on
     /// 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,
+    /// §3 M3: approve applying to a confirm-required environment
+    /// (`[project.environments]`). Repeatable; each must match the `--env`.
+    #[arg(long = "approve")]
+    approve: Vec<String>,
 }
 
 #[derive(Args)]
@@ -485,6 +508,12 @@ enum AppsCmd {
     },
 }
 
+#[derive(Subcommand)]
+enum ProjectsCmd {
+    /// List all projects + how many group nodes each owns.
+    Ls,
+}
+
 #[derive(Subcommand)]
 enum GroupsCmd {
     /// List all groups (flat).
@@ -1416,11 +1445,21 @@ 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.approve,
                     args.env.as_deref(),
                     mode,
                 )
@@ -1433,6 +1472,8 @@ async fn main() -> ExitCode {
                     args.prune,
                     args.yes,
                     args.force,
+                    args.takeover,
+                    &args.approve,
                     mode,
                 )
                 .await
@@ -1502,6 +1543,9 @@ async fn main() -> ExitCode {
                     cmd: DomainsCmd::Rm { app, domain_id },
                 },
         } => cmds::apps_domains::rm(&app, &domain_id).await,
+        Cmd::Projects {
+            cmd: ProjectsCmd::Ls,
+        } => cmds::projects::ls(mode).await,
         Cmd::Groups { cmd: GroupsCmd::Ls } => cmds::groups::ls(mode).await,
         Cmd::Groups {
             cmd: GroupsCmd::Tree,

crates/picloud-cli/src/manifest.rs

@@ -37,6 +37,12 @@ pub struct Manifest {
     pub app: Option<ManifestApp>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub group: Option<ManifestGroup>,
+    /// `[project]` (§7 multi-repo ownership) — the repo-root that OWNS the nodes
+    /// it applies. Independent of the app/group node kind; declared in the
+    /// repo's root manifest and threaded onto every apply from it. The first
+    /// apply with a new slug claims each group node it touches.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub project: Option<ManifestProject>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub scripts: Vec<ManifestScript>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
@@ -263,6 +269,41 @@ pub struct ManifestGroup {
     pub collections: Vec<CollectionDecl>,
 }
 
+/// A `[project]` block (§7 multi-repo ownership): the identity of the repo-root
+/// managing these nodes. The `slug` is committed (stable across clones); the
+/// server assigns the UUID and the first apply registers it.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ManifestProject {
+    pub slug: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+    /// `parent_group` (§7/§6) — the pre-existing group this repo binds UNDER
+    /// (its attach point). Applies are refused for any node not strictly within
+    /// that subtree. Absent = instance root = no ceiling (the default).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub parent_group: Option<String>,
+    /// §3 M3 per-env approval policy: `[project.environments]` maps an env name
+    /// to its policy. A `confirm = true` env requires an explicit `--approve
+    /// <env>` on `pic apply --env <env>` (a blanket `--yes` does NOT cover it —
+    /// §4.2 "CI must opt in per environment"). Client-side gating only; never
+    /// sent to the server.
+    #[serde(default, skip_serializing)]
+    pub environments: std::collections::BTreeMap<String, EnvPolicy>,
+}
+
+/// §3 M3: the apply-gating policy for one environment.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct EnvPolicy {
+    /// Require an explicit `--approve <env>` to apply to this env. NOT
+    /// `#[serde(default)]`: an env listed with an empty policy (`prod = {}`)
+    /// must fail to load (`missing field 'confirm'`) rather than silently
+    /// parse as un-gated — listing an env to protect it, then having the gate
+    /// quietly absent, is the failure mode to avoid.
+    pub confirm: bool,
+}
+
 /// The store kind of a shared collection (§11.6).
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
@@ -596,6 +637,7 @@ mod tests {
                 extension_points: vec!["theme".into()],
             }),
             group: None,
+            project: None,
             scripts: vec![
                 ManifestScript {
                     name: "create-post".into(),
@@ -759,6 +801,7 @@ mod tests {
                 extension_points: vec![],
             }),
             group: None,
+            project: None,
             scripts: vec![],
             routes: vec![],
             triggers: ManifestTriggers::default(),
@@ -904,13 +947,16 @@ mod tests {
         )
         .expect("group cron template parses");
         assert_eq!(cron.triggers.cron.len(), 1);
-        // ...but an email template on a group is still rejected (per-app secret).
-        let err = Manifest::parse(
+        // §4.5 M5.5: a group email template is now ALLOWED too — it materializes
+        // per descendant app and resolves its `inbound_secret_ref` against the
+        // GROUP's own secret store at apply (the server rejects it there if the
+        // secret is unset). Parse no longer refuses it.
+        let email = Manifest::parse(
             "[group]\nslug = \"acme\"\nname = \"ACME\"\n\n\
              [[triggers.email]]\nscript = \"shared\"\ninbound_secret_ref = \"sig\"\n",
         )
-        .expect_err("group email template is rejected");
-        assert!(err.to_string().contains("email"), "got: {err}");
+        .expect("group email template parses");
+        assert_eq!(email.triggers.email.len(), 1);
 
         // Neither / both is rejected.
         Manifest::parse("[vars]\nx = 1\n").expect_err("no [app] or [group]");
@@ -918,6 +964,49 @@ mod tests {
             .expect_err("both [app] and [group]");
     }
 
+    #[test]
+    fn project_block_parses_independently_of_node_kind() {
+        // §7: [project] is orthogonal to the app/group XOR — either node kind
+        // may declare the owning project.
+        let a = Manifest::parse(
+            "[project]\nslug = \"platform\"\nname = \"Platform\"\n\n\
+             [app]\nslug = \"web\"\nname = \"Web\"\n",
+        )
+        .expect("[app] + [project] parses");
+        assert_eq!(a.project.as_ref().unwrap().slug, "platform");
+        assert_eq!(
+            a.project.as_ref().unwrap().name.as_deref(),
+            Some("Platform")
+        );
+
+        let g = Manifest::parse(
+            "[project]\nslug = \"platform\"\n\n[group]\nslug = \"acme\"\nname = \"ACME\"\n",
+        )
+        .expect("[group] + [project] parses");
+        assert_eq!(g.project.as_ref().unwrap().slug, "platform");
+        assert!(
+            g.project.as_ref().unwrap().name.is_none(),
+            "name is optional"
+        );
+
+        // Absent [project] → None (backward-compatible).
+        assert!(Manifest::parse("[app]\nslug=\"w\"\nname=\"W\"\n")
+            .unwrap()
+            .project
+            .is_none());
+        // Unknown key inside [project] is rejected (deny_unknown_fields).
+        Manifest::parse("[project]\nslug=\"p\"\nbogus=1\n[app]\nslug=\"w\"\nname=\"W\"\n")
+            .expect_err("unknown [project] key rejected");
+
+        // §6/§7 M2: the optional attach-point `parent_group`.
+        let m = Manifest::parse(
+            "[project]\nslug = \"p\"\nparent_group = \"acme\"\n\n\
+             [group]\nslug = \"team\"\nname = \"T\"\n",
+        )
+        .expect("parent_group parses");
+        assert_eq!(m.project.unwrap().parent_group.as_deref(), Some("acme"));
+    }
+
     #[test]
     fn overlay_vars_override_base_per_key() {
         let mut base = sample();

crates/picloud-cli/tests/apply_ownership.rs

@@ -0,0 +1,489 @@
+//! §7 multi-repo ownership — the claim, end to end via `pic`.
+//!
+//! A `[project]` claims a group node on first apply; the SAME project re-applies
+//! freely; a DIFFERENT project applying to that node is refused (409) unless it
+//! `--takeover`s (which requires group-admin). An app inherits ownership from
+//! its nearest claimed ancestor group. `pic groups ls` shows the owning
+//! project's slug. The pure claim policy is unit-tested in manager-core; this
+//! journey pins the wire + CLI + authz interaction.
+
+use std::fs;
+use std::path::Path;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::{AppGuard, GroupGuard, ScriptGuard};
+
+fn manifest_dir() -> TempDir {
+    let dir = TempDir::new().expect("tempdir");
+    fs::create_dir_all(dir.path().join("scripts")).expect("scripts dir");
+    fs::write(
+        dir.path().join("scripts/shared.rhai"),
+        r#"log::info("shared"); "ok""#,
+    )
+    .unwrap();
+    dir
+}
+
+/// A `[project]` + `[group]` manifest (with one group script) in `dir`.
+fn write_group_manifest(dir: &Path, project: &str, group: &str) {
+    let m = format!(
+        "[project]\nslug = \"{project}\"\nname = \"Proj\"\n\n\
+         [group]\nslug = \"{group}\"\nname = \"Grp\"\n\n\
+         [[scripts]]\nname = \"shared\"\nfile = \"scripts/shared.rhai\"\n"
+    );
+    fs::write(dir.join("picloud.toml"), m).unwrap();
+}
+
+fn group_script_id(env: &common::TestEnv, group: &str) -> String {
+    let ls = common::pic_as(env)
+        .args(["scripts", "ls", "--group", group])
+        .output()
+        .expect("scripts ls");
+    let table = String::from_utf8(ls.stdout).unwrap();
+    table
+        .lines()
+        .map(common::cells)
+        .find(|c| c.get(2) == Some(&"shared"))
+        .and_then(|c| c.first().map(|s| (*s).to_string()))
+        .unwrap_or_else(|| panic!("group script not found:\n{table}"))
+}
+
+/// The `owner` cell (index 3: slug, name, parent, OWNER, created_at) for
+/// `group` in `pic groups ls`.
+fn owner_cell(env: &common::TestEnv, group: &str) -> String {
+    let ls = common::pic_as(env)
+        .args(["groups", "ls"])
+        .output()
+        .expect("groups ls");
+    let table = String::from_utf8(ls.stdout).unwrap();
+    table
+        .lines()
+        .map(common::cells)
+        .find(|c| c.first() == Some(&group))
+        .and_then(|c| c.get(3).map(|s| (*s).to_string()))
+        .unwrap_or_else(|| panic!("group `{group}` not in groups ls:\n{table}"))
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn claim_conflict_takeover_and_app_inheritance() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let group = common::unique_slug("own-grp");
+    let proj_a = common::unique_slug("plat");
+    let proj_b = common::unique_slug("teamb");
+
+    // The group must pre-exist (groups pre-exist; the manifest owns content).
+    let _g = GroupGuard::new(&env.url, &env.token, &group);
+    common::pic_as(&env)
+        .args(["groups", "create", &group])
+        .assert()
+        .success();
+
+    // (1) First apply by project A CLAIMS the group.
+    let dir_a = manifest_dir();
+    write_group_manifest(dir_a.path(), &proj_a, &group);
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(dir_a.path().join("picloud.toml"))
+        .assert()
+        .success();
+    let _gs = ScriptGuard::new(&env.url, &env.token, &group_script_id(&env, &group));
+    assert_eq!(
+        owner_cell(&env, &group),
+        proj_a,
+        "the owner column must show the claiming project"
+    );
+
+    // (2) The SAME project re-applies with no conflict (idempotent).
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(dir_a.path().join("picloud.toml"))
+        .assert()
+        .success();
+
+    // (3) A DIFFERENT project applying to the owned group is refused (409),
+    // naming the current owner.
+    let dir_b = manifest_dir();
+    write_group_manifest(dir_b.path(), &proj_b, &group);
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(dir_b.path().join("picloud.toml"))
+        .output()
+        .expect("apply B");
+    assert!(!out.status.success(), "a second project must be refused");
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("owned by project") && err.contains(&proj_a),
+        "the conflict must name the owner `{proj_a}`:\n{err}"
+    );
+
+    // (4) --takeover (admin holds group-admin implicitly) reassigns to B.
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(dir_b.path().join("picloud.toml"))
+        .arg("--takeover")
+        .assert()
+        .success();
+    assert_eq!(
+        owner_cell(&env, &group),
+        proj_b,
+        "takeover must reassign the owner"
+    );
+
+    // (5) An app under the claimed group INHERITS ownership: the owning project
+    // applies fine; a foreign or absent project is refused.
+    let app = common::unique_slug("own-app");
+    let _a = AppGuard::new(&env.url, &env.token, &app);
+    common::pic_as(&env)
+        .args(["apps", "create", &app, "--group", &group])
+        .assert()
+        .success();
+    let app_dir = manifest_dir();
+    let app_manifest = |proj: Option<&str>| -> String {
+        let head = proj
+            .map(|p| format!("[project]\nslug = \"{p}\"\n\n"))
+            .unwrap_or_default();
+        format!("{head}[app]\nslug = \"{app}\"\nname = \"App\"\n")
+    };
+    // Project B owns the group → ok.
+    fs::write(
+        app_dir.path().join("picloud.toml"),
+        app_manifest(Some(&proj_b)),
+    )
+    .unwrap();
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(app_dir.path().join("picloud.toml"))
+        .assert()
+        .success();
+    // Project A no longer owns the group → refused.
+    fs::write(
+        app_dir.path().join("picloud.toml"),
+        app_manifest(Some(&proj_a)),
+    )
+    .unwrap();
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(app_dir.path().join("picloud.toml"))
+        .output()
+        .expect("apply app A");
+    assert!(
+        !out.status.success(),
+        "an app under a claimed group must match its owner"
+    );
+    // No [project] under a claimed subtree → refused.
+    fs::write(app_dir.path().join("picloud.toml"), app_manifest(None)).unwrap();
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(app_dir.path().join("picloud.toml"))
+        .output()
+        .expect("apply app none");
+    assert!(
+        !out.status.success(),
+        "a no-project app under a claimed subtree is refused"
+    );
+
+    // (6) --takeover is capability-gated: a member with only EDITOR on the group
+    // can reconcile it but must NOT be able to take it over (needs group-admin).
+    let mem = common::member::member_user(fx, &common::unique_slug("own-mem"));
+    common::member::grant_group_membership(fx, &group, &mem.id, "editor");
+    let menv = common::custom_env(&env.url, &mem.token);
+    common::seed_credentials(&menv, &mem.username);
+    let proj_c = common::unique_slug("teamc");
+    let dir_c = manifest_dir();
+    write_group_manifest(dir_c.path(), &proj_c, &group);
+    let out = common::pic_as(&menv)
+        .args(["apply", "--file"])
+        .arg(dir_c.path().join("picloud.toml"))
+        .arg("--takeover")
+        .output()
+        .expect("member takeover");
+    assert!(
+        !out.status.success(),
+        "a member without group-admin must not be able to --takeover"
+    );
+    // The owner is unchanged — the failed takeover rolled back.
+    assert_eq!(
+        owner_cell(&env, &group),
+        proj_b,
+        "a refused takeover must not change ownership"
+    );
+}
+
+/// The `name` cell (index 1: slug, NAME, owned_groups, created_at) for `project`
+/// in `pic projects ls`.
+fn project_name_cell(env: &common::TestEnv, project: &str) -> String {
+    let ls = common::pic_as(env)
+        .args(["projects", "ls"])
+        .output()
+        .expect("projects ls");
+    let table = String::from_utf8(ls.stdout).unwrap();
+    table
+        .lines()
+        .map(common::cells)
+        .find(|c| c.first() == Some(&project))
+        .and_then(|c| c.get(1).map(|s| (*s).to_string()))
+        .unwrap_or_else(|| panic!("project `{project}` not in projects ls:\n{table}"))
+}
+
+/// A name-less `[project]` re-apply must PRESERVE the display name set on the
+/// first apply — an omitted optional field never clobbers the stored name back
+/// to the slug (regression for the `ON CONFLICT DO UPDATE SET name` path).
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn reapply_without_name_preserves_project_name() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let group = common::unique_slug("name-grp");
+    let proj = common::unique_slug("name-proj");
+    let _g = GroupGuard::new(&env.url, &env.token, &group);
+    common::pic_as(&env)
+        .args(["groups", "create", &group])
+        .assert()
+        .success();
+
+    let dir = manifest_dir();
+    let apply = |m: &str| {
+        fs::write(dir.path().join("picloud.toml"), m).unwrap();
+        common::pic_as(&env)
+            .args(["apply", "--file"])
+            .arg(dir.path().join("picloud.toml"))
+            .assert()
+            .success();
+    };
+
+    // First apply declares a display name.
+    apply(&format!(
+        "[project]\nslug = \"{proj}\"\nname = \"Acme Platform\"\n\n\
+         [group]\nslug = \"{group}\"\nname = \"Grp\"\n"
+    ));
+    assert_eq!(
+        project_name_cell(&env, &proj),
+        "Acme Platform",
+        "the first apply records the declared name"
+    );
+
+    // A re-apply that OMITS `name` must not overwrite it with the slug.
+    apply(&format!(
+        "[project]\nslug = \"{proj}\"\n\n\
+         [group]\nslug = \"{group}\"\nname = \"Grp\"\n"
+    ));
+    assert_eq!(
+        project_name_cell(&env, &proj),
+        "Acme Platform",
+        "a name-less re-apply must preserve the stored name, not clobber it to the slug"
+    );
+}
+
+/// §6/§7 M2 — `[project] parent_group` is the ceiling: applies are refused for
+/// any node not strictly within the attach point's subtree.
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn attach_point_ceiling_bounds_the_subtree() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    // acme (root) → team (child); plus a sibling `outsider` root.
+    let acme = common::unique_slug("acme");
+    let team = common::unique_slug("team");
+    let outsider = common::unique_slug("outsider");
+    let _a = GroupGuard::new(&env.url, &env.token, &acme);
+    let _t = GroupGuard::new(&env.url, &env.token, &team);
+    let _o = GroupGuard::new(&env.url, &env.token, &outsider);
+    common::pic_as(&env)
+        .args(["groups", "create", &acme])
+        .assert()
+        .success();
+    common::pic_as(&env)
+        .args(["groups", "create", &team, "--parent", &acme])
+        .assert()
+        .success();
+    common::pic_as(&env)
+        .args(["groups", "create", &outsider])
+        .assert()
+        .success();
+
+    let proj = common::unique_slug("attach-p");
+    let dir = manifest_dir();
+    let apply = |m: &str| -> std::process::Output {
+        fs::write(dir.path().join("picloud.toml"), m).unwrap();
+        common::pic_as(&env)
+            .args(["apply", "--file"])
+            .arg(dir.path().join("picloud.toml"))
+            .output()
+            .expect("apply")
+    };
+
+    // A group node strictly BELOW the attach point → ok.
+    let out = apply(&format!(
+        "[project]\nslug = \"{proj}\"\nparent_group = \"{acme}\"\n\n\
+         [group]\nslug = \"{team}\"\nname = \"Team\"\n"
+    ));
+    assert!(
+        out.status.success(),
+        "a node below the attach point applies: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+
+    // The attach point ITSELF → refused (you can't apply above your local root).
+    let out = apply(&format!(
+        "[project]\nslug = \"{proj}\"\nparent_group = \"{acme}\"\n\n\
+         [group]\nslug = \"{acme}\"\nname = \"Acme\"\n"
+    ));
+    assert!(
+        !out.status.success(),
+        "applying the attach point itself must be refused"
+    );
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("attach point"),
+        "the refusal must mention the attach point:\n{err}"
+    );
+
+    // A SIBLING subtree (not under acme) → refused.
+    let out = apply(&format!(
+        "[project]\nslug = \"{proj}\"\nparent_group = \"{acme}\"\n\n\
+         [group]\nslug = \"{outsider}\"\nname = \"Out\"\n"
+    ));
+    assert!(
+        !out.status.success(),
+        "a sibling subtree is outside the attach point"
+    );
+}
+
+/// §7 M3 — `pic plan` previews the ownership outcome (claim / conflict) and the
+/// cross-repo blast radius (descendant apps owned by other projects) BEFORE any
+/// apply.
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn plan_previews_ownership_and_blast_radius() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    // acme (root) → team_a, team_b; an app under each, each team claimed by a
+    // DIFFERENT project.
+    let acme = common::unique_slug("br-acme");
+    let team_a = common::unique_slug("br-a");
+    let team_b = common::unique_slug("br-b");
+    let app_a = common::unique_slug("br-app-a");
+    let app_b = common::unique_slug("br-app-b");
+    let plat = common::unique_slug("plat");
+    let teamb = common::unique_slug("teamb");
+    let platform = common::unique_slug("platform");
+    let _g0 = GroupGuard::new(&env.url, &env.token, &acme);
+    let _g1 = GroupGuard::new(&env.url, &env.token, &team_a);
+    let _g2 = GroupGuard::new(&env.url, &env.token, &team_b);
+    let _a1 = AppGuard::new(&env.url, &env.token, &app_a);
+    let _a2 = AppGuard::new(&env.url, &env.token, &app_b);
+    for (slug, parent) in [
+        (&acme, None),
+        (&team_a, Some(&acme)),
+        (&team_b, Some(&acme)),
+    ] {
+        let mut c = common::pic_as(&env);
+        c.args(["groups", "create", slug]);
+        if let Some(p) = parent {
+            c.args(["--parent", p]);
+        }
+        c.assert().success();
+    }
+    common::pic_as(&env)
+        .args(["apps", "create", &app_a, "--group", &team_a])
+        .assert()
+        .success();
+    common::pic_as(&env)
+        .args(["apps", "create", &app_b, "--group", &team_b])
+        .assert()
+        .success();
+
+    // Claim team_a by `plat`, team_b by `teamb` (empty group applies).
+    let dir = manifest_dir();
+    let claim = |project: &str, group: &str| {
+        let m = format!(
+            "[project]\nslug = \"{project}\"\n\n[group]\nslug = \"{group}\"\nname = \"G\"\n"
+        );
+        fs::write(dir.path().join("picloud.toml"), m).unwrap();
+        common::pic_as(&env)
+            .args(["apply", "--file"])
+            .arg(dir.path().join("picloud.toml"))
+            .assert()
+            .success();
+    };
+    claim(&plat, &team_a);
+    claim(&teamb, &team_b);
+
+    // `pic projects ls` lists both registered projects with their owned-group
+    // counts (plat owns exactly team_a).
+    let projects = String::from_utf8(
+        common::pic_as(&env)
+            .args(["projects", "ls"])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    let plat_row = projects
+        .lines()
+        .map(common::cells)
+        .find(|c| c.first() == Some(&plat.as_str()))
+        .unwrap_or_else(|| panic!("plat not in projects ls:\n{projects}"));
+    assert_eq!(
+        plat_row.get(2).copied(),
+        Some("1"),
+        "plat owns exactly one group:\n{projects}"
+    );
+    assert!(
+        projects.contains(teamb.as_str()),
+        "teamb must be listed:\n{projects}"
+    );
+
+    // Plan acme with project `platform`: acme is unclaimed → action `claim`; the
+    // blast radius lists the OTHER projects' descendant apps (plat + teamb).
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{platform}\"\n\n[group]\nslug = \"{acme}\"\nname = \"Acme\"\n"
+        ),
+    )
+    .unwrap();
+    let out = common::pic_as(&env)
+        .args(["plan", "--file"])
+        .arg(dir.path().join("picloud.toml"))
+        .output()
+        .expect("plan acme");
+    let stdout = String::from_utf8_lossy(&out.stdout);
+    assert!(
+        stdout.contains("claim"),
+        "acme is unclaimed → the preview must show `claim`:\n{stdout}"
+    );
+    assert!(
+        stdout.contains("blast_radius") && stdout.contains(&plat) && stdout.contains(&teamb),
+        "the blast radius must list the other projects' apps (plat + teamb):\n{stdout}"
+    );
+
+    // Plan team_a with `platform`: it is owned by `plat` → action `conflict`.
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!("[project]\nslug = \"{platform}\"\n\n[group]\nslug = \"{team_a}\"\nname = \"A\"\n"),
+    )
+    .unwrap();
+    let out = common::pic_as(&env)
+        .args(["plan", "--file"])
+        .arg(dir.path().join("picloud.toml"))
+        .output()
+        .expect("plan team_a");
+    let stdout = String::from_utf8_lossy(&out.stdout);
+    assert!(
+        stdout.contains("conflict") && stdout.contains(&plat),
+        "planning a foreign-owned group must preview a conflict naming the owner:\n{stdout}"
+    );
+}

crates/picloud-cli/tests/cli.rs

@@ -16,6 +16,7 @@ mod common;
 mod admins;
 mod api_keys;
 mod apply;
+mod apply_ownership;
 mod apps;
 mod auth;
 mod collections;
@@ -23,8 +24,10 @@ mod config;
 mod dead_letters;
 mod email_queue;
 mod enabled;
+mod env_approval;
 mod env_overlay;
 mod extension_points;
+mod group_create;
 mod group_modules;
 mod group_routes;
 mod group_scripts;
@@ -48,6 +51,7 @@ mod shared_topics;
 mod shared_triggers;
 mod staleness;
 mod stateful_templates;
+mod structural_divergence;
 mod suppress;
 mod tree;
 mod triggers;

crates/picloud-cli/tests/common/member.rs

@@ -78,6 +78,29 @@ pub fn grant_membership(fx: &Fixture, app_slug: &str, user_id: &str, role: &str)
     );
 }
 
+/// `POST /api/v1/admin/groups/{slug}/members` — grant `role` to `user_id` on a
+/// GROUP (hierarchy-aware). §7 takeover tests use this to give a member an
+/// `editor` group role — enough to reconcile a group node, but NOT to
+/// `--takeover` (which needs `GroupAdmin`).
+pub fn grant_group_membership(fx: &Fixture, group_slug: &str, user_id: &str, role: &str) {
+    let client = reqwest::blocking::Client::new();
+    let resp = client
+        .post(format!(
+            "{}/api/v1/admin/groups/{}/members",
+            fx.url, group_slug
+        ))
+        .bearer_auth(&fx.admin_token)
+        .json(&json!({ "user_id": user_id, "role": role }))
+        .send()
+        .expect("grant group membership");
+    assert!(
+        resp.status().is_success(),
+        "grant group membership failed: {} {}",
+        resp.status(),
+        resp.text().unwrap_or_default(),
+    );
+}
+
 /// `PATCH /api/v1/admin/apps/{slug}/members/{user_id}` — promote/demote.
 pub fn update_membership(fx: &Fixture, app_slug: &str, user_id: &str, role: &str) {
     let client = reqwest::blocking::Client::new();

crates/picloud-cli/tests/env_approval.rs

@@ -0,0 +1,206 @@
+//! §3 M3 — per-env approval gating (`[project.environments]`).
+//!
+//! A `[project.environments]` block marks some envs confirm-required. Applying
+//! to such an env with `pic apply --env <e>` is refused unless it is explicitly
+//! `--approve <e>`d — and a blanket `--yes` does NOT cover it (§4.2, "CI must
+//! opt in per environment"). An unlisted or `confirm = false` env applies
+//! freely. The gate is client-side, so a refused apply never reaches the server.
+
+use std::fs;
+use std::path::Path;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::GroupGuard;
+
+/// A repo whose `[project]` gates `production` (confirm-required) but not
+/// `staging`, managing one pre-existing `[group]`. Minimal per-env overlay files
+/// exist so `--env` can load them.
+fn gated_repo(dir: &Path, project: &str, group: &str) {
+    fs::write(
+        dir.join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{project}\"\n\n\
+             [project.environments]\nproduction = {{ confirm = true }}\n\
+             staging = {{ confirm = false }}\n\n\
+             [group]\nslug = \"{group}\"\nname = \"Env Gated\"\n"
+        ),
+    )
+    .unwrap();
+    fs::write(
+        dir.join("picloud.production.toml"),
+        "# production overlay\n",
+    )
+    .unwrap();
+    fs::write(dir.join("picloud.staging.toml"), "# staging overlay\n").unwrap();
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn confirm_required_env_needs_explicit_approve() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let group = common::unique_slug("ea-grp");
+    let project = common::unique_slug("ea-proj");
+    let _g = GroupGuard::new(&env.url, &env.token, &group);
+    common::pic_as(&env)
+        .args(["groups", "create", &group])
+        .assert()
+        .success();
+
+    let dir = TempDir::new().unwrap();
+    gated_repo(dir.path(), &project, &group);
+    let manifest = dir.path().join("picloud.toml");
+    let apply = |args: &[&str]| -> std::process::Output {
+        let mut c = common::pic_as(&env);
+        c.args(["apply", "--file"]).arg(&manifest).args(args);
+        c.output().expect("apply")
+    };
+
+    // production is confirm-required → a bare `--env production` is refused,
+    // and the message names `--approve`. (Client-side: never hits the server.)
+    let out = apply(&["--env", "production"]);
+    assert!(!out.status.success(), "production must require approval");
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("approve") && err.contains("production"),
+        "the refusal must point at --approve production:\n{err}"
+    );
+
+    // A blanket `--yes` does NOT cover a gated env.
+    assert!(
+        !apply(&["--env", "production", "--yes"]).status.success(),
+        "--yes must not bypass a confirm-required env"
+    );
+
+    // `--approve production` lets it through.
+    assert!(
+        apply(&["--env", "production", "--approve", "production"])
+            .status
+            .success(),
+        "an explicit --approve production must apply"
+    );
+
+    // staging is `confirm = false` → applies freely, no approval needed.
+    assert!(
+        apply(&["--env", "staging"]).status.success(),
+        "an un-gated env must apply without --approve"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn an_env_listed_with_an_empty_policy_is_a_hard_error() {
+    // `production = {}` (no `confirm`) must FAIL to load, not silently parse as
+    // un-gated — you can't half-declare a gate. `confirm` is a required field.
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let group = common::unique_slug("eae-grp");
+    let project = common::unique_slug("eae-proj");
+    let _g = GroupGuard::new(&env.url, &env.token, &group);
+
+    let dir = TempDir::new().unwrap();
+    let manifest = dir.path().join("picloud.toml");
+    fs::write(
+        &manifest,
+        format!(
+            "[project]\nslug = \"{project}\"\n\n\
+             [project.environments]\nproduction = {{}}\n\n\
+             [group]\nslug = \"{group}\"\nname = \"Empty Policy\"\n"
+        ),
+    )
+    .unwrap();
+    fs::write(dir.path().join("picloud.production.toml"), "# overlay\n").unwrap();
+
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest)
+        .args(["--env", "production"])
+        .output()
+        .expect("apply");
+    assert!(
+        !out.status.success(),
+        "an env with an empty policy must be a load error, not a silent no-gate"
+    );
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("confirm"),
+        "the error must name the missing `confirm` field:\n{err}"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn leaf_file_apply_honors_the_root_env_gate() {
+    // `pic apply --file <leaf>` where the leaf carries no `[project]` still
+    // honors the gate declared in the repo root (found by walking up).
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base = common::unique_slug("eal-base");
+    let leaf = common::unique_slug("eal-leaf");
+    let project = common::unique_slug("eal-proj");
+    let _b = GroupGuard::new(&env.url, &env.token, &base);
+    let _l = GroupGuard::new(&env.url, &env.token, &leaf);
+    common::pic_as(&env)
+        .args(["groups", "create", &base])
+        .assert()
+        .success();
+    common::pic_as(&env)
+        .args(["groups", "create", &leaf, "--parent", &base])
+        .assert()
+        .success();
+
+    // Root manifest carries the gating `[project]`; the leaf (in a subdir) has
+    // no `[project]` of its own and is applied on its own with `--file`.
+    let dir = TempDir::new().unwrap();
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{project}\"\n\n\
+             [project.environments]\nproduction = {{ confirm = true }}\n\n\
+             [group]\nslug = \"{base}\"\nname = \"Base\"\n"
+        ),
+    )
+    .unwrap();
+    let sub = dir.path().join("sub");
+    fs::create_dir_all(&sub).unwrap();
+    let leaf_manifest = sub.join("picloud.toml");
+    fs::write(
+        &leaf_manifest,
+        format!("[group]\nslug = \"{leaf}\"\nname = \"Leaf\"\n"),
+    )
+    .unwrap();
+    fs::write(sub.join("picloud.production.toml"), "# overlay\n").unwrap();
+
+    // Without --approve: refused, because the gate is discovered up-tree.
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&leaf_manifest)
+        .args(["--env", "production"])
+        .output()
+        .expect("apply");
+    assert!(
+        !out.status.success(),
+        "a leaf --file apply must honor the root's env gate"
+    );
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("approve") && err.contains("production"),
+        "the refusal must point at --approve production:\n{err}"
+    );
+
+    // With --approve production: passes the gate and applies (leaf pre-exists).
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&leaf_manifest)
+        .args(["--env", "production", "--approve", "production"])
+        .assert()
+        .success();
+}

crates/picloud-cli/tests/group_create.rs

@@ -0,0 +1,265 @@
+//! §6 M1 — declarative group CREATE, end to end via `pic apply --dir`.
+//!
+//! A repo declares its group subtree by directory nesting: a `[group]` node's
+//! parent is the nearest ancestor directory that also holds a `[group]`, and the
+//! topmost group binds to the repo's `[project] parent_group` (attach point).
+//! The tree apply creates the missing groups (parents-first) in one transaction,
+//! claims them for the project, and is a no-op on re-apply. Creating a group is
+//! capability-gated (a member without group-admin is refused).
+
+use std::fs;
+use std::path::Path;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::GroupGuard;
+
+/// A two-level repo on disk: root `[group] {top}` (claimed by `{project}`,
+/// attached under `{attach}`) with a nested `sub/` `[group] {sub}`.
+fn nested_repo(dir: &Path, project: &str, attach: &str, top: &str, sub: &str) {
+    fs::write(
+        dir.join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{project}\"\nparent_group = \"{attach}\"\n\n\
+             [group]\nslug = \"{top}\"\nname = \"Top\"\n"
+        ),
+    )
+    .unwrap();
+    let subdir = dir.join("sub");
+    fs::create_dir_all(&subdir).unwrap();
+    fs::write(
+        subdir.join("picloud.toml"),
+        format!("[group]\nslug = \"{sub}\"\nname = \"Sub\"\n"),
+    )
+    .unwrap();
+}
+
+/// The `parent` (index 2) and `owner` (index 3) cells for `group` in
+/// `pic groups ls` — columns: slug, name, parent, owner, created_at.
+fn parent_and_owner(env: &common::TestEnv, group: &str) -> (String, String) {
+    let ls = common::pic_as(env)
+        .args(["groups", "ls"])
+        .output()
+        .expect("groups ls");
+    let table = String::from_utf8(ls.stdout).unwrap();
+    let row = table
+        .lines()
+        .map(common::cells)
+        .find(|c| c.first() == Some(&group))
+        .unwrap_or_else(|| panic!("group `{group}` not in groups ls:\n{table}"));
+    (
+        row.get(2).copied().unwrap_or_default().to_string(),
+        row.get(3).copied().unwrap_or_default().to_string(),
+    )
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn tree_apply_creates_nested_groups_under_the_attach_point() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base = common::unique_slug("gc-base");
+    let top = common::unique_slug("gc-top");
+    let sub = common::unique_slug("gc-sub");
+    let project = common::unique_slug("gc-proj");
+
+    // The attach point pre-exists; `top` and `sub` do NOT — the tree creates them.
+    let _b = GroupGuard::new(&env.url, &env.token, &base);
+    common::pic_as(&env)
+        .args(["groups", "create", &base])
+        .assert()
+        .success();
+    // Teardown order (reverse of declaration): sub → top → base.
+    let _gtop = GroupGuard::new(&env.url, &env.token, &top);
+    let _gsub = GroupGuard::new(&env.url, &env.token, &sub);
+
+    let dir = TempDir::new().unwrap();
+    nested_repo(dir.path(), &project, &base, &top, &sub);
+
+    // Plan previews both group CREATES (and, with a [project], a claim).
+    let out = common::pic_as(&env)
+        .args(["plan", "--dir"])
+        .arg(dir.path())
+        .output()
+        .unwrap();
+    let plan = String::from_utf8(out.stdout).unwrap();
+    let plan_err = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        plan.contains(&top) && plan.contains(&sub),
+        "plan should mention both to-create groups:\nSTDOUT:\n{plan}\nSTDERR:\n{plan_err}"
+    );
+
+    // Apply creates `top` under `base`, then `sub` under `top`, in one tx.
+    common::pic_as(&env)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .assert()
+        .success();
+
+    let (top_parent, top_owner) = parent_and_owner(&env, &top);
+    assert_eq!(top_parent, base, "top's parent is the attach point");
+    assert_eq!(
+        top_owner, project,
+        "a created group is claimed by the project"
+    );
+    let (sub_parent, sub_owner) = parent_and_owner(&env, &sub);
+    assert_eq!(
+        sub_parent, top,
+        "sub's parent is its same-tree ancestor group"
+    );
+    assert_eq!(
+        sub_owner, project,
+        "the nested created group is also claimed"
+    );
+
+    // Re-apply is a clean no-op (both groups already exist, parents match).
+    let replan = String::from_utf8(
+        common::pic_as(&env)
+            .args(["plan", "--dir"])
+            .arg(dir.path())
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !replan.contains("create") && !replan.contains("update"),
+        "re-plan of an already-created tree must be a no-op:\n{replan}"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn creating_a_group_requires_group_admin_on_the_parent() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base = common::unique_slug("gcm-base");
+    let top = common::unique_slug("gcm-top");
+    let project = common::unique_slug("gcm-proj");
+
+    let _b = GroupGuard::new(&env.url, &env.token, &base);
+    common::pic_as(&env)
+        .args(["groups", "create", &base])
+        .assert()
+        .success();
+
+    // A member with only EDITOR on `base` may reconcile content but must not be
+    // able to CREATE a subgroup under it (that needs group-admin).
+    let mem = common::member::member_user(fx, &common::unique_slug("gcm-mem"));
+    common::member::grant_group_membership(fx, &base, &mem.id, "editor");
+    let menv = common::custom_env(&env.url, &mem.token);
+    common::seed_credentials(&menv, &mem.username);
+
+    let dir = TempDir::new().unwrap();
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{project}\"\nparent_group = \"{base}\"\n\n\
+             [group]\nslug = \"{top}\"\nname = \"Top\"\n"
+        ),
+    )
+    .unwrap();
+
+    let out = common::pic_as(&menv)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .output()
+        .expect("member apply");
+    assert!(
+        !out.status.success(),
+        "an editor without group-admin must not create a group"
+    );
+
+    // The group was NOT created (the tx rolled back).
+    let ls = String::from_utf8(
+        common::pic_as(&env)
+            .args(["groups", "ls"])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !ls.lines()
+            .map(common::cells)
+            .any(|c| c.first() == Some(&top.as_str())),
+        "a denied create must leave no group behind:\n{ls}"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn tree_apply_denies_group_content_write_without_the_cap() {
+    // A tree apply that writes content (here a `[vars]` entry) into a
+    // PRE-EXISTING group requires the caller hold that group's write cap. This
+    // pins the invariant the in-tx content-write re-check hardens: a group that
+    // isn't structurally created by this apply must pass content authz. (The
+    // race the re-check closes — a group appearing between the API-layer check
+    // and reconcile — isn't reproducible at the journey level; end to end a
+    // member without the cap is refused and the var is not written.)
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base = common::unique_slug("gcw-base");
+    let grp = common::unique_slug("gcw-grp");
+    let project = common::unique_slug("gcw-proj");
+    let _b = GroupGuard::new(&env.url, &env.token, &base);
+    let _g = GroupGuard::new(&env.url, &env.token, &grp);
+
+    // `base` is the attach point; `grp` pre-exists directly under it.
+    common::pic_as(&env)
+        .args(["groups", "create", &base])
+        .assert()
+        .success();
+    common::pic_as(&env)
+        .args(["groups", "create", &grp, "--parent", &base])
+        .assert()
+        .success();
+
+    // A member with NO membership on `grp` cannot write its vars.
+    let mem = common::member::member_user(fx, &common::unique_slug("gcw-mem"));
+    let menv = common::custom_env(&env.url, &mem.token);
+    common::seed_credentials(&menv, &mem.username);
+
+    // A repo that declares the pre-existing `grp` (attached under `base`) with a
+    // var — a group content write.
+    let dir = TempDir::new().unwrap();
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{project}\"\nparent_group = \"{base}\"\n\n\
+             [group]\nslug = \"{grp}\"\nname = \"Grp\"\n\n[vars]\nregion = \"eu\"\n"
+        ),
+    )
+    .unwrap();
+
+    let out = common::pic_as(&menv)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .output()
+        .expect("member apply");
+    assert!(
+        !out.status.success(),
+        "a member without the group's write cap must be refused a content write"
+    );
+
+    // The var was NOT written (the tx rolled back / never ran).
+    let vars = String::from_utf8(
+        common::pic_as(&env)
+            .args(["vars", "ls", "--group", &grp])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !vars.contains("region"),
+        "a denied content write must leave no var behind:\n{vars}"
+    );
+}

crates/picloud-cli/tests/structural_divergence.rs

@@ -0,0 +1,169 @@
+//! §6 M2 — structural-divergence detection + declarative reparent.
+//!
+//! Once the manifest declares a group's parent (by directory nesting), a
+//! `pic apply --dir` compares it to the server's actual parent. A divergence is
+//! REFUSED by default (Terraform detect-and-refuse); `--force-local-structure`
+//! reparents the group to match the manifest, `--adopt-server-structure` keeps
+//! the server's placement and reconciles content in place. `pic plan` previews
+//! the divergence first.
+
+use std::fs;
+use std::path::Path;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::GroupGuard;
+
+/// A repo that declares `mover` NESTED under `alt` (so its manifest parent is
+/// `alt`), with `alt` at the attach point `base`.
+fn diverging_repo(dir: &Path, project: &str, base: &str, alt: &str, mover: &str) {
+    fs::write(
+        dir.join("picloud.toml"),
+        format!(
+            "[project]\nslug = \"{project}\"\nparent_group = \"{base}\"\n\n\
+             [group]\nslug = \"{alt}\"\nname = \"Alt\"\n"
+        ),
+    )
+    .unwrap();
+    let sub = dir.join("mover");
+    fs::create_dir_all(&sub).unwrap();
+    fs::write(
+        sub.join("picloud.toml"),
+        format!("[group]\nslug = \"{mover}\"\nname = \"Mover\"\n"),
+    )
+    .unwrap();
+}
+
+fn server_parent(env: &common::TestEnv, group: &str) -> String {
+    let ls = common::pic_as(env)
+        .args(["groups", "ls"])
+        .output()
+        .expect("groups ls");
+    let table = String::from_utf8(ls.stdout).unwrap();
+    table
+        .lines()
+        .map(common::cells)
+        .find(|c| c.first() == Some(&group))
+        .and_then(|c| c.get(2).map(|s| (*s).to_string()))
+        .unwrap_or_else(|| panic!("group `{group}` not in groups ls:\n{table}"))
+}
+
+/// Pre-create base → {alt, mover} (both directly under base), so a repo that
+/// nests `mover` under `alt` diverges.
+fn seed(env: &common::TestEnv, base: &str, alt: &str, mover: &str) {
+    common::pic_as(env)
+        .args(["groups", "create", base])
+        .assert()
+        .success();
+    common::pic_as(env)
+        .args(["groups", "create", alt, "--parent", base])
+        .assert()
+        .success();
+    common::pic_as(env)
+        .args(["groups", "create", mover, "--parent", base])
+        .assert()
+        .success();
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn divergence_refused_by_default_then_force_local_reparents() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base = common::unique_slug("sd-base");
+    let alt = common::unique_slug("sd-alt");
+    let mover = common::unique_slug("sd-mover");
+    let project = common::unique_slug("sd-proj");
+    let _b = GroupGuard::new(&env.url, &env.token, &base);
+    let _a = GroupGuard::new(&env.url, &env.token, &alt);
+    let _m = GroupGuard::new(&env.url, &env.token, &mover);
+    seed(&env, &base, &alt, &mover);
+
+    let dir = TempDir::new().unwrap();
+    diverging_repo(dir.path(), &project, &base, &alt, &mover);
+
+    // Plan previews the divergence (server parent `base`, manifest parent `alt`).
+    let plan = String::from_utf8(
+        common::pic_as(&env)
+            .args(["plan", "--dir"])
+            .arg(dir.path())
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        plan.contains("diverged") && plan.contains(&base) && plan.contains(&alt),
+        "plan should show `mover` diverged (server={base}, manifest={alt}):\n{plan}"
+    );
+
+    // A bare apply is REFUSED on the divergence (422).
+    let out = common::pic_as(&env)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .output()
+        .expect("apply");
+    assert!(!out.status.success(), "a divergence must refuse by default");
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("structural")
+            || err.contains("force-local")
+            || err.contains("different parent"),
+        "the refusal must name the resolution flags:\n{err}"
+    );
+    assert_eq!(
+        server_parent(&env, &mover),
+        base,
+        "a refused apply must not move the group"
+    );
+
+    // `--force-local-structure` reparents `mover` to `alt` (the manifest shape).
+    common::pic_as(&env)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .arg("--force-local-structure")
+        .assert()
+        .success();
+    assert_eq!(
+        server_parent(&env, &mover),
+        alt,
+        "--force-local-structure must reparent to the manifest's parent"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn adopt_server_structure_keeps_the_server_placement() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base = common::unique_slug("sda-base");
+    let alt = common::unique_slug("sda-alt");
+    let mover = common::unique_slug("sda-mover");
+    let project = common::unique_slug("sda-proj");
+    let _b = GroupGuard::new(&env.url, &env.token, &base);
+    let _a = GroupGuard::new(&env.url, &env.token, &alt);
+    let _m = GroupGuard::new(&env.url, &env.token, &mover);
+    seed(&env, &base, &alt, &mover);
+
+    let dir = TempDir::new().unwrap();
+    diverging_repo(dir.path(), &project, &base, &alt, &mover);
+
+    // `--adopt-server-structure` proceeds without reparenting — `mover` stays
+    // under `base` (its server placement), and content reconciles in place.
+    common::pic_as(&env)
+        .args(["apply", "--dir"])
+        .arg(dir.path())
+        .arg("--adopt-server-structure")
+        .assert()
+        .success();
+    assert_eq!(
+        server_parent(&env, &mover),
+        base,
+        "--adopt-server-structure must keep the server's parent"
+    );
+}

crates/picloud/src/lib.rs

@@ -13,30 +13,31 @@ use picloud_manager_core::{
     admin_router, admins_router, api_keys_router, app_members_router, apply_router, apps_api,
     apps_router, attach_principal_if_present, auth_router, dead_letters_router, dev_emails_router,
     email_inbound_router, files_admin_router, groups_router, kv_admin_router, migrations,
-    rebuild_route_table, require_authenticated, route_admin_router, secrets_router, topics_router,
-    triggers_router, vars_router, AbandonedRepo, AdminPrincipalResolver, AdminSessionRepository,
-    AdminState, AdminUserRepository, AdminsState, ApiKeyRepository, ApiKeysState,
-    AppDomainRepository, AppMembersRepository, AppMembersState, AppRepository, ApplyService,
-    AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, DevEmailState, Dispatcher,
-    DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig,
-    FilesServiceImpl, FsFilesRepo, FsGroupFilesRepo, GroupDocsServiceImpl, GroupFilesServiceImpl,
-    GroupKvServiceImpl, GroupMembersRepository, GroupPubsubServiceImpl, GroupQueueServiceImpl,
-    GroupRepository, GroupsState, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState,
-    KvServiceImpl, OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo,
-    PostgresAdminSessionRepository, PostgresAdminUserRepository, PostgresApiKeyRepository,
-    PostgresAppDomainRepository, PostgresAppMembersRepository, PostgresAppRepository,
-    PostgresAppSecretsRepo, PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo,
-    PostgresAppUserRepository, PostgresAppUserRoleRepo, PostgresAppUserSessionRepository,
-    PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, PostgresDeadLetterService,
-    PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink,
-    PostgresGroupCollectionResolver, PostgresGroupDocsRepo, PostgresGroupKvRepo,
-    PostgresGroupMembersRepository, PostgresGroupQueueRepo, PostgresGroupRepository,
-    PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository,
-    PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo,
-    PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver,
-    RouteAdminState, SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl,
-    SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo,
-    TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState, VarsServiceImpl,
+    projects_router, rebuild_route_table, require_authenticated, route_admin_router,
+    secrets_router, topics_router, triggers_router, vars_router, AbandonedRepo,
+    AdminPrincipalResolver, AdminSessionRepository, AdminState, AdminUserRepository, AdminsState,
+    ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository, AppMembersState,
+    AppRepository, ApplyService, AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState,
+    DevEmailState, Dispatcher, DocsServiceImpl, EmailInboundState, EmailServiceImpl,
+    FilesAdminState, FilesConfig, FilesServiceImpl, FsFilesRepo, FsGroupFilesRepo,
+    GroupDocsServiceImpl, GroupFilesServiceImpl, GroupKvServiceImpl, GroupMembersRepository,
+    GroupPubsubServiceImpl, GroupQueueServiceImpl, GroupRepository, GroupsState, HttpConfig,
+    HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, OutboxEventEmitter,
+    OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository,
+    PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository,
+    PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserInvitationRepo,
+    PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, PostgresAppUserRoleRepo,
+    PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo,
+    PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository,
+    PostgresExecutionLogSink, PostgresGroupCollectionResolver, PostgresGroupDocsRepo,
+    PostgresGroupKvRepo, PostgresGroupMembersRepository, PostgresGroupQueueRepo,
+    PostgresGroupRepository, PostgresKvRepo, PostgresOutboxRepo, PostgresProjectRepository,
+    PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo,
+    PostgresTopicRepo, PostgresTriggerRepo, PostgresVarsRepo, PrincipalResolver, ProjectRepository,
+    ProjectsState, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver, RouteAdminState,
+    SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl, SecretsState,
+    SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo, TriggersState,
+    UsersServiceConfig, UsersServiceImpl, VarsApiState, VarsServiceImpl,
 };
 use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS;
 use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable};
@@ -116,6 +117,8 @@ pub async fn build_app(
     let apps_repo: Arc<dyn AppRepository> = Arc::new(PostgresAppRepository::new(pool.clone()));
     let groups_repo: Arc<dyn GroupRepository> =
         Arc::new(PostgresGroupRepository::new(pool.clone()));
+    let projects_repo: Arc<dyn ProjectRepository> =
+        Arc::new(PostgresProjectRepository::new(pool.clone()));
     let group_members_repo: Arc<dyn GroupMembersRepository> =
         Arc::new(PostgresGroupMembersRepository::new(pool.clone()));
     let domains_repo: Arc<dyn AppDomainRepository> =
@@ -537,6 +540,7 @@ pub async fn build_app(
         vars: Arc::new(PostgresVarsRepo::new(pool.clone())),
         apps: apps_repo.clone(),
         groups: groups_repo.clone(),
+        projects: projects_repo.clone(),
         modules: modules.clone(),
         domains: domains_repo.clone(),
         authz: authz.clone(),
@@ -694,6 +698,9 @@ pub async fn build_app(
         .merge(api_keys_router(api_keys_state))
         .merge(triggers_router(triggers_state))
         .merge(apply_router(apply_service))
+        .merge(projects_router(ProjectsState {
+            projects: projects_repo.clone(),
+        }))
         .merge(picloud_manager_core::queues_api::queues_router(
             picloud_manager_core::queues_api::QueuesState {
                 queues: queue_repo.clone(),

crates/picloud/tests/api.rs

@@ -891,12 +891,12 @@ async fn version_includes_public_base_url(pool: PgPool) {
     assert!(v["public_base_url"].is_string());
     assert_eq!(v["api"], 1);
     // `schema` is migrations::latest_version() — the highest embedded
-    // migration number, currently 65 (…0064_shared_topic_queue_kinds,
-    // 0065_group_queues from the v1.2 D2/D3 deferrals). This test is
+    // migration number, currently 66 (…0065_group_queues, then
+    // 0066_projects wiring the §7 multi-repo ownership seam). This test is
     // #[ignore]-gated so it doesn't run in the default `cargo test`; pinned to
     // current reality so an unintended schema change is still caught. Bump it
     // whenever a migration lands.
-    assert_eq!(v["schema"], 65);
+    assert_eq!(v["schema"], 66);
     assert_eq!(v["sdk"], "1.10");
 }
 

crates/shared/src/group.rs

@@ -11,7 +11,7 @@
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
-use crate::GroupId;
+use crate::{GroupId, ProjectId};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Group {
@@ -26,6 +26,10 @@ pub struct Group {
     /// Per-subtree structure version, bumped on every structural mutation
     /// (reparent/rename/delete). Not an authz input.
     pub structure_version: i64,
+    /// §7 multi-repo ownership: the project that declaratively manages this
+    /// node, or `None` when the node is UI/API-owned (no manifest claims it).
+    /// Apps inherit ownership from their nearest claimed ancestor group.
+    pub owner_project: Option<ProjectId>,
     pub created_at: DateTime<Utc>,
     pub updated_at: DateTime<Utc>,
 }

crates/shared/src/ids.rs

@@ -58,3 +58,4 @@ id_type!(AppUserId);
 id_type!(InvitationId);
 id_type!(QueueMessageId);
 id_type!(GroupId);
+id_type!(ProjectId);

crates/shared/src/lib.rs

@@ -29,6 +29,7 @@ pub mod kv;
 pub mod log_sink;
 pub mod modules;
 pub mod outbox_writer;
+pub mod project;
 pub mod pubsub;
 pub mod queue;
 pub mod realtime;
@@ -77,8 +78,8 @@ pub use group_pubsub::{GroupPubsubError, GroupPubsubService, NoopGroupPubsubServ
 pub use group_queue::{GroupQueueError, GroupQueueService, NoopGroupQueueService};
 pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService};
 pub use ids::{
-    AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, GroupId, InvitationId, QueueMessageId,
-    RequestId, ScriptId, TriggerId,
+    AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, GroupId, InvitationId, ProjectId,
+    QueueMessageId, RequestId, ScriptId, TriggerId,
 };
 pub use inbox::{
     InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult, NoopInboxResolver,
@@ -90,6 +91,7 @@ pub use modules::{
     ModuleResolution, ModuleScript, ModuleSource, ModuleSourceError, NoopModuleSource,
 };
 pub use outbox_writer::{HttpDispatchPayload, NewHttpOutbox, OutboxWriter, OutboxWriterError};
+pub use project::Project;
 pub use pubsub::{
     topic_matches, validate_topic_pattern, NoopPubsubService, PubsubError, PubsubService,
 };

crates/shared/src/project.rs

@@ -0,0 +1,27 @@
+//! Projects: the §7 multi-repo ownership registry.
+//!
+//! A project is a repo-root that declaratively MANAGES a slice of the shared
+//! group tree. Each group node is owned by exactly one project (recorded in
+//! `groups.owner_project`); apps inherit ownership from their nearest claimed
+//! ancestor group. Identity is a committed `[project] slug` — stable across
+//! clones; the server assigns the UUID.
+//!
+//! See docs/design/groups-and-project-tool.md §7.
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use crate::{AdminUserId, ProjectId};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Project {
+    pub id: ProjectId,
+    /// Instance-global identity, declared in the repo's root manifest and
+    /// frozen at registration.
+    pub slug: String,
+    pub name: String,
+    /// The admin who first registered the project (`None` once their account
+    /// is deleted — the FK is `ON DELETE SET NULL`).
+    pub created_by: Option<AdminUserId>,
+    pub created_at: DateTime<Utc>,
+}

docs/design/groups-and-project-tool.md

@@ -764,6 +764,94 @@ non-orphaning:
 **Corollary:** don't co-own a node — split config downward. Shared config lives *higher* (owned by a
 platform/shared repo attaching at root); team-specific bits go into subgroups each team owns.
 
+**Status — M1 shipped (the ownership claim).** The `owner_project` seam (0047) is now live, backed by a
+first-class `projects` table (`0066`, UUID pk + unique slug; `owner_project` FKs it `ON DELETE SET NULL`).
+A `[project]` block (slug + optional name) in the repo's root manifest declares identity; the first apply
+with a new slug registers the project and **claims** each group node it touches. The claim is a gate run
+inside the apply transaction, under the per-node advisory lock, *before* the diff — so a conflict
+short-circuits with a 409 before any write. Pure policy (`decide_group_claim` / `decide_app_owner` in
+`apply_service`): unclaimed→claim, owner→no-op, foreign→conflict unless `--takeover` (which additionally
+requires `GroupAdmin` — ownership ⟂ RBAC), no-project-into-a-claimed-subtree→conflict. Apps carry no
+`owner_project`; an app inherits ownership from its **nearest claimed ancestor group** (the ancestor walk
+is the boundary), and an unclaimed subtree stays open (backward-compatible — nothing changes until a repo
+first declares `[project]`). Visibility: `pic groups ls` shows an `owner` column; `--takeover` on `pic
+apply`. Pinned by `apply_service` unit tests + the `apply_ownership` journey.
+
+**Status — M2 shipped (the attach-point ceiling, point 2).** A `[project] parent_group = "<slug>"` binds the
+repo UNDER a pre-existing group; applies are refused (422 `OutsideAttachPoint`) for any node not strictly
+within that subtree. `check_within_attach` requires the attach group be a **proper** ancestor of a group node
+(so you can't apply the attach point itself, only its descendants) or an ancestor (inclusive) of an app node's
+group — resolved via `groups.ancestors`, enforced read-only before the claim in both `apply_owner` and
+`apply_tree`. Absent = instance root = no ceiling (default). Pinned by the `apply_ownership` journey's
+attach-ceiling case.
+
+**Status — M3 shipped (plan preview + `pic projects ls`).** `pic plan` now carries the declaring `[project]`
+and returns an `ownership` preview per node — `claim` / `owned` / `conflict` (owner named) / `unclaimed`
+(pure `preview_ownership`, unit-tested) — plus, for a group node, the cross-repo **blast radius**: descendant
+apps owned by OTHER projects the change fans out to (`group_blast_radius`, a subtree CTE with per-group
+memoized nearest-claimed resolution). The attach ceiling is previewed at plan too (consistent with apply).
+Read-only `pic projects ls` lists every project + its owned-group count (`GET /api/v1/admin/projects`,
+`list_with_counts`). Pinned by the `apply_ownership` journey's plan-preview case. **With M3, the multi-repo
+ownership track (§7 M1–M3) is complete. Deferred (separate later work):** §6 structural-divergence detection
+(`--adopt-server-structure` / `--force-local-structure`); declarative group create/reparent (lifting "groups
+pre-exist"); per-env approval gating (`[project.environments]`).
+
+**§6 group-tree M1 shipped — declarative group CREATE.** A `[group]` node's PARENT is now inferred from
+directory nesting (the nearest ancestor directory holding a `[group]`; the topmost group binds to the repo's
+`[project] parent_group` attach point, else the instance root). `discover::build_tree` emits `parent` (plus the
+group's `name`/`description`) per group node, sorted parents-first by directory depth; `TreeNode` gained those
+fields (`#[serde(default)]`, so a pre-§6 CLI still reconciles existing groups). Server-side, `apply_tree` runs a
+**Phase 0** (`create_missing_groups_tx`) that resolves-or-creates every group node IN the apply tx
+(parents-first, a same-tree parent resolved before its child via `read_group_id_by_slug_tx`), under the coarse
+`GROUP_STRUCTURAL_LOCK_KEY` (taken only when it creates), enforcing the **attach-point ceiling** and **RBAC**
+(`InstanceCreateGroup` at the root / `GroupAdmin(parent)` for a subgroup) per create; the group is then CLAIMED
+in Phase A alongside existing groups (a fresh group is `decide_group_claim` → `Claim`). `prepare_tree` gained a
+caller-supplied `resolved_ids` map (existing + just-created) and returns a `to_create` list the plan path
+previews (empty current → a full-create plan; ownership → `claim`) — a to-create group's token part is identical
+to its post-create part, so plan/apply tokens match across a create. `authz_tree` skips a to-create group (its
+authorization is the service-side create gate). Reparent of an EXISTING diverged group + the detect-and-refuse
+flags are M2. Pinned by `tests/group_create.rs`.
+
+**§6 group-tree M2 shipped — structural-divergence detection + declarative reparent.** With the declared parent now
+on the wire, `apply_tree` compares each EXISTING group node's server parent against the manifest's (directory-
+nesting) parent. A divergence is resolved by a request `StructureMode` (default `Refuse`, so a pre-M2 CLI never
+reshapes the tree): `Refuse` → `ApplyError::StructuralDivergence` (422, naming the resolution 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 the 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 / RBAC path; both are recorded as
+`structurally_changed` so the pool-based attach re-check skips their uncommitted rows). `pic plan` previews the
+divergence (`divergence_preview` → a `structure` row: `diverged`, 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 folds `structure_version`). Pinned by `tests/structural_divergence.rs`.
+
+**§3 M3 shipped — per-env approval gating (`[project.environments]`).** A `[project.environments]` block on the
+root `[project]` maps an env name to `{ confirm = bool }` (`ManifestProject.environments`, `#[serde(skip_serializing)]`
+— purely CLI-side). `pic apply --env <e>` is REFUSED (client-side, before any request) when `<e>` is confirm-required
+unless it is explicitly `--approve <e>`d; a blanket `--yes` does NOT cover a gated env (§4.2 "CI must opt in per
+environment"); an unlisted / `confirm = false` env applies freely. `require_env_approval` gates both the single
+(`run`) and tree (`run_tree`) apply paths; `--approve <env>` is a repeatable flag. Value overlays
+(`picloud.<env>.toml`) already merge client-side, so this milestone is the confirm-policy gate only. Pinned by
+`tests/env_approval.rs`. **Deferred (documented, not built):** server-side audited approval-override capability
+(§4.2 → `authz::can`); env-as-app mapping + declarative server-side `@E` config scoping.
+
+**§6/§3 group-tree Tier 1 (M1 create · M2 divergence/reparent · M3 per-env approval) is COMPLETE.**
+
+**Review follow-ups (post-Tier-1 hardening).** A code review of the Tier-1 branch surfaced four fixes: (1)
+M1's `authz_tree` skip for a to-create group opened a race window (a group created by a racer between the
+API-layer authz check and the in-tx reconcile could have its content written unauthorized) — closed by an
+in-tx content-write re-check in `apply_tree` for every group NOT structurally created/reparented by that apply
+(`require_group_node_writes` moved from `apply_api` onto `ApplyService`; a group WE create is covered by
+create-RBAC and its uncommitted row is invisible to the authz cascade, so it's skipped). (2) The M3 approval
+gate could silently fail open — now `EnvPolicy.confirm` is a REQUIRED field (an empty `prod = {}` is a load
+error, not a silent no-gate), a single-file `pic apply --file <leaf>` discovers the governing `[project]` by
+walking up the directory tree, and a non-root `[project].environments` is dropped with an explicit warning.
+(3) Divergence detection is unified — the plan-path `divergence_preview` reuses the already-loaded group rows
+(no per-node re-fetch) and compares parent slugs case-insensitively so it can't disagree with the apply path's
+id-based check. (4) `reconcile_group_structure_tx`'s create/reparent branches are extracted into helpers.
+Pinned by new cases in `tests/group_create.rs` + `tests/env_approval.rs`.
+
 ---
 
 ## 8. Diagrams