feat(hierarchies): multi-repo single-owner ownership + structural prune (M3)

§7 of the groups/project-tool design. A group node is now authoritatively
managed by exactly one project-root; `pic apply --dir` claims, conflicts,
takes over, and (with --prune) structurally reaps owned nodes.

- Migration 0052: `projects(id, key UNIQUE)` + promote the inert
  `groups.owner_project` (0047) to a real FK (ON DELETE SET NULL) + index.
- CLI mints a stable, gitignored project key in `.picloud/project.json`
  (`pic init`, or lazily on first tree plan/apply) and presents it on every
  tree request; `pic apply --dir --takeover` flag.
- Server: prepare_tree resolves ownership read-only (plan surfaces conflicts
  + prune candidates; token folds each group's owner key). apply_tree upserts
  the project in-tx, claims created groups on insert, reconciles existing-group
  ownership under the per-node advisory lock (first-commit-wins), and prunes
  owned-but-undeclared groups leaf-first (delete=RESTRICT, never another repo's
  or a UI-owned node).
- Authz (§7.4, ownership ⟂ RBAC): takeover requires GroupAdmin per contested
  node — enforced in authz_tree (pre-tx) AND re-verified in-tx at the ownership
  decision, so --force (which waives the staleness token) can't open a
  takeover-without-admin window. The attacker-supplied project key is
  length/charset-validated server-side.
- Tests: tests/ownership.rs (claim → conflict → takeover → flip; non-admin
  takeover → 403; prune-owned-only); format_conflicts unit test; schema golden
  reblessed. tree_shape M2 journey updated to reparent in-place (a fresh dir is
  now a distinct project and would correctly conflict).

Closes the M3 milestone; reviewed (4 findings: 1 authz-bypass via --force +
key validation + 2 LOW, all fixed). M4 (trigger/route templates) remains.

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

crates/manager-core/migrations/0052_owner_project.sql

@@ -0,0 +1,39 @@
+-- Phase 5 / M3 (v1.2 Hierarchies): multi-repo single-owner ownership (§7).
+--
+-- A group node is authoritatively MANAGED by at most one project-root (the
+-- repo whose manifest declares it as something it applies, not merely
+-- references). `groups.owner_project` has carried this seam since 0047 as an
+-- inert, FK-less UUID. M3 makes it a real reference: a `projects` table keyed
+-- by a stable, gitignored project key the CLI mints in `.picloud/`, and a
+-- foreign key from `groups.owner_project` to it.
+--
+-- Ownership is recorded at GROUP granularity; apps inherit their owning project
+-- from their group (apps are never claimed directly). A NULL `owner_project`
+-- means the node is UI/API-owned — no manifest fights it (§7.5). First apply
+-- claims; a second project's apply to an owned node is refused unless
+-- `--takeover` (group-admin gated). Ownership ⟂ RBAC: owning the manifest does
+-- NOT grant write — the actor still needs the usual group capabilities (§7.4).
+--
+-- ON DELETE SET NULL: deleting a project row (not something the platform does
+-- today) reverts its nodes to UI-owned rather than cascading away real groups.
+
+CREATE TABLE projects (
+    id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    -- Stable, opaque key minted by `pic init` and persisted (gitignored) in the
+    -- repo's `.picloud/`. The CLI presents it on every tree apply; the server
+    -- maps it to this row (upsert-by-key), so the same repo keeps the same
+    -- project identity across clones/CI without committing any server id.
+    key        TEXT NOT NULL UNIQUE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Promote the inert `groups.owner_project` (0047:38) to a real FK now that the
+-- referent exists. Existing rows are all NULL (no projects yet), so this adds
+-- no validation burden.
+ALTER TABLE groups
+    ADD CONSTRAINT groups_owner_project_fkey
+    FOREIGN KEY (owner_project) REFERENCES projects(id) ON DELETE SET NULL;
+
+-- The ownership reconcile reads "which groups does project P own?" (to find
+-- owned-but-undeclared nodes for structural prune) and "who owns group G?".
+CREATE INDEX groups_owner_project_idx ON groups (owner_project);

crates/manager-core/src/apply_api.rs

@@ -178,6 +178,16 @@ async fn group_apply_handler(
 // must hold the relevant read/write caps for EVERY node touched.
 // ----------------------------------------------------------------------------
 
+#[derive(Deserialize)]
+struct TreePlanRequest {
+    bundle: TreeBundle,
+    /// Stable project key from the repo's `.picloud/` link state (§7, M3).
+    /// Lets `plan` surface ownership conflicts and prune candidates. Optional
+    /// for legacy callers (then no ownership diagnosis is reported).
+    #[serde(default)]
+    project_key: Option<String>,
+}
+
 #[derive(Deserialize)]
 struct TreeApplyRequest {
     bundle: TreeBundle,
@@ -185,15 +195,25 @@ struct TreeApplyRequest {
     prune: bool,
     #[serde(default)]
     expected_token: Option<String>,
+    /// Stable project key (§7, M3): the apply claims unclaimed declared groups
+    /// for this project and refuses ones another project owns.
+    #[serde(default)]
+    project_key: Option<String>,
+    /// Take over declared groups owned by another project (group-admin gated).
+    #[serde(default)]
+    allow_takeover: bool,
 }
 
 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, false).await?;
+    Ok(Json(
+        svc.plan_tree(&req.bundle, req.project_key.as_deref())
+            .await?,
+    ))
 }
 
 async fn tree_apply_handler(
@@ -201,13 +221,22 @@ async fn tree_apply_handler(
     Extension(principal): Extension<Principal>,
     Json(req): Json<TreeApplyRequest>,
 ) -> Result<Json<ApplyReport>, ApplyError> {
-    authz_tree(&svc, &principal, &req.bundle, Some(req.prune)).await?;
+    authz_tree(
+        &svc,
+        &principal,
+        &req.bundle,
+        Some(req.prune),
+        req.allow_takeover,
+    )
+    .await?;
     let report = svc
         .apply_tree(
             &req.bundle,
             req.prune,
-            principal.user_id,
+            &principal,
             req.expected_token.as_deref(),
+            req.project_key.as_deref(),
+            req.allow_takeover,
         )
         .await?;
     Ok(Json(report))
@@ -222,6 +251,7 @@ async fn authz_tree(
     principal: &Principal,
     bundle: &TreeBundle,
     apply_prune: Option<bool>,
+    allow_takeover: bool,
 ) -> Result<(), ApplyError> {
     for node in &bundle.nodes {
         match node.kind {
@@ -283,6 +313,28 @@ async fn authz_tree(
                     Some(prune) => {
                         require_group_node_writes(svc, principal, group_id, &node.bundle, prune)
                             .await?;
+                        // Takeover gate (§7.4): seizing a group that another
+                        // project owns needs GroupAdmin specifically — a second,
+                        // independent gate on top of the write caps. We gate it
+                        // for any already-claimed node under `--takeover` (a
+                        // superset of genuinely-contested, never laxer); claiming
+                        // an unclaimed node, or applying without `--takeover`,
+                        // does not require admin. The authoritative owner rewrite
+                        // happens in-tx in `apply_tree` (which knows our project).
+                        if allow_takeover
+                            && crate::group_repo::get_group_owner(&svc.pool, group_id)
+                                .await
+                                .map_err(|e| ApplyError::Backend(e.to_string()))?
+                                .is_some()
+                        {
+                            require(
+                                svc.authz.as_ref(),
+                                principal,
+                                Capability::GroupAdmin(group_id),
+                            )
+                            .await
+                            .map_err(map_authz)?;
+                        }
                         // Reparent: an explicit parent differing from the live
                         // one needs GroupAdmin at the group, the SOURCE parent,
                         // and the DESTINATION parent — mirroring the interactive
@@ -484,6 +536,9 @@ impl IntoResponse for ApplyError {
                 json!({ "error": self.to_string() }),
             ),
             Self::StateMoved => (StatusCode::CONFLICT, json!({ "error": self.to_string() })),
+            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");

crates/manager-core/src/apply_service.rs

@@ -32,15 +32,16 @@ use std::sync::Arc;
 use picloud_orchestrator_core::routing::{pattern, RouteTable};
 use picloud_shared::{
     AdminUserId, AppId, DispatchMode, DocsEventOp, FilesEventOp, GroupId, HostKind, KvEventOp,
-    MasterKey, PathKind, Route, Script, ScriptId, ScriptKind, ScriptSandbox, ScriptValidator,
-    TriggerId,
+    MasterKey, PathKind, Principal, Route, Script, ScriptId, ScriptKind, ScriptSandbox,
+    ScriptValidator, TriggerId,
 };
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
+use uuid::Uuid;
 
 use crate::app_domain_repo::AppDomainRepository;
 use crate::app_repo::AppRepository;
-use crate::authz::AuthzRepo;
+use crate::authz::{require, AuthzDenied, AuthzRepo, Capability};
 use crate::repo::{
     delete_script_tx, insert_script_tx, update_script_tx, NewScript, ScriptPatch, ScriptRepository,
     ScriptRepositoryError,
@@ -426,6 +427,23 @@ pub struct NodePlan {
 pub struct TreePlanResult {
     pub nodes: Vec<NodePlan>,
     pub state_token: String,
+    /// Group nodes this manifest declares that another project already owns
+    /// (§7). A read-only `plan` surfaces them so CI sees the conflict before an
+    /// `apply` refuses; `apply` proceeds only with `--takeover`.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub conflicts: Vec<OwnershipConflict>,
+    /// Slugs of groups this project owns that are absent from the manifest —
+    /// what `apply --prune` would delete (leaf-first, RESTRICT-guarded).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub structural_prunes: Vec<String>,
+}
+
+/// One ownership conflict: a declared group already claimed by another project.
+#[derive(Debug, Clone, Serialize)]
+pub struct OwnershipConflict {
+    pub slug: String,
+    /// The owning project's key (so the operator knows whose node it is).
+    pub owner_key: String,
 }
 
 // ----------------------------------------------------------------------------
@@ -467,6 +485,11 @@ pub enum ApplyError {
     StateMoved,
     #[error("forbidden")]
     Forbidden,
+    #[error(
+        "one or more declared groups are owned by another project; \
+         re-run with `--takeover` to claim them (group-admin required): {0}"
+    )]
+    OwnershipConflict(String),
     #[error("authorization repo error: {0}")]
     AuthzRepo(String),
     #[error("backend: {0}")]
@@ -1077,15 +1100,19 @@ impl ApplyService {
     ///
     /// # Errors
     /// `Invalid` for a malformed/unknown bundle; `Backend` for repo failures.
-    pub async fn plan_tree(&self, bundle: &TreeBundle) -> Result<TreePlanResult, ApplyError> {
-        let pt = self.prepare_tree(bundle).await?;
+    pub async fn plan_tree(
+        &self,
+        bundle: &TreeBundle,
+        project_key: Option<&str>,
+    ) -> Result<TreePlanResult, ApplyError> {
+        let pt = self.prepare_tree(bundle, project_key).await?;
         let mut nodes: Vec<NodePlan> = pt
             .prepared
-            .into_iter()
+            .iter()
             .map(|p| NodePlan {
                 kind: p.node.kind,
                 slug: p.node.slug.clone(),
-                plan: p.plan,
+                plan: p.plan.clone(),
             })
             .collect();
         // To-create group nodes show all-Create content — their structural
@@ -1100,6 +1127,8 @@ impl ApplyService {
         Ok(TreePlanResult {
             nodes,
             state_token: pt.token,
+            conflicts: pt.conflicts,
+            structural_prunes: pt.prune_candidates.into_iter().map(|(_, s)| s).collect(),
         })
     }
 
@@ -1112,24 +1141,37 @@ impl ApplyService {
     /// # Errors
     /// `Invalid` for a bad bundle; `StateMoved` if the bound token is stale;
     /// `Backend` for repo/transaction failures.
-    #[allow(clippy::too_many_lines)]
+    #[allow(clippy::too_many_lines, clippy::too_many_arguments)]
     pub async fn apply_tree(
         &self,
         bundle: &TreeBundle,
         prune: bool,
-        actor: AdminUserId,
+        principal: &Principal,
         expected_token: Option<&str>,
+        project_key: Option<&str>,
+        allow_takeover: bool,
     ) -> Result<ApplyReport, ApplyError> {
-        let pt = self.prepare_tree(bundle).await?;
+        let actor = principal.user_id;
+        let pt = self.prepare_tree(bundle, project_key).await?;
         if let Some(expected) = expected_token {
             if pt.token != expected {
                 return Err(ApplyError::StateMoved);
             }
         }
+        // Ownership pre-check (§7): refuse, before opening the tx, if any
+        // declared group is owned by another project and `--takeover` was not
+        // given. The in-tx re-read below is the authoritative gate against a
+        // concurrent claim; this is the fast, clear path for the common case.
+        if !allow_takeover && !pt.conflicts.is_empty() {
+            return Err(ApplyError::OwnershipConflict(format_conflicts(
+                &pt.conflicts,
+            )));
+        }
         let PreparedTree {
             prepared,
             creates,
             reparents,
+            existing_group_nodes,
             ..
         } = pt;
 
@@ -1155,8 +1197,21 @@ impl ApplyService {
         let mut group_script_index: HashMap<GroupId, HashMap<String, ScriptId>> = HashMap::new();
         let mut any_app = false;
 
+        // Resolve (upsert) this repo's project row inside the tx, so created and
+        // claimed group nodes can be stamped with a stable owner id (§7, M3). A
+        // tree apply with no project key (legacy callers) skips all ownership.
+        let our_project_id = match project_key {
+            Some(k) => Some(
+                crate::group_repo::upsert_project_tx(&mut tx, k)
+                    .await
+                    .map_err(map_group_err)?,
+            ),
+            None => None,
+        };
+
         // Phase 0 — structure reconcile (M2): create absent group nodes and
         // reparent moved ones, in-tx so the whole apply stays all-or-nothing.
+        // Created groups are claimed by this project on insert (M3).
         if !creates.is_empty() || !reparents.is_empty() {
             self.reconcile_tree_structure_tx(
                 &mut tx,
@@ -1164,12 +1219,54 @@ impl ApplyService {
                 &reparents,
                 prune,
                 actor,
+                our_project_id,
                 &mut group_script_index,
                 &mut report,
             )
             .await?;
         }
 
+        // Phase 0.5 — ownership reconcile (§7, M3): for each EXISTING declared
+        // group, re-read the owner under this tx's lock (authoritative) and
+        // claim it (NULL owner) or take it over (`--takeover`). A conflict that
+        // appeared after `plan` and wasn't authorized aborts the whole apply.
+        if let Some(pid) = our_project_id {
+            for (gid, slug) in &existing_group_nodes {
+                let owner = crate::group_repo::get_group_owner_tx(&mut tx, *gid)
+                    .await
+                    .map_err(map_group_err)?;
+                match owner {
+                    Some((oid, _)) if oid == pid => {} // already ours — no write
+                    Some((_, okey)) => {
+                        if !allow_takeover {
+                            return Err(ApplyError::OwnershipConflict(format!(
+                                "{slug} (owned by project {okey})"
+                            )));
+                        }
+                        // Authoritative takeover gate (§7.4): GroupAdmin is
+                        // re-verified HERE, at the in-tx ownership decision, not
+                        // only in the pre-tx `authz_tree` pool read — otherwise
+                        // `--force` (which waives the staleness token) would let
+                        // a non-admin seize a group that another project claimed
+                        // after the pre-tx check saw it unowned. Ownership ⟂ RBAC.
+                        require(self.authz.as_ref(), principal, Capability::GroupAdmin(*gid))
+                            .await
+                            .map_err(map_authz_denied)?;
+                        crate::group_repo::set_group_owner_tx(&mut tx, *gid, pid)
+                            .await
+                            .map_err(map_group_err)?;
+                        report.groups_taken_over += 1;
+                    }
+                    None => {
+                        crate::group_repo::set_group_owner_tx(&mut tx, *gid, pid)
+                            .await
+                            .map_err(map_group_err)?;
+                        report.groups_claimed += 1;
+                    }
+                }
+            }
+        }
+
         // Phase A — groups first: reconcile each and record its post-create
         // `name -> id` so descendant app nodes can resolve inherited bindings.
         for p in &prepared {
@@ -1225,6 +1322,23 @@ impl ApplyService {
             }
         }
 
+        // Phase C — structural prune (§7, M3): with `--prune`, delete groups
+        // THIS project owns that the manifest no longer declares, leaf-first
+        // (delete_tx is RESTRICT, so a group still holding apps/child-groups
+        // aborts the apply with a clear error rather than orphaning data).
+        if prune {
+            if let Some(pid) = our_project_id {
+                self.structural_prune_tx(
+                    &mut tx,
+                    pid,
+                    &existing_group_nodes,
+                    &creates,
+                    &mut report,
+                )
+                .await?;
+            }
+        }
+
         tx.commit()
             .await
             .map_err(|e| ApplyError::Backend(e.to_string()))?;
@@ -1246,11 +1360,10 @@ impl ApplyService {
     /// `name -> id` is recorded so a descendant app can bind to it later in the
     /// same tx. Runs under the coarse structural lock so the cycle guard is sound.
     ///
-    /// Scope: create + reparent only. Deleting a server group that's absent
-    /// from the manifest is NOT done here even under `--prune` (group delete is
-    /// RESTRICT + destructive, and one repo shouldn't reap another's nodes) —
-    /// structural prune lands with the ownership layer (M3). Use
-    /// `pic groups rm` to delete a group explicitly.
+    /// Scope: create + reparent. A created group is stamped with `owner_project`
+    /// (this repo claims what it creates, §7 M3). Deleting an owned group that
+    /// the manifest dropped is a separate prune phase (`structural_prune_tx`),
+    /// scoped to project-owned, now-empty nodes.
     #[allow(clippy::too_many_arguments)]
     async fn reconcile_tree_structure_tx(
         &self,
@@ -1259,6 +1372,7 @@ impl ApplyService {
         reparents: &[(GroupId, Option<GroupId>)],
         prune: bool,
         actor: AdminUserId,
+        owner_project: Option<Uuid>,
         group_script_index: &mut HashMap<GroupId, HashMap<String, ScriptId>>,
         report: &mut ApplyReport,
     ) -> Result<(), ApplyError> {
@@ -1299,9 +1413,16 @@ impl ApplyService {
                         }
                     }
                 };
-                let g = crate::group_repo::create_group_tx(tx, &c.slug, &c.name, None, parent_id)
-                    .await
-                    .map_err(map_group_err)?;
+                let g = crate::group_repo::create_group_tx(
+                    tx,
+                    &c.slug,
+                    &c.name,
+                    None,
+                    parent_id,
+                    owner_project,
+                )
+                .await
+                .map_err(map_group_err)?;
                 created_ids.insert(c.slug.to_lowercase(), g.id);
                 let empty = CurrentState::default();
                 let plan = compute_diff(&empty, &c.node.bundle);
@@ -1341,6 +1462,80 @@ impl ApplyService {
         Ok(())
     }
 
+    /// Phase C of a tree apply (§7, M3): delete groups THIS project owns that the
+    /// manifest no longer declares, leaf-first. `delete_group_tx` is RESTRICT, so
+    /// a candidate still holding apps or non-pruned child groups aborts the whole
+    /// apply (no silent orphaning). Scoped to project-owned nodes: another repo's
+    /// groups and UI-owned (unclaimed) groups are never touched.
+    async fn structural_prune_tx(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        project_id: Uuid,
+        existing_group_nodes: &[(GroupId, String)],
+        creates: &[GroupCreate<'_>],
+        report: &mut ApplyReport,
+    ) -> Result<(), ApplyError> {
+        // Declared groups (existing nodes + just-created) are kept; every other
+        // group this project owns is a prune candidate.
+        let mut declared: HashSet<String> = existing_group_nodes
+            .iter()
+            .map(|(_, s)| s.to_lowercase())
+            .collect();
+        declared.extend(creates.iter().map(|c| c.slug.to_lowercase()));
+
+        let owned = crate::group_repo::list_owned_groups_tx(tx, project_id)
+            .await
+            .map_err(map_group_err)?;
+        let mut candidates: Vec<(GroupId, String, Option<GroupId>)> = owned
+            .into_iter()
+            .filter(|(_, slug, _)| !declared.contains(&slug.to_lowercase()))
+            .collect();
+        if candidates.is_empty() {
+            return Ok(());
+        }
+        // Serialize against concurrent reparents while we delete.
+        crate::group_repo::acquire_structural_lock_tx(tx)
+            .await
+            .map_err(map_group_err)?;
+
+        // Delete leaf-first: defer any candidate that is still the parent of
+        // another candidate until that child is gone.
+        while !candidates.is_empty() {
+            let parents: HashSet<GroupId> = candidates.iter().filter_map(|(_, _, p)| *p).collect();
+            let before = candidates.len();
+            let mut remaining: Vec<(GroupId, String, Option<GroupId>)> = Vec::new();
+            for (gid, slug, parent) in std::mem::take(&mut candidates) {
+                if parents.contains(&gid) {
+                    remaining.push((gid, slug, parent));
+                    continue;
+                }
+                // Re-confirm ownership under the lock: a concurrent `--takeover`
+                // could have flipped this candidate to another project between
+                // the candidate SELECT and now (prune nodes are undeclared, so
+                // the bound-plan token doesn't cover them). Skip if no longer ours.
+                let still_owned = crate::group_repo::get_group_owner_tx(tx, gid)
+                    .await
+                    .map_err(map_group_err)?
+                    .map(|(oid, _)| oid)
+                    == Some(project_id);
+                if !still_owned {
+                    continue;
+                }
+                crate::group_repo::delete_group_tx(tx, gid)
+                    .await
+                    .map_err(map_group_err)?;
+                report.groups_pruned += 1;
+            }
+            if remaining.len() == before {
+                return Err(ApplyError::Invalid(
+                    "structural prune could not order owned groups for deletion (cycle?)".into(),
+                ));
+            }
+            candidates = remaining;
+        }
+        Ok(())
+    }
+
     /// Resolve nodes to owners, validate each (apps' route/trigger targets may
     /// bind to in-tree ancestor group scripts), compute each node's plan, and
     /// fold a combined bound-plan token. Shared by `plan_tree` / `apply_tree`.
@@ -1348,10 +1543,14 @@ impl ApplyService {
     async fn prepare_tree<'a>(
         &self,
         bundle: &'a TreeBundle,
+        project_key: Option<&str>,
     ) -> Result<PreparedTree<'a>, ApplyError> {
         if bundle.nodes.is_empty() {
             return Err(ApplyError::Invalid("project tree has no nodes".into()));
         }
+        if let Some(k) = project_key {
+            validate_project_key(k)?;
+        }
         // Classify group nodes: existing (resolved id + live parent) vs to-create
         // (absent → created in apply Phase 0). App nodes always pre-exist.
         let mut group_id_by_slug: HashMap<String, GroupId> = HashMap::new();
@@ -1510,12 +1709,72 @@ impl ApplyService {
                 token_parts.push(format!("gs|{gid}|{}", g.structure_version));
             }
         }
+
+        // Ownership (§7, M3): resolve this repo's project, then classify each
+        // EXISTING group node as ours / unclaimed / owned-by-another. Folding
+        // the owner key into the token means a claim or takeover by a different
+        // repo between plan and apply trips StateMoved. To-create groups carry
+        // no conflict (a brand-new node is claimed by whoever creates it).
+        let our_project_id = match project_key {
+            Some(k) => crate::group_repo::get_project_id_by_key(&self.pool, k)
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?,
+            None => None,
+        };
+        let mut existing_group_nodes: Vec<(GroupId, String)> = Vec::new();
+        let mut declared_group_slugs: HashSet<String> = to_create_slugs.clone();
+        let mut conflicts: Vec<OwnershipConflict> = Vec::new();
+        for n in &bundle.nodes {
+            if n.kind != NodeKind::Group {
+                continue;
+            }
+            let lc = n.slug.to_lowercase();
+            declared_group_slugs.insert(lc.clone());
+            let Some(gid) = group_id_by_slug.get(&lc).copied() else {
+                continue; // to-create node, handled in Phase 0
+            };
+            existing_group_nodes.push((gid, n.slug.clone()));
+            let owner = crate::group_repo::get_group_owner(&self.pool, gid)
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?;
+            match &owner {
+                Some((_, key)) => token_parts.push(format!("own|{lc}|{key}")),
+                None => token_parts.push(format!("own|{lc}|none")),
+            }
+            if let Some((oid, okey)) = owner {
+                if Some(oid) != our_project_id {
+                    conflicts.push(OwnershipConflict {
+                        slug: n.slug.clone(),
+                        owner_key: okey,
+                    });
+                }
+            }
+        }
+
+        // Structural-prune candidates: groups THIS project owns that the
+        // manifest omits. Computed read-only here (so `plan` can list them);
+        // `apply --prune` re-derives the set in-tx and deletes leaf-first.
+        let mut prune_candidates: Vec<(GroupId, String)> = Vec::new();
+        if let Some(pid) = our_project_id {
+            for (gid, slug) in crate::group_repo::list_owned_groups(&self.pool, pid)
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?
+            {
+                if !declared_group_slugs.contains(&slug.to_lowercase()) {
+                    prune_candidates.push((gid, slug));
+                }
+            }
+        }
+
         token_parts.sort_unstable();
         Ok(PreparedTree {
             prepared,
             creates,
             reparents,
             token: combine_tokens(&token_parts),
+            existing_group_nodes,
+            conflicts,
+            prune_candidates,
         })
     }
 
@@ -3105,6 +3364,15 @@ pub struct ApplyReport {
     pub groups_created: u32,
     #[serde(default)]
     pub groups_reparented: u32,
+    /// Group nodes this apply claimed (NULL owner → this project), M3 §7.
+    #[serde(default)]
+    pub groups_claimed: u32,
+    /// Group nodes this apply took over from another project (`--takeover`).
+    #[serde(default)]
+    pub groups_taken_over: u32,
+    /// Owned-but-undeclared groups deleted by structural prune (`--prune`).
+    #[serde(default)]
+    pub groups_pruned: u32,
     #[serde(skip_serializing_if = "Vec::is_empty")]
     pub warnings: Vec<String>,
 }
@@ -3145,6 +3413,14 @@ struct PreparedTree<'a> {
     /// Existing groups to reparent: `(group id, new parent id)`.
     reparents: Vec<(GroupId, Option<GroupId>)>,
     token: String,
+    /// Ownership (§7, M3). Existing group nodes in the bundle: `(id, slug)` — the
+    /// apply path re-reads each owner in-tx (authoritative) to claim / detect
+    /// takeover. (This repo's own project id is re-derived in-tx via an upsert.)
+    existing_group_nodes: Vec<(GroupId, String)>,
+    /// Declared groups already owned by ANOTHER project (read-only diagnosis).
+    conflicts: Vec<OwnershipConflict>,
+    /// Owned-but-undeclared groups: `(id, slug)` — structural-prune candidates.
+    prune_candidates: Vec<(GroupId, String)>,
 }
 
 /// FNV-1a over a set of already-sorted per-resource token parts — the same
@@ -3254,6 +3530,45 @@ fn map_repo(e: ScriptRepositoryError) -> ApplyError {
     }
 }
 
+/// Validate an attacker-supplied project key (§7, M3). The key is a bearer
+/// handle for ownership, so it must be bounded and opaque before it's reflected
+/// into error messages or upserted. The CLI mints a v4 UUID; we accept any
+/// short, printable, delimiter-free token (ASCII alphanumeric plus `-`/`_`).
+fn validate_project_key(key: &str) -> Result<(), ApplyError> {
+    if key.is_empty() || key.len() > 128 {
+        return Err(ApplyError::Invalid(
+            "project key must be 1–128 characters".into(),
+        ));
+    }
+    if !key
+        .bytes()
+        .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+    {
+        return Err(ApplyError::Invalid(
+            "project key may contain only ASCII letters, digits, '-' and '_'".into(),
+        ));
+    }
+    Ok(())
+}
+
+/// Render an ownership-conflict list as `slug (owned by project key), …` for
+/// the `OwnershipConflict` error message.
+fn format_conflicts(conflicts: &[OwnershipConflict]) -> String {
+    conflicts
+        .iter()
+        .map(|c| format!("{} (owned by project {})", c.slug, c.owner_key))
+        .collect::<Vec<_>>()
+        .join(", ")
+}
+
+/// Map an in-service authz denial to the apply error surface (403 vs 500).
+fn map_authz_denied(e: AuthzDenied) -> ApplyError {
+    match e {
+        AuthzDenied::Denied => ApplyError::Forbidden,
+        AuthzDenied::Repo(r) => ApplyError::AuthzRepo(r.to_string()),
+    }
+}
+
 fn map_group_err(e: crate::group_repo::GroupRepositoryError) -> ApplyError {
     use crate::group_repo::GroupRepositoryError as E;
     // Conflict (cycle, slug clash, non-empty delete) and not-found are caller
@@ -3879,6 +4194,26 @@ mod tests {
         assert_ne!(state_token(&base), state_token(&added));
     }
 
+    #[test]
+    fn format_conflicts_lists_each_node_and_owner() {
+        // The conflict message drives the CLI error a developer sees; pin it.
+        let c = vec![
+            OwnershipConflict {
+                slug: "team-a".into(),
+                owner_key: "key-x".into(),
+            },
+            OwnershipConflict {
+                slug: "team-b".into(),
+                owner_key: "key-y".into(),
+            },
+        ];
+        assert_eq!(
+            format_conflicts(&c),
+            "team-a (owned by project key-x), team-b (owned by project key-y)"
+        );
+        assert_eq!(format_conflicts(&[]), "");
+    }
+
     #[test]
     fn state_token_ignores_dead_letter_triggers() {
         // The diff ignores dead-letter triggers (not manifest-representable),

crates/manager-core/src/group_repo.rs

@@ -192,7 +192,8 @@ impl GroupRepository for PostgresGroupRepository {
         parent_id: Option<GroupId>,
     ) -> Result<Group, GroupRepositoryError> {
         let mut tx = self.pool.begin().await?;
-        let g = create_group_tx(&mut tx, slug, name, description, parent_id).await?;
+        // Interactive create: UI/API-owned (no project claim — §7.5).
+        let g = create_group_tx(&mut tx, slug, name, description, parent_id, None).await?;
         tx.commit().await?;
         Ok(g)
     }
@@ -278,15 +279,17 @@ pub(crate) async fn create_group_tx(
     name: &str,
     description: Option<&str>,
     parent_id: Option<GroupId>,
+    owner_project: Option<Uuid>,
 ) -> Result<Group, GroupRepositoryError> {
     let res = sqlx::query_as::<_, GroupRow>(&format!(
-        "INSERT INTO groups (slug, name, description, parent_id) \
-         VALUES ($1, $2, $3, $4) RETURNING {GROUP_COLS}"
+        "INSERT INTO groups (slug, name, description, parent_id, owner_project) \
+         VALUES ($1, $2, $3, $4, $5) RETURNING {GROUP_COLS}"
     ))
     .bind(slug)
     .bind(name)
     .bind(description)
     .bind(parent_id.map(GroupId::into_inner))
+    .bind(owner_project)
     .fetch_one(&mut **tx)
     .await;
     match res {
@@ -391,6 +394,135 @@ pub(crate) async fn delete_group_tx(
     }
 }
 
+// ----------------------------------------------------------------------------
+// Project ownership (§7, M3). A `projects` row is keyed by the stable, opaque
+// key the CLI mints in `.picloud/`; `groups.owner_project` references it. These
+// helpers are free functions (pool + tx variants) so the apply path can claim
+// ownership inside the single apply transaction (first-commit-wins), while the
+// read-only plan path can surface conflicts without writing.
+// ----------------------------------------------------------------------------
+
+/// Resolve a project key to its id, if the project has ever been persisted.
+/// `None` means no apply has claimed under this key yet (so `plan` treats every
+/// node as claimable). Read-only — used by the plan path.
+pub(crate) async fn get_project_id_by_key(
+    pool: &PgPool,
+    key: &str,
+) -> Result<Option<Uuid>, GroupRepositoryError> {
+    let row: Option<(Uuid,)> = sqlx::query_as("SELECT id FROM projects WHERE key = $1")
+        .bind(key)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row.map(|r| r.0))
+}
+
+/// Upsert the project keyed by `key` inside the apply tx and return its id.
+/// Idempotent: re-applying the same repo reuses the same project identity.
+pub(crate) async fn upsert_project_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    key: &str,
+) -> Result<Uuid, GroupRepositoryError> {
+    let row: (Uuid,) = sqlx::query_as(
+        "INSERT INTO projects (key) VALUES ($1) \
+         ON CONFLICT (key) DO UPDATE SET key = EXCLUDED.key RETURNING id",
+    )
+    .bind(key)
+    .fetch_one(&mut **tx)
+    .await?;
+    Ok(row.0)
+}
+
+/// The project that owns `group_id`, as `(project_id, project_key)`. `None` when
+/// the node is UI/API-owned (no claim). Read-only — used by plan + the in-tx
+/// authoritative re-check. Joins through `projects` so the caller gets the
+/// human-facing key for a conflict message.
+pub(crate) async fn get_group_owner(
+    pool: &PgPool,
+    group_id: GroupId,
+) -> Result<Option<(Uuid, String)>, GroupRepositoryError> {
+    let row: Option<(Uuid, String)> = sqlx::query_as(
+        "SELECT p.id, p.key FROM groups g \
+         JOIN projects p ON p.id = g.owner_project WHERE g.id = $1",
+    )
+    .bind(group_id.into_inner())
+    .fetch_optional(pool)
+    .await?;
+    Ok(row)
+}
+
+/// The same owner lookup, but inside the apply tx — the authoritative read that
+/// the claim/conflict decision keys on (so a concurrent claim that committed
+/// after `plan` is seen under this tx's per-node advisory lock).
+pub(crate) async fn get_group_owner_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    group_id: GroupId,
+) -> Result<Option<(Uuid, String)>, GroupRepositoryError> {
+    let row: Option<(Uuid, String)> = sqlx::query_as(
+        "SELECT p.id, p.key FROM groups g \
+         JOIN projects p ON p.id = g.owner_project WHERE g.id = $1",
+    )
+    .bind(group_id.into_inner())
+    .fetch_optional(&mut **tx)
+    .await?;
+    Ok(row)
+}
+
+/// Stamp (claim or take over) `group_id`'s owning project inside the apply tx.
+/// Bumps `structure_version` so an ownership flip trips a bound plan token.
+pub(crate) async fn set_group_owner_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    group_id: GroupId,
+    project_id: Uuid,
+) -> Result<(), GroupRepositoryError> {
+    let res = sqlx::query(
+        "UPDATE groups SET owner_project = $2, \
+             structure_version = structure_version + 1, updated_at = NOW() WHERE id = $1",
+    )
+    .bind(group_id.into_inner())
+    .bind(project_id)
+    .execute(&mut **tx)
+    .await?;
+    if res.rows_affected() == 0 {
+        return Err(GroupRepositoryError::NotFound(group_id));
+    }
+    Ok(())
+}
+
+/// Every group `(id, slug)` owned by `project_id` — the candidate set for
+/// structural prune (those absent from the manifest are the ones to delete).
+/// Read-only; the apply path filters and deletes leaf-first under `delete_tx`.
+pub(crate) async fn list_owned_groups(
+    pool: &PgPool,
+    project_id: Uuid,
+) -> Result<Vec<(GroupId, String)>, GroupRepositoryError> {
+    let rows: Vec<(Uuid, String)> =
+        sqlx::query_as("SELECT id, slug FROM groups WHERE owner_project = $1")
+            .bind(project_id)
+            .fetch_all(pool)
+            .await?;
+    Ok(rows
+        .into_iter()
+        .map(|(id, slug)| (id.into(), slug))
+        .collect())
+}
+
+/// The same owned-group enumeration inside the apply tx, with each node's
+/// `parent_id` so the prune can delete leaf-first (children before parents).
+pub(crate) async fn list_owned_groups_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    project_id: Uuid,
+) -> Result<Vec<(GroupId, String, Option<GroupId>)>, GroupRepositoryError> {
+    let rows: Vec<(Uuid, String, Option<Uuid>)> =
+        sqlx::query_as("SELECT id, slug, parent_id FROM groups WHERE owner_project = $1")
+            .bind(project_id)
+            .fetch_all(&mut **tx)
+            .await?;
+    Ok(rows
+        .into_iter()
+        .map(|(id, slug, parent)| (id.into(), slug, parent.map(GroupId::from)))
+        .collect())
+}
+
 #[derive(sqlx::FromRow)]
 struct GroupRow {
     id: Uuid,

crates/manager-core/tests/expected_schema.txt

@@ -270,6 +270,11 @@ 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()
+  key: text NOT 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
@@ -492,6 +497,7 @@ indexes on group_members:
   group_members_user_id_idx: public.group_members USING btree (user_id)
 
 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)
@@ -508,6 +514,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_key_key: public.projects USING btree (key)
+  projects_pkey: public.projects USING btree (id)
+
 indexes on pubsub_trigger_details:
   pubsub_trigger_details_pkey: public.pubsub_trigger_details USING btree (trigger_id)
 
@@ -695,6 +705,7 @@ constraints on group_members:
   [PRIMARY KEY] group_members_pkey: PRIMARY KEY (group_id, user_id)
 
 constraints on groups:
+  [FOREIGN KEY] groups_owner_project_fkey: 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)
@@ -712,6 +723,10 @@ 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:
+  [PRIMARY KEY] projects_pkey: PRIMARY KEY (id)
+  [UNIQUE] projects_key_key: UNIQUE (key)
+
 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)
@@ -825,3 +840,4 @@ constraints on vars:
   0049: group secrets
   0050: group scripts
   0051: extension points
+  0052: owner project

crates/picloud-cli/Cargo.toml

@@ -34,6 +34,7 @@ toml = "0.8"
 directories = "5"
 rpassword = "7"
 anyhow = "1"
+uuid = { version = "1", features = ["v4"] }
 
 [dev-dependencies]
 assert_cmd = "2"

crates/picloud-cli/src/client.rs

@@ -1242,27 +1242,43 @@ impl Client {
     }
 
     /// `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> {
+    /// `project_key` (§7, M3) lets the server report ownership conflicts and
+    /// structural-prune candidates for this repo.
+    pub async fn plan_tree(
+        &self,
+        bundle: &serde_json::Value,
+        project_key: Option<&str>,
+    ) -> Result<TreePlanDto> {
+        let body = serde_json::json!({
+            "bundle": bundle,
+            "project_key": project_key,
+        });
         let resp = self
             .request(Method::POST, "/api/v1/admin/tree/plan")
-            .json(bundle)
+            .json(&body)
             .send()
             .await?;
         decode(resp).await
     }
 
     /// `POST /api/v1/admin/tree/apply` — reconcile a whole project tree in one
-    /// transaction (Phase 5).
+    /// transaction (Phase 5). `project_key`/`allow_takeover` drive ownership
+    /// (§7, M3): the apply claims unclaimed declared groups for this project and
+    /// refuses (or, with `allow_takeover`, seizes) ones another project owns.
     pub async fn apply_tree(
         &self,
         bundle: &serde_json::Value,
         prune: bool,
         expected_token: Option<&str>,
+        project_key: Option<&str>,
+        allow_takeover: bool,
     ) -> Result<ApplyReportDto> {
         let body = serde_json::json!({
             "bundle": bundle,
             "prune": prune,
             "expected_token": expected_token,
+            "project_key": project_key,
+            "allow_takeover": allow_takeover,
         });
         let resp = self
             .request(Method::POST, "/api/v1/admin/tree/apply")
@@ -1270,12 +1286,11 @@ impl Client {
             .send()
             .await?;
         if resp.status() == reqwest::StatusCode::CONFLICT {
+            // 409 covers both bound-plan staleness AND an ownership conflict; the
+            // server's message is self-explanatory for each, so surface it raw.
             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
     }
@@ -1365,6 +1380,20 @@ pub struct TreePlanDto {
     pub nodes: Vec<NodePlanDto>,
     #[serde(default)]
     pub state_token: String,
+    /// Declared groups owned by another project (§7, M3) — surfaced so CI sees
+    /// the conflict before `apply` refuses it.
+    #[serde(default)]
+    pub conflicts: Vec<OwnershipConflictDto>,
+    /// Slugs of owned groups absent from the manifest — what `--prune` deletes.
+    #[serde(default)]
+    pub structural_prunes: Vec<String>,
+}
+
+/// A single ownership conflict (`pic plan --dir`).
+#[derive(Debug, Deserialize)]
+pub struct OwnershipConflictDto {
+    pub slug: String,
+    pub owner_key: String,
 }
 
 #[derive(Debug, Deserialize)]
@@ -1421,6 +1450,12 @@ pub struct ApplyReportDto {
     #[serde(default)]
     pub groups_reparented: u32,
     #[serde(default)]
+    pub groups_claimed: u32,
+    #[serde(default)]
+    pub groups_taken_over: u32,
+    #[serde(default)]
+    pub groups_pruned: u32,
+    #[serde(default)]
     pub warnings: Vec<String>,
 }
 

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

@@ -118,6 +118,7 @@ pub async fn run_tree(
     prune: bool,
     yes: bool,
     force: bool,
+    takeover: bool,
     env: Option<&str>,
     mode: OutputMode,
 ) -> Result<()> {
@@ -138,8 +139,19 @@ pub async fn run_tree(
             .map(|l| l.state_token)
     };
 
+    // This repo's stable project key (§7, M3): the server claims declared
+    // groups for it and refuses ones another project owns (unless --takeover).
+    let project_key = crate::linkstate::ensure_project_key(dir)
+        .context("establishing project identity in .picloud/")?;
+
     let report = client
-        .apply_tree(&bundle, prune, expected_token.as_deref())
+        .apply_tree(
+            &bundle,
+            prune,
+            expected_token.as_deref(),
+            Some(&project_key),
+            takeover,
+        )
         .await?;
     crate::linkstate::clear_plan(dir, token_key);
 
@@ -181,14 +193,28 @@ pub async fn run_tree(
             ),
         );
     // Structural changes only happen on a tree apply; omit the line otherwise.
-    if report.groups_created > 0 || report.groups_reparented > 0 {
+    if report.groups_created > 0
+        || report.groups_reparented > 0
+        || report.groups_pruned > 0
+        || report.groups_claimed > 0
+        || report.groups_taken_over > 0
+    {
         block.field(
             "groups",
             format!(
-                "+{} reparented {}",
-                report.groups_created, report.groups_reparented
+                "+{} reparented {} pruned {}",
+                report.groups_created, report.groups_reparented, report.groups_pruned
             ),
         );
+        if report.groups_claimed > 0 || report.groups_taken_over > 0 {
+            block.field(
+                "ownership",
+                format!(
+                    "claimed {} taken-over {}",
+                    report.groups_claimed, report.groups_taken_over
+                ),
+            );
+        }
     }
     for w in &report.warnings {
         block.field("warning", w.clone());
@@ -211,7 +237,8 @@ fn confirm_prune(slug: &str) -> Result<()> {
     }
     eprint!(
         "apply --prune will DELETE live scripts/routes/triggers on `{slug}` that are \
-         absent from the manifest. This cannot be undone. Continue? [y/N] "
+         absent from the manifest — and, for a tree apply, entire owned groups absent \
+         from the manifest. This cannot be undone. Continue? [y/N] "
     );
     std::io::stderr().flush().ok();
     let mut answer = String::new();

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

@@ -87,6 +87,9 @@ pub fn run(
     }
     fs::write(&manifest_path, body).with_context(|| format!("writing {MANIFEST_FILE}"))?;
     ensure_gitignored(dir)?;
+    // Mint the repo's stable project identity (§7, M3) so the first tree apply
+    // claims its groups under a key that survives clones/CI. Gitignored.
+    crate::linkstate::ensure_project_key(dir).context("minting project key in .picloud/")?;
 
     let mut block = KvBlock::new();
     block

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

@@ -50,7 +50,10 @@ 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?;
+    // Mint/read this repo's project key so the server can report ownership
+    // (§7, M3). Best-effort: a read-only dir still plans (ownership unreported).
+    let project_key = crate::linkstate::ensure_project_key(dir).ok();
+    let plan = client.plan_tree(&bundle, project_key.as_deref()).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}");
@@ -85,6 +88,23 @@ fn render_tree(plan: &TreePlanDto, mode: OutputMode) {
         }
     }
     table.print(mode);
+
+    // Ownership diagnostics (§7, M3) — surfaced to stderr in non-JSON output
+    // (JSON consumers read `conflicts`/`structural_prunes` from the payload).
+    if mode != OutputMode::Json {
+        if !plan.conflicts.is_empty() {
+            eprintln!("\nownership conflicts (apply blocked without --takeover):");
+            for c in &plan.conflicts {
+                eprintln!("  - {} is owned by project {}", c.slug, c.owner_key);
+            }
+        }
+        if !plan.structural_prunes.is_empty() {
+            eprintln!("\ngroups absent from the manifest (deleted only with --prune):");
+            for slug in &plan.structural_prunes {
+                eprintln!("  - {slug}");
+            }
+        }
+    }
 }
 
 /// Assemble the wire bundle: scripts carry inlined source (read from

crates/picloud-cli/src/linkstate.rs

@@ -13,6 +13,59 @@ use serde::{Deserialize, Serialize};
 
 const DIR: &str = ".picloud";
 const PLAN_FILE: &str = "plan.json";
+const PROJECT_FILE: &str = "project.json";
+
+/// The repo's stable project identity (§7, M3): a random, opaque key minted on
+/// first need and persisted (gitignored) in `.picloud/`. It is the server-side
+/// ownership handle — the same repo keeps the same project across clones/CI
+/// because the key travels in the working tree, never in a committed file.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProjectLink {
+    pub key: String,
+}
+
+fn project_path(base: &Path) -> PathBuf {
+    base.join(DIR).join(PROJECT_FILE)
+}
+
+/// Read the project key, if one has been minted under `base/.picloud/`.
+#[must_use]
+pub fn read_project_key(base: &Path) -> Option<String> {
+    let body = fs::read(project_path(base)).ok()?;
+    serde_json::from_slice::<ProjectLink>(&body)
+        .ok()
+        .map(|l| l.key)
+}
+
+/// Read the project key, minting and persisting a fresh one if absent. Used by
+/// the tree `plan`/`apply` paths so a repo that never ran `pic init` still gets
+/// a stable ownership identity on first use. The new key is a random v4 UUID.
+pub fn ensure_project_key(base: &Path) -> Result<String> {
+    if let Some(k) = read_project_key(base) {
+        return Ok(k);
+    }
+    let key = uuid::Uuid::new_v4().to_string();
+    write_project_key(base, &key)?;
+    Ok(key)
+}
+
+/// Persist `key` as the project identity, creating `.picloud/` (self-ignored)
+/// if needed. Idempotent overwrite — callers usually go through
+/// [`ensure_project_key`].
+pub fn write_project_key(base: &Path, key: &str) -> Result<()> {
+    let dir = base.join(DIR);
+    fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
+    let ignore = dir.join(".gitignore");
+    if !ignore.exists() {
+        fs::write(&ignore, "*\n").context("writing .picloud/.gitignore")?;
+    }
+    let body = serde_json::to_vec_pretty(&ProjectLink {
+        key: key.to_string(),
+    })
+    .context("encoding .picloud/project.json")?;
+    fs::write(project_path(base), body).context("writing .picloud/project.json")?;
+    Ok(())
+}
 
 /// The recorded result of the last `pic plan`, scoped to the app it was for.
 #[derive(Debug, Clone, Serialize, Deserialize)]

crates/picloud-cli/src/main.rs

@@ -220,6 +220,10 @@ struct ApplyArgs {
     /// since the last `pic plan`).
     #[arg(long)]
     force: bool,
+    /// Tree apply only (`--dir`): seize declared groups currently owned by
+    /// another project (§7). Requires group-admin on each contested node.
+    #[arg(long, requires = "dir")]
+    takeover: bool,
     /// Merge the `picloud.<env>.toml` overlay (per-env slug + secrets) on
     /// top of the base manifest before applying.
     #[arg(long)]
@@ -1346,6 +1350,7 @@ async fn main() -> ExitCode {
                     args.prune,
                     args.yes,
                     args.force,
+                    args.takeover,
                     args.env.as_deref(),
                     mode,
                 )

crates/picloud-cli/tests/cli.rs

@@ -32,6 +32,7 @@ mod init;
 mod invoke;
 mod logs;
 mod output;
+mod ownership;
 mod plan;
 mod prune;
 mod pull;

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

@@ -78,6 +78,26 @@ pub fn grant_membership(fx: &Fixture, app_slug: &str, user_id: &str, role: &str)
     );
 }
 
+/// `POST /api/v1/admin/groups/{slug}/members` — grant `role` on a group.
+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/ownership.rs

@@ -0,0 +1,207 @@
+//! M3 multi-repo single-owner ownership (§7) via `pic apply --dir`:
+//!   * the first repo to apply a group CLAIMS it (its `.picloud/` project key
+//!     becomes the owner),
+//!   * a SECOND repo (different project key) applying the same group is
+//!     REJECTED with an ownership conflict,
+//!   * `--takeover` lets the second repo seize it (admin-gated; the fixture
+//!     token is instance owner), flipping ownership — proven by the first repo
+//!     now being the one rejected,
+//!   * `--prune` deletes an owned, now-undeclared, empty group; a group owned
+//!     by another repo is never touched.
+//!
+//! Each project lives in its own `TempDir`, so `pic` mints a distinct project
+//! key per repo (`.picloud/project.json`) — exactly the two-repo topology §7
+//! governs.
+
+use std::fs;
+use std::path::Path;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::GroupGuard;
+use crate::common::member;
+
+/// A single-group project dir: group `slug` under `parent`, no scripts (so a
+/// structural prune can delete it — a group holding scripts is RESTRICT-pinned).
+fn group_dir(slug: &str, parent: &str) -> TempDir {
+    let dir = TempDir::new().expect("tempdir");
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!("[group]\nslug = \"{slug}\"\nname = \"Owned Group\"\nparent = \"{parent}\"\n"),
+    )
+    .unwrap();
+    dir
+}
+
+fn apply(env: &common::TestEnv, dir: &Path, extra: &[&str]) -> std::process::Output {
+    let mut cmd = common::pic_as(env);
+    cmd.args(["apply", "--dir"]).arg(dir).args(extra);
+    cmd.output().expect("apply --dir")
+}
+
+fn ls_groups(env: &common::TestEnv) -> String {
+    String::from_utf8(
+        common::pic_as(env)
+            .args(["groups", "ls"])
+            .output()
+            .expect("groups ls")
+            .stdout,
+    )
+    .unwrap()
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn first_repo_claims_second_is_rejected_then_takeover_flips_ownership() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("own-g");
+    let _g = GroupGuard::new(&env.url, &env.token, &slug);
+
+    // Repo A claims the group on first apply (it does not pre-exist → created
+    // AND claimed for project A in one tx).
+    let repo_a = group_dir(&slug, "root");
+    let a1 = apply(&env, repo_a.path(), &[]);
+    assert!(
+        a1.status.success(),
+        "repo A claim apply failed: {}",
+        String::from_utf8_lossy(&a1.stderr)
+    );
+    assert!(
+        ls_groups(&env).contains(&slug),
+        "group should exist after claim"
+    );
+
+    // Repo B (a different dir → a different project key) is rejected: the group
+    // is owned by project A.
+    let repo_b = group_dir(&slug, "root");
+    let b1 = apply(&env, repo_b.path(), &[]);
+    assert!(
+        !b1.status.success(),
+        "repo B apply must be rejected (group owned by A)"
+    );
+    let b1_err = String::from_utf8_lossy(&b1.stderr).to_lowercase();
+    assert!(
+        b1_err.contains("owned by another project") || b1_err.contains("takeover"),
+        "conflict message should name the ownership clash:\n{b1_err}"
+    );
+
+    // Repo B with --takeover succeeds (the fixture token is instance owner, so
+    // it holds GroupAdmin on every node) and flips ownership to project B.
+    let b2 = apply(&env, repo_b.path(), &["--takeover"]);
+    assert!(
+        b2.status.success(),
+        "repo B --takeover should succeed: {}",
+        String::from_utf8_lossy(&b2.stderr)
+    );
+
+    // Proof the flip took: repo A — formerly the owner — is now the one
+    // rejected without --takeover.
+    let a2 = apply(&env, repo_a.path(), &[]);
+    assert!(
+        !a2.status.success(),
+        "after takeover, repo A must now be rejected (B owns the group)"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn takeover_without_group_admin_is_forbidden() {
+    // §7.4 — ownership ⟂ RBAC: `--takeover` needs GroupAdmin specifically, a
+    // second gate beyond the write caps. A non-admin member (editor) of the
+    // contested group cannot seize it for their repo.
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("own-ta");
+    let _g = GroupGuard::new(&env.url, &env.token, &slug);
+
+    // Admin (repo A) claims the group.
+    let repo_a = group_dir(&slug, "root");
+    assert!(
+        apply(&env, repo_a.path(), &[]).status.success(),
+        "admin claim apply should succeed"
+    );
+
+    // A member with `editor` (not group_admin) on the group.
+    let m = member::member_user(fx, &common::unique_username("ta"));
+    member::grant_group_membership(fx, &slug, &m.id, "editor");
+    let member_env = common::custom_env(&fx.url, &m.token);
+    common::seed_credentials(&member_env, &m.username);
+
+    // The member's repo B (different project key) attempts a takeover → 403.
+    let repo_b = group_dir(&slug, "root");
+    let out = apply(&member_env, repo_b.path(), &["--takeover"]);
+    assert!(
+        !out.status.success(),
+        "non-admin --takeover must be forbidden"
+    );
+    let err = String::from_utf8_lossy(&out.stderr).to_lowercase();
+    assert!(
+        err.contains("forbidden") || err.contains("403"),
+        "takeover denial should be an authz error:\n{err}"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn prune_deletes_owned_undeclared_group_only() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let parent = common::unique_slug("own-par");
+    let child = common::unique_slug("own-child");
+    // Drop order: child (registered last → dropped first), then parent.
+    let _gp = GroupGuard::new(&env.url, &env.token, &parent);
+    let _gc = GroupGuard::new(&env.url, &env.token, &child);
+
+    // Repo declares parent + child (both empty); apply claims both.
+    let dir = TempDir::new().expect("tempdir");
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!("[group]\nslug = \"{parent}\"\nname = \"Parent\"\nparent = \"root\"\n"),
+    )
+    .unwrap();
+    fs::create_dir_all(dir.path().join("sub")).unwrap();
+    fs::write(
+        dir.path().join("sub/picloud.toml"),
+        format!("[group]\nslug = \"{child}\"\nname = \"Child\"\nparent = \"{parent}\"\n"),
+    )
+    .unwrap();
+    let out = apply(&env, dir.path(), &[]);
+    assert!(
+        out.status.success(),
+        "initial claim apply failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let groups = ls_groups(&env);
+    assert!(
+        groups.contains(&parent) && groups.contains(&child),
+        "both groups should exist"
+    );
+
+    // Drop the child manifest from the SAME repo (keeping its project key) so
+    // the child becomes owned-but-undeclared, then apply --prune: the empty
+    // child is deleted; the parent (still declared) is kept.
+    fs::remove_dir_all(dir.path().join("sub")).unwrap();
+    let out2 = apply(&env, dir.path(), &["--prune", "--yes"]);
+    assert!(
+        out2.status.success(),
+        "prune apply failed: {}",
+        String::from_utf8_lossy(&out2.stderr)
+    );
+    let groups = ls_groups(&env);
+    assert!(
+        groups.contains(&parent),
+        "parent must survive prune:\n{groups}"
+    );
+    assert!(
+        !groups.contains(&child),
+        "owned, undeclared, empty child must be pruned:\n{groups}"
+    );
+}

crates/picloud-cli/tests/tree_shape.rs

@@ -83,8 +83,18 @@ fn tree_apply_creates_then_reparents_a_group() {
         "re-plan of an applied tree must be a no-op:\n{plan_txt}"
     );
 
-    // --- Reparent: move the child under the instance root. ---
-    let dir2 = group_dir(&child, "root");
+    // --- Reparent: move the child under the instance root. Rewrite the SAME
+    // repo's manifest (M3: a stable `.picloud/` project key per repo — a fresh
+    // dir would be a different project and conflict on the owned group). ---
+    fs::write(
+        dir.path().join("picloud.toml"),
+        format!(
+            "[group]\nslug = \"{child}\"\nname = \"Child Group\"\nparent = \"root\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n"
+        ),
+    )
+    .unwrap();
+    let dir2 = &dir;
     let out = common::pic_as(&env)
         .args(["apply", "--dir"])
         .arg(dir2.path())