feat(apply): project on the plan wire + ownership preview annotation

§7 M3 (part 1): `pic plan` now surfaces the ownership outcome BEFORE apply.

- plan / plan_owner / plan_tree take the declaring `[project]`; each result
  carries an `ownership` preview (pure `preview_ownership`, unit-tested):
  claim (unclaimed + project), owned (already this project), conflict (owned by
  X — named), or unclaimed. A group node previews its OWN owner; an app node its
  nearest claimed ancestor (read-only, on the pool).
- the attach-point ceiling (M2) is now previewed at plan too, so `pic plan`
  outside the subtree fails early — consistent with apply.
- wire: PlanRequest / TreePlanRequest wrap { bundle, project } (project
  `#[serde(default)]` → a pre-M3 CLI still plans); the plan DTOs + `pic plan`
  gain an `ownership` row (mode-agnostic: shows in tsv + json).

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

crates/manager-core/src/apply_api.rs

@@ -179,6 +179,15 @@ async fn group_extension_points_handler(
     Ok(Json(report))
 }
 
+/// A `plan` request (§7 M3): the desired bundle plus the optional declaring
+/// `[project]`, so the plan can preview the ownership outcome + attach ceiling.
+#[derive(Deserialize)]
+pub struct PlanRequest {
+    pub bundle: Bundle,
+    #[serde(default)]
+    pub project: Option<ProjectDecl>,
+}
+
 #[derive(Deserialize)]
 pub struct ApplyRequest {
     pub bundle: Bundle,
@@ -225,7 +234,7 @@ async fn plan_handler(
     State(svc): State<ApplyService>,
     Extension(principal): Extension<Principal>,
     Path(id_or_slug): Path<String>,
-    Json(bundle): Json<Bundle>,
+    Json(req): Json<PlanRequest>,
 ) -> Result<Json<PlanResult>, ApplyError> {
     let app_id = resolve_app_id(svc.apps.as_ref(), &id_or_slug).await?;
     // NOTE: the returned `Plan` discloses live secret NAMES (not values). That
@@ -237,7 +246,7 @@ async fn plan_handler(
     require(svc.authz.as_ref(), &principal, Capability::AppRead(app_id))
         .await
         .map_err(map_authz)?;
-    let plan = svc.plan(app_id, &bundle).await?;
+    let plan = svc.plan(app_id, &req.bundle, req.project.as_ref()).await?;
     Ok(Json(plan))
 }
 
@@ -251,7 +260,7 @@ async fn group_plan_handler(
     State(svc): State<ApplyService>,
     Extension(principal): Extension<Principal>,
     Path(id_or_slug): Path<String>,
-    Json(bundle): Json<Bundle>,
+    Json(req): Json<PlanRequest>,
 ) -> Result<Json<PlanResult>, ApplyError> {
     let group_id = resolve_group_id(svc.groups.as_ref(), &id_or_slug).await?;
     // Plan discloses the group's script + var + secret names; viewer-tier reads.
@@ -262,7 +271,13 @@ async fn group_plan_handler(
     )
     .await
     .map_err(map_authz)?;
-    let plan = svc.plan_owner(ApplyOwner::Group(group_id), &bundle).await?;
+    let plan = svc
+        .plan_owner(
+            ApplyOwner::Group(group_id),
+            &req.bundle,
+            req.project.as_ref(),
+        )
+        .await?;
     Ok(Json(plan))
 }
 
@@ -310,13 +325,22 @@ struct TreeApplyRequest {
     takeover: bool,
 }
 
+#[derive(Deserialize)]
+struct TreePlanRequest {
+    bundle: TreeBundle,
+    #[serde(default)]
+    project: Option<ProjectDecl>,
+}
+
 async fn tree_plan_handler(
     State(svc): State<ApplyService>,
     Extension(principal): Extension<Principal>,
-    Json(bundle): Json<TreeBundle>,
+    Json(req): Json<TreePlanRequest>,
 ) -> Result<Json<TreePlanResult>, ApplyError> {
-    authz_tree(&svc, &principal, &bundle, None).await?;
+    authz_tree(&svc, &principal, &req.bundle, None).await?;
-    Ok(Json(svc.plan_tree(&bundle).await?))
+    Ok(Json(
+        svc.plan_tree(&req.bundle, req.project.as_ref()).await?,
+    ))
 }
 
 async fn tree_apply_handler(

crates/manager-core/src/apply_service.rs

@@ -476,6 +476,10 @@ pub struct PlanResult {
     #[serde(flatten)]
     pub plan: Plan,
     pub state_token: String,
+    /// §7 M3: the ownership outcome this apply would produce (claim / owned /
+    /// conflict / unclaimed). Present only when a `[project]` was declared.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ownership: Option<OwnershipPreview>,
 }
 
 // ----------------------------------------------------------------------------
@@ -531,6 +535,39 @@ pub struct OwnershipClaim<'a> {
     pub principal: &'a Principal,
 }
 
+/// §7 M3 plan preview: what applying this node WOULD do about ownership, so
+/// `pic plan` surfaces a conflict before `pic apply`. `action` ∈ `claim`
+/// (unclaimed → this project claims it), `owned` (already this project),
+/// `conflict` (owned by `owner`), `unclaimed` (no project, or an app in open
+/// territory). `owner` is the current owner's slug for `owned`/`conflict`.
+#[derive(Debug, Clone, Serialize)]
+pub struct OwnershipPreview {
+    pub action: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub owner: Option<String>,
+}
+
+/// Pure preview policy (read-only, slug-only — no registration): compares the
+/// node's current owner slug against this apply's declared project slug.
+fn preview_ownership(
+    is_group: bool,
+    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),
+    };
+    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)),
+    }
+}
+
 /// One node's diff within a tree plan.
 #[derive(Debug, Clone, Serialize)]
 pub struct NodePlan {
@@ -538,6 +575,9 @@ pub struct NodePlan {
     pub slug: String,
     #[serde(flatten)]
     pub plan: Plan,
+    /// §7 M3: this node's ownership outcome under the tree's `[project]`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ownership: Option<OwnershipPreview>,
 }
 
 /// What `plan_tree` returns: a per-node diff plus ONE combined bound-plan token
@@ -712,8 +752,14 @@ impl ApplyService {
     ///
     /// # Errors
     /// `Invalid` for a malformed bundle; `Backend` for repo failures.
-    pub async fn plan(&self, app_id: AppId, bundle: &Bundle) -> Result<PlanResult, ApplyError> {
+    pub async fn plan(
-        self.plan_owner(ApplyOwner::App(app_id), bundle).await
+        &self,
+        app_id: AppId,
+        bundle: &Bundle,
+        project: Option<&ProjectDecl>,
+    ) -> Result<PlanResult, ApplyError> {
+        self.plan_owner(ApplyOwner::App(app_id), bundle, project)
+            .await
     }
 
     /// Owner-generic plan (Phase 5): diff `bundle` against an app OR group node.
@@ -726,6 +772,7 @@ impl ApplyService {
         &self,
         owner: ApplyOwner,
         bundle: &Bundle,
+        project: Option<&ProjectDecl>,
     ) -> Result<PlanResult, ApplyError> {
         let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
         self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
@@ -734,14 +781,24 @@ impl ApplyService {
         if let ApplyOwner::App(app_id) = owner {
             self.validate_route_hosts(app_id, bundle).await?;
         }
+        // §6/§7 M3: preview the attach-point ceiling (same rule as apply) so a
+        // `pic plan` outside the subtree fails early, consistent with apply.
+        if let Some(pg_slug) = project.and_then(|p| p.parent_group.as_deref()) {
+            let pg = self.resolve_group(pg_slug).await?;
+            self.check_within_attach(owner, pg, pg_slug).await?;
+        }
         let current = self.load_current(owner).await?;
         let current_names = self.resolve_current_target_names(&current).await?;
         validate_email_secrets_present(bundle, &current.secret_names)?;
+        let ownership = self
+            .ownership_preview(owner, project.map(|p| p.slug.as_str()))
+            .await?;
         Ok(PlanResult {
             plan: compute_diff_with_names(&current, bundle, &current_names),
             // Fingerprint of the live state this plan was computed against, so
             // `apply` can refuse if the node changed underneath it (§4.2).
             state_token: state_token_with_names(&current, &current_names),
+            ownership,
         })
     }
 
@@ -1527,16 +1584,30 @@ impl ApplyService {
     ///
     /// # Errors
     /// `Invalid` for a malformed/unknown bundle; `Backend` for repo failures.
-    pub async fn plan_tree(&self, bundle: &TreeBundle) -> Result<TreePlanResult, ApplyError> {
+    pub async fn plan_tree(
+        &self,
+        bundle: &TreeBundle,
+        project: Option<&ProjectDecl>,
+    ) -> Result<TreePlanResult, ApplyError> {
         let (prepared, token) = self.prepare_tree(bundle).await?;
-        let nodes = prepared
+        // §6/§7 M3: attach-point ceiling preview (consistent with apply_tree).
-            .into_iter()
+        if let Some(pg_slug) = project.and_then(|p| p.parent_group.as_deref()) {
-            .map(|p| NodePlan {
+            let pg = self.resolve_group(pg_slug).await?;
+            for p in &prepared {
+                self.check_within_attach(p.owner, pg, pg_slug).await?;
+            }
+        }
+        let declared = project.map(|p| p.slug.as_str());
+        let mut nodes = Vec::with_capacity(prepared.len());
+        for p in prepared {
+            let ownership = self.ownership_preview(p.owner, declared).await?;
+            nodes.push(NodePlan {
                 kind: p.node.kind,
                 slug: p.node.slug.clone(),
                 plan: p.plan,
-            })
+                ownership,
-            .collect();
+            });
+        }
         Ok(TreePlanResult {
             nodes,
             state_token: token,
@@ -2017,6 +2088,67 @@ impl ApplyService {
         Ok(())
     }
 
+    /// A group's OWN owning-project slug (not inherited), read on the pool.
+    async fn read_group_owner_pool(&self, gid: GroupId) -> Result<Option<String>, ApplyError> {
+        let g = self
+            .groups
+            .get_by_id(gid)
+            .await
+            .map_err(|e| ApplyError::Backend(e.to_string()))?;
+        match g.and_then(|g| g.owner_project) {
+            None => Ok(None),
+            Some(pid) => Ok(self
+                .projects
+                .get_by_id(pid)
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?
+                .map(|p| p.slug)),
+        }
+    }
+
+    /// §7 M3 preview: the ownership outcome a node's apply would produce, or
+    /// `None` when no `[project]` is declared (nothing to preview). Read-only.
+    /// A group node previews against its OWN owner; an app node against its
+    /// nearest claimed ancestor.
+    async fn ownership_preview(
+        &self,
+        owner: ApplyOwner,
+        declared: Option<&str>,
+    ) -> Result<Option<OwnershipPreview>, ApplyError> {
+        if declared.is_none() {
+            return Ok(None);
+        }
+        let (is_group, current) = match owner {
+            ApplyOwner::Group(g) => (true, self.read_group_owner_pool(g).await?),
+            ApplyOwner::App(a) => {
+                let current = match self
+                    .apps
+                    .get_by_id(a)
+                    .await
+                    .map_err(|e| ApplyError::Backend(e.to_string()))?
+                {
+                    Some(app) => {
+                        let chain = self
+                            .groups
+                            .ancestors(app.group_id)
+                            .await
+                            .map_err(|e| ApplyError::Backend(e.to_string()))?;
+                        self.nearest_claimed_from_chain(&chain)
+                            .await?
+                            .map(|(_, s)| s)
+                    }
+                    None => None,
+                };
+                (false, current)
+            }
+        };
+        Ok(Some(preview_ownership(
+            is_group,
+            current.as_deref(),
+            declared,
+        )))
+    }
+
     async fn resolve_app(&self, ident: &str) -> Result<picloud_shared::App, ApplyError> {
         crate::app_repo::resolve_app(self.apps.as_ref(), ident)
             .await
@@ -5218,4 +5350,27 @@ mod tests {
         assert!(decide_app_owner(Some((owner, "platform".into())), Some(me)).is_err());
         assert!(decide_app_owner(Some((owner, "platform".into())), None).is_err());
     }
+
+    #[test]
+    fn ownership_preview_reflects_claim_owned_conflict() {
+        // Group node: unclaimed + project → claim; already ours → owned; foreign
+        // → conflict (owner named); no project → unclaimed.
+        assert_eq!(preview_ownership(true, None, Some("p")).action, "claim");
+        assert_eq!(preview_ownership(true, None, None).action, "unclaimed");
+        let owned = preview_ownership(true, Some("p"), Some("p"));
+        assert_eq!(owned.action, "owned");
+        assert_eq!(owned.owner.as_deref(), Some("p"));
+        let conflict = preview_ownership(true, Some("plat"), Some("teamb"));
+        assert_eq!(conflict.action, "conflict");
+        assert_eq!(conflict.owner.as_deref(), Some("plat"));
+        // App node never claims — unclaimed subtree stays unclaimed.
+        assert_eq!(
+            preview_ownership(false, None, Some("p")).action,
+            "unclaimed"
+        );
+        assert_eq!(
+            preview_ownership(false, Some("plat"), Some("teamb")).action,
+            "conflict"
+        );
+    }
 }

crates/picloud-cli/src/client.rs

@@ -1220,14 +1220,16 @@ impl Client {
         kind: NodeKind,
         slug: &str,
         bundle: &serde_json::Value,
+        project: Option<&crate::manifest::ManifestProject>,
     ) -> Result<PlanDto> {
         let slug = seg(slug);
+        let body = serde_json::json!({ "bundle": bundle, "project": project });
         let resp = self
             .request(
                 Method::POST,
                 &format!("/api/v1/admin/{}/{slug}/plan", kind.path()),
             )
-            .json(bundle)
+            .json(&body)
             .send()
             .await?;
         decode(resp).await
@@ -1274,10 +1276,15 @@ 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> {
+    pub async fn plan_tree(
+        &self,
+        bundle: &serde_json::Value,
+        project: Option<&crate::manifest::ManifestProject>,
+    ) -> Result<TreePlanDto> {
+        let body = serde_json::json!({ "bundle": bundle, "project": project });
         let resp = self
             .request(Method::POST, "/api/v1/admin/tree/plan")
-            .json(bundle)
+            .json(&body)
             .send()
             .await?;
         decode(resp).await
@@ -1527,6 +1534,9 @@ pub struct PlanDto {
     /// in `.picloud/` and replayed to `apply` for the bound-plan check.
     #[serde(default)]
     pub state_token: String,
+    /// §7 M3: the ownership outcome this apply would produce.
+    #[serde(default)]
+    pub ownership: Option<OwnershipPreviewDto>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -1537,6 +1547,15 @@ pub struct ChangeDto {
     pub detail: Option<String>,
 }
 
+/// §7 M3 plan preview: `action` ∈ claim/owned/conflict/unclaimed; `owner` is
+/// the current owner's slug (for owned/conflict).
+#[derive(Debug, Deserialize)]
+pub struct OwnershipPreviewDto {
+    pub action: String,
+    #[serde(default)]
+    pub owner: Option<String>,
+}
+
 /// Response of `POST .../tree/plan` (Phase 5): a per-node diff + one combined
 /// bound-plan token for the whole subtree.
 #[derive(Debug, Deserialize)]
@@ -1567,6 +1586,9 @@ pub struct NodePlanDto {
     pub collections: Vec<ChangeDto>,
     #[serde(default)]
     pub suppressions: Vec<ChangeDto>,
+    /// §7 M3: this node's ownership outcome under the tree's `[project]`.
+    #[serde(default)]
+    pub ownership: Option<OwnershipPreviewDto>,
 }
 
 /// Response of `POST .../apply`: counts of what changed.

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

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