feat(stateful-templates): materialize cron+queue templates + hooks (M5.2-M5.4)

materialize::rematerialize_stateful_templates reconciles a group cron/queue
template into one app-owned copy per descendant (materialized_from = template),
via the all-apps app_chain CTE ⋈ group-owned stateful templates. A precise
create/delete diff preserves cron last_fired_at on unchanged copies; a queue
copy is skipped-with-warning when the app already fills that queue's consumer
slot (the one-consumer invariant). Called full-live at the route-rebuild
chokepoints: apply (single + tree), app create/delete (apps_api), group
reparent (groups_api) — AppsState/GroupsState gain a pool. Pinned by
tests/stateful_templates.rs (per-app copy, idempotent, new-app materializes,
reparent de-materializes).

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

crates/manager-core/src/apply_service.rs

@@ -1420,6 +1420,11 @@ impl ApplyService {
                     .push("route table refresh failed; it will self-heal".into());
             }
         }
+        // §4.5 M5: re-materialize stateful (cron/queue) group templates into
+        // per-descendant app rows. Post-commit + best-effort like the route
+        // rebuild; per-app skips (e.g. a queue slot already taken) surface as
+        // warnings.
+        report.warnings.extend(self.rematerialize().await);
         Ok(report)
     }
 
@@ -1559,6 +1564,8 @@ impl ApplyService {
                 .warnings
                 .push("route table refresh failed; it will self-heal".into());
         }
+        // §4.5 M5: one post-commit materialization pass covers the whole tree.
+        report.warnings.extend(self.rematerialize().await);
         Ok(report)
     }
 
@@ -1851,6 +1858,19 @@ impl ApplyService {
             .map_err(|e| ApplyError::Backend(e.to_string()))
     }
 
+    /// §4.5 M5: reconcile materialized stateful-template copies. Best-effort —
+    /// a failure is logged (the copies self-heal on the next mutation) and any
+    /// per-app skip warnings are returned for the apply report.
+    async fn rematerialize(&self) -> Vec<String> {
+        match crate::materialize::rematerialize_stateful_templates(&self.pool).await {
+            Ok(warnings) => warnings,
+            Err(e) => {
+                tracing::warn!(error = %e, "apply: stateful-template materialization failed");
+                vec!["stateful-template materialization failed; it will self-heal".into()]
+            }
+        }
+    }
+
     /// Resolve a referenced secret to plaintext, then re-seal it for an
     /// email trigger's inbound-secret column. The value never appears in
     /// the manifest — only the secret's name does.

crates/manager-core/src/apps_api.rs

@@ -52,6 +52,10 @@ pub struct AppsState {
     /// Group tree — resolves an app's parent group at create time
     /// (defaults to the instance root).
     pub groups: Arc<dyn GroupRepository>,
+    /// §4.5 M5: creating/deleting an app changes which ancestor-group stateful
+    /// templates it inherits, so its materialized cron/queue copies are
+    /// reconciled here (full-live, mirroring the route_table rebuild).
+    pub pool: sqlx::PgPool,
 }
 
 pub fn apps_router(state: AppsState) -> Router {
@@ -565,6 +569,11 @@ async fn refresh_route_cache(state: &AppsState) {
     {
         tracing::warn!(error = %e, "apps: route table refresh failed; it will self-heal");
     }
+    // §4.5 M5: reconcile materialized stateful-template copies for the changed
+    // app set (best-effort; self-heals on the next mutation).
+    if let Err(e) = crate::materialize::rematerialize_stateful_templates(&state.pool).await {
+        tracing::warn!(error = %e, "apps: stateful-template materialization failed; it will self-heal");
+    }
 }
 
 pub async fn refresh_domain_cache(state: &AppsState) -> Result<(), AppsApiError> {

crates/manager-core/src/groups_api.rs

@@ -50,6 +50,10 @@ pub struct GroupsState {
     /// apps inherit — so the route snapshot is rebuilt after a reparent.
     pub routes: Arc<dyn RouteRepository>,
     pub route_table: Arc<RouteTable>,
+    /// §4.5 M5: reparenting moves a subtree, changing which ancestor-group
+    /// stateful templates its apps inherit — their materialized cron/queue
+    /// copies are reconciled after a reparent.
+    pub pool: sqlx::PgPool,
 }
 
 pub fn groups_router(state: GroupsState) -> Router {
@@ -295,6 +299,11 @@ async fn reparent_group(
     {
         tracing::warn!(error = %e, "groups: route table refresh after reparent failed; it will self-heal");
     }
+    // §4.5 M5: the moved subtree inherits a different set of ancestor-group
+    // stateful templates — reconcile its materialized copies.
+    if let Err(e) = crate::materialize::rematerialize_stateful_templates(&s.pool).await {
+        tracing::warn!(error = %e, "groups: stateful-template materialization after reparent failed; it will self-heal");
+    }
     Ok(Json(moved))
 }
 

crates/manager-core/src/lib.rs

@@ -70,6 +70,7 @@ pub mod kv_repo;
 pub mod kv_service;
 pub mod log_sink;
 pub mod login_rate_limit;
+pub mod materialize;
 pub mod migrations;
 pub mod module_source;
 pub mod outbox_event_emitter;

crates/manager-core/src/materialize.rs

@@ -0,0 +1,186 @@
+//! §4.5 M5: materialization of STATEFUL group trigger templates.
+//!
+//! A group-owned cron / queue / email trigger is a TEMPLATE that is never
+//! dispatched directly (the scheduler / queue consumer / email webhook all
+//! filter `t.app_id IS NOT NULL`). Instead this reconciler expands each template
+//! into an app-owned copy — `materialized_from = template.id` — for every
+//! descendant app, and removes copies whose (app, template) pair no longer
+//! inherits. The copy owns its per-app state (cron `last_fired_at`, the queue
+//! advisory lock, the email sealed secret), so the unchanged dispatch paths work
+//! against it as if it were hand-authored.
+//!
+//! Full-live like `rebuild_route_table`: called at the same chokepoints (apply,
+//! app create/delete, group reparent). A precise create/delete diff (not
+//! delete-all-then-recreate) preserves `last_fired_at` on unchanged cron copies.
+//!
+//! Returns per-app WARNINGS (e.g. a queue whose consumer slot is already taken,
+//! or an email whose descendant lacks the referenced secret) — the caller
+//! surfaces them; materialization of the other pairs still proceeds.
+
+use std::collections::HashSet;
+
+use sqlx::PgPool;
+use uuid::Uuid;
+
+/// The stateful kinds that materialize (M5.2 cron; M5.4 queue; M5.5 email).
+const MATERIALIZED_KINDS: &[&str] = &["cron", "queue"];
+
+/// (descendant app, source template, kind) that SHOULD have a materialized copy.
+#[derive(sqlx::FromRow)]
+struct ShouldRow {
+    effective_app_id: Uuid,
+    template_id: Uuid,
+    kind: String,
+}
+
+/// Reconcile all materialized stateful-template copies. Idempotent; safe to call
+/// on every tree/apply mutation. Best-effort per pair — a pair that can't
+/// materialize (queue slot taken, missing email secret) yields a warning and is
+/// skipped, not an error.
+pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result<Vec<String>, sqlx::Error> {
+    let mut tx = pool.begin().await?;
+
+    // Every (descendant app × ancestor-group stateful template) that should
+    // exist — the all-apps `app_chain` CTE (each app × its ancestor chain) ⋈
+    // group-owned stateful templates.
+    let kinds: Vec<String> = MATERIALIZED_KINDS
+        .iter()
+        .map(|s| (*s).to_string())
+        .collect();
+    let should: Vec<ShouldRow> = sqlx::query_as(
+        "WITH RECURSIVE app_chain AS ( \
+             SELECT a.id AS effective_app_id, a.group_id AS owner_group, \
+                    a.group_id AS next_group, 0 AS depth \
+             FROM apps a WHERE a.group_id IS NOT NULL \
+           UNION ALL \
+             SELECT ac.effective_app_id, g.parent_id, g.parent_id, ac.depth + 1 \
+             FROM groups g JOIN app_chain ac ON g.id = ac.next_group \
+             WHERE ac.depth < 64 AND g.parent_id IS NOT NULL \
+         ) \
+         SELECT DISTINCT ac.effective_app_id, t.id AS template_id, t.kind \
+         FROM app_chain ac \
+         JOIN triggers t ON t.group_id = ac.owner_group \
+         WHERE t.group_id IS NOT NULL AND t.enabled = TRUE AND t.kind = ANY($1)",
+    )
+    .bind(&kinds)
+    .fetch_all(&mut *tx)
+    .await?;
+
+    let should_set: HashSet<(Uuid, Uuid)> = should
+        .iter()
+        .map(|r| (r.effective_app_id, r.template_id))
+        .collect();
+
+    // Existing managed copies.
+    let existing: Vec<(Uuid, Uuid)> = sqlx::query_as(
+        "SELECT app_id, materialized_from FROM triggers WHERE materialized_from IS NOT NULL",
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    let existing_set: HashSet<(Uuid, Uuid)> = existing.iter().copied().collect();
+
+    // Delete stale copies (template gone or app no longer a descendant).
+    for (app_id, template_id) in &existing_set {
+        if !should_set.contains(&(*app_id, *template_id)) {
+            sqlx::query("DELETE FROM triggers WHERE app_id = $1 AND materialized_from = $2")
+                .bind(app_id)
+                .bind(template_id)
+                .execute(&mut *tx)
+                .await?;
+        }
+    }
+
+    // Create missing copies.
+    let mut warnings = Vec::new();
+    for r in &should {
+        if existing_set.contains(&(r.effective_app_id, r.template_id)) {
+            continue;
+        }
+        if let Some(w) =
+            materialize_one(&mut tx, r.effective_app_id, r.template_id, &r.kind).await?
+        {
+            warnings.push(w);
+        }
+    }
+
+    tx.commit().await?;
+    Ok(warnings)
+}
+
+/// Create one app-owned copy of a group template + its per-kind detail. Returns
+/// `Some(warning)` when the pair is intentionally skipped (e.g. a queue slot the
+/// app already fills). `name` uses the column default (a fresh UUID) so it can't
+/// collide with the app's own trigger names.
+async fn materialize_one(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    app_id: Uuid,
+    template_id: Uuid,
+    kind: &str,
+) -> Result<Option<String>, sqlx::Error> {
+    // §M5.4: a queue copy would violate the one-consumer-per-(app_id, queue_name)
+    // invariant if the app already has a consumer (hand-authored or another
+    // template) on that queue — skip with a warning.
+    if kind == "queue" {
+        let taken: Option<(Uuid,)> = sqlx::query_as(
+            "SELECT t.id FROM triggers t \
+             JOIN queue_trigger_details d ON d.trigger_id = t.id \
+             WHERE t.app_id = $1 AND t.kind = 'queue' \
+               AND d.queue_name = ( \
+                   SELECT queue_name FROM queue_trigger_details WHERE trigger_id = $2)",
+        )
+        .bind(app_id)
+        .bind(template_id)
+        .fetch_optional(&mut **tx)
+        .await?;
+        if taken.is_some() {
+            return Ok(Some(format!(
+                "queue template not materialized for one app — it already has a \
+                 consumer on that queue (app_id={app_id})"
+            )));
+        }
+    }
+
+    // Parent row: copy the template's settings, owned by the app, linked back.
+    let new_id: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers ( \
+            app_id, script_id, kind, enabled, dispatch_mode, \
+            retry_max_attempts, retry_backoff, retry_base_ms, \
+            registered_by_principal, materialized_from) \
+         SELECT $1, script_id, kind, enabled, dispatch_mode, \
+                retry_max_attempts, retry_backoff, retry_base_ms, \
+                registered_by_principal, id \
+         FROM triggers WHERE id = $2 RETURNING id",
+    )
+    .bind(app_id)
+    .bind(template_id)
+    .fetch_one(&mut **tx)
+    .await?;
+
+    match kind {
+        "cron" => {
+            // A fresh copy starts with last_fired_at NULL → fires on next due.
+            sqlx::query(
+                "INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
+                 SELECT $1, schedule, timezone FROM cron_trigger_details WHERE trigger_id = $2",
+            )
+            .bind(new_id.0)
+            .bind(template_id)
+            .execute(&mut **tx)
+            .await?;
+        }
+        "queue" => {
+            sqlx::query(
+                "INSERT INTO queue_trigger_details \
+                    (trigger_id, queue_name, visibility_timeout_secs) \
+                 SELECT $1, queue_name, visibility_timeout_secs \
+                 FROM queue_trigger_details WHERE trigger_id = $2",
+            )
+            .bind(new_id.0)
+            .bind(template_id)
+            .execute(&mut **tx)
+            .await?;
+        }
+        _ => {}
+    }
+    Ok(None)
+}

crates/manager-core/tests/stateful_templates.rs

@@ -0,0 +1,198 @@
+//! §4.5 M5 integration test: materialization of STATEFUL group trigger
+//! templates. A group cron template materializes one app-owned copy per
+//! descendant app; a new app materializes on reconcile; an app leaving the
+//! subtree de-materializes; a group queue template skips (with a warning) an app
+//! that already fills that queue's consumer slot.
+//!
+//! Drives `rematerialize_stateful_templates` directly. Skips when `DATABASE_URL`
+//! is unset.
+
+#![allow(clippy::too_many_lines, clippy::many_single_char_names)]
+
+use picloud_manager_core::materialize::rematerialize_stateful_templates;
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("stateful_templates: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&url)
+        .await
+        .expect("connect");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("migrate");
+    Some(pool)
+}
+
+/// Count materialized copies of `template_id` owned by `app_id`.
+async fn copies(pool: &PgPool, app_id: Uuid, template_id: Uuid) -> i64 {
+    let (n,): (i64,) = sqlx::query_as(
+        "SELECT COUNT(*) FROM triggers WHERE app_id = $1 AND materialized_from = $2",
+    )
+    .bind(app_id)
+    .bind(template_id)
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    n
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn cron_template_materializes_per_descendant_and_reconciles() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let sfx = Uuid::new_v4().simple().to_string();
+    let admin = {
+        let r: (Uuid,) = sqlx::query_as(
+            "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
+        )
+        .bind(format!("mat-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        r.0
+    };
+    // Group G with a handler + a cron TEMPLATE.
+    let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+        .bind(format!("mat-g-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    let handler: (Uuid,) = sqlx::query_as(
+        "INSERT INTO scripts (name, source, group_id) VALUES ($1, 'x', $2) RETURNING id",
+    )
+    .bind(format!("nightly-{sfx}"))
+    .bind(g.0)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    let tmpl: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers \
+            (group_id, script_id, kind, enabled, dispatch_mode, retry_max_attempts, \
+             retry_backoff, retry_base_ms, registered_by_principal, name) \
+         VALUES ($1, $2, 'cron', TRUE, 'async', 3, 'exponential', 1000, $3, $4) RETURNING id",
+    )
+    .bind(g.0)
+    .bind(handler.0)
+    .bind(admin)
+    .bind(format!("t-{sfx}"))
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    sqlx::query(
+        "INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
+         VALUES ($1, '0 0 * * * *', 'UTC')",
+    )
+    .bind(tmpl.0)
+    .execute(&pool)
+    .await
+    .unwrap();
+
+    // Two apps under G.
+    let mk_app = |slug: String| {
+        let pool = pool.clone();
+        let g = g.0;
+        async move {
+            let r: (Uuid,) = sqlx::query_as(
+                "INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id",
+            )
+            .bind(slug)
+            .bind(g)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+            r.0
+        }
+    };
+    let a = mk_app(format!("mat-a-{sfx}")).await;
+    let b = mk_app(format!("mat-b-{sfx}")).await;
+
+    // Reconcile → one copy per app.
+    let w = rematerialize_stateful_templates(&pool).await.unwrap();
+    assert!(w.is_empty(), "no warnings expected: {w:?}");
+    assert_eq!(copies(&pool, a, tmpl.0).await, 1, "app A materialized");
+    assert_eq!(copies(&pool, b, tmpl.0).await, 1, "app B materialized");
+
+    // Idempotent: a second reconcile does not duplicate.
+    rematerialize_stateful_templates(&pool).await.unwrap();
+    assert_eq!(copies(&pool, a, tmpl.0).await, 1, "no duplicate copy");
+
+    // The copy is app-owned + linked back to the template + carries the detail.
+    let (kind, sched): (String, String) = sqlx::query_as(
+        "SELECT t.kind, d.schedule FROM triggers t \
+         JOIN cron_trigger_details d ON d.trigger_id = t.id \
+         WHERE t.app_id = $1 AND t.materialized_from = $2",
+    )
+    .bind(a)
+    .bind(tmpl.0)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(kind, "cron");
+    assert_eq!(sched, "0 0 * * * *");
+
+    // A new app under G materializes on the next reconcile.
+    let c = mk_app(format!("mat-c-{sfx}")).await;
+    rematerialize_stateful_templates(&pool).await.unwrap();
+    assert_eq!(copies(&pool, c, tmpl.0).await, 1, "new app materialized");
+
+    // Reparent app A into a SIBLING group (not under G) → its copy
+    // de-materializes.
+    let g2: (Uuid,) =
+        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(format!("mat-g2-{sfx}"))
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    sqlx::query("UPDATE apps SET group_id = $2 WHERE id = $1")
+        .bind(a)
+        .bind(g2.0)
+        .execute(&pool)
+        .await
+        .unwrap();
+    rematerialize_stateful_templates(&pool).await.unwrap();
+    assert_eq!(
+        copies(&pool, a, tmpl.0).await,
+        0,
+        "an app that left the subtree de-materializes"
+    );
+    assert_eq!(
+        copies(&pool, b, tmpl.0).await,
+        1,
+        "a still-descendant app keeps its copy"
+    );
+
+    // Cleanup (materialized copies CASCADE via app delete / template delete).
+    let _ = sqlx::query("DELETE FROM triggers WHERE materialized_from = $1")
+        .bind(tmpl.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM triggers WHERE id = $1")
+        .bind(tmpl.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM apps WHERE id = ANY($1)")
+        .bind(vec![a, b, c])
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM scripts WHERE id = $1")
+        .bind(handler.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM groups WHERE id = ANY($1)")
+        .bind(vec![g.0, g2.0])
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM admin_users WHERE id = $1")
+        .bind(admin)
+        .execute(&pool)
+        .await;
+}