test(stateful-templates): group email template materialization (M5.5)

Drive rematerialize directly with a group email template carrying a fake
sealed secret: a descendant app gets one copy whose inbound-secret
ciphertext + nonce byte-equal the template's (verbatim copy, no reseal),
the reconcile is idempotent, and reparenting the app out of the subtree
de-materializes the copy.

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

crates/manager-core/tests/stateful_templates.rs

@@ -2,7 +2,9 @@
 //! 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.
+//! that already fills that queue's consumer slot. A group EMAIL template
+//! materializes a per-app copy whose sealed inbound secret byte-equals the
+//! template's (shared-group-secret model — a verbatim copy, no reseal).
 //!
 //! Drives `rematerialize_stateful_templates` directly. Skips when `DATABASE_URL`
 //! is unset.
@@ -196,3 +198,141 @@ async fn cron_template_materializes_per_descendant_and_reconciles() {
         .execute(&pool)
         .await;
 }
+
+/// §M5.5: a group EMAIL template materializes a per-descendant-app copy whose
+/// sealed inbound secret is a verbatim byte-copy of the template's (no reseal),
+/// and de-materializes when the app leaves the subtree.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn email_template_copies_sealed_secret_per_descendant() {
+    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-em-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        r.0
+    };
+    // Group G with a handler + an email TEMPLATE carrying a (fake) sealed secret.
+    // The materializer copies the ciphertext bytes verbatim, so real crypto is
+    // not needed to exercise the copy path.
+    let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+        .bind(format!("mat-em-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!("inbound-{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, 'email', TRUE, 'async', 3, 'exponential', 1000, $3, $4) RETURNING id",
+    )
+    .bind(g.0)
+    .bind(handler.0)
+    .bind(admin)
+    .bind(format!("t-em-{sfx}"))
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    let ct: Vec<u8> = vec![1, 2, 3, 4, 5, 6, 7, 8];
+    let nonce: Vec<u8> = vec![9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9];
+    sqlx::query(
+        "INSERT INTO email_trigger_details \
+            (trigger_id, inbound_secret_encrypted, inbound_secret_nonce) VALUES ($1, $2, $3)",
+    )
+    .bind(tmpl.0)
+    .bind(&ct)
+    .bind(&nonce)
+    .execute(&pool)
+    .await
+    .unwrap();
+
+    // An app under G.
+    let a: (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("mat-em-a-{sfx}"))
+            .bind(g.0)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+
+    // Reconcile → one email copy, secret bytes verbatim from the template.
+    let w = rematerialize_stateful_templates(&pool).await.unwrap();
+    assert!(w.is_empty(), "no warnings expected: {w:?}");
+    assert_eq!(copies(&pool, a.0, tmpl.0).await, 1, "app materialized");
+    let (copy_ct, copy_nonce): (Vec<u8>, Vec<u8>) = sqlx::query_as(
+        "SELECT d.inbound_secret_encrypted, d.inbound_secret_nonce \
+         FROM triggers t JOIN email_trigger_details d ON d.trigger_id = t.id \
+         WHERE t.app_id = $1 AND t.materialized_from = $2",
+    )
+    .bind(a.0)
+    .bind(tmpl.0)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(copy_ct, ct, "inbound secret ciphertext copied verbatim");
+    assert_eq!(copy_nonce, nonce, "inbound secret nonce copied verbatim");
+
+    // Idempotent.
+    rematerialize_stateful_templates(&pool).await.unwrap();
+    assert_eq!(copies(&pool, a.0, tmpl.0).await, 1, "no duplicate copy");
+
+    // Reparent the app out of the subtree → the copy de-materializes.
+    let g2: (Uuid,) =
+        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(format!("mat-em-g2-{sfx}"))
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    sqlx::query("UPDATE apps SET group_id = $2 WHERE id = $1")
+        .bind(a.0)
+        .bind(g2.0)
+        .execute(&pool)
+        .await
+        .unwrap();
+    rematerialize_stateful_templates(&pool).await.unwrap();
+    assert_eq!(
+        copies(&pool, a.0, tmpl.0).await,
+        0,
+        "an app that left the subtree de-materializes its email copy"
+    );
+
+    // Cleanup.
+    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 = $1")
+        .bind(a.0)
+        .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;
+}