feat(apply): declarative shared-collection reconcile (§11.6 C4)

Thread group-collection markers through the apply engine — a name-only
resource modeled field-for-field on extension_points (§5.5): Bundle.collections,
CurrentState.collection_names, Plan.collections (+ is_noop), diff_collections
(declared→Create / live-undeclared→Delete / else NoOp), load_current loads the
names, reconcile_node_tx inserts on Create + deletes on prune via
group_collection_repo, state_token folds in coll|<name>, validate_bundle
rejects empty/duplicate names (no reserved-name guard — a collection is a data
namespace, not an importable module), ApplyReport gains collections_created/
deleted. Pruning a marker removes only the declaration; the group_kv_entries
data survives until the owning group is deleted.

The apply engine stays owner-generic; restricting declarations to [group] nodes
is enforced at the CLI manifest layer (next commit).

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

crates/manager-core/src/apply_service.rs

@@ -88,6 +88,12 @@ pub struct Bundle {
     /// optional default body is a co-located `kind = module` script.
     #[serde(default)]
     pub extension_points: Vec<String>,
+    /// Declared shared group-collection *names* (§11.6) — KV collections this
+    /// node offers as cross-app-shared. Name-only, like `extension_points`; the
+    /// data lives in `group_kv_entries`. Authored on `[group]` nodes only (the
+    /// CLI rejects it on `[app]`); the apply engine stays owner-generic.
+    #[serde(default)]
+    pub collections: Vec<String>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -322,6 +328,8 @@ pub struct Plan {
     pub vars: Vec<ResourceChange>,
     #[serde(default)]
     pub extension_points: Vec<ResourceChange>,
+    #[serde(default)]
+    pub collections: Vec<ResourceChange>,
 }
 
 impl Plan {
@@ -335,6 +343,7 @@ impl Plan {
             .chain(&self.secrets)
             .chain(&self.vars)
             .chain(&self.extension_points)
+            .chain(&self.collections)
             .all(|c| c.op == Op::NoOp)
     }
 }
@@ -413,6 +422,8 @@ pub struct CurrentState {
     pub vars: Vec<(String, serde_json::Value)>,
     /// Extension-point marker names declared directly at this node (§5.5).
     pub extension_point_names: Vec<String>,
+    /// Shared group-collection names declared directly at this node (§11.6).
+    pub collection_names: Vec<String>,
 }
 
 /// One row of the read-only extension-point report (§5.5).
@@ -810,6 +821,23 @@ impl ApplyService {
             }
         }
 
+        // 3d. Shared group-collection markers (§11.6) — insert each Create
+        // (idempotent). Name-only identity; deletes happen in prune. Pruning a
+        // marker hides the store but does NOT drop its `group_kv_entries` data
+        // (that dies only with the owning group).
+        for ch in &plan.collections {
+            if ch.op == Op::Create {
+                crate::group_collection_repo::insert_collection_tx(
+                    &mut *tx,
+                    owner.as_script_owner(),
+                    &ch.key,
+                )
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?;
+                report.collections_created += 1;
+            }
+        }
+
         // 4. Prune (only with --prune): delete stale triggers, then scripts
         // (route removals already happened in the route delete-pass above).
         // Secret pruning is deliberately deferred (destructive + irreversible):
@@ -899,6 +927,22 @@ impl ApplyService {
                     report.extension_points_deleted += 1;
                 }
             }
+
+            // Shared group-collection markers are prunable config too (§11.6).
+            // Only the marker is removed here — the data survives until the
+            // owning group is deleted.
+            for ch in &plan.collections {
+                if ch.op == Op::Delete {
+                    crate::group_collection_repo::delete_collection_tx(
+                        &mut *tx,
+                        owner.as_script_owner(),
+                        &ch.key,
+                    )
+                    .await
+                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
+                    report.collections_deleted += 1;
+                }
+            }
         }
         Ok(name_to_id)
     }
@@ -1924,6 +1968,23 @@ impl ApplyService {
                 )));
             }
         }
+
+        // Shared group collections (§11.6): unique names. No reserved-name
+        // guard — a collection name is data namespace, not an importable
+        // module, so it can't shadow an SDK namespace.
+        let mut collection_names: HashSet<String> = HashSet::new();
+        for name in &bundle.collections {
+            if name.is_empty() {
+                return Err(ApplyError::Invalid(
+                    "shared collection name must not be empty".into(),
+                ));
+            }
+            if !collection_names.insert(name.to_lowercase()) {
+                return Err(ApplyError::Invalid(format!(
+                    "duplicate shared collection `{name}`"
+                )));
+            }
+        }
         Ok(())
     }
 
@@ -1976,6 +2037,11 @@ impl ApplyService {
             crate::extension_point_repo::list_for_owner(&self.pool, owner.as_script_owner())
                 .await
                 .map_err(|e| ApplyError::Backend(e.to_string()))?;
+        // Shared group-collection markers declared directly at this node (§11.6).
+        let collection_names =
+            crate::group_collection_repo::list_for_owner(&self.pool, owner.as_script_owner())
+                .await
+                .map_err(|e| ApplyError::Backend(e.to_string()))?;
         Ok(CurrentState {
             scripts,
             routes,
@@ -1983,6 +2049,7 @@ impl ApplyService {
             secret_names,
             vars,
             extension_point_names,
+            collection_names,
         })
     }
 
@@ -2047,6 +2114,7 @@ fn compute_diff_with_names(
         secrets: diff_secrets(current, bundle),
         vars: diff_vars(current, bundle),
         extension_points: diff_extension_points(current, bundle),
+        collections: diff_collections(current, bundle),
     }
 }
 
@@ -2562,6 +2630,42 @@ fn diff_extension_points(current: &CurrentState, bundle: &Bundle) -> Vec<Resourc
     out
 }
 
+/// Diff shared group-collection markers by name (§11.6). Same shape as
+/// `diff_extension_points`: a declared-but-absent name is a Create, a
+/// live-but-undeclared name is a Delete (pruned under `--prune`). Pruning a
+/// marker hides the store but never drops its `group_kv_entries` data.
+fn diff_collections(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
+    let live: HashSet<&str> = current
+        .collection_names
+        .iter()
+        .map(String::as_str)
+        .collect();
+    let declared: HashSet<&str> = bundle.collections.iter().map(String::as_str).collect();
+
+    let mut out = Vec::new();
+    for name in &bundle.collections {
+        out.push(ResourceChange {
+            op: if live.contains(name.as_str()) {
+                Op::NoOp
+            } else {
+                Op::Create
+            },
+            key: name.clone(),
+            detail: None,
+        });
+    }
+    for name in &current.collection_names {
+        if !declared.contains(name.as_str()) {
+            out.push(ResourceChange {
+                op: Op::Delete,
+                key: name.clone(),
+                detail: Some("on server, not declared".into()),
+            });
+        }
+    }
+    out
+}
+
 /// Reject any email trigger whose referenced secret isn't set, so `plan` and
 /// `apply` give the same answer (apply also fails in `resolve_and_seal`, but
 /// only after taking the lock — surfacing it here keeps plan honest).
@@ -2852,6 +2956,10 @@ pub struct ApplyReport {
     pub extension_points_created: u32,
     #[serde(default)]
     pub extension_points_deleted: u32,
+    #[serde(default)]
+    pub collections_created: u32,
+    #[serde(default)]
+    pub collections_deleted: u32,
     #[serde(skip_serializing_if = "Vec::is_empty")]
     pub warnings: Vec<String>,
 }
@@ -3022,7 +3130,8 @@ fn state_token_with_names(
             + current.triggers.len()
             + current.secret_names.len()
             + current.vars.len()
-            + current.extension_point_names.len(),
+            + current.extension_point_names.len()
+            + current.collection_names.len(),
     );
     for s in &current.scripts {
         parts.push(format!(
@@ -3081,6 +3190,10 @@ fn state_token_with_names(
     for n in &current.extension_point_names {
         parts.push(format!("ep|{n}"));
     }
+    // Shared group-collection markers are name-only too (§11.6).
+    for n in &current.collection_names {
+        parts.push(format!("coll|{n}"));
+    }
     // Order-independent: sort the per-resource tokens before hashing.
     parts.sort_unstable();
     let mut h: u64 = 0xcbf2_9ce4_8422_2325;
@@ -3162,6 +3275,7 @@ mod tests {
             secrets: vec!["S".into()],
             vars: std::collections::BTreeMap::new(),
             extension_points: Vec::new(),
+            collections: Vec::new(),
         };
         let plan = compute_diff(&current, &bundle);
         assert_eq!(plan.scripts.len(), 1);
@@ -3185,6 +3299,7 @@ mod tests {
             secrets: vec!["S".into()],
             vars: std::collections::BTreeMap::new(),
             extension_points: Vec::new(),
+            collections: Vec::new(),
         };
         let plan = compute_diff(&current, &bundle);
         assert!(plan.is_noop(), "expected all no-op, got {plan:?}");
@@ -3203,6 +3318,7 @@ mod tests {
             secrets: vec![],
             vars: std::collections::BTreeMap::new(),
             extension_points: Vec::new(),
+            collections: Vec::new(),
         };
         let plan = compute_diff(&current, &bundle);
         assert_eq!(plan.scripts[0].op, Op::Update);
@@ -3222,6 +3338,7 @@ mod tests {
             secrets: vec![],
             vars: std::collections::BTreeMap::new(),
             extension_points: Vec::new(),
+            collections: Vec::new(),
         };
         let plan = compute_diff(&current, &bundle);
         assert_eq!(plan.scripts[0].op, Op::Delete);
@@ -3269,6 +3386,7 @@ mod tests {
             secrets: vec![],
             vars: std::collections::BTreeMap::new(),
             extension_points: Vec::new(),
+            collections: Vec::new(),
         };
         let plan = compute_diff(&current, &bundle);
         assert_eq!(plan.routes[0].op, Op::Update);
@@ -3282,6 +3400,7 @@ mod tests {
             secrets: vec![],
             vars: std::collections::BTreeMap::new(),
             extension_points: Vec::new(),
+            collections: Vec::new(),
         }
     }