BUZZ-301: Add dummy user deletion workflow

Signed-off-by: Ravneet Arora <rarora@squareup.com>

crates/buzz-admin/src/main.rs

@@ -22,6 +22,7 @@
 mod communities;
 mod deletions;
 mod nip_fi;
+mod user_deletions;
 
 use std::future::Future;
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -126,6 +127,11 @@ enum Command {
         #[command(subcommand)]
         command: deletions::DeletionsCommand,
     },
+    /// Synthetic multi-community user deletion demonstrations; no real erasure.
+    UserDeletions {
+        #[command(subcommand)]
+        command: user_deletions::UserDeletionsCommand,
+    },
     /// Emit missing kind:39000/39001/39002 channel discovery events, or
     /// republish only a targeted channel's kind:39002 roster.
     ///
@@ -203,6 +209,7 @@ async fn run(cli: Cli) -> Result<i32> {
         } => cmd_list_product_feedback(limit).await,
         Command::Communities { command } => communities::run(command).await,
         Command::Deletions { command } => deletions::run(command).await,
+        Command::UserDeletions { command } => user_deletions::run(command).await,
         Command::NipFi { command } => nip_fi::run(command).await,
         Command::ReconcileChannels { channel, relay_key } => {
             reconcile_channels(channel, relay_key).await?;

crates/buzz-admin/src/user_deletions.rs

@@ -0,0 +1,33 @@
+//! Thin `buzz-admin user-deletions` adapter.
+
+pub use buzz_deletion::user::Command as UserDeletionsCommand;
+
+/// Delegate to the shared deletion engine's PostgreSQL-only demonstration path.
+pub async fn run(command: UserDeletionsCommand) -> anyhow::Result<i32> {
+    buzz_deletion::user::run(command).await
+}
+
+#[cfg(test)]
+mod tests {
+    use clap::Parser;
+
+    #[test]
+    fn user_commands_are_registered_without_a_continuous_worker() {
+        for args in [
+            vec![
+                "buzz-admin",
+                "user-deletions",
+                "submit",
+                "--target",
+                "demo:alice",
+                "--requested-by",
+                "operator-a",
+            ],
+            vec!["buzz-admin", "user-deletions", "list"],
+            vec!["buzz-admin", "user-deletions", "drain"],
+        ] {
+            assert!(crate::Cli::try_parse_from(args).is_ok());
+        }
+        assert!(crate::Cli::try_parse_from(["buzz-admin", "user-deletions", "worker"]).is_err());
+    }
+}

crates/buzz-db/src/lib.rs

@@ -66,10 +66,10 @@ pub(crate) use runtime::{
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, artifact, channel,
-    channel_members, community, deletion, dm, event, feed, git_repo, moderation, nip_fi_binding,
-    operator_listener, partition, personal_read, product_feedback, push, reaction, read_state,
-    relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, replaceable,
-    storage_accounting, thread, thread_window, usage, user, workflow,
+    channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation,
+    nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction,
+    read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
+    replaceable, storage_accounting, thread, thread_window, usage, user, workflow,
 };
 
 pub use allowlist::AllowlistEntry;

crates/buzz-db/src/runtime/migration.rs

@@ -489,6 +489,10 @@ mod postgres_tests {
             "storage_taxonomy_sweeps",
             "community_serving_write_leases",
             "community_deletion_executor_heartbeats",
+            "entity_deletion_requests",
+            "entity_deletion_stage_history",
+            "entity_deletion_approvals",
+            "entity_deletion_executor_heartbeats",
             "relay_operators",
             "relay_admin_actions",
             "relay_admin_outbox",
@@ -709,7 +713,12 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 66);
+        assert_eq!(migrations.len(), 67);
+        assert_eq!(migrations[66].version, 67);
+        assert!(migrations[66]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE entity_deletion_stage_history"));
         assert_eq!(migrations[64].version, 65);
         let retire_ttl = strip_sql_comments(migrations[64].sql.as_str());
         for statement in [

crates/buzz-db/src/store/deletion.rs

@@ -702,11 +702,11 @@ pub struct LeaseToken {
 
 /// A claimed request with its durable token.
 #[derive(Debug, Clone)]
-pub struct ClaimedDeletion {
+pub struct ClaimedDeletion<R = DeletionRequest, L = LeaseToken> {
     /// Request snapshot.
-    pub request: DeletionRequest,
+    pub request: R,
     /// Required token.
-    pub lease: LeaseToken,
+    pub lease: L,
 }
 
 /// Short-lived durable lease for an external serving side effect.

crates/buzz-db/src/store/entity_deletion/lifecycle.rs

@@ -0,0 +1,218 @@
+use std::time::Duration;
+
+use sqlx::{Postgres, Transaction};
+use uuid::Uuid;
+
+use crate::Result;
+
+use super::{
+    actor, lock_owned, lock_request, safety, EntityClaim, EntityDeletionStore, EntityLeaseToken,
+    EntityRequest,
+};
+
+fn seconds(duration: Duration) -> Result<i64> {
+    i64::try_from(duration.as_secs())
+        .ok()
+        .filter(|v| *v > 0)
+        .ok_or_else(|| safety("lease/retry duration must be a positive whole number of seconds"))
+}
+
+pub(super) async fn heartbeat_row(
+    tx: &mut Transaction<'_, Postgres>,
+    token: &EntityLeaseToken,
+    mode: &str,
+) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO entity_deletion_executor_heartbeats (executor_id, mode, request_id)
+         VALUES ($1, $2, $3)
+         ON CONFLICT (executor_id) DO UPDATE
+         SET mode = EXCLUDED.mode, request_id = EXCLUDED.request_id,
+             heartbeat_at = now(), draining = false, stopped_at = NULL",
+    )
+    .bind(&token.owner)
+    .bind(mode)
+    .bind(token.request_id)
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
+impl EntityDeletionStore {
+    /// Claim one approved due request; expired owners are replaced atomically.
+    pub async fn claim(
+        &self,
+        id: Option<Uuid>,
+        owner: &str,
+        duration: Duration,
+    ) -> Result<Option<EntityClaim>> {
+        actor(owner)?;
+        let duration = seconds(duration)?;
+        let request: Option<EntityRequest> = sqlx::query_as(
+            "WITH candidate AS (
+                 SELECT r.id FROM entity_deletion_requests r
+                 WHERE ($1::uuid IS NULL OR r.id = $1)
+                   AND r.status IN ('approved', 'running') AND r.blocked_at IS NULL
+                   AND r.next_attempt_at <= clock_timestamp()
+                   AND (r.lease_until IS NULL OR r.lease_until <= clock_timestamp())
+                   AND EXISTS (SELECT 1 FROM entity_deletion_approvals a
+                     WHERE a.request_id = r.id AND a.inventory_digest = r.inventory_digest)
+                 ORDER BY r.created_at, r.id FOR UPDATE SKIP LOCKED LIMIT 1
+             )
+             UPDATE entity_deletion_requests r
+             SET status = 'running', lease_owner = $2, lease_generation = lease_generation + 1,
+                 lease_until = clock_timestamp() + make_interval(secs => $3),
+                 attempts = attempts + 1, updated_at = now()
+             FROM candidate WHERE r.id = candidate.id RETURNING r.*",
+        )
+        .bind(id)
+        .bind(owner)
+        .bind(duration)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(request.map(|request| {
+            let lease = EntityLeaseToken {
+                request_id: request.id,
+                owner: owner.into(),
+                generation: request.lease_generation,
+            };
+            EntityClaim { request, lease }
+        }))
+    }
+
+    /// Renew parent ownership and save executor liveness in the same transaction.
+    pub async fn heartbeat(
+        &self,
+        token: &EntityLeaseToken,
+        mode: &str,
+        duration: Duration,
+    ) -> Result<()> {
+        let duration = seconds(duration)?;
+        let mut tx = self.pool.begin().await?;
+        lock_owned(&mut tx, token).await?;
+        sqlx::query(
+            "UPDATE entity_deletion_requests
+             SET lease_until = clock_timestamp() + make_interval(secs => $2), updated_at = now()
+             WHERE id = $1",
+        )
+        .bind(token.request_id)
+        .bind(duration)
+        .execute(&mut *tx)
+        .await?;
+        heartbeat_row(&mut tx, token, mode).await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    /// Release only this generation and record executor shutdown.
+    pub async fn stop_executor(&self, token: Option<&EntityLeaseToken>, owner: &str) -> Result<()> {
+        let mut tx = self.pool.begin().await?;
+        if let Some(token) = token {
+            sqlx::query(
+                "UPDATE entity_deletion_requests SET lease_owner = NULL, lease_until = NULL,
+                     updated_at = now() WHERE id = $1 AND lease_owner = $2 AND lease_generation = $3",
+            ).bind(token.request_id).bind(&token.owner).bind(token.generation)
+                .execute(&mut *tx).await?;
+        }
+        sqlx::query(
+            "UPDATE entity_deletion_executor_heartbeats SET draining = true, stopped_at = now(),
+                 heartbeat_at = now() WHERE executor_id = $1",
+        )
+        .bind(owner)
+        .execute(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    /// Persist failure at the current stage and release ownership for a bounded retry.
+    pub async fn record_failure(
+        &self,
+        token: &EntityLeaseToken,
+        stage_id: Uuid,
+        error: &str,
+        permanent: bool,
+        delay: Duration,
+    ) -> Result<()> {
+        let delay = seconds(delay)?;
+        let message: String = error.chars().take(4096).collect();
+        let mut tx = self.pool.begin().await?;
+        let request = lock_owned(&mut tx, token).await?;
+        let affected = sqlx::query(
+            "UPDATE entity_deletion_stage_history
+             SET status = 'failed', started_at = COALESCE(started_at, now()),
+                 last_error = $3, updated_at = now()
+             WHERE request_id = $1 AND id = $2 AND is_current",
+        )
+        .bind(token.request_id)
+        .bind(stage_id)
+        .bind(&message)
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+        if affected != 1 {
+            return Err(safety("failed stage is no longer current"));
+        }
+        let blocked = permanent || request.retry_count >= 4;
+        sqlx::query(
+            "UPDATE entity_deletion_requests
+             SET retry_count = retry_count + 1, last_error = $2, last_error_at = now(),
+                 next_attempt_at = clock_timestamp() + make_interval(secs => $3),
+                 blocked_at = CASE WHEN $4 THEN now() ELSE NULL END,
+                 blocked_reason = CASE WHEN $4 THEN $2 ELSE NULL END,
+                 lease_owner = NULL, lease_until = NULL, updated_at = now()
+             WHERE id = $1",
+        )
+        .bind(token.request_id)
+        .bind(&message)
+        .bind(delay)
+        .bind(blocked)
+        .execute(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    /// Resume blocked work without changing completed history or the current step.
+    pub async fn unblock(
+        &self,
+        id: Uuid,
+        unblocked_by: &str,
+        reason: &str,
+    ) -> Result<EntityRequest> {
+        actor(unblocked_by)?;
+        if reason.trim().is_empty() || reason.len() > 4096 {
+            return Err(safety(
+                "unblock requires a nonempty reason of at most 4096 bytes",
+            ));
+        }
+        let mut tx = self.pool.begin().await?;
+        let request = lock_request(&mut tx, id).await?;
+        if request.blocked_at.is_none() || request.lease_owner.is_some() {
+            return Err(safety("only an unowned blocked request can be unblocked"));
+        }
+        let affected = sqlx::query(
+            "UPDATE entity_deletion_stage_history SET metadata_json = metadata_json || $2,
+                 updated_at = now() WHERE request_id = $1 AND is_current",
+        )
+        .bind(id)
+        .bind(serde_json::json!({"last_unblock": {
+            "by": unblocked_by, "reason": reason, "at": chrono::Utc::now()
+        }}))
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+        if affected != 1 {
+            return Err(safety("blocked request has no current stage"));
+        }
+        let request = sqlx::query_as(
+            "UPDATE entity_deletion_requests SET blocked_at = NULL, blocked_reason = NULL,
+                 retry_count = 0, next_attempt_at = now(), last_error = NULL,
+                 last_error_at = NULL, updated_at = now() WHERE id = $1 RETURNING *",
+        )
+        .bind(id)
+        .fetch_one(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(request)
+    }
+}

crates/buzz-db/src/store/entity_deletion/manager.rs

@@ -0,0 +1,159 @@
+use uuid::Uuid;
+
+use crate::Result;
+
+use super::{
+    actor, lock_request, safety, DemoInventory, EntityDeletionStore, EntityRequest, Inspection,
+};
+
+impl EntityDeletionStore {
+    /// Submit a synthetic target, converging on an identical previous submission.
+    pub async fn submit(
+        &self,
+        target: &str,
+        requested_by: &str,
+        reason: Option<&str>,
+    ) -> Result<EntityRequest> {
+        DemoInventory::for_target(target)?;
+        actor(requested_by)?;
+        if reason.is_some_and(|r| r.len() > 4096) {
+            return Err(safety("reason exceeds 4096 bytes"));
+        }
+        let mut tx = self.pool.begin().await?;
+        sqlx::query(
+            "INSERT INTO entity_deletion_requests (target, requested_by, reason)
+             VALUES ($1, $2, $3) ON CONFLICT (entity_type, target) DO NOTHING",
+        )
+        .bind(target)
+        .bind(requested_by)
+        .bind(reason)
+        .execute(&mut *tx)
+        .await?;
+        let request: EntityRequest = sqlx::query_as(
+            "SELECT * FROM entity_deletion_requests WHERE entity_type = 'user' AND target = $1",
+        )
+        .bind(target)
+        .fetch_one(&mut *tx)
+        .await?;
+        if request.requested_by != requested_by || request.reason.as_deref() != reason {
+            return Err(safety(format!(
+                "target already has request {}; inspect it before resubmission",
+                request.id
+            )));
+        }
+        tx.commit().await?;
+        Ok(request)
+    }
+
+    /// Freeze the fixed non-owned fixture and its plan in one transaction.
+    pub async fn prepare_inventory(&self, id: Uuid) -> Result<EntityRequest> {
+        let mut tx = self.pool.begin().await?;
+        let request = lock_request(&mut tx, id).await?;
+        if request.status != "submitted" {
+            request.inventory()?;
+            tx.commit().await?;
+            return Ok(request);
+        }
+        let inventory = DemoInventory::for_target(&request.target)?;
+        let request = sqlx::query_as(
+            "UPDATE entity_deletion_requests
+             SET inventory_manifest = $2, inventory_digest = $3, inventory_frozen_at = now(),
+                 status = 'inventoried', updated_at = now() WHERE id = $1 RETURNING *",
+        )
+        .bind(id)
+        .bind(serde_json::to_value(&inventory)?)
+        .bind(inventory.digest()?)
+        .fetch_one(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(request)
+    }
+
+    /// Approve the frozen inventory and create the first pending stage atomically.
+    pub async fn approve(&self, id: Uuid, approved_by: &str) -> Result<EntityRequest> {
+        actor(approved_by)?;
+        let mut tx = self.pool.begin().await?;
+        let request = lock_request(&mut tx, id).await?;
+        if request.requested_by == approved_by {
+            return Err(safety("approval requires a different operator"));
+        }
+        let inventory = request.inventory()?;
+        let previous: Option<String> = sqlx::query_scalar(
+            "SELECT approved_by FROM entity_deletion_approvals WHERE request_id = $1",
+        )
+        .bind(id)
+        .fetch_optional(&mut *tx)
+        .await?;
+        if let Some(previous) = previous {
+            if previous != approved_by {
+                return Err(safety("approval is already recorded by another operator"));
+            }
+            tx.commit().await?;
+            return Ok(request);
+        }
+        if request.status != "inventoried" {
+            return Err(safety("request must be inventoried before approval"));
+        }
+        sqlx::query(
+            "INSERT INTO entity_deletion_approvals (request_id, inventory_digest, approved_by)
+             VALUES ($1, $2, $3)",
+        )
+        .bind(id)
+        .bind(inventory.digest()?)
+        .bind(approved_by)
+        .execute(&mut *tx)
+        .await?;
+        super::stages::insert_stage(&mut tx, id, &inventory, 1).await?;
+        let request = sqlx::query_as(
+            "UPDATE entity_deletion_requests SET status = 'approved', updated_at = now()
+             WHERE id = $1 RETURNING *",
+        )
+        .bind(id)
+        .fetch_one(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(request)
+    }
+
+    /// Load a request from the authoritative writer database.
+    pub async fn get(&self, id: Uuid) -> Result<EntityRequest> {
+        sqlx::query_as("SELECT * FROM entity_deletion_requests WHERE id = $1")
+            .bind(id)
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or_else(|| crate::DbError::NotFound(format!("entity deletion request {id}")))
+    }
+
+    /// List bounded operator-global requests, newest first.
+    pub async fn list(&self, limit: u16) -> Result<Vec<EntityRequest>> {
+        Ok(sqlx::query_as(
+            "SELECT * FROM entity_deletion_requests ORDER BY created_at DESC, id LIMIT $1",
+        )
+        .bind(i64::from(limit.clamp(1, 1000)))
+        .fetch_all(&self.pool)
+        .await?)
+    }
+
+    /// Inspect a consistent snapshot while serializing with request mutations.
+    pub async fn inspect(&self, id: Uuid) -> Result<Inspection> {
+        let mut tx = self.pool.begin().await?;
+        let request = lock_request(&mut tx, id).await?;
+        let approval =
+            sqlx::query_as("SELECT * FROM entity_deletion_approvals WHERE request_id = $1")
+                .bind(id)
+                .fetch_optional(&mut *tx)
+                .await?;
+        let stage_history = sqlx::query_as(
+            "SELECT * FROM entity_deletion_stage_history WHERE request_id = $1 ORDER BY sequence",
+        )
+        .bind(id)
+        .fetch_all(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(Inspection {
+            request,
+            approval,
+            stage_history,
+        })
+    }
+}

crates/buzz-db/src/store/entity_deletion/mod.rs

@@ -0,0 +1,93 @@
+//! Operator-global workflow state for synthetic, non-owned user deletion demos.
+//! Domain handlers must never infer permission to erase data from these records.
+
+use sqlx::{PgPool, Postgres, Transaction};
+use uuid::Uuid;
+
+use crate::{Db, DbError, Result};
+
+mod lifecycle;
+mod manager;
+mod model;
+mod stages;
+
+pub use model::{
+    Approval, DemoCommunity, DemoInventory, EntityClaim, EntityLeaseToken, EntityRequest,
+    Inspection, PlannedStage, StageRecord,
+};
+
+#[cfg(test)]
+mod postgres_tests;
+
+/// PostgreSQL adapter for operator-driven entity workflow demonstrations.
+#[derive(Clone)]
+pub struct EntityDeletionStore {
+    pool: PgPool,
+}
+
+impl Db {
+    /// Use the writer pool for durable entity deletion workflow state.
+    pub fn entity_deletion_store(&self) -> EntityDeletionStore {
+        EntityDeletionStore {
+            pool: self.pool().clone(),
+        }
+    }
+}
+
+fn safety(message: impl Into<String>) -> DbError {
+    DbError::DeletionSafety(message.into())
+}
+
+fn stale(id: Uuid) -> DbError {
+    DbError::AccessDenied(format!("stale entity deletion lease: {id}"))
+}
+
+/// Whether a mutation was rejected because execution ownership is stale.
+pub fn is_stale_lease(error: &DbError) -> bool {
+    matches!(error, DbError::AccessDenied(message) if message.starts_with("stale entity deletion lease:"))
+}
+
+fn actor(value: &str) -> Result<&str> {
+    if value.is_empty() || value.len() > 256 || value.trim() != value {
+        return Err(safety(
+            "operator/executor identity must be 1..256 bytes without surrounding whitespace",
+        ));
+    }
+    Ok(value)
+}
+
+async fn lock_request(tx: &mut Transaction<'_, Postgres>, id: Uuid) -> Result<EntityRequest> {
+    sqlx::query_as("SELECT * FROM entity_deletion_requests WHERE id = $1 FOR UPDATE")
+        .bind(id)
+        .fetch_optional(&mut **tx)
+        .await?
+        .ok_or_else(|| DbError::NotFound(format!("entity deletion request {id}")))
+}
+
+async fn lock_owned(
+    tx: &mut Transaction<'_, Postgres>,
+    token: &EntityLeaseToken,
+) -> Result<EntityRequest> {
+    // Keep the row lock until the child mutation commits. A concurrent claim
+    // cannot change the generation between ownership verification and progress.
+    let request = lock_request(tx, token.request_id).await?;
+    let valid: bool = sqlx::query_scalar(
+        "SELECT EXISTS (
+            SELECT 1 FROM entity_deletion_requests r
+            JOIN entity_deletion_approvals a
+              ON a.request_id = r.id AND a.inventory_digest = r.inventory_digest
+            WHERE r.id = $1 AND r.lease_owner = $2 AND r.lease_generation = $3
+              AND r.lease_until > clock_timestamp()
+              AND r.status = 'running' AND r.blocked_at IS NULL
+        )",
+    )
+    .bind(token.request_id)
+    .bind(&token.owner)
+    .bind(token.generation)
+    .fetch_one(&mut **tx)
+    .await?;
+    if !valid {
+        return Err(stale(token.request_id));
+    }
+    Ok(request)
+}

crates/buzz-db/src/store/entity_deletion/model.rs

@@ -0,0 +1,249 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+use crate::{deletion::ClaimedDeletion, Result};
+
+use super::safety;
+
+/// A participant community in the fixed synthetic inventory.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DemoCommunity {
+    /// Synthetic provenance; no real community row is required.
+    pub id: Uuid,
+    /// Reserved example hostname.
+    pub host: String,
+    /// Only participant relationships are supported by this demonstration.
+    pub relationship: String,
+}
+
+/// An ordered action in the frozen, approved demonstration plan.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct PlannedStage {
+    /// None identifies a request-wide step.
+    pub target_community_id: Option<Uuid>,
+    /// Explicit action selected by the dummy handler.
+    pub stage: String,
+}
+
+/// Fixed three-community fixture; never a discovery result for a real user.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DemoInventory {
+    /// Fixture format version.
+    pub version: u32,
+    /// Synthetic target, prefixed with `demo:`.
+    pub target: String,
+    /// Three non-owned participant communities.
+    pub communities: Vec<DemoCommunity>,
+    /// Bounded deterministic execution order included in the approval digest.
+    pub stages: Vec<PlannedStage>,
+}
+
+impl DemoInventory {
+    /// Build the only supported fixture, without reading user or membership data.
+    pub fn for_target(target: &str) -> Result<Self> {
+        let slug = target.strip_prefix("demo:").unwrap_or_default();
+        if slug.is_empty()
+            || slug.len() > 64
+            || !slug.as_bytes()[0].is_ascii_alphanumeric()
+            || !slug
+                .bytes()
+                .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-')
+        {
+            return Err(safety(
+                "target must be demo:<1..64 lowercase letters, digits, underscores or hyphens>",
+            ));
+        }
+        let communities: Vec<_> = (1..=3)
+            .map(|n| DemoCommunity {
+                id: Uuid::from_u128(n),
+                host: format!("participant-{n}.invalid"),
+                relationship: "participant".into(),
+            })
+            .collect();
+        let mut stages = vec![PlannedStage {
+            target_community_id: None,
+            stage: "gate_user".into(),
+        }];
+        for community in &communities {
+            stages.push(PlannedStage {
+                target_community_id: Some(community.id),
+                stage: "gate_community".into(),
+            });
+        }
+        for community in &communities {
+            for stage in ["drain", "delete_data", "verify_community"] {
+                stages.push(PlannedStage {
+                    target_community_id: Some(community.id),
+                    stage: stage.into(),
+                });
+            }
+        }
+        for stage in ["verify_user", "post_delete"] {
+            stages.push(PlannedStage {
+                target_community_id: None,
+                stage: stage.into(),
+            });
+        }
+        Ok(Self {
+            version: 1,
+            target: target.into(),
+            communities,
+            stages,
+        })
+    }
+
+    /// Reject edited fixtures, including owned communities or arbitrary actions.
+    pub fn validate(&self) -> Result<()> {
+        if *self != Self::for_target(&self.target)? {
+            return Err(safety(
+                "only the fixed non-owned participant demo inventory is supported",
+            ));
+        }
+        Ok(())
+    }
+
+    /// Digest the validated typed manifest so JSON object ordering is irrelevant.
+    pub fn digest(&self) -> Result<Vec<u8>> {
+        self.validate()?;
+        Ok(Sha256::digest(serde_json::to_vec(self)?).to_vec())
+    }
+}
+
+/// Overall lifecycle and the sole ownership record for a demonstration.
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct EntityRequest {
+    /// Request identifier.
+    pub id: Uuid,
+    /// Currently always user.
+    pub entity_type: String,
+    /// Synthetic user label.
+    pub target: String,
+    /// Overall lifecycle; granular progress is in stage history.
+    pub status: String,
+    /// Submitting operator.
+    pub requested_by: String,
+    /// Operator's optional reason.
+    pub reason: Option<String>,
+    /// Frozen synthetic inventory and stage plan.
+    pub inventory_manifest: Option<Value>,
+    /// Digest covered by approval.
+    pub inventory_digest: Option<Vec<u8>>,
+    /// Inventory freeze time.
+    pub inventory_frozen_at: Option<DateTime<Utc>>,
+    /// Executor holding the parent lease.
+    pub lease_owner: Option<String>,
+    /// Monotonic ownership generation.
+    pub lease_generation: i64,
+    /// Ownership deadline, checked using database time.
+    pub lease_until: Option<DateTime<Utc>>,
+    /// Total claims.
+    pub attempts: i32,
+    /// Consecutive failures of the current stage.
+    pub retry_count: i32,
+    /// Earliest automatic retry.
+    pub next_attempt_at: DateTime<Utc>,
+    /// Latest workflow error.
+    pub last_error: Option<String>,
+    /// Latest error time.
+    pub last_error_at: Option<DateTime<Utc>>,
+    /// Manual intervention marker.
+    pub blocked_at: Option<DateTime<Utc>>,
+    /// Why automatic attempts stopped.
+    pub blocked_reason: Option<String>,
+    /// Submission time.
+    pub created_at: DateTime<Utc>,
+    /// Last lifecycle change.
+    pub updated_at: DateTime<Utc>,
+    /// Demonstration completion time.
+    pub completed_at: Option<DateTime<Utc>>,
+}
+
+impl EntityRequest {
+    /// Decode and validate the approved fixture before dispatching any stage.
+    pub fn inventory(&self) -> Result<DemoInventory> {
+        let inventory: DemoInventory = serde_json::from_value(
+            self.inventory_manifest
+                .clone()
+                .ok_or_else(|| safety("inventory is not prepared"))?,
+        )?;
+        if inventory.target != self.target
+            || self.inventory_digest.as_deref() != Some(inventory.digest()?.as_slice())
+        {
+            return Err(safety("inventory does not match request target/digest"));
+        }
+        Ok(inventory)
+    }
+}
+
+/// Ownership proof carried by user workflow operations; no community-wide fence.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct EntityLeaseToken {
+    /// Parent request.
+    pub request_id: Uuid,
+    /// Executor identity.
+    pub owner: String,
+    /// Claim generation.
+    pub generation: i64,
+}
+
+/// The same claim container used by community deletion, with user record types.
+pub type EntityClaim = ClaimedDeletion<EntityRequest, EntityLeaseToken>;
+
+/// Current or completed granular stage, with one row reused across retries.
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct StageRecord {
+    /// Stage record identifier.
+    pub id: Uuid,
+    /// Parent request.
+    pub request_id: Uuid,
+    /// One-based position in the approved plan.
+    pub sequence: i32,
+    /// Community provenance, or None for request-wide work.
+    pub target_community_id: Option<Uuid>,
+    /// Action name.
+    pub stage: String,
+    /// pending, running, failed, or completed.
+    pub status: String,
+    /// The request's one current unfinished stage.
+    pub is_current: bool,
+    /// First attempt start time.
+    pub started_at: Option<DateTime<Utc>>,
+    /// Successful completion time.
+    pub finished_at: Option<DateTime<Utc>>,
+    /// Stage details and result.
+    pub metadata_json: Value,
+    /// Last stage failure, retained for diagnosis after retry.
+    pub last_error: Option<String>,
+    /// Latest stage record update.
+    pub updated_at: DateTime<Utc>,
+}
+
+/// Independent approval of the exact frozen inventory.
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct Approval {
+    /// Parent request.
+    pub request_id: Uuid,
+    /// Approved manifest digest.
+    pub inventory_digest: Vec<u8>,
+    /// Approving operator.
+    pub approved_by: String,
+    /// Approval time.
+    pub approved_at: DateTime<Utc>,
+}
+
+/// Consistent request, approval and stage-history snapshot for inspection.
+#[derive(Debug, Serialize)]
+pub struct Inspection {
+    /// Overall lifecycle and ownership.
+    pub request: EntityRequest,
+    /// Approval, if recorded.
+    pub approval: Option<Approval>,
+    /// Reached stages in plan order.
+    pub stage_history: Vec<StageRecord>,
+}

crates/buzz-db/src/store/entity_deletion/postgres_tests.rs

@@ -0,0 +1,514 @@
+use std::time::Duration;
+
+use serde_json::json;
+use sqlx::PgPool;
+
+use super::*;
+
+const LEASE: Duration = Duration::from_secs(120);
+
+async fn store() -> EntityDeletionStore {
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .expect("connect isolated DB");
+    EntityDeletionStore { pool }
+}
+
+async fn approved(store: &EntityDeletionStore) -> EntityRequest {
+    let target = format!("demo:{}", Uuid::new_v4().simple());
+    let request = store
+        .submit(&target, "operator-a", None)
+        .await
+        .expect("submit");
+    store
+        .prepare_inventory(request.id)
+        .await
+        .expect("inventory");
+    store
+        .approve(request.id, "operator-b")
+        .await
+        .expect("approve")
+}
+
+async fn claim(store: &EntityDeletionStore, id: Uuid, owner: &str) -> EntityClaim {
+    store
+        .claim(Some(id), owner, LEASE)
+        .await
+        .expect("claim")
+        .expect("eligible")
+}
+
+async fn due(store: &EntityDeletionStore, id: Uuid) {
+    sqlx::query("UPDATE entity_deletion_requests SET next_attempt_at = now() WHERE id = $1")
+        .bind(id)
+        .execute(&store.pool)
+        .await
+        .expect("make retry due");
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn inventory_approval_and_completed_history_are_immutable() {
+    let store = store().await;
+    assert!(store.submit("real-user", "operator-a", None).await.is_err());
+    let request = store
+        .submit("demo:alice", "operator-a", None)
+        .await
+        .expect("submit");
+    assert_eq!(
+        store
+            .submit("demo:alice", "operator-a", None)
+            .await
+            .expect("repeat")
+            .id,
+        request.id
+    );
+    assert!(store
+        .submit("demo:alice", "someone-else", None)
+        .await
+        .is_err());
+    assert!(store
+        .claim(Some(request.id), "worker", LEASE)
+        .await
+        .expect("unapproved")
+        .is_none());
+    assert!(store.approve(request.id, "operator-b").await.is_err());
+    store
+        .prepare_inventory(request.id)
+        .await
+        .expect("inventory");
+    assert!(sqlx::query(
+        "INSERT INTO entity_deletion_approvals (request_id, inventory_digest, approved_by)
+         SELECT id, inventory_digest, requested_by FROM entity_deletion_requests WHERE id = $1",
+    )
+    .bind(request.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_requests SET target = 'demo:retargeted' WHERE id = $1"
+    )
+    .bind(request.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_requests SET requested_by = 'replacement' WHERE id = $1"
+    )
+    .bind(request.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(store.approve(request.id, "operator-a").await.is_err());
+    store
+        .approve(request.id, "operator-b")
+        .await
+        .expect("approve");
+    store
+        .approve(request.id, "operator-b")
+        .await
+        .expect("idempotent approve");
+    assert_eq!(
+        store
+            .inspect(request.id)
+            .await
+            .expect("inspect")
+            .stage_history
+            .len(),
+        1
+    );
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_requests SET inventory_manifest = '{}' WHERE id = $1"
+    )
+    .bind(request.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(
+        sqlx::query("DELETE FROM entity_deletion_approvals WHERE request_id = $1")
+            .bind(request.id)
+            .execute(&store.pool)
+            .await
+            .is_err()
+    );
+    assert!(sqlx::query("INSERT INTO entity_deletion_stage_history (request_id, sequence, stage) VALUES ($1, 2, 'drain')")
+        .bind(request.id).execute(&store.pool).await.is_err());
+    let claim = claim(&store, request.id, "worker").await;
+    let stage = store.start_stage(&claim.lease).await.expect("start");
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_stage_history SET stage = 'delete_data' WHERE id = $1"
+    )
+    .bind(stage.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(sqlx::query("UPDATE entity_deletion_stage_history SET started_at = now() + interval '1 second' WHERE id = $1")
+        .bind(stage.id).execute(&store.pool).await.is_err());
+    store
+        .complete_stage(&claim.lease, stage.id, &json!({"simulated": true}))
+        .await
+        .expect("complete");
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_stage_history SET metadata_json = '{}' WHERE id = $1"
+    )
+    .bind(stage.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(
+        sqlx::query("DELETE FROM entity_deletion_stage_history WHERE id = $1")
+            .bind(stage.id)
+            .execute(&store.pool)
+            .await
+            .is_err()
+    );
+    let mut inventory = claim.request.inventory().expect("fixture");
+    inventory.communities[0].relationship = "owner".into();
+    assert!(
+        inventory.validate().is_err(),
+        "owned-community fixtures are unsupported"
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn claim_generation_guards_children_heartbeat_and_release() {
+    let store = store().await;
+    let request = approved(&store).await;
+    let (a, b) = tokio::join!(
+        store.claim(Some(request.id), "a", LEASE),
+        store.claim(Some(request.id), "b", LEASE),
+    );
+    let a = a.expect("claim a");
+    let b = b.expect("claim b");
+    assert_ne!(a.is_some(), b.is_some());
+    let first = a.or(b).expect("one claim");
+    let stage = store.start_stage(&first.lease).await.expect("start");
+    store
+        .heartbeat(&first.lease, "run", LEASE)
+        .await
+        .expect("heartbeat");
+    let heartbeat: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM entity_deletion_executor_heartbeats WHERE executor_id = $1",
+    )
+    .bind(&first.lease.owner)
+    .fetch_one(&store.pool)
+    .await
+    .expect("heartbeat row");
+    assert_eq!(heartbeat, 1);
+    assert_eq!(
+        store
+            .get(request.id)
+            .await
+            .expect("request")
+            .lease_generation,
+        first.lease.generation
+    );
+    sqlx::query("UPDATE entity_deletion_requests SET lease_until = now() - interval '1 second' WHERE id = $1")
+        .bind(request.id).execute(&store.pool).await.expect("expire");
+    assert!(is_stale_lease(
+        &store
+            .heartbeat(&first.lease, "run", LEASE)
+            .await
+            .expect_err("expired")
+    ));
+    let second = claim(&store, request.id, "successor").await;
+    assert_eq!(second.lease.generation, first.lease.generation + 1);
+    assert!(is_stale_lease(
+        &store
+            .complete_stage(&first.lease, stage.id, &json!({}))
+            .await
+            .expect_err("stale child write")
+    ));
+    assert!(is_stale_lease(
+        &store
+            .start_stage(&first.lease)
+            .await
+            .expect_err("stale start")
+    ));
+    store
+        .stop_executor(Some(&first.lease), &first.lease.owner)
+        .await
+        .expect("stale release");
+    assert_eq!(
+        store
+            .get(request.id)
+            .await
+            .expect("request")
+            .lease_owner
+            .as_deref(),
+        Some("successor")
+    );
+    let resumed = store.start_stage(&second.lease).await.expect("resume");
+    assert_eq!(resumed.id, stage.id);
+    assert_eq!(resumed.started_at, stage.started_at);
+    store
+        .complete_stage(&second.lease, stage.id, &json!({}))
+        .await
+        .expect("successor advances");
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn stage_and_successor_creation_roll_back_together() {
+    let store = store().await;
+    let request = approved(&store).await;
+    let claim = claim(&store, request.id, "worker").await;
+    let stage = store.start_stage(&claim.lease).await.expect("start");
+    sqlx::raw_sql(
+        "CREATE FUNCTION reject_demo_successor() RETURNS trigger LANGUAGE plpgsql AS $$
+         BEGIN RAISE EXCEPTION 'injected successor insert failure'; END $$;
+         CREATE TRIGGER reject_demo_successor BEFORE INSERT ON entity_deletion_stage_history
+         FOR EACH ROW EXECUTE FUNCTION reject_demo_successor();",
+    )
+    .execute(&store.pool)
+    .await
+    .expect("inject next-row failure");
+    assert!(store
+        .complete_stage(&claim.lease, stage.id, &json!({"result": "done"}))
+        .await
+        .is_err());
+    let inspection = store.inspect(request.id).await.expect("inspect rollback");
+    assert_eq!(inspection.stage_history.len(), 1);
+    let current = &inspection.stage_history[0];
+    assert_eq!(current.status, "running");
+    assert!(current.is_current);
+    assert!(current.finished_at.is_none());
+    assert_eq!(current.metadata_json, json!({}));
+    sqlx::raw_sql("DROP TRIGGER reject_demo_successor ON entity_deletion_stage_history; DROP FUNCTION reject_demo_successor();")
+        .execute(&store.pool).await.expect("remove fault");
+    store
+        .complete_stage(&claim.lease, stage.id, &json!({"result": "done"}))
+        .await
+        .expect("advance");
+    let inspection = store.inspect(request.id).await.expect("inspect");
+    assert_eq!(inspection.stage_history.len(), 2);
+    assert_eq!(
+        inspection
+            .stage_history
+            .iter()
+            .filter(|s| s.is_current)
+            .count(),
+        1
+    );
+    assert_eq!(inspection.stage_history[1].sequence, 2);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn community_b_failure_resumes_without_repeating_a_and_retries_are_bounded() {
+    let store = store().await;
+    let request = approved(&store).await;
+    let first = claim(&store, request.id, "worker-1").await;
+    for sequence in 1..9 {
+        let stage = store.start_stage(&first.lease).await.expect("start");
+        assert_eq!(stage.sequence, sequence);
+        store
+            .complete_stage(&first.lease, stage.id, &json!({"sequence": sequence}))
+            .await
+            .expect("advance");
+    }
+    let failed = store.start_stage(&first.lease).await.expect("B delete");
+    assert_eq!(failed.target_community_id, Some(Uuid::from_u128(2)));
+    assert_eq!(failed.stage, "delete_data");
+    store
+        .record_failure(
+            &first.lease,
+            failed.id,
+            "temporary",
+            false,
+            Duration::from_secs(30),
+        )
+        .await
+        .expect("fail");
+    assert!(store
+        .claim(Some(request.id), "too-early", LEASE)
+        .await
+        .expect("claim delayed")
+        .is_none());
+    let snapshot = store.inspect(request.id).await.expect("inspect failed");
+    assert_eq!(
+        snapshot
+            .stage_history
+            .iter()
+            .filter(|s| s.status == "completed")
+            .count(),
+        8
+    );
+    assert_eq!(
+        snapshot.stage_history.last().expect("B stage").status,
+        "failed"
+    );
+    assert!(snapshot.request.lease_owner.is_none());
+    for attempt in 2..=5 {
+        due(&store, request.id).await;
+        let retry = claim(&store, request.id, &format!("worker-{attempt}")).await;
+        let stage = store
+            .start_stage(&retry.lease)
+            .await
+            .expect("retry same step");
+        assert_eq!(stage.id, failed.id);
+        assert_eq!(stage.started_at, failed.started_at);
+        store
+            .record_failure(
+                &retry.lease,
+                stage.id,
+                "temporary",
+                false,
+                Duration::from_secs(30),
+            )
+            .await
+            .expect("fail retry");
+    }
+    assert!(store
+        .get(request.id)
+        .await
+        .expect("blocked")
+        .blocked_at
+        .is_some());
+    due(&store, request.id).await;
+    assert!(store
+        .claim(Some(request.id), "blocked", LEASE)
+        .await
+        .expect("blocked claim")
+        .is_none());
+    store
+        .unblock(request.id, "operator-c", "dependency repaired")
+        .await
+        .expect("unblock");
+    let retry = claim(&store, request.id, "recovered").await;
+    for sequence in 9..=15 {
+        let stage = store.start_stage(&retry.lease).await.expect("resume");
+        assert_eq!(stage.sequence, sequence);
+        store
+            .complete_stage(&retry.lease, stage.id, &json!({"simulated": true}))
+            .await
+            .expect("complete");
+    }
+    let finished = store.inspect(request.id).await.expect("finished");
+    assert_eq!(finished.request.status, "demo_completed");
+    assert_eq!(finished.request.retry_count, 0);
+    assert!(finished.request.lease_owner.is_none());
+    assert!(finished
+        .stage_history
+        .iter()
+        .all(|s| !s.is_current && s.status == "completed"));
+    assert_eq!(finished.stage_history.len(), 15);
+    for (before, after) in snapshot.stage_history[..8]
+        .iter()
+        .zip(&finished.stage_history[..8])
+    {
+        assert_eq!(
+            serde_json::to_value(before).expect("before"),
+            serde_json::to_value(after).expect("after")
+        );
+    }
+    assert_eq!(
+        finished.stage_history[8].last_error.as_deref(),
+        Some("temporary")
+    );
+    assert_eq!(
+        finished.stage_history[8].metadata_json["last_unblock"]["by"],
+        "operator-c"
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn desired_schema_classifies_workflow_tables_without_community_fences() {
+    let store = store().await;
+    let names = [
+        "entity_deletion_requests",
+        "entity_deletion_stage_history",
+        "entity_deletion_approvals",
+        "entity_deletion_executor_heartbeats",
+    ];
+    let classified: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM _operator_global_tables WHERE table_name = ANY($1)",
+    )
+    .bind(names.as_slice())
+    .fetch_one(&store.pool)
+    .await
+    .expect("classifications");
+    assert_eq!(classified, 4);
+    let fences: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM pg_trigger t JOIN pg_class c ON c.oid = t.tgrelid
+         WHERE c.relname = ANY($1) AND t.tgname LIKE 'community_write_fence_%'",
+    )
+    .bind(names.as_slice())
+    .fetch_one(&store.pool)
+    .await
+    .expect("fences");
+    assert_eq!(fences, 0);
+    let checks: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM pg_constraint WHERE conname IN
+         ('entity_deletion_request_shape', 'entity_deletion_stage_shape') AND convalidated",
+    )
+    .fetch_one(&store.pool)
+    .await
+    .expect("shape checks");
+    assert_eq!(checks, 2);
+    let request = approved(&store).await;
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_requests SET lease_owner = 'partial-lease' WHERE id = $1"
+    )
+    .bind(request.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    assert!(sqlx::query(
+        "UPDATE entity_deletion_stage_history SET status = 'completed' WHERE request_id = $1"
+    )
+    .bind(request.id)
+    .execute(&store.pool)
+    .await
+    .is_err());
+    // Adding operator state must not break the existing destructive catalog.
+    crate::Db::from_pool(store.pool.clone())
+        .validate_deletion_catalog()
+        .await
+        .expect("community catalog");
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn migration_schema_entity_workflow_upgrade_preserves_existing_data() {
+    let store = store().await;
+    crate::migration::run_migrations_through(&store.pool, 66)
+        .await
+        .expect("previous schema");
+    let id = Uuid::new_v4();
+    sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'upgrade-fixture.invalid')")
+        .bind(id)
+        .execute(&store.pool)
+        .await
+        .expect("existing community");
+    crate::migration::run_migrations(&store.pool)
+        .await
+        .expect("upgrade");
+    let request = approved(&store).await;
+    crate::migration::run_migrations(&store.pool)
+        .await
+        .expect("migration rerun");
+    assert_eq!(
+        store
+            .inspect(request.id)
+            .await
+            .expect("preserved request")
+            .stage_history
+            .len(),
+        1
+    );
+    let host: String = sqlx::query_scalar("SELECT host FROM communities WHERE id = $1")
+        .bind(id)
+        .fetch_one(&store.pool)
+        .await
+        .expect("preserved community");
+    assert_eq!(host, "upgrade-fixture.invalid");
+    let claimed = claim(&store, request.id, "migration-worker").await;
+    let stage = store.start_stage(&claimed.lease).await.expect("start");
+    store
+        .complete_stage(&claimed.lease, stage.id, &json!({}))
+        .await
+        .expect("advance");
+}

crates/buzz-db/src/store/entity_deletion/stages.rs

@@ -0,0 +1,133 @@
+use serde_json::Value;
+use sqlx::{Postgres, Transaction};
+use uuid::Uuid;
+
+use crate::Result;
+
+use super::{
+    lock_owned, safety, DemoInventory, EntityDeletionStore, EntityLeaseToken, EntityRequest,
+    StageRecord,
+};
+
+pub(super) async fn insert_stage(
+    tx: &mut Transaction<'_, Postgres>,
+    id: Uuid,
+    inventory: &DemoInventory,
+    sequence: i32,
+) -> Result<()> {
+    let index = usize::try_from(sequence - 1).map_err(|_| safety("invalid stage sequence"))?;
+    let stage = inventory
+        .stages
+        .get(index)
+        .ok_or_else(|| safety("stage is outside the approved plan"))?;
+    sqlx::query(
+        "INSERT INTO entity_deletion_stage_history (request_id, sequence, target_community_id, stage)
+         VALUES ($1, $2, $3, $4)",
+    ).bind(id).bind(sequence).bind(stage.target_community_id).bind(&stage.stage)
+        .execute(&mut **tx).await?;
+    Ok(())
+}
+
+async fn current(
+    tx: &mut Transaction<'_, Postgres>,
+    request: &EntityRequest,
+) -> Result<StageRecord> {
+    let stage: StageRecord = sqlx::query_as(
+        "SELECT * FROM entity_deletion_stage_history WHERE request_id = $1 AND is_current FOR UPDATE",
+    ).bind(request.id).fetch_optional(&mut **tx).await?
+        .ok_or_else(|| safety("runnable request has no current stage"))?;
+    let inventory = request.inventory()?;
+    let index =
+        usize::try_from(stage.sequence - 1).map_err(|_| safety("invalid stage sequence"))?;
+    let planned = inventory
+        .stages
+        .get(index)
+        .ok_or_else(|| safety("stage is outside the approved plan"))?;
+    if stage.stage != planned.stage || stage.target_community_id != planned.target_community_id {
+        return Err(safety("current stage differs from the approved plan"));
+    }
+    Ok(stage)
+}
+
+impl EntityDeletionStore {
+    /// Read the current stage for dispatch; starting it still validates ownership.
+    pub async fn current_stage(&self, id: Uuid) -> Result<StageRecord> {
+        sqlx::query_as(
+            "SELECT * FROM entity_deletion_stage_history WHERE request_id = $1 AND is_current",
+        )
+        .bind(id)
+        .fetch_optional(&self.pool)
+        .await?
+        .ok_or_else(|| safety("runnable request has no current stage"))
+    }
+
+    /// Start or resume the current stage under the parent ownership lock.
+    pub async fn start_stage(&self, token: &EntityLeaseToken) -> Result<StageRecord> {
+        let mut tx = self.pool.begin().await?;
+        let request = lock_owned(&mut tx, token).await?;
+        let stage = current(&mut tx, &request).await?;
+        let stage = sqlx::query_as(
+            "UPDATE entity_deletion_stage_history
+             SET status = 'running', started_at = COALESCE(started_at, clock_timestamp()), updated_at = now()
+             WHERE id = $1 RETURNING *",
+        ).bind(stage.id).fetch_one(&mut *tx).await?;
+        tx.commit().await?;
+        Ok(stage)
+    }
+
+    /// Commit a stage result and its successor (or final completion) atomically.
+    pub async fn complete_stage(
+        &self,
+        token: &EntityLeaseToken,
+        stage_id: Uuid,
+        metadata: &Value,
+    ) -> Result<EntityRequest> {
+        if !metadata.is_object() || serde_json::to_vec(metadata)?.len() > 16_384 {
+            return Err(safety(
+                "stage metadata must be an object of at most 16384 bytes",
+            ));
+        }
+        let mut tx = self.pool.begin().await?;
+        let request = lock_owned(&mut tx, token).await?;
+        let stage = current(&mut tx, &request).await?;
+        if stage.id != stage_id || stage.status != "running" {
+            return Err(safety("only the current running stage can complete"));
+        }
+        let inventory = request.inventory()?;
+        let completed: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM entity_deletion_stage_history
+             WHERE request_id = $1 AND sequence < $2 AND status = 'completed'",
+        )
+        .bind(request.id)
+        .bind(stage.sequence)
+        .fetch_one(&mut *tx)
+        .await?;
+        if completed != i64::from(stage.sequence - 1) {
+            return Err(safety("earlier stages have not all completed"));
+        }
+        sqlx::query(
+            "UPDATE entity_deletion_stage_history SET status = 'completed', is_current = false,
+                 finished_at = clock_timestamp(), metadata_json = metadata_json || $2, updated_at = now()
+             WHERE id = $1",
+        ).bind(stage.id).bind(metadata).execute(&mut *tx).await?;
+        let finished = usize::try_from(stage.sequence).ok() == Some(inventory.stages.len());
+        if !finished {
+            insert_stage(&mut tx, request.id, &inventory, stage.sequence + 1).await?;
+        }
+        let request = sqlx::query_as(
+            "UPDATE entity_deletion_requests
+             SET status = CASE WHEN $2 THEN 'demo_completed' ELSE status END,
+                 completed_at = CASE WHEN $2 THEN clock_timestamp() ELSE NULL END,
+                 lease_owner = CASE WHEN $2 THEN NULL ELSE lease_owner END,
+                 lease_until = CASE WHEN $2 THEN NULL ELSE lease_until END,
+                 retry_count = 0, last_error = NULL, last_error_at = NULL, updated_at = now()
+             WHERE id = $1 RETURNING *",
+        )
+        .bind(request.id)
+        .bind(finished)
+        .fetch_one(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(request)
+    }
+}

crates/buzz-db/src/store/mod.rs

@@ -18,6 +18,8 @@ pub mod community;
 pub mod deletion;
 /// Direct message channel persistence.
 pub mod dm;
+/// Operator-global synthetic user deletion workflow persistence.
+pub mod entity_deletion;
 /// Event storage and retrieval.
 pub mod event;
 /// App-owned push enqueue and channel TTL refresh for inserted events.

crates/buzz-deletion/src/executor.rs

@@ -0,0 +1,406 @@
+//! Shared request selection, execution, heartbeats, cancellation and recovery.
+//! Dispatch is explicit; each store owns its transaction and safety semantics.
+
+use super::*;
+use buzz_db::entity_deletion::{EntityClaim, EntityRequest, StageRecord};
+
+#[derive(Clone)]
+pub(super) enum WorkflowServices {
+    Community(Services),
+    User(user::Services),
+}
+
+impl From<Services> for WorkflowServices {
+    fn from(services: Services) -> Self {
+        Self::Community(services)
+    }
+}
+
+impl From<user::Services> for WorkflowServices {
+    fn from(services: user::Services) -> Self {
+        Self::User(services)
+    }
+}
+
+#[derive(Clone)]
+enum Execution {
+    Community {
+        services: Services,
+        claim: ClaimedDeletion,
+    },
+    User {
+        services: user::Services,
+        claim: EntityClaim,
+        stage: StageRecord,
+    },
+}
+
+enum Selection {
+    Ready(Box<Execution>),
+    Empty,
+    Finished(i32),
+}
+
+#[cfg(test)]
+pub(super) async fn execute_claim(
+    services: &Services,
+    mode: LoopMode,
+    claim: ClaimedDeletion,
+    shutdown: &CancellationToken,
+) -> Result<RunOutput> {
+    let mut execution = Execution::Community {
+        services: services.clone(),
+        claim,
+    };
+    match execution.run(mode, shutdown).await? {
+        Output::Community(output) => Ok(output),
+        Output::User { .. } => anyhow::bail!("community execution returned a user result"),
+    }
+}
+
+#[derive(Serialize)]
+#[serde(untagged)]
+enum Output {
+    Community(RunOutput),
+    User {
+        request: Box<EntityRequest>,
+        demonstration: &'static str,
+    },
+}
+
+impl Output {
+    fn failed(&self) -> bool {
+        match self {
+            Self::Community(output) => {
+                output.last_error.is_some() || output.blocked_reason.is_some()
+            }
+            Self::User { request, .. } => {
+                request.last_error.is_some() || request.blocked_reason.is_some()
+            }
+        }
+    }
+}
+
+pub(super) async fn run_loop(
+    services: impl Into<WorkflowServices>,
+    mode: LoopMode,
+    request_id: Option<Uuid>,
+    executor_id: String,
+) -> Result<i32> {
+    let services = services.into();
+    let shutdown = shutdown_token();
+    loop {
+        if shutdown.is_cancelled() {
+            services.stop(&executor_id).await?;
+            return Ok(0);
+        }
+        let mut execution = match services
+            .claim(mode, request_id, &executor_id, &shutdown)
+            .await?
+        {
+            Selection::Ready(execution) => execution,
+            Selection::Finished(code) => return Ok(code),
+            Selection::Empty => {
+                if mode == LoopMode::Run {
+                    anyhow::bail!("deletion request is not runnable, is blocked, or is leased by another executor");
+                }
+                return Ok(0);
+            }
+        };
+        let output = execution.run(mode, &shutdown).await?;
+        print_json(&output)?;
+        let failed = output.failed();
+        let stopped = execution.stop_requested();
+        if mode == LoopMode::Run || shutdown.is_cancelled() || failed || stopped {
+            return Ok(i32::from(failed));
+        }
+    }
+}
+
+impl WorkflowServices {
+    async fn stop(&self, owner: &str) -> Result<()> {
+        match self {
+            Self::Community(s) => s.store.stop_executor(None, owner).await?,
+            Self::User(s) => s.store.stop_executor(None, owner).await?,
+        }
+        Ok(())
+    }
+
+    async fn claim(
+        &self,
+        mode: LoopMode,
+        id: Option<Uuid>,
+        owner: &str,
+        shutdown: &CancellationToken,
+    ) -> Result<Selection> {
+        match self {
+            Self::User(services) => {
+                let Some(claim) = services
+                    .store
+                    .claim(id, owner, DEFAULT_LEASE_DURATION)
+                    .await?
+                else {
+                    return Ok(Selection::Empty);
+                };
+                let stage = services.store.current_stage(claim.request.id).await?;
+                Ok(Selection::Ready(Box::new(Execution::User {
+                    services: services.clone(),
+                    claim,
+                    stage,
+                })))
+            }
+            Self::Community(services) => {
+                let claim = match id {
+                    Some(id) => {
+                        services
+                            .store
+                            .claim_specific(id, owner, DEFAULT_LEASE_DURATION)
+                            .await?
+                    }
+                    None => {
+                        services
+                            .store
+                            .claim_next(owner, DEFAULT_LEASE_DURATION)
+                            .await?
+                    }
+                };
+                let claim = if claim.is_none() && mode == LoopMode::Drain && id.is_none() {
+                    services
+                        .store
+                        .claim_next_owner_submission(owner, DEFAULT_LEASE_DURATION)
+                        .await?
+                } else {
+                    claim
+                };
+                let Some(mut claim) = claim else {
+                    return Ok(Selection::Empty);
+                };
+                if claim.request.stage == DeletionStage::Submitted {
+                    let request = claim.request.clone();
+                    match prepare_owner_claim_with(
+                        services, mode, claim, shutdown, HEARTBEAT_INTERVAL,
+                        |lost| async move { build_inventory(services, &request, Some(&lost)).await },
+                    ).await? {
+                        OwnerPreparationOutcome::Prepared(prepared) => claim = *prepared,
+                        OwnerPreparationOutcome::Stopped => return Ok(Selection::Finished(0)),
+                        OwnerPreparationOutcome::Failed(output) => {
+                            // Preserve the original drain's nonzero exit and JSON evidence.
+                            print_json(&output)?;
+                            return Ok(Selection::Finished(1));
+                        }
+                    }
+                }
+                Ok(Selection::Ready(Box::new(Execution::Community {
+                    services: services.clone(),
+                    claim,
+                })))
+            }
+        }
+    }
+}
+
+impl Execution {
+    async fn run(&mut self, mode: LoopMode, shutdown: &CancellationToken) -> Result<Output> {
+        let mut completed = 0u16;
+        loop {
+            if shutdown.is_cancelled() {
+                self.stop(mode).await?;
+                return self.output().await;
+            }
+            self.heartbeat(mode).await?;
+            match self.run_stage(mode, shutdown).await {
+                StageOutcome::Completed => {}
+                StageOutcome::Shutdown => {
+                    self.stop(mode).await?;
+                    return self.output().await;
+                }
+                StageOutcome::Failed(error) => {
+                    self.record_failure(&error).await?;
+                    return self.output().await;
+                }
+            }
+            completed += 1;
+            if self.refresh().await? {
+                if let Self::User {
+                    services, claim, ..
+                } = self
+                {
+                    services
+                        .store
+                        .stop_executor(Some(&claim.lease), &claim.lease.owner)
+                        .await?;
+                }
+                return self.output().await;
+            }
+            if let Self::User { services, .. } = self {
+                if services
+                    .controls
+                    .stop_after
+                    .is_some_and(|limit| completed >= limit)
+                {
+                    self.stop(mode).await?;
+                    return self.output().await;
+                }
+            }
+        }
+    }
+
+    fn stop_requested(&self) -> bool {
+        matches!(self, Self::User { services, .. } if services.controls.stop_after.is_some())
+    }
+
+    async fn heartbeat(&self, mode: LoopMode) -> Result<()> {
+        match self {
+            Self::Community { services, claim } => {
+                services
+                    .store
+                    .heartbeat(&claim.lease, mode.as_str(), DEFAULT_LEASE_DURATION, false)
+                    .await?;
+            }
+            Self::User {
+                services, claim, ..
+            } => {
+                services
+                    .store
+                    .heartbeat(&claim.lease, mode.as_str(), DEFAULT_LEASE_DURATION)
+                    .await?;
+            }
+        }
+        Ok(())
+    }
+
+    async fn stop(&self, mode: LoopMode) -> Result<()> {
+        match self {
+            Self::Community { services, claim } => {
+                stop_claim_executor(services, mode, &claim.lease).await?
+            }
+            Self::User {
+                services, claim, ..
+            } => {
+                services
+                    .store
+                    .stop_executor(Some(&claim.lease), &claim.lease.owner)
+                    .await?;
+            }
+        }
+        Ok(())
+    }
+
+    async fn output(&self) -> Result<Output> {
+        match self {
+            Self::Community { services, claim } => Ok(Output::Community(run_output(
+                services.store.get(claim.request.id).await?,
+            ))),
+            Self::User {
+                services, claim, ..
+            } => Ok(Output::User {
+                request: Box::new(services.store.get(claim.request.id).await?),
+                demonstration: "Synthetic workflow only; no user data was deleted.",
+            }),
+        }
+    }
+
+    async fn refresh(&mut self) -> Result<bool> {
+        match self {
+            Self::Community { services, claim } => {
+                let request = services.store.get(claim.request.id).await?;
+                let terminal = request.stage == DeletionStage::RetentionPending
+                    || request.blocked_reason.is_some();
+                claim.lease.fence_generation = request.fence_generation;
+                claim.request = request;
+                Ok(terminal)
+            }
+            Self::User {
+                services,
+                claim,
+                stage,
+            } => {
+                claim.request = services.store.get(claim.request.id).await?;
+                let terminal =
+                    claim.request.status == "demo_completed" || claim.request.blocked_at.is_some();
+                if !terminal {
+                    *stage = services.store.current_stage(claim.request.id).await?;
+                }
+                Ok(terminal)
+            }
+        }
+    }
+
+    async fn record_failure(&self, error: &anyhow::Error) -> Result<()> {
+        match self {
+            Self::Community { services, claim } => {
+                let request = services.store.get(claim.request.id).await?;
+                if request.lease_owner.as_deref() == Some(&claim.lease.owner)
+                    && request.lease_generation == claim.lease.generation
+                {
+                    record_stage_failure(services, &claim.lease, claim.request.stage, error)
+                        .await?;
+                }
+            }
+            Self::User {
+                services,
+                claim,
+                stage,
+            } => {
+                match services
+                    .store
+                    .record_failure(
+                        &claim.lease,
+                        stage.id,
+                        &format!("{error:#}"),
+                        is_permanent_error(error),
+                        RETRY_DELAY,
+                    )
+                    .await
+                {
+                    Ok(()) => {}
+                    Err(error) if buzz_db::entity_deletion::is_stale_lease(&error) => {}
+                    Err(error) => return Err(error.into()),
+                }
+            }
+        }
+        Ok(())
+    }
+
+    async fn run_stage(&self, mode: LoopMode, shutdown: &CancellationToken) -> StageOutcome {
+        let heartbeat_execution = self.clone();
+        let cancel = CancellationToken::new();
+        let heartbeat_cancel = cancel.clone();
+        let lost = CancellationToken::new();
+        let heartbeat_lost = lost.clone();
+        let heartbeat = tokio::spawn(async move {
+            let mut interval = tokio::time::interval(heartbeat_interval());
+            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+            interval.tick().await;
+            loop {
+                tokio::select! {
+                    _ = heartbeat_cancel.cancelled() => return,
+                    _ = interval.tick() => {
+                        if heartbeat_execution.heartbeat(mode).await.is_err() {
+                            heartbeat_lost.cancel();
+                            return;
+                        }
+                    }
+                }
+            }
+        });
+        let operation = async {
+            match self {
+                Self::Community { services, claim } => execute_stage(services, claim, &lost).await,
+                Self::User {
+                    services,
+                    claim,
+                    stage,
+                } => user::execute_stage(services, claim, stage, &lost).await,
+            }
+        };
+        let result = await_stage(operation, shutdown, &lost).await;
+        cancel.cancel();
+        match heartbeat.await {
+            Ok(()) => result,
+            Err(error) => {
+                StageOutcome::Failed(anyhow::anyhow!("deletion heartbeat task failed: {error}"))
+            }
+        }
+    }
+}

crates/buzz-deletion/src/lib.rs

@@ -1,6 +1,18 @@
 #![deny(unsafe_code)]
 #![warn(missing_docs)]
-//! Shared durable whole-community deletion engine and store adapters.
+//! Shared durable deletion engine and store adapters.
+
+mod executor;
+/// Synthetic multi-community user deletion commands.
+pub mod user;
+
+use executor::run_loop;
+
+#[cfg(test)]
+use executor::execute_claim;
+
+#[cfg(test)]
+mod user_postgres_tests;
 
 use std::future::Future;
 #[cfg(test)]
@@ -546,6 +558,10 @@ fn resolve_submit_host(host: Option<&str>, relay_url: Option<&str>) -> Result<St
 }
 
 async fn connect_store() -> Result<DeletionStore> {
+    Ok(store(&connect_database().await?))
+}
+
+async fn connect_database() -> Result<Db> {
     let database_url = required_env("DATABASE_URL")?;
     // The scheduled drain starts in a fresh pod whose first connection can
     // outlast the relay's three-second budget.
@@ -559,7 +575,7 @@ async fn connect_store() -> Result<DeletionStore> {
         "deletion",
     )
     .await?;
-    Ok(store(&db))
+    Ok(db)
 }
 
 fn resolve_s3_region(buzz_region: Option<String>, aws_region: Option<String>) -> String {
@@ -962,91 +978,6 @@ async fn freeze_destructive_manifest(
     Ok(manifest)
 }
 
-async fn run_loop(
-    services: Services,
-    mode: LoopMode,
-    request_id: Option<Uuid>,
-    executor_id: String,
-) -> Result<i32> {
-    let shutdown = shutdown_token();
-    let mut ran = false;
-    loop {
-        if shutdown.is_cancelled() {
-            services
-                .store
-                .stop_executor(None, &executor_id)
-                .await
-                .context("record executor drain")?;
-            return Ok(0);
-        }
-        let claim = match request_id {
-            Some(id) => {
-                services
-                    .store
-                    .claim_specific(id, &executor_id, DEFAULT_LEASE_DURATION)
-                    .await?
-            }
-            None => {
-                services
-                    .store
-                    .claim_next(&executor_id, DEFAULT_LEASE_DURATION)
-                    .await?
-            }
-        };
-        let claim = if claim.is_none() && mode == LoopMode::Drain && request_id.is_none() {
-            services
-                .store
-                .claim_next_owner_submission(&executor_id, DEFAULT_LEASE_DURATION)
-                .await?
-        } else {
-            claim
-        };
-        let Some(mut claim) = claim else {
-            if mode == LoopMode::Run && !ran {
-                anyhow::bail!(
-                    "deletion request is not runnable, is blocked, or is leased by another executor"
-                );
-            }
-            return Ok(0);
-        };
-        ran = true;
-        if claim.request.stage == DeletionStage::Submitted {
-            let preparation_request = claim.request.clone();
-            let preparation_services = &services;
-            match prepare_owner_claim_with(
-                &services,
-                mode,
-                claim,
-                &shutdown,
-                HEARTBEAT_INTERVAL,
-                |heartbeat_lost| async move {
-                    build_inventory(
-                        preparation_services,
-                        &preparation_request,
-                        Some(&heartbeat_lost),
-                    )
-                    .await
-                },
-            )
-            .await?
-            {
-                OwnerPreparationOutcome::Prepared(prepared) => claim = *prepared,
-                OwnerPreparationOutcome::Stopped => return Ok(0),
-                OwnerPreparationOutcome::Failed(output) => {
-                    print_json(&output)?;
-                    return Ok(1);
-                }
-            }
-        }
-        let output = execute_claim(&services, mode, claim, &shutdown).await?;
-        print_json(&output)?;
-        let failed = output.last_error.is_some() || output.blocked_reason.is_some();
-        if mode == LoopMode::Run || shutdown.is_cancelled() || failed {
-            return Ok(i32::from(failed));
-        }
-    }
-}
-
 enum OwnerPreparationOutcome {
     Prepared(Box<ClaimedDeletion>),
     Stopped,
@@ -1254,57 +1185,6 @@ async fn record_stage_failure(
     }
 }
 
-async fn execute_claim(
-    services: &Services,
-    mode: LoopMode,
-    mut claim: ClaimedDeletion,
-    shutdown: &CancellationToken,
-) -> Result<RunOutput> {
-    let token = claim.lease.clone();
-    loop {
-        if shutdown.is_cancelled() {
-            stop_claim_executor(services, mode, &token).await?;
-            let request = services.store.get(token.request_id).await?;
-            return Ok(run_output(request));
-        }
-        services
-            .store
-            .heartbeat(&token, mode.as_str(), DEFAULT_LEASE_DURATION, false)
-            .await?;
-        let stage_result = run_stage_with_heartbeat(services, mode, &claim, shutdown).await;
-        match stage_result {
-            StageOutcome::Completed => {}
-            StageOutcome::Shutdown => {
-                stop_claim_executor(services, mode, &token).await?;
-                let request = services.store.get(token.request_id).await?;
-                return Ok(run_output(request));
-            }
-            StageOutcome::Failed(error) => {
-                let request = services.store.get(token.request_id).await?;
-                if request.lease_owner.as_deref() != Some(&token.owner)
-                    || request.lease_generation != token.generation
-                {
-                    return Ok(run_output(request));
-                }
-                if !record_stage_failure(services, &token, claim.request.stage, &error).await? {
-                    // Ownership expired between the stage failure and durable
-                    // error recording. A successor now owns retry/block policy.
-                    let request = services.store.get(token.request_id).await?;
-                    return Ok(run_output(request));
-                }
-                let request = services.store.get(token.request_id).await?;
-                return Ok(run_output(request));
-            }
-        }
-        let request = services.store.get(token.request_id).await?;
-        if request.stage == DeletionStage::RetentionPending || request.blocked_reason.is_some() {
-            return Ok(run_output(request));
-        }
-        claim.request = request.clone();
-        claim.lease.fence_generation = request.fence_generation;
-    }
-}
-
 enum StageOutcome {
     Completed,
     Shutdown,
@@ -1330,61 +1210,6 @@ where
     }
 }
 
-async fn run_stage_with_heartbeat(
-    services: &Services,
-    mode: LoopMode,
-    claim: &ClaimedDeletion,
-    shutdown: &CancellationToken,
-) -> StageOutcome {
-    let heartbeat_services = services.clone();
-    let heartbeat_token = claim.lease.clone();
-    let heartbeat_mode = mode.as_str();
-    let heartbeat_shutdown = CancellationToken::new();
-    let heartbeat_cancel = heartbeat_shutdown.clone();
-    let heartbeat_error = CancellationToken::new();
-    let heartbeat_error_signal = heartbeat_error.clone();
-    let heartbeat = tokio::spawn(async move {
-        let mut interval = tokio::time::interval(heartbeat_interval());
-        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-        interval.tick().await;
-        loop {
-            tokio::select! {
-                _ = heartbeat_cancel.cancelled() => return,
-                _ = interval.tick() => {
-                    if heartbeat_services
-                        .store
-                        .heartbeat(
-                            &heartbeat_token,
-                            heartbeat_mode,
-                            DEFAULT_LEASE_DURATION,
-                            false,
-                        )
-                        .await
-                        .is_err()
-                    {
-                        heartbeat_error_signal.cancel();
-                        return;
-                    }
-                }
-            }
-        }
-    });
-
-    let stage = await_stage(
-        execute_stage(services, claim, &heartbeat_error),
-        shutdown,
-        &heartbeat_error,
-    )
-    .await;
-    heartbeat_shutdown.cancel();
-    match heartbeat.await {
-        Ok(()) => stage,
-        Err(error) => {
-            StageOutcome::Failed(anyhow::anyhow!("deletion heartbeat task failed: {error}"))
-        }
-    }
-}
-
 async fn run_guarded_external_step<F, Fut, T>(
     services: &Services,
     token: &LeaseToken,

crates/buzz-deletion/src/user.rs

@@ -0,0 +1,199 @@
+//! PostgreSQL-only operator commands and dummy actions for user deletion.
+
+use anyhow::Result;
+use buzz_db::entity_deletion::{EntityClaim, EntityDeletionStore, StageRecord};
+use clap::{Args, Subcommand};
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+use crate::{default_executor_id, print_json, LoopMode};
+
+/// Commands for synthetic user-deletion demonstrations; no real erasure occurs.
+#[derive(Debug, Subcommand)]
+pub enum Command {
+    /// Save a synthetic request. Repeating the same submission returns its row.
+    Submit {
+        /// Synthetic user label, for example demo:alice.
+        #[arg(long)]
+        target: String,
+        /// Submitting operator identity.
+        #[arg(long)]
+        requested_by: String,
+        /// Optional operator reason.
+        #[arg(long)]
+        reason: Option<String>,
+    },
+    /// Prepare and freeze the three-participant-community demonstration plan.
+    Inventory {
+        /// Request UUID.
+        id: Uuid,
+    },
+    /// List demonstration requests.
+    List {
+        /// Maximum number of rows.
+        #[arg(long, default_value_t = 100, value_parser = clap::value_parser!(u16).range(1..=1000))]
+        limit: u16,
+    },
+    /// Show inventory, approval, ownership, and all reached stages.
+    Inspect {
+        /// Request UUID.
+        id: Uuid,
+    },
+    /// Independently approve the frozen demonstration inventory.
+    Approve {
+        /// Request UUID.
+        id: Uuid,
+        /// Must differ from the submitting operator.
+        #[arg(long)]
+        approved_by: String,
+    },
+    /// Run an approved demonstration through its remaining dummy stages.
+    Run {
+        /// Request UUID.
+        id: Uuid,
+        /// Executor identity, otherwise a unique process identity.
+        #[arg(long)]
+        executor_id: Option<String>,
+        /// Controls for demonstrating failure and recovery.
+        #[command(flatten)]
+        controls: DemoControls,
+    },
+    /// Drain eligible synthetic requests; exits when none are runnable.
+    Drain {
+        /// Executor identity.
+        #[arg(long)]
+        executor_id: Option<String>,
+    },
+    /// Resume blocked work after operator remediation.
+    Unblock {
+        /// Request UUID.
+        id: Uuid,
+        /// Operator identity recorded in current-stage metadata.
+        #[arg(long)]
+        unblocked_by: String,
+        /// Explanation of the remediation.
+        #[arg(long)]
+        reason: String,
+    },
+}
+
+/// Bounded, explicit failure controls for an artificial workflow.
+#[derive(Debug, Clone, Default, Args)]
+pub struct DemoControls {
+    /// Fail at this one-based plan sequence on each invocation using this flag.
+    #[arg(long, value_parser = clap::value_parser!(u16).range(1..=15))]
+    pub fail_at: Option<u16>,
+    /// Release ownership after this many successful steps in this invocation.
+    #[arg(long, value_parser = clap::value_parser!(u16).range(1..=15))]
+    pub stop_after: Option<u16>,
+    /// Delay each dummy action to exercise heartbeats or process interruption.
+    #[arg(long, default_value_t = 0, value_parser = clap::value_parser!(u64).range(0..=60000))]
+    pub step_delay_ms: u64,
+}
+
+#[derive(Clone)]
+pub(crate) struct Services {
+    pub store: EntityDeletionStore,
+    pub controls: DemoControls,
+}
+
+/// Execute one user-deletion command with a PostgreSQL-only connection.
+pub async fn run(command: Command) -> Result<i32> {
+    let db = crate::connect_database().await?;
+    run_with_store(command, db.entity_deletion_store()).await
+}
+
+pub(crate) async fn run_with_store(command: Command, store: EntityDeletionStore) -> Result<i32> {
+    match command {
+        Command::Submit {
+            target,
+            requested_by,
+            reason,
+        } => {
+            print_json(
+                &store
+                    .submit(&target, &requested_by, reason.as_deref())
+                    .await?,
+            )?;
+        }
+        Command::Inventory { id } => print_json(&store.prepare_inventory(id).await?)?,
+        Command::List { limit } => print_json(&store.list(limit).await?)?,
+        Command::Inspect { id } => print_json(&store.inspect(id).await?)?,
+        Command::Approve { id, approved_by } => {
+            print_json(&store.approve(id, &approved_by).await?)?
+        }
+        Command::Unblock {
+            id,
+            unblocked_by,
+            reason,
+        } => {
+            print_json(&store.unblock(id, &unblocked_by, &reason).await?)?;
+        }
+        Command::Run {
+            id,
+            executor_id,
+            controls,
+        } => {
+            return crate::executor::run_loop(
+                Services { store, controls },
+                LoopMode::Run,
+                Some(id),
+                executor_id.unwrap_or_else(default_executor_id),
+            )
+            .await;
+        }
+        Command::Drain { executor_id } => {
+            return crate::executor::run_loop(
+                Services {
+                    store,
+                    controls: DemoControls::default(),
+                },
+                LoopMode::Drain,
+                None,
+                executor_id.unwrap_or_else(default_executor_id),
+            )
+            .await;
+        }
+    }
+    Ok(0)
+}
+
+pub(crate) async fn execute_stage(
+    services: &Services,
+    claim: &EntityClaim,
+    expected: &StageRecord,
+    heartbeat_lost: &CancellationToken,
+) -> Result<()> {
+    let stage = services.store.start_stage(&claim.lease).await?;
+    if stage.id != expected.id {
+        anyhow::bail!("current user stage changed before dispatch");
+    }
+    let scope = stage
+        .target_community_id
+        .map_or_else(|| "request".into(), |id| id.to_string());
+    let message = format!(
+        "DEMO ONLY: would execute {} for {} in {}; no user data is deleted",
+        stage.stage, claim.request.target, scope,
+    );
+    eprintln!("{message}");
+    tokio::select! {
+        biased;
+        _ = heartbeat_lost.cancelled() => return Err(crate::DeletionLeaseLost.into()),
+        _ = tokio::time::sleep(std::time::Duration::from_millis(services.controls.step_delay_ms)) => {}
+    }
+    if services.controls.fail_at.map(i32::from) == Some(stage.sequence) {
+        return Err(crate::transient(format!(
+            "injected demo failure at stage {}",
+            stage.sequence
+        )));
+    }
+    services
+        .store
+        .complete_stage(
+            &claim.lease,
+            stage.id,
+            &serde_json::json!({"message": message, "simulated": true}),
+        )
+        .await?;
+    Ok(())
+}

crates/buzz-deletion/src/user_postgres_tests.rs

@@ -0,0 +1,216 @@
+use super::*;
+use user::{Command, DemoControls};
+
+async fn fixture() -> (Db, buzz_db::entity_deletion::EntityRequest) {
+    let url = std::env::var("BUZZ_TEST_DATABASE_URL")
+        .or_else(|_| std::env::var("DATABASE_URL"))
+        .expect("isolated PostgreSQL URL");
+    let pool = sqlx::PgPool::connect(&url).await.expect("connect");
+    let db = Db::from_pool(pool);
+    let store = db.entity_deletion_store();
+    let target = format!("demo:{}", Uuid::new_v4().simple());
+    user::run_with_store(
+        Command::Submit {
+            target: target.clone(),
+            requested_by: "submitter".into(),
+            reason: None,
+        },
+        store.clone(),
+    )
+    .await
+    .expect("submit command");
+    let request = store
+        .list(100)
+        .await
+        .expect("list")
+        .into_iter()
+        .find(|r| r.target == target)
+        .expect("submitted request");
+    user::run_with_store(Command::Inventory { id: request.id }, store.clone())
+        .await
+        .expect("inventory command");
+    assert!(user::run_with_store(
+        Command::Run {
+            id: request.id,
+            executor_id: Some("unapproved".into()),
+            controls: DemoControls::default()
+        },
+        store.clone(),
+    )
+    .await
+    .is_err());
+    user::run_with_store(
+        Command::Approve {
+            id: request.id,
+            approved_by: "approver".into(),
+        },
+        store.clone(),
+    )
+    .await
+    .expect("approve command");
+    (db, store.get(request.id).await.expect("request"))
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn shared_executor_runs_three_communities_and_resumes_b() {
+    let (db, request) = fixture().await;
+    let store = db.entity_deletion_store();
+    // Canary data at the synthetic community IDs makes accidental real cleanup
+    // observable, including updates that would preserve row counts.
+    for n in 1..=3 {
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(Uuid::from_u128(n))
+            .bind(format!("canary-{n}.invalid"))
+            .execute(db.pool())
+            .await
+            .expect("community canary");
+        sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)")
+            .bind(Uuid::from_u128(n))
+            .bind(vec![7u8; 32])
+            .execute(db.pool())
+            .await
+            .expect("user canary");
+        sqlx::query(
+            "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')",
+        )
+        .bind(Uuid::from_u128(n))
+        .bind(hex::encode([7u8; 32]))
+        .execute(db.pool())
+        .await
+        .expect("membership canary");
+    }
+    let before: serde_json::Value = sqlx::query_scalar(
+        "SELECT jsonb_build_object(
+             'communities', (SELECT jsonb_agg(to_jsonb(c) ORDER BY id) FROM communities c),
+             'users', (SELECT jsonb_agg(to_jsonb(u) ORDER BY community_id, pubkey) FROM users u),
+             'members', (SELECT jsonb_agg(to_jsonb(m) ORDER BY community_id, pubkey) FROM relay_members m),
+             'events', (SELECT count(*) FROM events),
+             'community_requests', (SELECT count(*) FROM community_deletion_requests))",
+    )
+    .fetch_one(db.pool())
+    .await
+    .expect("domain snapshot");
+    let result = user::run_with_store(
+        Command::Run {
+            id: request.id,
+            executor_id: Some("first-process".into()),
+            controls: DemoControls {
+                fail_at: Some(9),
+                ..Default::default()
+            },
+        },
+        store.clone(),
+    )
+    .await
+    .expect("failed demonstration");
+    assert_eq!(result, 1);
+    let snapshot = store.inspect(request.id).await.expect("history");
+    assert_eq!(snapshot.stage_history.len(), 9);
+    assert_eq!(snapshot.stage_history[8].stage, "delete_data");
+    assert_eq!(
+        snapshot.stage_history[8].target_community_id,
+        Some(Uuid::from_u128(2))
+    );
+    assert_eq!(snapshot.stage_history[8].status, "failed");
+    assert!(snapshot.request.lease_owner.is_none());
+    assert_eq!(
+        snapshot.stage_history[..4]
+            .iter()
+            .map(|s| s.stage.as_str())
+            .collect::<Vec<_>>(),
+        [
+            "gate_user",
+            "gate_community",
+            "gate_community",
+            "gate_community"
+        ]
+    );
+    assert!(snapshot.stage_history[4..7]
+        .iter()
+        .all(|s| s.target_community_id == Some(Uuid::from_u128(1)) && s.status == "completed"));
+    sqlx::query("UPDATE entity_deletion_requests SET next_attempt_at = now() WHERE id = $1")
+        .bind(request.id)
+        .execute(db.pool())
+        .await
+        .expect("make retry due");
+    assert_eq!(
+        user::run_with_store(
+            Command::Drain {
+                executor_id: Some("replacement-process".into()),
+            },
+            store.clone()
+        )
+        .await
+        .expect("drain retry"),
+        0
+    );
+    let finished = store.inspect(request.id).await.expect("finished");
+    assert_eq!(finished.request.status, "demo_completed");
+    assert_eq!(finished.request.lease_generation, 2);
+    assert_eq!(finished.stage_history.len(), 15);
+    assert!(finished
+        .stage_history
+        .iter()
+        .all(|s| !s.is_current && s.metadata_json["simulated"] == true));
+    for (a, b) in snapshot.stage_history[..8]
+        .iter()
+        .zip(&finished.stage_history[..8])
+    {
+        assert_eq!(
+            serde_json::to_value(a).expect("before"),
+            serde_json::to_value(b).expect("after")
+        );
+    }
+    let after: serde_json::Value = sqlx::query_scalar(
+        "SELECT jsonb_build_object(
+             'communities', (SELECT jsonb_agg(to_jsonb(c) ORDER BY id) FROM communities c),
+             'users', (SELECT jsonb_agg(to_jsonb(u) ORDER BY community_id, pubkey) FROM users u),
+             'members', (SELECT jsonb_agg(to_jsonb(m) ORDER BY community_id, pubkey) FROM relay_members m),
+             'events', (SELECT count(*) FROM events),
+             'community_requests', (SELECT count(*) FROM community_deletion_requests))",
+    )
+    .fetch_one(db.pool())
+    .await
+    .expect("domain snapshot");
+    assert_eq!(before, after);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn voluntary_stop_retains_one_pending_stage_for_next_executor() {
+    let (db, request) = fixture().await;
+    let store = db.entity_deletion_store();
+    user::run_with_store(
+        Command::Run {
+            id: request.id,
+            executor_id: Some("stopping-process".into()),
+            controls: DemoControls {
+                stop_after: Some(7),
+                ..Default::default()
+            },
+        },
+        store.clone(),
+    )
+    .await
+    .expect("stop after A");
+    let stopped = store.inspect(request.id).await.expect("stopped");
+    assert_eq!(stopped.request.status, "running");
+    assert!(stopped.request.lease_owner.is_none());
+    assert_eq!(stopped.stage_history.len(), 8);
+    assert_eq!(stopped.stage_history[7].status, "pending");
+    assert!(stopped.stage_history[7].is_current);
+    user::run_with_store(
+        Command::Run {
+            id: request.id,
+            executor_id: Some("next-process".into()),
+            controls: DemoControls::default(),
+        },
+        store.clone(),
+    )
+    .await
+    .expect("resume");
+    let finished = store.inspect(request.id).await.expect("finished");
+    assert_eq!(finished.request.status, "demo_completed");
+    assert_eq!(finished.stage_history[7].id, stopped.stage_history[7].id);
+}

docs/operator-user-deletion-demo.md

@@ -0,0 +1,156 @@
+# User deletion workflow demonstration
+
+This exercises real PostgreSQL requests, approvals, leases, stage history and
+recovery for a synthetic user in three participant communities. Every action
+prints `DEMO ONLY`. It does not discover or delete real users, change memberships,
+gate access, contact storage, or delete owned communities.
+
+## Prepare an isolated development database
+
+Activate Hermit and start a separate PostgreSQL instance. Use a different
+container name and port from a running Buzz development environment.
+
+```bash
+. ./bin/activate-hermit
+docker run --detach --name buzz-user-deletion-demo \
+  --publish 127.0.0.1:55432:5432 \
+  --env POSTGRES_USER=buzz \
+  --env POSTGRES_PASSWORD=buzz-local-demo \
+  --env POSTGRES_DB=buzz postgres:17-alpine
+export DATABASE_URL=postgres://buzz:buzz-local-demo@127.0.0.1:55432/buzz
+cargo build -p buzz-admin
+./target/debug/buzz-admin migrate
+```
+
+These are disposable local credentials. If the port is occupied, choose another
+one in both the Docker command and URL. No Redis or media-storage configuration
+is required by `user-deletions`.
+
+## Submit, prepare and approve
+
+An operator manually verifies the incoming request in the real operational
+process. This demonstration uses `demo:alice`; it is not an account lookup.
+Only `demo:` labels are accepted, and inventory is always the fixed participant
+fixture. A different label creates an independent demonstration.
+
+```bash
+./target/debug/buzz-admin user-deletions list
+./target/debug/buzz-admin user-deletions submit \
+  --target demo:alice --requested-by operator-a
+```
+
+Copy the returned `id` into `REQUEST_ID`. Submission alone creates a `submitted`
+request. Repeating the identical target/operator/reason returns that same
+request without resetting its progress.
+
+```bash
+REQUEST_ID=<returned-uuid>
+./target/debug/buzz-admin user-deletions inventory "$REQUEST_ID"
+./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID"
+./target/debug/buzz-admin user-deletions approve "$REQUEST_ID" \
+  --approved-by operator-b
+```
+
+Inventory saves the three targets and ordered plan with a digest. Approval must
+come from a different operator and covers the saved digest. Approval also creates
+the first pending/current stage. A request cannot run before approval.
+
+## Execute and inspect
+
+```bash
+./target/debug/buzz-admin user-deletions run "$REQUEST_ID"
+./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID"
+```
+
+The 15 stages are:
+
+| Sequence | Target | Action |
+| --- | --- | --- |
+| 1 | Request | `gate_user` |
+| 2–4 | Communities A, B, C | `gate_community` for each |
+| 5–7 | Community A | `drain`, `delete_data`, `verify_community` |
+| 8–10 | Community B | `drain`, `delete_data`, `verify_community` |
+| 11–13 | Community C | `drain`, `delete_data`, `verify_community` |
+| 14 | Request | `verify_user` |
+| 15 | Request | `post_delete` |
+
+All gates are simulated. They do not enforce access restrictions.
+
+`entity_deletion_requests` holds overall status, inventory and worker ownership.
+`entity_deletion_stage_history` holds each reached stage's target, status,
+`is_current`, `started_at`, `finished_at`, `metadata_json` and latest error.
+The final parent status is `demo_completed`, with no current stage or live lease.
+Completion output explicitly says no user data was deleted.
+
+## Demonstrate recovery
+
+Use a fresh synthetic target for each scenario and repeat submission, inventory
+and approval first.
+
+To stop after A finishes:
+
+```bash
+./target/debug/buzz-admin user-deletions run "$REQUEST_ID" --stop-after 7
+./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID"
+./target/debug/buzz-admin user-deletions run "$REQUEST_ID"
+```
+
+The stop releases ownership and leaves B's `drain` row pending/current.
+The next invocation claims a new generation and continues that row.
+
+To fail during B's data-removal stage:
+
+```bash
+./target/debug/buzz-admin user-deletions run "$REQUEST_ID" --fail-at 9
+./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID"
+# Wait until the saved next_attempt_at (normally 30 seconds after failure).
+./target/debug/buzz-admin user-deletions drain
+```
+
+The failed command exits nonzero. A's completed rows remain unchanged, B's current
+row records the error, and ownership is released. The next invocation without
+`--fail-at` resumes B. The flag injects a failure on every invocation where it is
+specified; it is not a persistent request setting.
+
+Five consecutive failures at the same stage block automatic claims. After
+remediation, unblock and retry:
+
+```bash
+./target/debug/buzz-admin user-deletions unblock "$REQUEST_ID" \
+  --unblocked-by operator-c --reason "Removed the injected failure"
+./target/debug/buzz-admin user-deletions run "$REQUEST_ID"
+```
+
+`--step-delay-ms 15000` allows observing periodic heartbeats or interrupting a
+worker. SIGTERM/Ctrl-C releases its matching lease. After an abrupt process exit,
+a later worker can claim the request once `lease_until` expires. An old generation
+cannot update stage progress. A printed action whose transition did not commit
+can print again when retried.
+
+## Repeatable checks
+
+The PostgreSQL lane uses an isolated database per test. Follow
+`crates/buzz-db/TESTING.md` for its environment and run:
+
+```bash
+scripts/test-postgres-test-discovery.sh
+scripts/postgres-test-run.sh -p buzz-db -p buzz-deletion --lib \
+  -E 'test(entity_deletion) | test(user_postgres_tests) | (package(buzz-deletion) & test(postgres_tests))'
+```
+
+Run the actual binary demonstration against a disposable, migrated database:
+
+```bash
+BUZZ_USER_DELETION_TEST_DATABASE_URL="$DATABASE_URL" \
+  python3 scripts/test-user-deletion-demo.py --binary ./target/debug/buzz-admin
+```
+
+That script creates unique synthetic requests and verifies approval rejection,
+completion, stopping/resumption, failure in B followed by a delayed retry,
+and heartbeat renewal followed by SIGTERM and resumption.
+It leaves the demonstration rows for inspection.
+
+Schema migrations may roll out normally while these tables are unused.
+Introducing real execution requires a separate change and cleanup of
+demonstration records. Owned-community policy, real identity discovery, real
+write gates and real cleanup handlers remain future work.

migrations/0067_entity_deletion_workflow.sql

@@ -0,0 +1,169 @@
+-- Durable synthetic user-deletion workflows; existing community state is unchanged.
+
+CREATE FUNCTION protect_entity_deletion_request()
+RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF (NEW.id, NEW.entity_type, NEW.target, NEW.requested_by, NEW.reason, NEW.created_at)
+        IS DISTINCT FROM
+        (OLD.id, OLD.entity_type, OLD.target, OLD.requested_by, OLD.reason, OLD.created_at)
+    THEN
+        RAISE EXCEPTION 'entity deletion target and provenance are immutable';
+    END IF;
+    IF OLD.inventory_frozen_at IS NOT NULL AND
+        (NEW.inventory_manifest, NEW.inventory_digest, NEW.inventory_frozen_at)
+        IS DISTINCT FROM
+        (OLD.inventory_manifest, OLD.inventory_digest, OLD.inventory_frozen_at)
+    THEN
+        RAISE EXCEPTION 'entity deletion inventory is immutable';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE FUNCTION protect_entity_deletion_approval()
+RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF TG_OP <> 'INSERT' THEN
+        RAISE EXCEPTION 'entity deletion approval evidence is immutable';
+    END IF;
+    IF NOT EXISTS (
+        SELECT 1 FROM entity_deletion_requests
+        WHERE id = NEW.request_id AND inventory_digest = NEW.inventory_digest
+          AND status = 'inventoried' AND requested_by <> NEW.approved_by
+    ) THEN
+        RAISE EXCEPTION 'approval requires a frozen inventory and a different operator';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE FUNCTION protect_entity_deletion_stage()
+RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF TG_OP = 'DELETE' THEN
+        RAISE EXCEPTION 'entity deletion stage history cannot be removed';
+    END IF;
+    IF OLD.status = 'completed' OR
+        (NEW.id, NEW.request_id, NEW.sequence, NEW.target_community_id, NEW.stage)
+        IS DISTINCT FROM
+        (OLD.id, OLD.request_id, OLD.sequence, OLD.target_community_id, OLD.stage)
+    THEN
+        RAISE EXCEPTION 'completed history and stage identity are immutable';
+    END IF;
+    IF OLD.started_at IS NOT NULL AND NEW.started_at IS DISTINCT FROM OLD.started_at THEN
+        RAISE EXCEPTION 'first stage start time is immutable';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+-- Operator-global workflow state. Initially used only by synthetic user demos.
+CREATE TABLE entity_deletion_requests (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    entity_type TEXT NOT NULL DEFAULT 'user' CHECK (entity_type = 'user'),
+    target TEXT NOT NULL CHECK (target ~ '^demo:[a-z0-9][a-z0-9_-]{0,63}$'),
+    status TEXT NOT NULL DEFAULT 'submitted'
+        CHECK (status IN ('submitted', 'inventoried', 'approved', 'running', 'demo_completed')),
+    requested_by TEXT NOT NULL CHECK (length(trim(requested_by)) BETWEEN 1 AND 256),
+    reason TEXT,
+    inventory_manifest JSONB,
+    inventory_digest BYTEA CHECK (inventory_digest IS NULL OR length(inventory_digest) = 32),
+    inventory_frozen_at TIMESTAMPTZ,
+    lease_owner TEXT,
+    lease_generation BIGINT NOT NULL DEFAULT 0 CHECK (lease_generation >= 0),
+    lease_until TIMESTAMPTZ,
+    attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    retry_count INTEGER NOT NULL DEFAULT 0 CHECK (retry_count >= 0),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_error TEXT,
+    last_error_at TIMESTAMPTZ,
+    blocked_at TIMESTAMPTZ,
+    blocked_reason TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    completed_at TIMESTAMPTZ,
+    CONSTRAINT entity_deletion_request_shape CHECK (
+        ((lease_owner IS NULL) = (lease_until IS NULL))
+        AND ((blocked_at IS NULL) = (blocked_reason IS NULL))
+        AND ((status = 'demo_completed') = (completed_at IS NOT NULL))
+        AND (status <> 'demo_completed' OR lease_owner IS NULL)
+        AND ((status = 'submitted') = (inventory_manifest IS NULL))
+        AND ((inventory_manifest IS NULL) = (inventory_digest IS NULL))
+        AND ((inventory_manifest IS NULL) = (inventory_frozen_at IS NULL))
+    ),
+    UNIQUE (entity_type, target),
+    UNIQUE (id, inventory_digest)
+);
+CREATE INDEX entity_deletion_requests_runnable
+    ON entity_deletion_requests (next_attempt_at, created_at, id)
+    WHERE status IN ('approved', 'running') AND blocked_at IS NULL;
+
+-- target_community_id is operator provenance, including synthetic references.
+-- It is not tenant data and intentionally has no FK to the communities table.
+CREATE TABLE entity_deletion_stage_history (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    request_id UUID NOT NULL REFERENCES entity_deletion_requests(id),
+    sequence INTEGER NOT NULL CHECK (sequence > 0),
+    target_community_id UUID,
+    stage TEXT NOT NULL CHECK (length(stage) BETWEEN 1 AND 64),
+    status TEXT NOT NULL DEFAULT 'pending'
+        CHECK (status IN ('pending', 'running', 'failed', 'completed')),
+    is_current BOOLEAN NOT NULL DEFAULT true,
+    started_at TIMESTAMPTZ,
+    finished_at TIMESTAMPTZ,
+    metadata_json JSONB NOT NULL DEFAULT '{}' CHECK (jsonb_typeof(metadata_json) = 'object'),
+    last_error TEXT,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT entity_deletion_stage_shape CHECK (
+        (is_current = (status <> 'completed'))
+        AND ((status = 'completed') = (finished_at IS NOT NULL))
+        AND ((status = 'pending') = (started_at IS NULL))
+        AND (finished_at IS NULL OR finished_at >= started_at)
+    ),
+    UNIQUE (request_id, sequence)
+);
+CREATE UNIQUE INDEX entity_deletion_stage_history_current
+    ON entity_deletion_stage_history (request_id) WHERE is_current;
+
+CREATE TABLE entity_deletion_approvals (
+    request_id UUID PRIMARY KEY,
+    inventory_digest BYTEA NOT NULL CHECK (length(inventory_digest) = 32),
+    approved_by TEXT NOT NULL CHECK (length(trim(approved_by)) BETWEEN 1 AND 256),
+    approved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    FOREIGN KEY (request_id, inventory_digest)
+        REFERENCES entity_deletion_requests(id, inventory_digest)
+);
+
+CREATE TABLE entity_deletion_executor_heartbeats (
+    executor_id TEXT PRIMARY KEY,
+    mode TEXT NOT NULL CHECK (mode IN ('run', 'drain')),
+    request_id UUID REFERENCES entity_deletion_requests(id),
+    started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    draining BOOLEAN NOT NULL DEFAULT false,
+    stopped_at TIMESTAMPTZ
+);
+
+CREATE TRIGGER protect_entity_deletion_request
+    BEFORE UPDATE ON entity_deletion_requests
+    FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_request();
+
+CREATE TRIGGER protect_entity_deletion_stage
+    BEFORE UPDATE OR DELETE ON entity_deletion_stage_history
+    FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_stage();
+
+CREATE TRIGGER protect_entity_deletion_approval
+    BEFORE INSERT OR UPDATE OR DELETE ON entity_deletion_approvals
+    FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_approval();
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_requests', 'deployment entity deletion workflow and synthetic demo state');
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_stage_history', 'deployment entity deletion workflow and synthetic demo state');
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_approvals', 'deployment entity deletion workflow and synthetic demo state');
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_executor_heartbeats', 'deployment entity deletion workflow and synthetic demo state');

schema/functions/protect_entity_deletion_approval.sql

@@ -0,0 +1,16 @@
+CREATE FUNCTION protect_entity_deletion_approval()
+RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF TG_OP <> 'INSERT' THEN
+        RAISE EXCEPTION 'entity deletion approval evidence is immutable';
+    END IF;
+    IF NOT EXISTS (
+        SELECT 1 FROM entity_deletion_requests
+        WHERE id = NEW.request_id AND inventory_digest = NEW.inventory_digest
+          AND status = 'inventoried' AND requested_by <> NEW.approved_by
+    ) THEN
+        RAISE EXCEPTION 'approval requires a frozen inventory and a different operator';
+    END IF;
+    RETURN NEW;
+END;
+$$;

schema/functions/protect_entity_deletion_request.sql

@@ -0,0 +1,19 @@
+CREATE FUNCTION protect_entity_deletion_request()
+RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF (NEW.id, NEW.entity_type, NEW.target, NEW.requested_by, NEW.reason, NEW.created_at)
+        IS DISTINCT FROM
+        (OLD.id, OLD.entity_type, OLD.target, OLD.requested_by, OLD.reason, OLD.created_at)
+    THEN
+        RAISE EXCEPTION 'entity deletion target and provenance are immutable';
+    END IF;
+    IF OLD.inventory_frozen_at IS NOT NULL AND
+        (NEW.inventory_manifest, NEW.inventory_digest, NEW.inventory_frozen_at)
+        IS DISTINCT FROM
+        (OLD.inventory_manifest, OLD.inventory_digest, OLD.inventory_frozen_at)
+    THEN
+        RAISE EXCEPTION 'entity deletion inventory is immutable';
+    END IF;
+    RETURN NEW;
+END;
+$$;

schema/functions/protect_entity_deletion_stage.sql

@@ -0,0 +1,19 @@
+CREATE FUNCTION protect_entity_deletion_stage()
+RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF TG_OP = 'DELETE' THEN
+        RAISE EXCEPTION 'entity deletion stage history cannot be removed';
+    END IF;
+    IF OLD.status = 'completed' OR
+        (NEW.id, NEW.request_id, NEW.sequence, NEW.target_community_id, NEW.stage)
+        IS DISTINCT FROM
+        (OLD.id, OLD.request_id, OLD.sequence, OLD.target_community_id, OLD.stage)
+    THEN
+        RAISE EXCEPTION 'completed history and stage identity are immutable';
+    END IF;
+    IF OLD.started_at IS NOT NULL AND NEW.started_at IS DISTINCT FROM OLD.started_at THEN
+        RAISE EXCEPTION 'first stage start time is immutable';
+    END IF;
+    RETURN NEW;
+END;
+$$;

schema/schema.sql

@@ -78,6 +78,10 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 \i functions/attach_community_write_fence.sql
 \i functions/profile_name_fields.sql
 
+\i functions/protect_entity_deletion_request.sql
+\i functions/protect_entity_deletion_approval.sql
+\i functions/protect_entity_deletion_stage.sql
+
 -- tables
 \i tables/public/communities.sql
 \i tables/public/channels.sql
@@ -138,11 +142,19 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 \i tables/public/artifact_heads.sql
 \i tables/public/artifact_revisions.sql
 
+\i tables/public/entity_deletion_requests.sql
+\i tables/public/entity_deletion_stage_history.sql
+\i tables/public/entity_deletion_approvals.sql
+\i tables/public/entity_deletion_executor_heartbeats.sql
+
 -- partitions
 \i partitions/events.sql
 \i partitions/delivery_log.sql
 
 -- triggers
+\i triggers/entity_deletion_requests.sql
+\i triggers/entity_deletion_stage_history.sql
+\i triggers/entity_deletion_approvals.sql
 \i triggers/channels.sql
 \i triggers/events.sql
 \i triggers/community_deletion_requests.sql
@@ -210,3 +222,9 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 
 -- Fence any community-scoped table that lacks an explicit attach.
 \i community_write_fence.sql
+
+-- Operator-global entity deletion state.
+\i seeds/entity_deletion_requests.sql
+\i seeds/entity_deletion_stage_history.sql
+\i seeds/entity_deletion_approvals.sql
+\i seeds/entity_deletion_executor_heartbeats.sql

schema/seeds/entity_deletion_approvals.sql

@@ -0,0 +1,2 @@
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_approvals', 'deployment entity deletion workflow and synthetic demo state');

schema/seeds/entity_deletion_executor_heartbeats.sql

@@ -0,0 +1,2 @@
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_executor_heartbeats', 'deployment entity deletion workflow and synthetic demo state');

schema/seeds/entity_deletion_requests.sql

@@ -0,0 +1,2 @@
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_requests', 'deployment entity deletion workflow and synthetic demo state');

schema/seeds/entity_deletion_stage_history.sql

@@ -0,0 +1,2 @@
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_stage_history', 'deployment entity deletion workflow and synthetic demo state');

schema/tables/public/entity_deletion_approvals.sql

@@ -0,0 +1,8 @@
+CREATE TABLE entity_deletion_approvals (
+    request_id UUID PRIMARY KEY,
+    inventory_digest BYTEA NOT NULL CHECK (length(inventory_digest) = 32),
+    approved_by TEXT NOT NULL CHECK (length(trim(approved_by)) BETWEEN 1 AND 256),
+    approved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    FOREIGN KEY (request_id, inventory_digest)
+        REFERENCES entity_deletion_requests(id, inventory_digest)
+);

schema/tables/public/entity_deletion_executor_heartbeats.sql

@@ -0,0 +1,9 @@
+CREATE TABLE entity_deletion_executor_heartbeats (
+    executor_id TEXT PRIMARY KEY,
+    mode TEXT NOT NULL CHECK (mode IN ('run', 'drain')),
+    request_id UUID REFERENCES entity_deletion_requests(id),
+    started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    draining BOOLEAN NOT NULL DEFAULT false,
+    stopped_at TIMESTAMPTZ
+);

schema/tables/public/entity_deletion_requests.sql

@@ -0,0 +1,40 @@
+-- Operator-global workflow state. Initially used only by synthetic user demos.
+CREATE TABLE entity_deletion_requests (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    entity_type TEXT NOT NULL DEFAULT 'user' CHECK (entity_type = 'user'),
+    target TEXT NOT NULL CHECK (target ~ '^demo:[a-z0-9][a-z0-9_-]{0,63}$'),
+    status TEXT NOT NULL DEFAULT 'submitted'
+        CHECK (status IN ('submitted', 'inventoried', 'approved', 'running', 'demo_completed')),
+    requested_by TEXT NOT NULL CHECK (length(trim(requested_by)) BETWEEN 1 AND 256),
+    reason TEXT,
+    inventory_manifest JSONB,
+    inventory_digest BYTEA CHECK (inventory_digest IS NULL OR length(inventory_digest) = 32),
+    inventory_frozen_at TIMESTAMPTZ,
+    lease_owner TEXT,
+    lease_generation BIGINT NOT NULL DEFAULT 0 CHECK (lease_generation >= 0),
+    lease_until TIMESTAMPTZ,
+    attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    retry_count INTEGER NOT NULL DEFAULT 0 CHECK (retry_count >= 0),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_error TEXT,
+    last_error_at TIMESTAMPTZ,
+    blocked_at TIMESTAMPTZ,
+    blocked_reason TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    completed_at TIMESTAMPTZ,
+    CONSTRAINT entity_deletion_request_shape CHECK (
+        ((lease_owner IS NULL) = (lease_until IS NULL))
+        AND ((blocked_at IS NULL) = (blocked_reason IS NULL))
+        AND ((status = 'demo_completed') = (completed_at IS NOT NULL))
+        AND (status <> 'demo_completed' OR lease_owner IS NULL)
+        AND ((status = 'submitted') = (inventory_manifest IS NULL))
+        AND ((inventory_manifest IS NULL) = (inventory_digest IS NULL))
+        AND ((inventory_manifest IS NULL) = (inventory_frozen_at IS NULL))
+    ),
+    UNIQUE (entity_type, target),
+    UNIQUE (id, inventory_digest)
+);
+CREATE INDEX entity_deletion_requests_runnable
+    ON entity_deletion_requests (next_attempt_at, created_at, id)
+    WHERE status IN ('approved', 'running') AND blocked_at IS NULL;

schema/tables/public/entity_deletion_stage_history.sql

@@ -0,0 +1,26 @@
+-- target_community_id is operator provenance, including synthetic references.
+-- It is not tenant data and intentionally has no FK to the communities table.
+CREATE TABLE entity_deletion_stage_history (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    request_id UUID NOT NULL REFERENCES entity_deletion_requests(id),
+    sequence INTEGER NOT NULL CHECK (sequence > 0),
+    target_community_id UUID,
+    stage TEXT NOT NULL CHECK (length(stage) BETWEEN 1 AND 64),
+    status TEXT NOT NULL DEFAULT 'pending'
+        CHECK (status IN ('pending', 'running', 'failed', 'completed')),
+    is_current BOOLEAN NOT NULL DEFAULT true,
+    started_at TIMESTAMPTZ,
+    finished_at TIMESTAMPTZ,
+    metadata_json JSONB NOT NULL DEFAULT '{}' CHECK (jsonb_typeof(metadata_json) = 'object'),
+    last_error TEXT,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT entity_deletion_stage_shape CHECK (
+        (is_current = (status <> 'completed'))
+        AND ((status = 'completed') = (finished_at IS NOT NULL))
+        AND ((status = 'pending') = (started_at IS NULL))
+        AND (finished_at IS NULL OR finished_at >= started_at)
+    ),
+    UNIQUE (request_id, sequence)
+);
+CREATE UNIQUE INDEX entity_deletion_stage_history_current
+    ON entity_deletion_stage_history (request_id) WHERE is_current;

schema/triggers/entity_deletion_approvals.sql

@@ -0,0 +1,3 @@
+CREATE TRIGGER protect_entity_deletion_approval
+    BEFORE INSERT OR UPDATE OR DELETE ON entity_deletion_approvals
+    FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_approval();

schema/triggers/entity_deletion_requests.sql

@@ -0,0 +1,3 @@
+CREATE TRIGGER protect_entity_deletion_request
+    BEFORE UPDATE ON entity_deletion_requests
+    FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_request();

schema/triggers/entity_deletion_stage_history.sql

@@ -0,0 +1,3 @@
+CREATE TRIGGER protect_entity_deletion_stage
+    BEFORE UPDATE OR DELETE ON entity_deletion_stage_history
+    FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_stage();

scripts/reconcile-schema-after-pgschema.sql

@@ -232,3 +232,64 @@ BEGIN
         RAISE EXCEPTION 'replica_heartbeat must contain its singleton row after pgschema apply';
     END IF;
 END $$;
+
+-- Entity deletion is operator-global, even when a stage targets a community.
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('entity_deletion_requests', 'deployment entity deletion workflow and synthetic demo state'),
+    ('entity_deletion_stage_history', 'deployment entity deletion workflow and synthetic demo state'),
+    ('entity_deletion_approvals', 'deployment entity deletion workflow and synthetic demo state'),
+    ('entity_deletion_executor_heartbeats', 'deployment entity deletion workflow and synthetic demo state')
+ON CONFLICT (table_name) DO NOTHING;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint
+                   WHERE conrelid = 'entity_deletion_requests'::regclass
+                     AND conname = 'entity_deletion_request_shape') THEN
+        ALTER TABLE entity_deletion_requests ADD CONSTRAINT entity_deletion_request_shape CHECK (
+        ((lease_owner IS NULL) = (lease_until IS NULL))
+        AND ((blocked_at IS NULL) = (blocked_reason IS NULL))
+        AND ((status = 'demo_completed') = (completed_at IS NOT NULL))
+        AND (status <> 'demo_completed' OR lease_owner IS NULL)
+        AND ((status = 'submitted') = (inventory_manifest IS NULL))
+        AND ((inventory_manifest IS NULL) = (inventory_digest IS NULL))
+        AND ((inventory_manifest IS NULL) = (inventory_frozen_at IS NULL))
+        );
+    END IF;
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint
+                   WHERE conrelid = 'entity_deletion_requests'::regclass
+                     AND conname = 'entity_deletion_request_shape' AND convalidated) THEN
+        RAISE EXCEPTION 'entity_deletion_request_shape must be validated';
+    END IF;
+END $$;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint
+                   WHERE conrelid = 'entity_deletion_stage_history'::regclass
+                     AND conname = 'entity_deletion_stage_shape') THEN
+        ALTER TABLE entity_deletion_stage_history ADD CONSTRAINT entity_deletion_stage_shape CHECK (
+        (is_current = (status <> 'completed'))
+        AND ((status = 'completed') = (finished_at IS NOT NULL))
+        AND ((status = 'pending') = (started_at IS NULL))
+        AND (finished_at IS NULL OR finished_at >= started_at)
+        );
+    END IF;
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint
+                   WHERE conrelid = 'entity_deletion_stage_history'::regclass
+                     AND conname = 'entity_deletion_stage_shape' AND convalidated) THEN
+        RAISE EXCEPTION 'entity_deletion_stage_shape must be validated';
+    END IF;
+END $$;
+
+DO $$
+BEGIN
+    IF (SELECT count(*) FROM _operator_global_tables WHERE table_name IN (
+        'entity_deletion_requests',
+        'entity_deletion_stage_history',
+        'entity_deletion_approvals',
+        'entity_deletion_executor_heartbeats'
+    )) <> 4 THEN
+        RAISE EXCEPTION 'entity deletion tables must be operator-global';
+    END IF;
+END $$;

scripts/test-user-deletion-demo.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+"""Exercise the actual user-deletion CLI against an explicit disposable database."""
+
+import argparse
+import json
+import os
+import subprocess
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--binary", type=Path, required=True)
+    args = parser.parse_args()
+    database_url = os.environ.get("BUZZ_USER_DELETION_TEST_DATABASE_URL")
+    if not database_url:
+        parser.error("set BUZZ_USER_DELETION_TEST_DATABASE_URL to an isolated, migrated database")
+    environment = dict(os.environ, DATABASE_URL=database_url)
+    binary = str(args.binary.resolve())
+
+    def command(*parts, expected=0):
+        result = subprocess.run(
+            [binary, "user-deletions", *parts],
+            env=environment,
+            capture_output=True,
+            text=True,
+            timeout=90,
+            check=False,
+        )
+        if result.returncode != expected:
+            raise AssertionError(
+                f"{parts}: expected exit {expected}, got {result.returncode}\n"
+                f"{result.stdout[-8000:]}\n{result.stderr[-8000:]}"
+            )
+        if expected != 0:
+            return None
+        return json.loads(result.stdout)
+
+    def prepare():
+        target = "demo:cli-" + uuid.uuid4().hex
+        request = command("submit", "--target", target, "--requested-by", "operator-a")
+        request_id = request["id"]
+        repeated = command("submit", "--target", target, "--requested-by", "operator-a")
+        assert repeated["id"] == request_id
+        command("run", request_id, expected=5)
+        inventory = command("inventory", request_id)
+        assert len(inventory["inventory_manifest"]["communities"]) == 3
+        assert all(c["relationship"] == "participant"
+                   for c in inventory["inventory_manifest"]["communities"])
+        command("approve", request_id, "--approved-by", "operator-a", expected=5)
+        command("approve", request_id, "--approved-by", "operator-b")
+        return request_id
+
+    def finished(request_id):
+        result = command("inspect", request_id)
+        assert result["request"]["status"] == "demo_completed"
+        assert result["request"]["lease_owner"] is None
+        history = result["stage_history"]
+        assert len(history) == 15
+        assert all(not s["is_current"] and s["status"] == "completed" for s in history)
+        assert all(s["metadata_json"]["simulated"] for s in history)
+        assert [s["stage"] for s in history[:4]] == [
+            "gate_user", "gate_community", "gate_community", "gate_community"
+        ]
+        for start, community in [(4, 1), (7, 2), (10, 3)]:
+            assert all(s["target_community_id"] == str(uuid.UUID(int=community))
+                       for s in history[start:start + 3])
+        return result
+
+    stopped_id = prepare()
+    command("run", stopped_id, "--stop-after", "7")
+    stopped = command("inspect", stopped_id)
+    assert stopped["request"]["lease_owner"] is None
+    assert len(stopped["stage_history"]) == 8
+    assert stopped["stage_history"][-1]["status"] == "pending"
+    command("run", stopped_id)
+    completed = finished(stopped_id)
+    assert completed["stage_history"][:7] == stopped["stage_history"][:7]
+    print(f"PASS CLI stop/resume: {stopped_id}", flush=True)
+
+    failed_id = prepare()
+    command("run", failed_id, "--fail-at", "9", expected=1)
+    failed = command("inspect", failed_id)
+    assert len(failed["stage_history"]) == 9
+    assert failed["stage_history"][-1]["status"] == "failed"
+    assert failed["request"]["lease_owner"] is None
+    retry_at = datetime.fromisoformat(failed["request"]["next_attempt_at"].replace("Z", "+00:00"))
+    delay = max(0, (retry_at - datetime.now(timezone.utc)).total_seconds()) + 0.2
+    assert delay < 35, "unexpected retry delay"
+    time.sleep(delay)
+    command("run", failed_id)
+    recovered = finished(failed_id)
+    assert recovered["stage_history"][:8] == failed["stage_history"][:8]
+    assert recovered["stage_history"][8]["id"] == failed["stage_history"][8]["id"]
+    assert recovered["request"]["lease_generation"] == 2
+    print(f"PASS CLI failure in B/retry: {failed_id}", flush=True)
+
+    interrupted_id = prepare()
+    process = subprocess.Popen(
+        [binary, "user-deletions", "run", interrupted_id, "--step-delay-ms", "30000"],
+        env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+    )
+    try:
+        # Cold database connection setup has its own retry budget. Once the
+        # stage starts, require renewal within two heartbeat intervals.
+        deadline = time.monotonic() + 45
+        initial = None
+        while time.monotonic() < deadline:
+            if process.poll() is not None:
+                stdout, stderr = process.communicate(timeout=5)
+                raise AssertionError(
+                    f"worker exited before heartbeat: {process.returncode}\n"
+                    f"{stdout[-8000:]}\n{stderr[-8000:]}"
+                )
+            current = command("inspect", interrupted_id)
+            if current["stage_history"][0]["status"] == "running":
+                if initial is None:
+                    initial = current
+                    deadline = time.monotonic() + 20
+                elif current["request"]["lease_until"] != initial["request"]["lease_until"]:
+                    break
+            time.sleep(0.25)
+        else:
+            raise AssertionError("worker did not renew its lease while a dummy action ran")
+        command("run", interrupted_id, expected=5)
+        process.terminate()
+        stdout, stderr = process.communicate(timeout=15)
+        assert process.returncode == 0, (stdout[-8000:], stderr[-8000:])
+    finally:
+        if process.poll() is None:
+            process.kill()
+            process.communicate(timeout=10)
+    interrupted = command("inspect", interrupted_id)
+    assert interrupted["request"]["lease_owner"] is None
+    assert interrupted["stage_history"][0]["is_current"]
+    command("run", interrupted_id)
+    resumed = finished(interrupted_id)
+    assert resumed["request"]["lease_generation"] == 2
+    assert resumed["stage_history"][0]["id"] == interrupted["stage_history"][0]["id"]
+    assert resumed["stage_history"][0]["started_at"] == interrupted["stage_history"][0]["started_at"]
+    print(f"PASS CLI heartbeat/SIGTERM/resume: {interrupted_id}", flush=True)
+    print("PASS: all actions were simulated; no owned-community path was exercised.")
+
+
+if __name__ == "__main__":
+    main()