refactor(apply): address ownership review follow-ups (#3–#6)

Four cleanups from the code review of the §7 ownership track, no behavior
change to any passing test (416 manager-core units + 133 CLI journeys green):

- #3 app-ownership read parity: `check_app_owner_single` now resolves the
  nearest-claimed ancestor WITHIN the apply tx (`nearest_claimed_in_tx`, the
  same mechanism the tree path uses) instead of on a separate pool connection,
  so the check and this apply's writes see one consistent snapshot. (Fully
  serializing against a concurrent claim on an out-of-tree ancestor stays part
  of the documented pool→tx follow-up.)

- #4 blast-radius N+1 → one query: `group_blast_radius` replaced the
  per-descendant-group `ancestors()` round-trips with a single recursive
  `subtree → app_chain → nearest` CTE that groups + counts server-side. Cost is
  now one query regardless of subtree size.

- #5 one ownership policy, two projections: `decide_group_claim` /
  `decide_app_owner` are generic over the identity key, and `preview_ownership`
  (plan side) now DERIVES its label from them (keyed on slugs, takeover-agnostic)
  rather than re-encoding the arms — plan and apply can no longer drift.

- #6 shared slug validator: `validate_project_slug` delegates to a new
  `groups_api::validate_slug_format`, which `groups_api::validate_slug` also now
  uses — the format rule lives once. Projects intentionally skip the
  reserved-word check (a project slug is never routed).

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

crates/manager-core/src/apply_service.rs

@@ -568,18 +568,34 @@ fn preview_ownership(
     current: Option<&str>,
     declared: Option<&str>,
 ) -> OwnershipPreview {
-    let mk = |action: &str, owner: Option<&str>| OwnershipPreview {
-        action: action.to_string(),
-        owner: owner.map(str::to_string),
-        blast_radius: Vec::new(),
+    // Project the SAME pure policy the apply path runs onto a display label,
+    // keyed on slugs (plan has no assigned project id yet) and takeover-agnostic
+    // (`takeover = false`; plan never mutates ownership). Deriving the preview
+    // here — rather than re-encoding the arms — keeps plan and apply from
+    // drifting: one decision, two projections.
+    let cur = current.map(|s| (s.to_string(), s.to_string()));
+    let decl = declared.map(str::to_string);
+    let action = if is_group {
+        match decide_group_claim(cur, decl, false) {
+            GroupClaimAction::Claim(_) => "claim",
+            // Noop covers both "already ours" and "unclaimed + no project"; the
+            // current owner disambiguates.
+            GroupClaimAction::Noop if current.is_none() => "unclaimed",
+            GroupClaimAction::Noop => "owned",
+            // Takeover can't arise with `takeover = false`; treat defensively.
+            GroupClaimAction::Conflict(_) | GroupClaimAction::Takeover(_) => "conflict",
+        }
+    } else {
+        match decide_app_owner(cur, decl) {
+            Ok(()) if current.is_none() => "unclaimed",
+            Ok(()) => "owned",
+            Err(_) => "conflict",
+        }
     };
-    match (current, declared) {
-        // A group node claims when unclaimed + a project is declared.
-        (None, Some(_)) if is_group => mk("claim", None),
-        // Otherwise unclaimed (an app never claims; no project → nothing).
-        (None, _) => mk("unclaimed", None),
-        (Some(c), Some(d)) if c == d => mk("owned", Some(c)),
-        (Some(c), _) => mk("conflict", Some(c)),
+    OwnershipPreview {
+        action: action.to_string(),
+        owner: current.map(str::to_string),
+        blast_radius: Vec::new(),
     }
 }
 
@@ -1505,7 +1521,7 @@ impl ApplyService {
                 .await?;
             }
             ApplyOwner::App(app_id) => {
-                self.check_app_owner_single(app_id, declared_project)
+                self.check_app_owner_single(&mut tx, app_id, declared_project)
                     .await?;
             }
         }
@@ -2014,12 +2030,18 @@ impl ApplyService {
     }
 
     /// 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).
+    /// nothing — it must match its nearest claimed ancestor group. The ancestor
+    /// STRUCTURE is read on the pool (`ancestors()`), but the ownership VALUE is
+    /// then read WITHIN the tx via `nearest_claimed_in_tx` — the same mechanism
+    /// the tree path uses — so the check and this apply's writes see one
+    /// consistent snapshot instead of a separate pool connection (`in_tree` is
+    /// empty; a single app node claims no groups itself). An unclaimed subtree
+    /// stays open. (Fully serializing against a concurrent claim on an
+    /// out-of-tree ancestor is part of the tracked pool→tx follow-up — see the
+    /// `apply_owner` advisory-lock note.)
     async fn check_app_owner_single(
         &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
         app_id: AppId,
         declared: Option<ProjectId>,
     ) -> Result<(), ApplyError> {
@@ -2036,7 +2058,8 @@ impl ApplyService {
             .ancestors(app.group_id)
             .await
             .map_err(|e| ApplyError::Backend(e.to_string()))?;
-        let nearest = self.nearest_claimed_from_chain(&chain).await?;
+        let chain_ids: Vec<GroupId> = chain.iter().map(|g| g.id).collect();
+        let nearest = nearest_claimed_in_tx(tx, &chain_ids, &HashMap::new()).await?;
         decide_app_owner(nearest, declared).map_err(ApplyError::OwnershipConflict)
     }
 
@@ -2176,48 +2199,51 @@ impl ApplyService {
         gid: GroupId,
         exclude: Option<&str>,
     ) -> Result<Vec<ProjectImpact>, ApplyError> {
-        // All apps in gid's subtree (gid + descendant groups).
-        let rows: Vec<(uuid::Uuid, uuid::Uuid)> = sqlx::query_as(
+        // One set-based query: every app in gid's subtree, resolved to its
+        // nearest claimed ancestor project, grouped and counted — replacing the
+        // per-descendant-group `ancestors()` round-trips (N+1) with a single
+        // recursive walk. `subtree` = gid + descendant groups; `app_chain` =
+        // (app, ancestor group, depth) for every subtree app walked to the root;
+        // `nearest` picks the shallowest claimed ancestor per app. Apps owned by
+        // `exclude` (the planning project) or by nothing are dropped.
+        let rows: Vec<(String, i64)> = sqlx::query_as(
             "WITH RECURSIVE subtree AS (
                  SELECT id FROM groups WHERE id = $1
                UNION ALL
                  SELECT g.id FROM groups g JOIN subtree s ON g.parent_id = s.id
+             ),
+             app_chain AS (
+                 SELECT a.id AS app_id, a.group_id AS gid, 0 AS depth
+                 FROM apps a WHERE a.group_id IN (SELECT id FROM subtree)
+               UNION ALL
+                 SELECT ac.app_id, g.parent_id, ac.depth + 1
+                 FROM app_chain ac JOIN groups g ON g.id = ac.gid
+                 WHERE g.parent_id IS NOT NULL
+             ),
+             nearest AS (
+                 SELECT DISTINCT ON (ac.app_id) ac.app_id, g.owner_project
+                 FROM app_chain ac JOIN groups g ON g.id = ac.gid
+                 WHERE g.owner_project IS NOT NULL
+                 ORDER BY ac.app_id, ac.depth ASC
              )
-             SELECT a.id, a.group_id FROM apps a WHERE a.group_id IN (SELECT id FROM subtree)",
+             SELECT p.slug, COUNT(*) AS n
+             FROM nearest n JOIN projects p ON p.id = n.owner_project
+             WHERE p.slug IS DISTINCT FROM $2
+             GROUP BY p.slug
+             ORDER BY p.slug",
         )
         .bind(gid.into_inner())
+        .bind(exclude)
         .fetch_all(&self.pool)
         .await
         .map_err(|e| ApplyError::Backend(e.to_string()))?;
 
-        let mut owner_by_group: HashMap<GroupId, Option<String>> = HashMap::new();
-        let mut counts: std::collections::BTreeMap<String, u32> = std::collections::BTreeMap::new();
-        for (_app, group_id) in rows {
-            let ag = GroupId::from(group_id);
-            let owner = if let Some(o) = owner_by_group.get(&ag) {
-                o.clone()
-            } else {
-                let chain = self
-                    .groups
-                    .ancestors(ag)
-                    .await
-                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
-                let o = self
-                    .nearest_claimed_from_chain(&chain)
-                    .await?
-                    .map(|(_, s)| s);
-                owner_by_group.insert(ag, o.clone());
-                o
-            };
-            if let Some(slug) = owner {
-                if Some(slug.as_str()) != exclude {
-                    *counts.entry(slug).or_default() += 1;
-                }
-            }
-        }
-        Ok(counts
+        Ok(rows
             .into_iter()
-            .map(|(project, apps)| ProjectImpact { project, apps })
+            .map(|(project, n)| ProjectImpact {
+                project,
+                apps: u32::try_from(n).unwrap_or(u32::MAX),
+            })
             .collect())
     }
 
@@ -3342,26 +3368,29 @@ async fn delete_group_var_tx(
 // ----------------------------------------------------------------------------
 
 /// What a group-node apply should do about ownership, given the current owner,
-/// the declared project, and whether `--takeover` was passed.
+/// the declared project, and whether `--takeover` was passed. Generic over the
+/// identity key `K` so the same policy serves both the apply path (`K =
+/// ProjectId`, comparing the assigned ids) and the plan preview (`K = String`,
+/// comparing slugs before any id is assigned).
 #[derive(Debug, PartialEq, Eq)]
-enum GroupClaimAction {
+enum GroupClaimAction<K> {
     /// Nothing to write (unclaimed + no project, or already owned by us).
     Noop,
     /// First-apply-claims: set `owner_project`.
-    Claim(ProjectId),
+    Claim(K),
     /// Reassign to us — capability-gated (`GroupAdmin`) at the call site.
-    Takeover(ProjectId),
+    Takeover(K),
     /// 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>,
+/// `(key, slug)`; `declared` is this apply's project key. See §7.3.
+fn decide_group_claim<K: Eq>(
+    current: Option<(K, String)>,
+    declared: Option<K>,
     takeover: bool,
-) -> GroupClaimAction {
+) -> GroupClaimAction<K> {
     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).
@@ -3388,11 +3417,12 @@ fn decide_group_claim(
 }
 
 /// 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>,
+/// their nearest claimed ancestor group (`nearest`, as `(key, slug)`); an
+/// unclaimed subtree is open (backward-compatible). Generic over the identity
+/// key `K` for the same reason as [`decide_group_claim`]. See §7.
+fn decide_app_owner<K: Eq>(
+    nearest: Option<(K, String)>,
+    declared: Option<K>,
 ) -> Result<(), String> {
     match (nearest, declared) {
         (None, _) => Ok(()),
@@ -3407,26 +3437,13 @@ fn decide_app_owner(
     }
 }
 
-/// §7: validate a `[project]` slug — the same rule as app/group slugs
-/// (`^[a-z0-9][a-z0-9-]{0,62}$`).
+/// §7: validate a `[project]` slug — the same FORMAT rule as app/group slugs
+/// (`^[a-z0-9][a-z0-9-]{0,62}$`), reusing the shared validator. A project slug
+/// is intentionally NOT reserved-word-checked: unlike a group/app slug it is
+/// never used in a route or host, so no path can collide with it.
 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(())
+    crate::groups_api::validate_slug_format(slug)
+        .map_err(|why| ApplyError::Invalid(format!("project slug {slug:?}: {why}")))
 }
 
 /// Register (or fetch) a project by slug inside the apply tx, returning its id.
@@ -5378,7 +5395,7 @@ mod tests {
             GroupClaimAction::Claim(p)
         );
         assert_eq!(
-            decide_group_claim(None, None, false),
+            decide_group_claim::<ProjectId>(None, None, false),
             GroupClaimAction::Noop
         );
     }
@@ -5419,7 +5436,7 @@ mod tests {
         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());
+        assert!(decide_app_owner::<ProjectId>(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.

crates/manager-core/src/groups_api.rs

@@ -451,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"));
     }