feat(apply): §7 ownership claim state machine + project/takeover wire + 409

The heart of multi-repo ownership: an apply now claims/verifies node ownership
before it reconciles, so two repos can't clobber each other's subtree.

- Pure policy (DB-free, unit-tested): `decide_group_claim` (unclaimed→claim,
  owner→noop, foreign→conflict unless --takeover, no-project-into-claimed→
  conflict) and `decide_app_owner` (an app must match its nearest claimed
  ancestor; an unclaimed subtree stays open — backward-compatible).
- Execution under the per-node advisory lock: `upsert_project_tx` (first apply
  registers the project), `read/write_group_owner_tx`, `claim_group_owner`
  (takeover additionally requires `GroupAdmin` — ownership ⟂ RBAC),
  `check_app_owner_single` + tree `nearest_claimed_in_tx` (in-tree Phase-A
  claim overlays committed ancestors). No structure_version bump — a claim
  isn't a diff change, so it must not churn a pending bound plan.
- `apply`/`apply_owner`/`apply_tree` take an `OwnershipClaim` (project +
  takeover + principal); the claim slots after the lock, before load_current,
  so a conflict short-circuits before any diff work.
- New `ApplyError::OwnershipConflict` → HTTP 409 (actionable); wired through
  `ApplyRequest`/`TreeApplyRequest` (both `#[serde(default)]`, so the existing
  CLI stays compatible until it learns [project]).

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

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,
+    ExtensionPointInfo, NodeKind, OwnershipClaim, PlanResult, ProjectDecl, RouteTemplateInfo,
-    TreePlanResult, TriggerTemplateInfo,
+    SuppressionInfo, TreeBundle, TreePlanResult, TriggerTemplateInfo,
 };
 use crate::authz::{require, AuthzDenied, Capability};
 use crate::group_repo::GroupRepository;
@@ -188,6 +188,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 +204,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?;
@@ -263,12 +274,17 @@ async fn group_apply_handler(
 ) -> 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?;
+    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,6 +303,11 @@ 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,
 }
 
 async fn tree_plan_handler(
@@ -304,11 +325,16 @@ 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(),
         )
         .await?;
@@ -504,6 +530,10 @@ impl IntoResponse for ApplyError {
                 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");

crates/manager-core/src/apply_service.rs

@@ -32,15 +32,15 @@ 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, ScriptOwner, ScriptSandbox,
+    MasterKey, PathKind, Principal, ProjectId, Route, Script, ScriptId, ScriptKind, ScriptOwner,
-    ScriptValidator, TriggerId,
+    ScriptSandbox, ScriptValidator, TriggerId,
 };
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
 
 use crate::app_domain_repo::AppDomainRepository;
 use crate::app_repo::AppRepository;
-use crate::authz::AuthzRepo;
+use crate::authz::{require, AuthzDenied, AuthzRepo, Capability};
 use crate::config_resolver::CHAIN_LEVELS_CTE;
 use crate::repo::{
     delete_script_tx, insert_script_tx, update_script_tx, NewScript, ScriptPatch, ScriptRepository,
@@ -507,6 +507,25 @@ pub struct TreeBundle {
     pub nodes: Vec<TreeNode>,
 }
 
+/// §7 multi-repo ownership: the `[project]` identity an apply declares. The
+/// slug is committed in the repo's root manifest (stable across clones); the
+/// first apply with a new slug registers the project (server-assigned UUID).
+#[derive(Debug, Clone, Deserialize)]
+pub struct ProjectDecl {
+    pub slug: String,
+    #[serde(default)]
+    pub name: Option<String>,
+}
+
+/// The ownership context threaded through an apply: the declared project (if
+/// any), whether the caller passed `--takeover`, and the principal (its
+/// `user_id` is the actor; a takeover additionally requires `GroupAdmin`).
+pub struct OwnershipClaim<'a> {
+    pub project: Option<ProjectDecl>,
+    pub takeover: bool,
+    pub principal: &'a Principal,
+}
+
 /// One node's diff within a tree plan.
 #[derive(Debug, Clone, Serialize)]
 pub struct NodePlan {
@@ -627,6 +646,11 @@ pub enum ApplyError {
     StateMoved,
     #[error("forbidden")]
     Forbidden,
+    /// §7 multi-repo ownership: this apply's `[project]` does not own a node it
+    /// touches (or the node is unclaimed and the apply declared no project).
+    /// Maps to 409 — actionable (`use --takeover`, `declare [project]`).
+    #[error("ownership conflict: {0}")]
+    OwnershipConflict(String),
     #[error("authorization repo error: {0}")]
     AuthzRepo(String),
     #[error("backend: {0}")]
@@ -1314,14 +1338,14 @@ impl ApplyService {
         app_id: AppId,
         bundle: &Bundle,
         prune: bool,
-        actor: AdminUserId,
+        claim: &OwnershipClaim<'_>,
         expected_token: Option<&str>,
     ) -> Result<ApplyReport, ApplyError> {
         self.apply_owner(
             ApplyOwner::App(app_id),
             bundle,
             prune,
-            actor,
+            claim,
             expected_token,
         )
         .await
@@ -1338,9 +1362,10 @@ impl ApplyService {
         owner: ApplyOwner,
         bundle: &Bundle,
         prune: bool,
-        actor: AdminUserId,
+        claim: &OwnershipClaim<'_>,
         expected_token: Option<&str>,
     ) -> Result<ApplyReport, ApplyError> {
+        let actor = claim.principal.user_id;
         let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
         self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
         self.check_imports_resolve(owner, bundle).await?;
@@ -1368,6 +1393,32 @@ impl ApplyService {
             .await
             .map_err(|e| ApplyError::Backend(e.to_string()))?;
 
+        // §7 ownership: register the declared project (first apply claims), then
+        // gate this node. A group node claims/verifies itself; an app node must
+        // match its nearest claimed ancestor. Runs under the lock above so two
+        // repos racing to claim the same group serialize. Short-circuits before
+        // any diff work on a conflict (409).
+        let declared_project = match &claim.project {
+            Some(decl) => Some(upsert_project_tx(&mut tx, decl, actor).await?),
+            None => None,
+        };
+        match owner {
+            ApplyOwner::Group(gid) => {
+                self.claim_group_owner(
+                    &mut tx,
+                    gid,
+                    declared_project,
+                    claim.takeover,
+                    claim.principal,
+                )
+                .await?;
+            }
+            ApplyOwner::App(app_id) => {
+                self.check_app_owner_single(app_id, declared_project)
+                    .await?;
+            }
+        }
+
         let current = self.load_current(owner).await?;
         // Phase 4: names of group scripts the current routes/triggers bind to,
         // so the token / diff / prune resolve inherited bindings (see method).
@@ -1481,14 +1532,17 @@ impl ApplyService {
     ///
     /// # Errors
     /// `Invalid` for a bad bundle; `StateMoved` if the bound token is stale;
+    /// `OwnershipConflict` (§7) if a node is owned by another project;
     /// `Backend` for repo/transaction failures.
+    #[allow(clippy::too_many_lines)]
     pub async fn apply_tree(
         &self,
         bundle: &TreeBundle,
         prune: bool,
-        actor: AdminUserId,
+        claim: &OwnershipClaim<'_>,
         expected_token: Option<&str>,
     ) -> Result<ApplyReport, ApplyError> {
+        let actor = claim.principal.user_id;
         let (prepared, token) = self.prepare_tree(bundle).await?;
         if let Some(expected) = expected_token {
             if token != expected {
@@ -1517,10 +1571,32 @@ impl ApplyService {
         let mut report = ApplyReport::default();
         let mut group_script_index: HashMap<GroupId, HashMap<String, ScriptId>> = HashMap::new();
 
+        // §7 ownership: one project for the whole tree; claim/verify every group
+        // node in Phase A (recording it in `claimed_in_tree`), then check every
+        // app in Phase B against its nearest claimed ancestor. A conflict aborts
+        // the whole transaction.
+        let declared_project = match &claim.project {
+            Some(decl) => Some(upsert_project_tx(&mut tx, decl, actor).await?),
+            None => None,
+        };
+        let declared_slug = claim.project.as_ref().map(|d| d.slug.clone());
+        let mut claimed_in_tree: HashMap<GroupId, (ProjectId, String)> = HashMap::new();
+
         // 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 {
             if let ApplyOwner::Group(gid) = p.owner {
+                self.claim_group_owner(
+                    &mut tx,
+                    gid,
+                    declared_project,
+                    claim.takeover,
+                    claim.principal,
+                )
+                .await?;
+                if let (Some(pid), Some(slug)) = (declared_project, &declared_slug) {
+                    claimed_in_tree.insert(gid, (pid, slug.clone()));
+                }
                 report
                     .warnings
                     .extend(disabled_target_warnings(&p.node.bundle));
@@ -1546,6 +1622,12 @@ impl ApplyService {
         // ids just created) and pre-existing ancestors, then reconcile.
         for p in &prepared {
             if let ApplyOwner::App(_) = p.owner {
+                // §7: an app must match its nearest claimed ancestor — an in-tree
+                // group (just claimed in Phase A) or a committed ancestor.
+                let nearest =
+                    nearest_claimed_in_tx(&mut tx, &p.app_chain, &claimed_in_tree).await?;
+                decide_app_owner(nearest, declared_project)
+                    .map_err(ApplyError::OwnershipConflict)?;
                 report
                     .warnings
                     .extend(disabled_target_warnings(&p.node.bundle));
@@ -1783,6 +1865,84 @@ impl ApplyService {
         Ok(out)
     }
 
+    // ------------------------------------------------------------------------
+    // §7 ownership claim — the ApplyService side (executes the pure verdict).
+    // ------------------------------------------------------------------------
+
+    /// Claim / verify ownership of a GROUP node inside the apply tx. Reads the
+    /// live owner under the caller's advisory lock, runs the pure policy, and
+    /// applies the verdict: first-apply claims, a re-apply by the owner is a
+    /// no-op, a conflict is a 409, and a `--takeover` reassign additionally
+    /// requires `GroupAdmin` on the target (ownership ⟂ RBAC, §7.4).
+    async fn claim_group_owner(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        gid: GroupId,
+        declared: Option<ProjectId>,
+        takeover: bool,
+        principal: &Principal,
+    ) -> Result<(), ApplyError> {
+        let current = read_group_owner_tx(tx, gid).await?;
+        match decide_group_claim(current, declared, takeover) {
+            GroupClaimAction::Noop => Ok(()),
+            GroupClaimAction::Claim(pid) => write_group_owner_tx(tx, gid, pid).await,
+            GroupClaimAction::Takeover(pid) => {
+                require(self.authz.as_ref(), principal, Capability::GroupAdmin(gid))
+                    .await
+                    .map_err(map_authz)?;
+                write_group_owner_tx(tx, gid, pid).await
+            }
+            GroupClaimAction::Conflict(msg) => Err(ApplyError::OwnershipConflict(msg)),
+        }
+    }
+
+    /// Verify an APP node's ownership (single-node path). The app claims
+    /// nothing — it must match its nearest claimed ancestor group. Resolves the
+    /// chain via `ancestors()` (which now carries `owner_project`); an unclaimed
+    /// subtree is open. The narrow read-then-commit window is accepted (an app
+    /// writes no ownership; hazard c).
+    async fn check_app_owner_single(
+        &self,
+        app_id: AppId,
+        declared: Option<ProjectId>,
+    ) -> Result<(), ApplyError> {
+        let Some(app) = self
+            .apps
+            .get_by_id(app_id)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?
+        else {
+            return Ok(()); // resolved earlier in the handler; defensive.
+        };
+        let chain = self
+            .groups
+            .ancestors(app.group_id)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?;
+        let nearest = self.nearest_claimed_from_chain(&chain).await?;
+        decide_app_owner(nearest, declared).map_err(ApplyError::OwnershipConflict)
+    }
+
+    /// The nearest ancestor (nearest-first `chain`) with an owner, as
+    /// `(id, slug)`. Resolves the owning project's slug lazily — only when a
+    /// claim exists — for the conflict message.
+    async fn nearest_claimed_from_chain(
+        &self,
+        chain: &[picloud_shared::Group],
+    ) -> Result<Option<(ProjectId, String)>, ApplyError> {
+        let Some(pid) = chain.iter().find_map(|g| g.owner_project) else {
+            return Ok(None);
+        };
+        let slug = self
+            .projects
+            .get_by_id(pid)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?
+            .map(|p| p.slug)
+            .unwrap_or_default();
+        Ok(Some((pid, slug)))
+    }
+
     async fn resolve_app(&self, ident: &str) -> Result<picloud_shared::App, ApplyError> {
         crate::app_repo::resolve_app(self.apps.as_ref(), ident)
             .await
@@ -2893,6 +3053,192 @@ async fn delete_group_var_tx(
     Ok(())
 }
 
+// ----------------------------------------------------------------------------
+// §7 multi-repo ownership — the claim state machine + its tx helpers.
+//
+// A project is single-owner-per-NODE: each group node is owned by exactly one
+// project (`groups.owner_project`); apps inherit ownership from their nearest
+// claimed ancestor group. The pure `decide_*` fns are the whole policy; the
+// apply path executes their verdict (the UPDATE / the takeover authz gate)
+// under the per-node advisory lock. See docs/design/groups-and-project-tool.md §7.
+// ----------------------------------------------------------------------------
+
+/// What a group-node apply should do about ownership, given the current owner,
+/// the declared project, and whether `--takeover` was passed.
+#[derive(Debug, PartialEq, Eq)]
+enum GroupClaimAction {
+    /// Nothing to write (unclaimed + no project, or already owned by us).
+    Noop,
+    /// First-apply-claims: set `owner_project`.
+    Claim(ProjectId),
+    /// Reassign to us — capability-gated (`GroupAdmin`) at the call site.
+    Takeover(ProjectId),
+    /// Refuse (409); the string is the actionable message.
+    Conflict(String),
+}
+
+/// Pure ownership policy for a GROUP node. `current` is the live owner as
+/// `(id, slug)`; `declared` is this apply's project id. See §7.3.
+fn decide_group_claim(
+    current: Option<(ProjectId, String)>,
+    declared: Option<ProjectId>,
+    takeover: bool,
+) -> GroupClaimAction {
+    match (current, declared) {
+        // Unclaimed: a declared project claims it; no project is a no-op (the
+        // node stays UI/API-owned — the backward-compatible path).
+        (None, Some(p)) => GroupClaimAction::Claim(p),
+        (None, None) => GroupClaimAction::Noop,
+        // Already ours: re-apply by the owning repo.
+        (Some((c, _)), Some(p)) if c == p => GroupClaimAction::Noop,
+        // Owned by someone else: takeover (gated) or refuse.
+        (Some((_, cslug)), Some(p)) => {
+            if takeover {
+                GroupClaimAction::Takeover(p)
+            } else {
+                GroupClaimAction::Conflict(format!(
+                    "group is owned by project '{cslug}'; use --takeover (requires group-admin) \
+                     to reassign it"
+                ))
+            }
+        }
+        // A no-project apply must not silently write into a claimed subtree.
+        (Some((_, cslug)), None) => GroupClaimAction::Conflict(format!(
+            "group is owned by project '{cslug}'; declare a matching [project] (or --takeover)"
+        )),
+    }
+}
+
+/// Pure ownership policy for an APP node. Apps never claim — they must match
+/// their nearest claimed ancestor group (`nearest`, as `(id, slug)`); an
+/// unclaimed subtree is open (backward-compatible). See §7.
+fn decide_app_owner(
+    nearest: Option<(ProjectId, String)>,
+    declared: Option<ProjectId>,
+) -> Result<(), String> {
+    match (nearest, declared) {
+        (None, _) => Ok(()),
+        (Some((c, _)), Some(p)) if c == p => Ok(()),
+        (Some((_, cslug)), Some(_)) => Err(format!(
+            "the nearest owning group belongs to project '{cslug}', not this apply's project; \
+             apply from '{cslug}', or --takeover its group from a [group] node"
+        )),
+        (Some((_, cslug)), None) => Err(format!(
+            "this app is under a subtree owned by project '{cslug}'; declare [project] '{cslug}'"
+        )),
+    }
+}
+
+/// §7: validate a `[project]` slug — the same rule as app/group slugs
+/// (`^[a-z0-9][a-z0-9-]{0,62}$`).
+fn validate_project_slug(slug: &str) -> Result<(), ApplyError> {
+    let bad = |why: &str| ApplyError::Invalid(format!("project slug {slug:?}: {why}"));
+    if slug.is_empty() || slug.len() > 63 {
+        return Err(bad("must be 1–63 characters"));
+    }
+    let first = slug.chars().next().expect("non-empty checked above");
+    if !(first.is_ascii_lowercase() || first.is_ascii_digit()) {
+        return Err(bad("must start with a lowercase letter or digit"));
+    }
+    if !slug
+        .chars()
+        .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
+    {
+        return Err(bad(
+            "may contain only lowercase letters, digits, and hyphens",
+        ));
+    }
+    Ok(())
+}
+
+/// Register (or fetch) a project by slug inside the apply tx, returning its id.
+/// First apply with a new slug inserts; a re-apply updates the display name
+/// (the manifest is authoritative) but preserves the original `created_by`.
+async fn upsert_project_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    decl: &ProjectDecl,
+    actor: AdminUserId,
+) -> Result<ProjectId, ApplyError> {
+    validate_project_slug(&decl.slug)?;
+    let name = decl.name.clone().unwrap_or_else(|| decl.slug.clone());
+    let (id,): (uuid::Uuid,) = sqlx::query_as(
+        "INSERT INTO projects (slug, name, created_by) VALUES ($1, $2, $3) \
+         ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name \
+         RETURNING id",
+    )
+    .bind(&decl.slug)
+    .bind(&name)
+    .bind(actor.into_inner())
+    .fetch_one(&mut **tx)
+    .await
+    .map_err(|e| ApplyError::Backend(e.to_string()))?;
+    Ok(id.into())
+}
+
+/// Read a group's current owner as `(project id, project slug)` inside the tx,
+/// or `None` when unclaimed. Serialized with the write via the caller's
+/// per-node advisory lock.
+async fn read_group_owner_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    gid: GroupId,
+) -> Result<Option<(ProjectId, String)>, ApplyError> {
+    let row: Option<(Option<uuid::Uuid>, Option<String>)> = sqlx::query_as(
+        "SELECT g.owner_project, p.slug FROM groups g \
+         LEFT JOIN projects p ON p.id = g.owner_project WHERE g.id = $1",
+    )
+    .bind(gid.into_inner())
+    .fetch_optional(&mut **tx)
+    .await
+    .map_err(|e| ApplyError::Backend(e.to_string()))?;
+    Ok(row.and_then(|(pid, slug)| pid.map(|p| (p.into(), slug.unwrap_or_default()))))
+}
+
+/// Assign `owner_project` for a group inside the tx. Deliberately does NOT bump
+/// `structure_version` — a claim is not a diff/inheritance change, so it must
+/// not invalidate a pending bound plan (hazard b in the design).
+async fn write_group_owner_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    gid: GroupId,
+    pid: ProjectId,
+) -> Result<(), ApplyError> {
+    sqlx::query("UPDATE groups SET owner_project = $1 WHERE id = $2")
+        .bind(pid.into_inner())
+        .bind(gid.into_inner())
+        .execute(&mut **tx)
+        .await
+        .map_err(|e| ApplyError::Backend(e.to_string()))?;
+    Ok(())
+}
+
+/// Nearest claimed ancestor for an app, resolved in-tx: an in-tree group
+/// (Phase-A overlay) wins over a committed ancestor. `app_chain` is nearest-
+/// first. Used by the tree path (the single-node path folds `ancestors()`).
+async fn nearest_claimed_in_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    app_chain: &[GroupId],
+    in_tree: &HashMap<GroupId, (ProjectId, String)>,
+) -> Result<Option<(ProjectId, String)>, ApplyError> {
+    for gid in app_chain {
+        if let Some(hit) = in_tree.get(gid) {
+            return Ok(Some(hit.clone()));
+        }
+        if let Some(hit) = read_group_owner_tx(tx, *gid).await? {
+            return Ok(Some(hit));
+        }
+    }
+    Ok(None)
+}
+
+/// Map an authz denial to the apply surface (`Denied` → 403, repo error → 500).
+/// Used by the takeover capability gate, keeping "lacks group-admin" (403)
+/// distinct from a 409 ownership conflict.
+fn map_authz(d: AuthzDenied) -> ApplyError {
+    match d {
+        AuthzDenied::Denied => ApplyError::Forbidden,
+        AuthzDenied::Repo(e) => ApplyError::AuthzRepo(e.to_string()),
+    }
+}
+
 /// Owner of an apply node (Phase 5): an app or a group. The apply engine is
 /// owner-generic — only script-create ownership, var writes, the current-state
 /// load, and the advisory lock differ; the diff and all other CRUD are shared.
@@ -4736,4 +5082,66 @@ mod tests {
             "queue identity is queue_name-only: {p:?}"
         );
     }
+
+    // ------------------------------------------------------------------------
+    // §7 ownership — the pure claim policy (the whole decision, DB-free).
+    // ------------------------------------------------------------------------
+
+    #[test]
+    fn group_claim_unclaimed_claims_only_with_a_project() {
+        let p = ProjectId::new();
+        // First apply with a project claims; without a project it stays UI-owned.
+        assert_eq!(
+            decide_group_claim(None, Some(p), false),
+            GroupClaimAction::Claim(p)
+        );
+        assert_eq!(
+            decide_group_claim(None, None, false),
+            GroupClaimAction::Noop
+        );
+    }
+
+    #[test]
+    fn group_claim_by_owner_is_a_noop() {
+        let p = ProjectId::new();
+        assert_eq!(
+            decide_group_claim(Some((p, "acme".into())), Some(p), false),
+            GroupClaimAction::Noop
+        );
+    }
+
+    #[test]
+    fn group_claim_foreign_owner_conflicts_unless_takeover() {
+        let owner = ProjectId::new();
+        let me = ProjectId::new();
+        // A different project is refused without --takeover…
+        assert!(matches!(
+            decide_group_claim(Some((owner, "platform".into())), Some(me), false),
+            GroupClaimAction::Conflict(_)
+        ));
+        // …and reassigns with it (the authz gate is applied at the call site).
+        assert_eq!(
+            decide_group_claim(Some((owner, "platform".into())), Some(me), true),
+            GroupClaimAction::Takeover(me)
+        );
+        // A no-project apply must not silently write into a claimed node.
+        assert!(matches!(
+            decide_group_claim(Some((owner, "platform".into())), None, false),
+            GroupClaimAction::Conflict(_)
+        ));
+    }
+
+    #[test]
+    fn app_owner_matches_nearest_or_conflicts() {
+        let owner = ProjectId::new();
+        let me = ProjectId::new();
+        // An unclaimed subtree is open (backward-compatible).
+        assert!(decide_app_owner(None, Some(me)).is_ok());
+        assert!(decide_app_owner(None, None).is_ok());
+        // Same project as the nearest claimed ancestor → ok.
+        assert!(decide_app_owner(Some((owner, "x".into())), Some(owner)).is_ok());
+        // A different project, or none, under a claimed subtree → conflict.
+        assert!(decide_app_owner(Some((owner, "platform".into())), Some(me)).is_err());
+        assert!(decide_app_owner(Some((owner, "platform".into())), None).is_err());
+    }
 }