feat(apply): atomic project-tree apply with cross-node inherited binding (Phase 5 C3)

Phase 5 C3 — the headline. A whole project subtree (group + app nodes) is
reconciled in ONE Postgres transaction, all-or-nothing.

`POST /api/v1/admin/tree/{plan,apply}` take a `TreeBundle { nodes: [{kind,
slug, bundle}] }`. The actor must hold the relevant read/write caps for EVERY
node (per-kind, widened on prune) — `authz_tree` + reusable
`require_{app,group}_node_writes` (now shared with the single-node handlers).

Engine:
  * The in-tx reconcile body of `apply_owner` is extracted to `reconcile_node_tx`
    (returns the node's post-create `name → script_id`), so single-node and
    tree apply share one implementation — behaviour-preserving (391 unit tests
    + app journeys green).
  * `prepare_tree` resolves each slug→owner, validates (an app's route/trigger
    target may bind to an in-tree ancestor group's script), computes each
    node's plan, and folds a combined bound-plan token = every node's
    `state_token` + every in-scope group's `structure_version` (so a reparent
    or content edit between plan and apply trips StateMoved).
  * `apply_tree`: lock every node key (sorted, deadlock-free); reconcile GROUPS
    first, recording each group's `name → id` in an in-memory index; then APPS,
    resolving inherited route/trigger targets nearest-ancestor-wins across the
    in-tree index (incl. scripts created earlier in THIS tx, invisible to the
    pool resolver) and pre-existing out-of-tree ancestors. One commit, one
    post-commit route refresh.

Live-validated against the dev DB:
  * Headline: a group node creates `shared`; an app node in the SAME apply
    binds `GET /greet` to it — the route ends up bound to the group-owned
    script, and re-plan is all-noop (idempotent).
  * Atomicity: one invalid node → 422 and ZERO rows written (the valid group
    node's script is not created).

CLI tree discovery + `pic plan/apply` tree mode is C4; journeys + docs are C5.

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

crates/manager-core/src/apply_api.rs

@@ -17,7 +17,8 @@ use serde_json::json;
 
 use crate::app_repo::AppRepository;
 use crate::apply_service::{
-    ApplyError, ApplyOwner, ApplyReport, ApplyService, Bundle, BundleTrigger, PlanResult,
+    ApplyError, ApplyOwner, ApplyReport, ApplyService, Bundle, BundleTrigger, NodeKind, PlanResult,
+    TreeBundle, TreePlanResult,
 };
 use crate::authz::{require, AuthzDenied, Capability};
 use crate::group_repo::GroupRepository;
@@ -29,6 +30,8 @@ pub fn apply_router(service: ApplyService) -> Router {
         .route("/apps/{id}/apply", post(apply_handler))
         .route("/groups/{id}/plan", post(group_plan_handler))
         .route("/groups/{id}/apply", post(group_apply_handler))
+        .route("/tree/plan", post(tree_plan_handler))
+        .route("/tree/apply", post(tree_apply_handler))
         .with_state(service)
 }
 
@@ -50,62 +53,7 @@ async fn apply_handler(
     Json(req): Json<ApplyRequest>,
 ) -> Result<Json<ApplyReport>, ApplyError> {
     let app_id = resolve_app_id(svc.apps.as_ref(), &id_or_slug).await?;
-    // Read is always needed; write caps are required for the resource kinds
-    // the bundle touches — and for ALL kinds when `prune` is set, since
-    // pruning deletes resources whose bundle section is empty (and a script
-    // delete cascades its routes/triggers).
-    require(svc.authz.as_ref(), &principal, Capability::AppRead(app_id))
-        .await
-        .map_err(map_authz)?;
-    if req.prune || !req.bundle.scripts.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::AppWriteScript(app_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    if req.prune || !req.bundle.routes.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::AppWriteRoute(app_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    if req.prune || !req.bundle.triggers.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::AppManageTriggers(app_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    if req.prune || !req.bundle.vars.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::AppVarsWrite(app_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    // Email triggers resolve and decrypt a stored secret by name server-side,
-    // which the secrets API guards with `AppSecretsRead`. Require it here too
-    // so apply can't bind a secret a principal couldn't otherwise read — the
-    // caps aren't strictly nested on the API-key scope path.
-    if req.bundle.triggers.iter().any(BundleTrigger::is_email) {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::AppSecretsRead(app_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
+    require_app_node_writes(&svc, &principal, app_id, &req.bundle, req.prune).await?;
     let report = svc
         .apply(
             app_id,
@@ -170,26 +118,7 @@ async fn group_apply_handler(
     Json(req): Json<ApplyRequest>,
 ) -> Result<Json<ApplyReport>, ApplyError> {
     let group_id = resolve_group_id(svc.groups.as_ref(), &id_or_slug).await?;
-    // Write caps per touched kind (+ all kinds on prune). Group scripts are
-    // editor-tier (`GroupScriptsWrite`), group vars likewise (`GroupVarsWrite`).
-    if req.prune || !req.bundle.scripts.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::GroupScriptsWrite(group_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
-    if req.prune || !req.bundle.vars.is_empty() {
-        require(
-            svc.authz.as_ref(),
-            &principal,
-            Capability::GroupVarsWrite(group_id),
-        )
-        .await
-        .map_err(map_authz)?;
-    }
+    require_group_node_writes(&svc, &principal, group_id, &req.bundle, req.prune).await?;
     let report = svc
         .apply_owner(
             ApplyOwner::Group(group_id),
@@ -202,6 +131,188 @@ async fn group_apply_handler(
     Ok(Json(report))
 }
 
+// ----------------------------------------------------------------------------
+// Tree apply (Phase 5): a whole project subtree in one transaction. The actor
+// must hold the relevant read/write caps for EVERY node touched.
+// ----------------------------------------------------------------------------
+
+#[derive(Deserialize)]
+struct TreeApplyRequest {
+    bundle: TreeBundle,
+    #[serde(default)]
+    prune: bool,
+    #[serde(default)]
+    expected_token: Option<String>,
+}
+
+async fn tree_plan_handler(
+    State(svc): State<ApplyService>,
+    Extension(principal): Extension<Principal>,
+    Json(bundle): Json<TreeBundle>,
+) -> Result<Json<TreePlanResult>, ApplyError> {
+    authz_tree(&svc, &principal, &bundle, None).await?;
+    Ok(Json(svc.plan_tree(&bundle).await?))
+}
+
+async fn tree_apply_handler(
+    State(svc): State<ApplyService>,
+    Extension(principal): Extension<Principal>,
+    Json(req): Json<TreeApplyRequest>,
+) -> Result<Json<ApplyReport>, ApplyError> {
+    authz_tree(&svc, &principal, &req.bundle, Some(req.prune)).await?;
+    let report = svc
+        .apply_tree(
+            &req.bundle,
+            req.prune,
+            principal.user_id,
+            req.expected_token.as_deref(),
+        )
+        .await?;
+    Ok(Json(report))
+}
+
+/// Per-node capability check for a tree plan/apply. `prune` widens an apply's
+/// write requirement to every kind. A plan (`prune` ignored, no writes) needs
+/// only the per-node read cap.
+async fn authz_tree(
+    svc: &ApplyService,
+    principal: &Principal,
+    bundle: &TreeBundle,
+    apply_prune: Option<bool>,
+) -> Result<(), ApplyError> {
+    for node in &bundle.nodes {
+        match node.kind {
+            NodeKind::App => {
+                let app_id = resolve_app_id(svc.apps.as_ref(), &node.slug).await?;
+                match apply_prune {
+                    Some(prune) => {
+                        require_app_node_writes(svc, principal, app_id, &node.bundle, prune)
+                            .await?;
+                    }
+                    None => {
+                        require(svc.authz.as_ref(), principal, Capability::AppRead(app_id))
+                            .await
+                            .map_err(map_authz)?;
+                    }
+                }
+            }
+            NodeKind::Group => {
+                let group_id = resolve_group_id(svc.groups.as_ref(), &node.slug).await?;
+                match apply_prune {
+                    Some(prune) => {
+                        require_group_node_writes(svc, principal, group_id, &node.bundle, prune)
+                            .await?;
+                    }
+                    None => {
+                        require(
+                            svc.authz.as_ref(),
+                            principal,
+                            Capability::GroupScriptsRead(group_id),
+                        )
+                        .await
+                        .map_err(map_authz)?;
+                    }
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+/// App-node write caps: per resource kind the bundle touches, widened to all
+/// kinds when `prune` (which deletes empty-section resources + cascades). Read
+/// is always needed. Shared by the single-app and tree apply paths.
+async fn require_app_node_writes(
+    svc: &ApplyService,
+    principal: &Principal,
+    app_id: picloud_shared::AppId,
+    bundle: &Bundle,
+    prune: bool,
+) -> Result<(), ApplyError> {
+    require(svc.authz.as_ref(), principal, Capability::AppRead(app_id))
+        .await
+        .map_err(map_authz)?;
+    if prune || !bundle.scripts.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::AppWriteScript(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    if prune || !bundle.routes.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::AppWriteRoute(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    if prune || !bundle.triggers.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::AppManageTriggers(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    if prune || !bundle.vars.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::AppVarsWrite(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    // Email triggers resolve + decrypt a stored secret by name server-side
+    // (`AppSecretsRead`), so require it so apply can't bind a secret the
+    // principal couldn't otherwise read.
+    if bundle.triggers.iter().any(BundleTrigger::is_email) {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::AppSecretsRead(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    Ok(())
+}
+
+/// Group-node write caps (editor-tier): scripts → `GroupScriptsWrite`, vars →
+/// `GroupVarsWrite`, both on prune.
+async fn require_group_node_writes(
+    svc: &ApplyService,
+    principal: &Principal,
+    group_id: GroupId,
+    bundle: &Bundle,
+    prune: bool,
+) -> Result<(), ApplyError> {
+    if prune || !bundle.scripts.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::GroupScriptsWrite(group_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    if prune || !bundle.vars.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            principal,
+            Capability::GroupVarsWrite(group_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    Ok(())
+}
+
 /// Resolve a slug-or-id path param to a `GroupId`, mapping miss → 404.
 async fn resolve_group_id(
     groups: &dyn GroupRepository,