From 9551934a42739b558421debea949fe5f11f3a3bb Mon Sep 17 00:00:00 2001
From: Ravneet Arora <rarora@squareup.com>
Date: Fri, 30 Oct 2026 17:46:54 -0400
Subject: [PATCH 1/2] Add checkpointed agent database cleanup handler

Signed-off-by: Ravneet Arora <rarora@squareup.com>
---
 .../store/user_deletion/agents/delivery.rs    |  73 ++++
 .../src/store/user_deletion/agents/mod.rs     | 255 +++++++++++
 .../src/store/user_deletion/agents/model.rs   | 149 +++++++
 .../store/user_deletion/agents/operations.rs  |  76 ++++
 .../store/user_deletion/agents/workflows.rs   |  63 +++
 .../src/store/user_deletion/discovery.rs      |  15 +-
 .../user_deletion/discovery_postgres_tests.rs |   3 +-
 .../src/store/user_deletion/execution.rs      | 105 +++++
 .../src/store/user_deletion/inventory.rs      |  78 ++++
 .../src/store/user_deletion/messages/mod.rs   | 194 ++++-----
 crates/buzz-db/src/store/user_deletion/mod.rs |   3 +
 crates/buzz-deletion/src/user.rs              |   3 +
 crates/buzz-deletion/src/user/agents.rs       |  91 ++++
 .../buzz-deletion/src/user/agents/README.md   | 156 +++++++
 .../src/user/agents/postgres_tests.rs         | 403 ++++++++++++++++++
 .../postgres_tests/recovery_postgres_tests.rs | 124 ++++++
 .../postgres_tests/safety_postgres_tests.rs   | 250 +++++++++++
 .../postgres_tests/safety_postgres_tests.rs   |   4 +-
 docs/operator-user-deletion-inventory.md      |  10 +-
 19 files changed, 1946 insertions(+), 109 deletions(-)
 create mode 100644 crates/buzz-db/src/store/user_deletion/agents/delivery.rs
 create mode 100644 crates/buzz-db/src/store/user_deletion/agents/mod.rs
 create mode 100644 crates/buzz-db/src/store/user_deletion/agents/model.rs
 create mode 100644 crates/buzz-db/src/store/user_deletion/agents/operations.rs
 create mode 100644 crates/buzz-db/src/store/user_deletion/agents/workflows.rs
 create mode 100644 crates/buzz-db/src/store/user_deletion/execution.rs
 create mode 100644 crates/buzz-deletion/src/user/agents.rs
 create mode 100644 crates/buzz-deletion/src/user/agents/README.md
 create mode 100644 crates/buzz-deletion/src/user/agents/postgres_tests.rs
 create mode 100644 crates/buzz-deletion/src/user/agents/postgres_tests/recovery_postgres_tests.rs
 create mode 100644 crates/buzz-deletion/src/user/agents/postgres_tests/safety_postgres_tests.rs

diff --git a/crates/buzz-db/src/store/user_deletion/agents/delivery.rs b/crates/buzz-db/src/store/user_deletion/agents/delivery.rs
new file mode 100644
index 00000000000..c23ac22aa25
--- /dev/null
+++ b/crates/buzz-db/src/store/user_deletion/agents/delivery.rs
@@ -0,0 +1,73 @@
+use super::*;
+
+pub(super) async fn purge(tx: &mut AdmittedTx, subject: &[u8], limit: u16) -> Result<bool> {
+    let community = tx.community();
+    let active: bool = sqlx::query_scalar(
+        "SELECT EXISTS (
+         SELECT 1 FROM push_wake_outbox WHERE community_id=$1 AND author=$2 AND lease_until>clock_timestamp()
+         UNION ALL SELECT 1 FROM operator_listener_outbox WHERE community_id=$1
+          AND (target_pubkey=$2 OR listener_pubkey=$2) AND lease_until>clock_timestamp())",
+    )
+    .bind(community.as_uuid())
+    .bind(subject)
+    .fetch_one(tx.conn())
+    .await?;
+    if active {
+        return Err(safety("agent delivery workers must drain before deletion"));
+    }
+    for query in [
+        "DELETE FROM push_wake_outbox WHERE (community_id,id) IN
+         (SELECT community_id,id FROM push_wake_outbox WHERE community_id=$1 AND author=$2 ORDER BY id LIMIT $3)",
+        "DELETE FROM operator_listener_outbox WHERE community_id=$1 AND id IN
+         (SELECT id FROM operator_listener_outbox WHERE community_id=$1 AND (target_pubkey=$2 OR listener_pubkey=$2)
+          ORDER BY id LIMIT $3)",
+        "DELETE FROM push_leases WHERE (community_id,author,installation_id) IN
+         (SELECT community_id,author,installation_id FROM push_leases WHERE community_id=$1 AND author=$2
+          ORDER BY installation_id LIMIT $3)",
+    ] {
+        if sqlx::query(query)
+            .bind(community.as_uuid())
+            .bind(subject)
+            .bind(i64::from(limit))
+            .execute(tx.conn())
+            .await?
+            .rows_affected() != 0
+        {
+            return Ok(false);
+        }
+    }
+    let subscription: Option<String> = sqlx::query_scalar(
+        "SELECT id FROM subscriptions WHERE community_id=$1 AND owner_pubkey=$2 ORDER BY id LIMIT 1 FOR UPDATE",
+    )
+    .bind(community.as_uuid())
+    .bind(subject)
+    .fetch_optional(tx.conn())
+    .await?;
+    if let Some(id) = subscription {
+        let removed = sqlx::query(
+            "DELETE FROM delivery_log WHERE community_id=$1 AND (delivered_at,id) IN
+             (SELECT delivered_at,id FROM delivery_log WHERE community_id=$1 AND subscription_id=$2
+              ORDER BY delivered_at,id LIMIT $3)",
+        )
+        .bind(community.as_uuid())
+        .bind(&id)
+        .bind(i64::from(limit))
+        .execute(tx.conn())
+        .await?
+        .rows_affected();
+        if removed == 0 {
+            sqlx::query(
+                "DELETE FROM subscriptions WHERE community_id=$1 AND id=$2 AND owner_pubkey=$3",
+            )
+            .bind(community.as_uuid())
+            .bind(id)
+            .bind(subject)
+            .execute(tx.conn())
+            .await?;
+        }
+        return Ok(false);
+    }
+    // Deployment-global listener registrations are retained for finalization,
+    // because the same agent key may legitimately participate in other tenants.
+    messages::purge_event_kinds(tx, subject, &[30350], limit).await
+}
diff --git a/crates/buzz-db/src/store/user_deletion/agents/mod.rs b/crates/buzz-db/src/store/user_deletion/agents/mod.rs
new file mode 100644
index 00000000000..6f7a132e274
--- /dev/null
+++ b/crates/buzz-db/src/store/user_deletion/agents/mod.rs
@@ -0,0 +1,255 @@
+//! Bounded agent database operations and atomic per-agent checkpoints.
+use super::{execution, messages, UserDeletionInventory, UserDeletionStore};
+use crate::entity_deletion::{lock_owned, validate_approved_inventory, EntityLease, EntityStage};
+use crate::{AdmittedTx, Result};
+use buzz_core::CommunityId;
+use chrono::{DateTime, Utc};
+use execution::safety;
+use uuid::Uuid;
+
+mod delivery;
+mod model;
+mod operations;
+mod workflows;
+pub use model::{AgentDbPhase, AgentDbProgress, AgentDbStatus, AgentsDbBatch, AgentsDbProgress};
+
+impl UserDeletionStore {
+    /// Execute one bounded operation for the next unfinished approved agent.
+    ///
+    /// `advance` is the trusted handler's phase policy, called only when the
+    /// current operation is exhausted. SQL effects and policy progress commit
+    /// together. This does not select workflow stages or release ownership.
+    /// Profiles/configuration, credentials, memberships, media and hosted resources
+    /// are deliberately retained for finalization. Callers must gate and drain
+    /// all affected writers before using this standalone operation.
+    pub async fn purge_agents_db_batch<F>(
+        &self,
+        lease: &EntityLease,
+        stage_id: Uuid,
+        community: CommunityId,
+        relay_pubkey: &[u8; 32],
+        batch_size: u16,
+        advance: F,
+    ) -> Result<AgentsDbBatch>
+    where
+        F: FnOnce(&mut AgentDbProgress, DateTime<Utc>),
+    {
+        if !(1..=messages::MAX_BATCH_SIZE).contains(&batch_size) {
+            return Err(safety("agent batch size must be 1..=100"));
+        }
+        let (mut tx, stage, inventory, owner) = execution::begin(
+            self,
+            lease,
+            stage_id,
+            community,
+            relay_pubkey,
+            "purge_agents_db",
+        )
+        .await?;
+        let agents = inventory.agents(*community.as_uuid(), &owner)?;
+        if agents.iter().any(|key| key == &hex::encode(relay_pubkey)) {
+            return Err(safety("cannot purge the relay signing identity"));
+        }
+        // The approved list is stable. Never silently include newly owned agents
+        // or falsely finish an old inventory which omitted them.
+        let unexpected: bool = sqlx::query_scalar(
+            "SELECT EXISTS (SELECT 1 FROM users WHERE community_id=$1
+             AND agent_owner_pubkey=decode($2,'hex') AND NOT (encode(pubkey,'hex')=ANY($3)))",
+        )
+        .bind(community.as_uuid())
+        .bind(&owner)
+        .bind(&agents)
+        .fetch_one(tx.conn())
+        .await?;
+        if unexpected {
+            return Err(safety(
+                "owned agents changed since approval; reconcile the inventory",
+            ));
+        }
+        let mut progress = AgentsDbProgress::read(&stage, &agents)?;
+        if let Some(agent) = progress.next_agent(&agents).map(str::to_owned) {
+            validate_agent(&mut tx, lease, &owner, &agent).await?;
+            progress.current_agent = Some(agent.clone());
+            let item = progress
+                .agents
+                .get_mut(&agent)
+                .ok_or_else(|| safety("missing agent progress"))?;
+            item.status = AgentDbStatus::Running;
+            item.last_error = None;
+            let subject = hex::decode(&agent).map_err(|_| safety("invalid agent key"))?;
+            let exhausted = match item.phase {
+                AgentDbPhase::Messages => {
+                    messages::purge_for_subject(
+                        &mut tx,
+                        community,
+                        &subject,
+                        &agent,
+                        relay_pubkey,
+                        batch_size,
+                    )
+                    .await?
+                    .exhausted
+                }
+                AgentDbPhase::PrivateState => {
+                    operations::private_state(&mut tx, &subject, batch_size).await?
+                }
+                AgentDbPhase::JobRecords => {
+                    messages::purge_event_kinds(
+                        &mut tx,
+                        &subject,
+                        operations::JOB_KINDS,
+                        batch_size,
+                    )
+                    .await?
+                }
+                AgentDbPhase::OwnedWorkflows => {
+                    workflows::purge(&mut tx, &subject, batch_size).await?
+                }
+                AgentDbPhase::DeliveryState => {
+                    delivery::purge(&mut tx, &subject, batch_size).await?
+                }
+                AgentDbPhase::VerifyDbContent => {
+                    operations::verify(&mut tx, &subject, &agent, relay_pubkey).await?;
+                    true
+                }
+            };
+            if exhausted {
+                let now = sqlx::query_scalar("SELECT clock_timestamp()")
+                    .fetch_one(tx.conn())
+                    .await?;
+                advance(item, now);
+            }
+            if item.status == AgentDbStatus::Completed {
+                progress.current_agent = None;
+            }
+        }
+        let exhausted = progress.next_agent(&agents).is_none();
+        // Validate the handler's serialized state before committing it.
+        let mut check = stage.clone();
+        check.metadata_json["progress"] = serde_json::to_value(&progress)?;
+        AgentsDbProgress::read(&check, &agents)?;
+        save_progress(&mut tx, stage_id, &progress).await?;
+        execution::verify_live(&mut tx, lease).await?;
+        tx.commit().await?;
+        Ok(AgentsDbBatch {
+            progress,
+            exhausted,
+        })
+    }
+
+    /// Save the current agent's failure and the normal stage/request retry state
+    /// in one transaction. An expired/superseded worker cannot record failure.
+    pub async fn fail_agents_db_stage(
+        &self,
+        lease: &EntityLease,
+        stage_id: Uuid,
+        message: &str,
+    ) -> Result<()> {
+        let mut tx = self.pool.begin().await?;
+        let request = lock_owned(&mut tx, lease).await?;
+        validate_approved_inventory(&mut tx, &request).await?;
+        let stage: EntityStage = sqlx::query_as(
+            "SELECT id,request_id,target_community_id,stage,status,is_current,started_at,
+             finished_at,metadata_json,last_error,updated_at FROM entity_deletion_stage_history
+             WHERE id=$1 FOR UPDATE",
+        )
+        .bind(stage_id)
+        .fetch_one(&mut *tx)
+        .await?;
+        let community = stage
+            .target_community_id
+            .ok_or_else(|| safety("missing agent community"))?;
+        let inventory = UserDeletionInventory::read(&request)?;
+        if stage.request_id != lease.request_id
+            || stage.stage != "purge_agents_db"
+            || stage.status != "running"
+            || !stage.is_current
+            || stage.metadata_json["inputs"]
+                != inventory.inputs(&request.target, Some(community))?
+        {
+            return Err(safety("expected current running agent stage"));
+        }
+        let agents = inventory.agents(community, &request.target)?;
+        let mut progress = AgentsDbProgress::read(&stage, &agents)?;
+        let message: String = message.chars().take(2048).collect();
+        if let Some(agent) = progress.next_agent(&agents).map(str::to_owned) {
+            progress.current_agent = Some(agent.clone());
+            let item = progress
+                .agents
+                .get_mut(&agent)
+                .ok_or_else(|| safety("missing agent progress"))?;
+            item.status = AgentDbStatus::Failed;
+            item.last_error = Some(message.clone());
+        }
+        sqlx::query(
+            "UPDATE entity_deletion_stage_history SET status='failed',last_error=$2,
+             metadata_json=jsonb_set(metadata_json,'{progress}',$3),updated_at=clock_timestamp() WHERE id=$1",
+        )
+        .bind(stage_id)
+        .bind(&message)
+        .bind(serde_json::to_value(progress)?)
+        .execute(&mut *tx)
+        .await?;
+        sqlx::query(
+            "UPDATE entity_deletion_requests SET status='retry_wait',retry_count=retry_count+1,
+             last_error=$2,next_attempt_at=clock_timestamp()+interval '30 seconds',updated_at=clock_timestamp()
+             WHERE id=$1",
+        )
+        .bind(lease.request_id)
+        .bind(message)
+        .execute(&mut *tx)
+        .await?;
+        lock_owned(&mut tx, lease).await?;
+        tx.commit().await?;
+        Ok(())
+    }
+}
+
+async fn validate_agent(
+    tx: &mut AdmittedTx,
+    lease: &EntityLease,
+    owner: &str,
+    agent: &str,
+) -> Result<()> {
+    let community = tx.community();
+    let proof: Option<(Uuid, Option<String>)> = sqlx::query_as(
+        "SELECT request_id,source_owner_pubkey FROM user_deletion_fences
+         WHERE scope_community_id=$1 AND pubkey=$2",
+    )
+    .bind(community.as_uuid())
+    .bind(agent)
+    .fetch_optional(tx.conn())
+    .await?;
+    if proof != Some((lease.request_id, Some(owner.to_owned()))) {
+        return Err(safety(
+            "approved agent must have this request's community fence and owner provenance",
+        ));
+    }
+    // Missing rows are allowed on recovery; conflicting live ownership is not.
+    let live: Option<Option<Vec<u8>>> = sqlx::query_scalar(
+        "SELECT agent_owner_pubkey FROM users WHERE community_id=$1 AND pubkey=decode($2,'hex')",
+    )
+    .bind(community.as_uuid())
+    .bind(agent)
+    .fetch_optional(tx.conn())
+    .await?;
+    let owner = hex::decode(owner).map_err(|_| safety("invalid owner"))?;
+    if live.is_some_and(|live| live.as_deref() != Some(owner.as_slice())) {
+        return Err(safety(
+            "agent ownership differs from the approved deletion scope",
+        ));
+    }
+    Ok(())
+}
+
+async fn save_progress(
+    tx: &mut AdmittedTx,
+    stage: Uuid,
+    progress: &AgentsDbProgress,
+) -> Result<()> {
+    sqlx::query(
+        "UPDATE entity_deletion_stage_history SET metadata_json=jsonb_set(metadata_json,'{progress}',$2),
+         updated_at=clock_timestamp() WHERE id=$1",
+    ).bind(stage).bind(serde_json::to_value(progress)?).execute(tx.conn()).await?;
+    Ok(())
+}
diff --git a/crates/buzz-db/src/store/user_deletion/agents/model.rs b/crates/buzz-db/src/store/user_deletion/agents/model.rs
new file mode 100644
index 00000000000..ac4340527f6
--- /dev/null
+++ b/crates/buzz-db/src/store/user_deletion/agents/model.rs
@@ -0,0 +1,149 @@
+//! Saved, bounded per-agent progress. Phase routing belongs to buzz-deletion.
+use super::*;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+
+/// Internal database cleanup operation, not a workflow stage.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AgentDbPhase {
+    /// Reuse message removal and dependent-record cleanup.
+    Messages,
+    /// Memory, metrics, lists, reminders and read progress.
+    PrivateState,
+    /// Agent-authored job protocol events.
+    JobRecords,
+    /// Agent-owned workflow definitions and execution records.
+    OwnedWorkflows,
+    /// Subscriptions, push state and recipient delivery records.
+    DeliveryState,
+    /// Independent absence checks for the supported database scope.
+    VerifyDbContent,
+}
+
+/// Saved outcome for a single approved agent.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AgentDbStatus {
+    /// No work started.
+    Pending,
+    /// Cleanup is in progress.
+    Running,
+    /// Last attempt failed; phase is the resume position.
+    Failed,
+    /// Supported database content was verified absent.
+    Completed,
+}
+
+/// One agent's durable resume position; never contains message bodies.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct AgentDbProgress {
+    /// Current state of this agent's database cleanup.
+    pub status: AgentDbStatus,
+    /// Internal phase to resume.
+    pub phase: AgentDbPhase,
+    /// Database-clock time of successful verification.
+    pub completed_at: Option<DateTime<Utc>>,
+    /// Most recent failure, bounded to 2048 characters.
+    pub last_error: Option<String>,
+}
+
+/// Progress stored under `entity_deletion_stage_history.metadata_json.progress`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct AgentsDbProgress {
+    /// Serialization contract version.
+    pub version: u8,
+    /// Current agent, or null before execution/after all agents finish.
+    pub current_agent: Option<String>,
+    /// Status for every approved agent in this community.
+    pub agents: BTreeMap<String, AgentDbProgress>,
+}
+
+impl AgentsDbProgress {
+    pub(super) fn read(stage: &EntityStage, agents: &[String]) -> Result<Self> {
+        let progress: Self = match stage.metadata_json.get("progress") {
+            Some(value) => serde_json::from_value(value.clone())
+                .map_err(|_| safety("invalid saved agent progress"))?,
+            None => Self {
+                version: 1,
+                current_agent: None,
+                agents: agents
+                    .iter()
+                    .map(|key| {
+                        (
+                            key.clone(),
+                            AgentDbProgress {
+                                status: AgentDbStatus::Pending,
+                                phase: AgentDbPhase::Messages,
+                                completed_at: None,
+                                last_error: None,
+                            },
+                        )
+                    })
+                    .collect(),
+            },
+        };
+        if progress.version != 1
+            || progress.agents.len() != agents.len()
+            || agents.iter().any(|key| !progress.agents.contains_key(key))
+        {
+            return Err(safety(
+                "saved agent progress differs from approved inventory",
+            ));
+        }
+        let first = progress.next_agent(agents);
+        if progress
+            .current_agent
+            .as_deref()
+            .is_some_and(|key| Some(key) != first)
+        {
+            return Err(safety(
+                "saved current agent is not the next unfinished agent",
+            ));
+        }
+        let mut unfinished = false;
+        for key in agents {
+            let item = &progress.agents[key];
+            if item.status == AgentDbStatus::Completed {
+                if unfinished
+                    || item.phase != AgentDbPhase::VerifyDbContent
+                    || item.completed_at.is_none()
+                    || item.last_error.is_some()
+                {
+                    return Err(safety("invalid completed agent progress"));
+                }
+            } else {
+                if item.completed_at.is_some()
+                    || (unfinished && item.status != AgentDbStatus::Pending)
+                    || (item.status == AgentDbStatus::Pending
+                        && item.phase != AgentDbPhase::Messages)
+                    || (matches!(item.status, AgentDbStatus::Running | AgentDbStatus::Failed)
+                        && progress.current_agent.as_deref() != Some(key.as_str()))
+                {
+                    return Err(safety("invalid unfinished agent progress"));
+                }
+                unfinished = true;
+            }
+        }
+        Ok(progress)
+    }
+
+    pub(super) fn next_agent<'a>(&self, agents: &'a [String]) -> Option<&'a str> {
+        agents
+            .iter()
+            .find(|key| self.agents[*key].status != AgentDbStatus::Completed)
+            .map(String::as_str)
+    }
+}
+
+/// Result of one bounded operation; the caller repeats while not exhausted.
+#[derive(Debug, Serialize)]
+pub struct AgentsDbBatch {
+    /// Exact progress committed with this batch.
+    pub progress: AgentsDbProgress,
+    /// All approved agents completed supported database cleanup.
+    pub exhausted: bool,
+}
diff --git a/crates/buzz-db/src/store/user_deletion/agents/operations.rs b/crates/buzz-db/src/store/user_deletion/agents/operations.rs
new file mode 100644
index 00000000000..f020f26fa9e
--- /dev/null
+++ b/crates/buzz-db/src/store/user_deletion/agents/operations.rs
@@ -0,0 +1,76 @@
+use super::*;
+
+// Profiles, emoji sets, owner-authored configuration and shared artifacts are
+// deliberately excluded. These remain inputs to media/identity finalization.
+pub(super) const PRIVATE_KINDS: &[i32] = &[
+    3, 10000, 10001, 10002, 10003, 30000, 30003, 30315, 30078, 30174, 44200, 30300,
+];
+pub(super) const JOB_KINDS: &[i32] = &[43001, 43002, 43003, 43004, 43005, 43006];
+
+pub(super) async fn private_state(tx: &mut AdmittedTx, subject: &[u8], limit: u16) -> Result<bool> {
+    if !messages::purge_event_kinds(tx, subject, PRIVATE_KINDS, limit).await? {
+        return Ok(false);
+    }
+    let community = tx.community();
+    let removed = sqlx::query(
+        "DELETE FROM personal_read_frontiers WHERE (community_id,actor,channel_id,root_id) IN
+         (SELECT community_id,actor,channel_id,root_id FROM personal_read_frontiers
+          WHERE community_id=$1 AND actor=$2 ORDER BY channel_id,root_id LIMIT $3)",
+    )
+    .bind(community.as_uuid())
+    .bind(subject)
+    .bind(i64::from(limit))
+    .execute(tx.conn())
+    .await?
+    .rows_affected();
+    if removed != 0 {
+        return Ok(false);
+    }
+    sqlx::query("DELETE FROM personal_read_accounts WHERE community_id=$1 AND actor=$2")
+        .bind(community.as_uuid())
+        .bind(subject)
+        .execute(tx.conn())
+        .await?;
+    Ok(true)
+}
+
+pub(super) async fn verify(
+    tx: &mut AdmittedTx,
+    subject: &[u8],
+    subject_hex: &str,
+    relay: &[u8; 32],
+) -> Result<()> {
+    if messages::has_remaining(tx, subject, subject_hex, relay).await? {
+        return Err(safety("agent verification found remaining message data"));
+    }
+    let community = tx.community();
+    let kinds = PRIVATE_KINDS
+        .iter()
+        .chain(JOB_KINDS)
+        .copied()
+        .chain([30620, 30350])
+        .collect::<Vec<_>>();
+    let remaining: bool = sqlx::query_scalar(
+        "SELECT EXISTS (
+         SELECT 1 FROM events WHERE community_id=$1 AND pubkey=$2 AND kind=ANY($3)
+         UNION ALL SELECT 1 FROM personal_read_accounts WHERE community_id=$1 AND actor=$2
+         UNION ALL SELECT 1 FROM personal_read_frontiers WHERE community_id=$1 AND actor=$2
+         UNION ALL SELECT 1 FROM workflows WHERE community_id=$1 AND owner_pubkey=$2
+         UNION ALL SELECT 1 FROM subscriptions WHERE community_id=$1 AND owner_pubkey=$2
+         UNION ALL SELECT 1 FROM push_leases WHERE community_id=$1 AND author=$2
+         UNION ALL SELECT 1 FROM push_wake_outbox WHERE community_id=$1 AND author=$2
+         UNION ALL SELECT 1 FROM operator_listener_outbox WHERE community_id=$1
+           AND (target_pubkey=$2 OR listener_pubkey=$2))",
+    )
+    .bind(community.as_uuid())
+    .bind(subject)
+    .bind(kinds)
+    .fetch_one(tx.conn())
+    .await?;
+    if remaining {
+        return Err(safety(
+            "agent verification found remaining supported database content",
+        ));
+    }
+    Ok(())
+}
diff --git a/crates/buzz-db/src/store/user_deletion/agents/workflows.rs b/crates/buzz-db/src/store/user_deletion/agents/workflows.rs
new file mode 100644
index 00000000000..4b09eb55985
--- /dev/null
+++ b/crates/buzz-db/src/store/user_deletion/agents/workflows.rs
@@ -0,0 +1,63 @@
+use super::*;
+
+pub(super) async fn purge(tx: &mut AdmittedTx, subject: &[u8], limit: u16) -> Result<bool> {
+    let community = tx.community();
+    let workflow: Option<(Uuid, bool)> = sqlx::query_as(
+        "SELECT id, definition::text ~* 'https?://|/media/' FROM workflows
+         WHERE community_id=$1 AND owner_pubkey=$2 ORDER BY id LIMIT 1 FOR UPDATE",
+    )
+    .bind(community.as_uuid())
+    .bind(subject)
+    .fetch_optional(tx.conn())
+    .await?;
+    if let Some((id, has_links)) = workflow {
+        if has_links {
+            return Err(safety("agent workflow has link references; preserve media/resource inputs before deletion"));
+        }
+        let active: bool = sqlx::query_scalar(
+            "SELECT EXISTS (SELECT 1 FROM workflow_runs WHERE community_id=$1 AND workflow_id=$2
+             AND status NOT IN ('completed','failed','cancelled'))",
+        )
+        .bind(community.as_uuid())
+        .bind(id)
+        .fetch_one(tx.conn())
+        .await?;
+        if active {
+            return Err(safety(
+                "agent-owned workflow runs must stop before deletion",
+            ));
+        }
+        // Child-first explicit deletes. Each call commits at most one bounded
+        // child batch; the workflow row remains the recovery anchor until empty.
+        for query in [
+            "DELETE FROM scheduled_workflow_fires WHERE (community_id,workflow_id,scheduled_for) IN
+             (SELECT community_id,workflow_id,scheduled_for FROM scheduled_workflow_fires
+              WHERE community_id=$1 AND workflow_id=$2 ORDER BY scheduled_for LIMIT $3)",
+            "DELETE FROM workflow_approvals WHERE (community_id,token) IN
+             (SELECT community_id,token FROM workflow_approvals WHERE community_id=$1
+              AND (workflow_id=$2 OR run_id IN (SELECT id FROM workflow_runs WHERE community_id=$1 AND workflow_id=$2))
+              ORDER BY token LIMIT $3)",
+            "DELETE FROM workflow_runs WHERE (community_id,id) IN
+             (SELECT community_id,id FROM workflow_runs WHERE community_id=$1 AND workflow_id=$2 ORDER BY id LIMIT $3)",
+        ] {
+            if sqlx::query(query)
+                .bind(community.as_uuid())
+                .bind(id)
+                .bind(i64::from(limit))
+                .execute(tx.conn())
+                .await?
+                .rows_affected() != 0
+            {
+                return Ok(false);
+            }
+        }
+        sqlx::query("DELETE FROM workflows WHERE community_id=$1 AND id=$2 AND owner_pubkey=$3")
+            .bind(community.as_uuid())
+            .bind(id)
+            .bind(subject)
+            .execute(tx.conn())
+            .await?;
+        return Ok(false);
+    }
+    messages::purge_event_kinds(tx, subject, &[30620], limit).await
+}
diff --git a/crates/buzz-db/src/store/user_deletion/discovery.rs b/crates/buzz-db/src/store/user_deletion/discovery.rs
index 27358e98ef8..f6cae7ad82f 100644
--- a/crates/buzz-db/src/store/user_deletion/discovery.rs
+++ b/crates/buzz-db/src/store/user_deletion/discovery.rs
@@ -60,10 +60,11 @@ struct Membership {
     community_id: Uuid,
     host: String,
     role: Option<String>,
+    agents: Vec<String>,
 }
 
 impl UserDeletionStore {
-    /// Generate only community identities and relationships for demo or real input.
+    /// Generate community relationships and their currently owned agent keys.
     /// `demo` selects the existing fixture; other inputs accept hex or npub.
     pub async fn generate_inventory(&self, pubkey: &str) -> Result<Value> {
         let mut tx = self.pool.begin().await?;
@@ -101,7 +102,10 @@ pub(super) async fn generate_inventory_on(
                  SELECT community_id FROM relay_members WHERE pubkey = $1 \
                  UNION SELECT community_id FROM current_channels \
              ) \
-             SELECT c.id AS community_id, c.host, m.role \
+             SELECT c.id AS community_id, c.host, m.role, \
+                 ARRAY(SELECT encode(u.pubkey, 'hex') FROM users u \
+                       WHERE u.community_id = c.id AND u.agent_owner_pubkey = decode($1, 'hex') \
+                       ORDER BY u.pubkey LIMIT 1025) AS agents \
              FROM matched_communities matched \
              JOIN communities c ON c.id = matched.community_id \
              LEFT JOIN relay_members m ON m.community_id = c.id AND m.pubkey = $1 \
@@ -116,9 +120,15 @@ pub(super) async fn generate_inventory_on(
             "inventory exceeds {MAX_COMMUNITIES} communities; no partial inventory returned"
         )));
     }
+    let mut agent_count = 0;
     let communities = rows
         .into_iter()
         .map(|row| {
+            agent_count += row.agents.len();
+            if row.agents.len() > super::inventory::MAX_AGENTS_PER_COMMUNITY
+                || agent_count > super::inventory::MAX_INVENTORY_AGENTS {
+                return Err(DbError::DeletionSafety("agent inventory exceeds its bounded scope".into()));
+            }
             let relationship = match row.role.as_deref() {
                 Some("owner") => "owner",
                 Some("admin" | "member") | None => "participant",
@@ -132,6 +142,7 @@ pub(super) async fn generate_inventory_on(
                 "community_id": row.community_id,
                 "label": row.host,
                 "relationship": relationship,
+                "agents": row.agents.into_iter().map(|pubkey| json!({"pubkey": pubkey})).collect::<Vec<_>>(),
             }))
         })
         .collect::<Result<Vec<_>>>()?;
diff --git a/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs
index 00b9c9df9eb..0eddda0ea57 100644
--- a/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs
+++ b/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs
@@ -128,7 +128,7 @@ async fn channel_memberships_discover_participants_deduplicate_and_preserve_owne
                 "participant"
             }
         );
-        assert_eq!(actual.as_object().unwrap().len(), 3);
+        assert_eq!(actual.as_object().unwrap().len(), 4);
     }
     let npub = nostr::PublicKey::parse(PUBKEY)
         .unwrap()
@@ -179,6 +179,7 @@ async fn current_inventory_includes_all_lifecycle_states_and_excludes_removed_me
         expected.push(json!({
             "community_id": id, "label": host,
             "relationship": if role == "owner" { "owner" } else { "participant" },
+            "agents": [],
         }));
     }
     let former = Uuid::new_v4();
diff --git a/crates/buzz-db/src/store/user_deletion/execution.rs b/crates/buzz-db/src/store/user_deletion/execution.rs
new file mode 100644
index 00000000000..cade06327c3
--- /dev/null
+++ b/crates/buzz-db/src/store/user_deletion/execution.rs
@@ -0,0 +1,105 @@
+//! Common transaction admission for standalone user and agent cleanup.
+use super::{normalize_user_deletion_pubkey, UserDeletionInventory, UserDeletionStore};
+use crate::entity_deletion::{lock_owned, validate_approved_inventory, EntityLease, EntityStage};
+use crate::{AdmittedTx, DbError, Result};
+use buzz_core::CommunityId;
+use uuid::Uuid;
+
+pub(super) fn safety(message: &str) -> DbError {
+    DbError::DeletionSafety(message.into())
+}
+
+pub(super) async fn begin(
+    store: &UserDeletionStore,
+    lease: &EntityLease,
+    stage_id: Uuid,
+    community: CommunityId,
+    relay_pubkey: &[u8; 32],
+    stage_name: &str,
+) -> Result<(AdmittedTx, EntityStage, UserDeletionInventory, String)> {
+    let mut tx = store.pool.begin().await?;
+    sqlx::query("SET LOCAL statement_timeout = '10s'")
+        .execute(&mut *tx)
+        .await?;
+    sqlx::query("SET LOCAL lock_timeout = '2s'")
+        .execute(&mut *tx)
+        .await?;
+    // Existing lock order: community, request, identities. A shared lock
+    // followed by an upgrade would deadlock two deletion workers.
+    sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
+        .bind(community.as_uuid())
+        .execute(&mut *tx)
+        .await?;
+    crate::deletion::guard_community_write(&mut tx, community).await?;
+    let request = lock_owned(&mut tx, lease).await?;
+    validate_approved_inventory(&mut tx, &request).await?;
+    let inventory = UserDeletionInventory::read(&request)?;
+    let expected_inputs = inventory.inputs(&request.target, Some(*community.as_uuid()))?;
+    let stage: EntityStage = sqlx::query_as(
+        "SELECT id, request_id, target_community_id, stage, status, is_current,
+             started_at, finished_at, metadata_json, last_error, updated_at
+             FROM entity_deletion_stage_history WHERE id = $1 FOR UPDATE",
+    )
+    .bind(stage_id)
+    .fetch_one(&mut *tx)
+    .await?;
+    if stage.request_id != lease.request_id
+        || stage.stage != stage_name
+        || stage.status != "running"
+        || !stage.is_current
+        || stage.target_community_id != Some(*community.as_uuid())
+        || stage.metadata_json["inputs"] != expected_inputs
+    {
+        return Err(safety(&format!(
+            "expected the current approved {stage_name} stage"
+        )));
+    }
+    let subject_hex = normalize_user_deletion_pubkey(&request.target)?;
+    let subject = hex::decode(&subject_hex).map_err(|_| safety("invalid subject"))?;
+    if subject == relay_pubkey {
+        return Err(safety("cannot purge the relay signing identity"));
+    }
+    let identity = crate::user_deletion_fences::FenceIdentity::user(
+        nostr::PublicKey::from_slice(&subject).map_err(|_| safety("invalid subject"))?,
+    );
+    let fence = crate::user_deletion_fences::find(&mut tx, identity)
+        .await?
+        .ok_or_else(|| safety("user must be fenced before message deletion"))?;
+    if fence.request_id != lease.request_id {
+        return Err(safety("user fence belongs to another deletion request"));
+    }
+    let active: bool = sqlx::query_scalar(
+        "SELECT EXISTS (SELECT 1 FROM community_serving_write_leases
+             WHERE community_id = $1 AND lease_until > clock_timestamp())",
+    )
+    .bind(community.as_uuid())
+    .fetch_one(&mut *tx)
+    .await?;
+    if active {
+        return Err(safety(
+            "external community writes must drain before message deletion",
+        ));
+    }
+    let tx = AdmittedTx::admit(
+        tx,
+        &crate::deletion::DeletionStore::new(store.pool.clone()),
+        community,
+    )
+    .await?;
+    Ok((tx, stage, inventory, subject_hex))
+}
+
+pub(super) async fn verify_live(tx: &mut AdmittedTx, lease: &EntityLease) -> Result<()> {
+    let unexpired: bool = sqlx::query_scalar(
+        "SELECT lease_expires_at > clock_timestamp() FROM entity_deletion_requests WHERE id=$1",
+    )
+    .bind(lease.request_id)
+    .fetch_one(tx.conn())
+    .await?;
+    if !unexpired {
+        return Err(DbError::AccessDenied(
+            "deletion lease expired during cleanup batch".into(),
+        ));
+    }
+    Ok(())
+}
diff --git a/crates/buzz-db/src/store/user_deletion/inventory.rs b/crates/buzz-db/src/store/user_deletion/inventory.rs
index 07941e512cd..93bdbfe6d9d 100644
--- a/crates/buzz-db/src/store/user_deletion/inventory.rs
+++ b/crates/buzz-db/src/store/user_deletion/inventory.rs
@@ -6,6 +6,11 @@ use serde_json::{json, Value};
 use std::collections::HashSet;
 use uuid::Uuid;
 
+/// Bound per-community agent progress stored in one stage row.
+pub const MAX_AGENTS_PER_COMMUNITY: usize = 1024;
+/// Bound the total approved agent scope in a request.
+pub const MAX_INVENTORY_AGENTS: usize = 4096;
+
 fn invalid(message: &str) -> DbError {
     DbError::DeletionSafety(message.into())
 }
@@ -40,6 +45,7 @@ impl UserDeletionInventory {
         }
         let mut seen = HashSet::new();
         let mut communities = Vec::new();
+        let mut agent_count = 0;
         for row in rows {
             if row["relationship"] == "owner" {
                 return Err(invalid("owned-community execution is not supported yet"));
@@ -58,11 +64,28 @@ impl UserDeletionInventory {
             {
                 return Err(invalid("duplicate community or unsupported deletion scope"));
             }
+            if row.get("agents").is_some() {
+                agent_count += read_agents(row, &request.target)?.len();
+                if agent_count > MAX_INVENTORY_AGENTS {
+                    return Err(invalid("inventory exceeds 4096 agents"));
+                }
+            }
             communities.push((id, row.clone()));
         }
         Ok(Self { communities })
     }
 
+    /// Read the exact approved agents for one community. Legacy inventories with
+    /// no agent field remain readable, but cannot authorize agent cleanup.
+    pub fn agents(&self, community: Uuid, owner: &str) -> Result<Vec<String>> {
+        let (_, row) = self
+            .communities
+            .iter()
+            .find(|(id, _)| *id == community)
+            .ok_or_else(|| invalid("agent community is outside inventory"))?;
+        read_agents(row, owner)
+    }
+
     /// Build handler inputs for a request-wide or community-scoped stage.
     pub fn inputs(&self, target: &str, community: Option<Uuid>) -> Result<Value> {
         let community = match community {
@@ -77,3 +100,58 @@ impl UserDeletionInventory {
         Ok(json!({"target": target, "community": community}))
     }
 }
+
+fn read_agents(row: &Value, owner: &str) -> Result<Vec<String>> {
+    let entries = row["agents"].as_array().ok_or_else(|| {
+        invalid("approved inventory has no agent list; prepare and approve agent scope first")
+    })?;
+    if entries.len() > MAX_AGENTS_PER_COMMUNITY {
+        return Err(invalid("inventory exceeds 1024 agents in one community"));
+    }
+    let mut seen = HashSet::new();
+    entries
+        .iter()
+        .map(|entry| {
+            let key = entry["pubkey"]
+                .as_str()
+                .ok_or_else(|| invalid("invalid inventory agent"))?;
+            let normalized = super::normalize_user_deletion_pubkey(key)?;
+            if key != normalized || key == owner || !seen.insert(key) {
+                return Err(invalid(
+                    "agent keys must be canonical, unique and different from the owner",
+                ));
+            }
+            Ok(key.to_owned())
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn agent_scope_requires_an_explicit_bounded_canonical_list() {
+        let owner = "01".repeat(32);
+        let agent = "ab".repeat(32);
+        assert!(read_agents(&json!({}), &owner).is_err());
+        assert!(read_agents(&json!({"agents": []}), &owner)
+            .unwrap()
+            .is_empty());
+        assert_eq!(
+            read_agents(&json!({"agents": [{"pubkey": agent}]}), &owner).unwrap(),
+            vec![agent.clone()]
+        );
+        for entries in [
+            json!([{"pubkey": owner}]),
+            json!([{"pubkey": agent}, {"pubkey": agent}]),
+            json!([{"pubkey": agent.to_uppercase()}]),
+            json!([{"pubkey": "not a public key"}]),
+            json!([{}]),
+            json!(null),
+            json!(vec![json!({"pubkey": agent}); MAX_AGENTS_PER_COMMUNITY + 1]),
+        ] {
+            assert!(read_agents(&json!({"agents": entries}), &owner).is_err());
+        }
+    }
+}
diff --git a/crates/buzz-db/src/store/user_deletion/messages/mod.rs b/crates/buzz-db/src/store/user_deletion/messages/mod.rs
index f7cdca2a3ab..f3cd70f74ea 100644
--- a/crates/buzz-db/src/store/user_deletion/messages/mod.rs
+++ b/crates/buzz-db/src/store/user_deletion/messages/mod.rs
@@ -4,8 +4,8 @@
 //! message cursor. A community deletion lock excludes admitted database writers
 //! for the short batch; it does not replace prior user gating or external drain.
 
-use super::{normalize_user_deletion_pubkey, UserDeletionInventory, UserDeletionStore};
-use crate::entity_deletion::{lock_owned, validate_approved_inventory, EntityLease, EntityStage};
+use super::UserDeletionStore;
+use crate::entity_deletion::EntityLease;
 use crate::{AdmittedTx, DbError, Result};
 use buzz_core::CommunityId;
 use serde::Serialize;
@@ -73,116 +73,58 @@ impl UserDeletionStore {
         if !(1..=MAX_BATCH_SIZE).contains(&batch_size) {
             return Err(safety("message batch size must be 1..=100"));
         }
-        let mut tx = self.pool.begin().await?;
-        sqlx::query("SET LOCAL statement_timeout = '10s'")
-            .execute(&mut *tx)
-            .await?;
-        sqlx::query("SET LOCAL lock_timeout = '2s'")
-            .execute(&mut *tx)
-            .await?;
-        // Existing lock order: community, request, identities. A shared lock
-        // followed by an upgrade would deadlock two deletion workers.
-        sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
-            .bind(community.as_uuid())
-            .execute(&mut *tx)
-            .await?;
-        crate::deletion::guard_community_write(&mut tx, community).await?;
-        let request = lock_owned(&mut tx, lease).await?;
-        validate_approved_inventory(&mut tx, &request).await?;
-        let inventory = UserDeletionInventory::read(&request)?;
-        let expected_inputs = inventory.inputs(&request.target, Some(*community.as_uuid()))?;
-        let stage: EntityStage = sqlx::query_as(
-            "SELECT id, request_id, target_community_id, stage, status, is_current,
-             started_at, finished_at, metadata_json, last_error, updated_at
-             FROM entity_deletion_stage_history WHERE id = $1 FOR UPDATE",
+        let (mut tx, _, _, subject_hex) = super::execution::begin(
+            self,
+            lease,
+            stage_id,
+            community,
+            relay_pubkey,
+            "purge_messages",
         )
-        .bind(stage_id)
-        .fetch_one(&mut *tx)
         .await?;
-        if stage.request_id != lease.request_id
-            || stage.stage != "purge_messages"
-            || stage.status != "running"
-            || !stage.is_current
-            || stage.target_community_id != Some(*community.as_uuid())
-            || stage.metadata_json["inputs"] != expected_inputs
-        {
-            return Err(safety("expected the current approved purge_messages stage"));
-        }
-        let subject_hex = normalize_user_deletion_pubkey(&request.target)?;
         let subject = hex::decode(&subject_hex).map_err(|_| safety("invalid subject"))?;
-        if subject == relay_pubkey {
-            return Err(safety("cannot purge the relay signing identity"));
-        }
-        let identity = crate::user_deletion_fences::FenceIdentity::user(
-            nostr::PublicKey::from_slice(&subject).map_err(|_| safety("invalid subject"))?,
-        );
-        let fence = crate::user_deletion_fences::find(&mut tx, identity)
-            .await?
-            .ok_or_else(|| safety("user must be fenced before message deletion"))?;
-        if fence.request_id != lease.request_id {
-            return Err(safety("user fence belongs to another deletion request"));
-        }
-        let active: bool = sqlx::query_scalar(
-            "SELECT EXISTS (SELECT 1 FROM community_serving_write_leases
-             WHERE community_id = $1 AND lease_until > clock_timestamp())",
-        )
-        .bind(community.as_uuid())
-        .fetch_one(&mut *tx)
-        .await?;
-        if active {
-            return Err(safety(
-                "external community writes must drain before message deletion",
-            ));
-        }
-        let mut tx = AdmittedTx::admit(
-            tx,
-            &crate::deletion::DeletionStore::new(self.pool.clone()),
+        let result = purge_for_subject(
+            &mut tx,
             community,
+            &subject,
+            &subject_hex,
+            relay_pubkey,
+            batch_size,
         )
         .await?;
-        let seed = select(&mut tx, community, &subject, relay_pubkey, None, 1).await?;
-        let mut result = MessageBatch::default();
-        if let Some(seed) = seed.first() {
-            result.root_id = Some(hex::encode(&seed.root_id));
-            let candidates = select(
-                &mut tx,
-                community,
-                &subject,
-                relay_pubkey,
-                Some(seed),
-                batch_size,
-            )
-            .await?;
-            cleanup::remove(&mut tx, community, &candidates, &mut result).await?;
-        } else {
-            // A historic reaction can outlive its source event. Drain these in
-            // bounded batches too; deleting only signed events would miss them.
-            result.reactions_deleted =
-                cleanup::remove_orphan_reactions(&mut tx, community, &subject, batch_size).await?;
-            result.workflow_runs_deleted =
-                cleanup::remove_orphan_runs(&mut tx, community, &subject_hex, batch_size).await?;
-            result.exhausted = result.reactions_deleted == 0 && result.workflow_runs_deleted == 0;
-        }
-        // Expiry during a long query must roll back even though the parent row
-        // lock prevented a replacement worker from claiming in the meantime.
-        // Ownership cannot change while the request row is locked. Only the
-        // clock can invalidate the lease after admission.
-        let unexpired: bool = sqlx::query_scalar(
-            "SELECT lease_expires_at > clock_timestamp() FROM entity_deletion_requests WHERE id=$1",
-        )
-        .bind(lease.request_id)
-        .fetch_one(tx.conn())
-        .await?;
-        if !unexpired {
-            return Err(DbError::AccessDenied(
-                "deletion lease expired during message batch".into(),
-            ));
-        }
+        super::execution::verify_live(&mut tx, lease).await?;
         tx.commit().await?;
         Ok(result)
     }
 }
 
+pub(super) async fn purge_for_subject(
+    tx: &mut AdmittedTx,
+    community: CommunityId,
+    subject: &[u8],
+    subject_hex: &str,
+    relay_pubkey: &[u8; 32],
+    batch_size: u16,
+) -> Result<MessageBatch> {
+    let seed = select(tx, community, subject, relay_pubkey, None, 1).await?;
+    let mut result = MessageBatch::default();
+    if let Some(seed) = seed.first() {
+        result.root_id = Some(hex::encode(&seed.root_id));
+        let candidates =
+            select(tx, community, subject, relay_pubkey, Some(seed), batch_size).await?;
+        cleanup::remove(tx, community, &candidates, &mut result).await?;
+    } else {
+        // A historic reaction can outlive its source event. Drain these in
+        // bounded batches too; deleting only signed events would miss them.
+        result.reactions_deleted =
+            cleanup::remove_orphan_reactions(tx, community, subject, batch_size).await?;
+        result.workflow_runs_deleted =
+            cleanup::remove_orphan_runs(tx, community, subject_hex, batch_size).await?;
+        result.exhausted = result.reactions_deleted == 0 && result.workflow_runs_deleted == 0;
+    }
+    Ok(result)
+}
+
 async fn select(
     tx: &mut AdmittedTx,
     community: CommunityId,
@@ -201,3 +143,53 @@ async fn select(
         .fetch_all(tx.conn())
         .await?)
 }
+
+/// Remove a bounded set of authored non-message events, preserving the same
+/// media, delivery, workflow-copy and thread-repair guarantees as messages.
+pub(super) async fn purge_event_kinds(
+    tx: &mut AdmittedTx,
+    subject: &[u8],
+    kinds: &[i32],
+    limit: u16,
+) -> Result<bool> {
+    let community = tx.community();
+    let rows: Vec<Candidate> = sqlx::query_as(
+        "WITH seed AS (SELECT channel_id FROM events WHERE community_id=$1 AND pubkey=$2
+         AND kind=ANY($3) ORDER BY created_at,id LIMIT 1)
+         SELECT e.id,e.channel_id,e.id AS root_id,
+         (e.content ~* 'https?://|/media/' OR EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t
+          WHERE t->>0 IN ('imeta','link-preview','url','thumb','image','x','emoji'))) AS has_media
+         FROM events e, seed WHERE e.community_id=$1 AND e.pubkey=$2 AND e.kind=ANY($3)
+         AND e.channel_id IS NOT DISTINCT FROM seed.channel_id ORDER BY e.created_at,e.id LIMIT $4",
+    )
+    .bind(community.as_uuid())
+    .bind(subject)
+    .bind(kinds)
+    .bind(i64::from(limit))
+    .fetch_all(tx.conn())
+    .await?;
+    if rows.is_empty() {
+        return Ok(true);
+    }
+    cleanup::remove(tx, community, &rows, &mut MessageBatch::default()).await?;
+    Ok(false)
+}
+
+pub(super) async fn has_remaining(
+    tx: &mut AdmittedTx,
+    subject: &[u8],
+    subject_hex: &str,
+    relay: &[u8; 32],
+) -> Result<bool> {
+    let community = tx.community();
+    if !select(tx, community, subject, relay, None, 1)
+        .await?
+        .is_empty()
+    {
+        return Ok(true);
+    }
+    Ok(sqlx::query_scalar(
+        "SELECT EXISTS (SELECT 1 FROM reactions WHERE community_id=$1 AND pubkey=$2
+         UNION ALL SELECT 1 FROM workflow_runs WHERE community_id=$1 AND lower(trigger_context->>'author')=$3)",
+    ).bind(community.as_uuid()).bind(subject).bind(subject_hex).fetch_one(tx.conn()).await?)
+}
diff --git a/crates/buzz-db/src/store/user_deletion/mod.rs b/crates/buzz-db/src/store/user_deletion/mod.rs
index aa3de5dbeec..e992af0a989 100644
--- a/crates/buzz-db/src/store/user_deletion/mod.rs
+++ b/crates/buzz-db/src/store/user_deletion/mod.rs
@@ -1,7 +1,10 @@
 //! User-specific submission and inventory preparation over the shared deletion tables.
 use crate::Db;
 use sqlx::PgPool;
+/// Standalone owned-agent database cleanup and durable per-agent progress.
+pub mod agents;
 mod discovery;
+mod execution;
 mod inventory;
 /// Standalone per-thread message cleanup.
 pub mod messages;
diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs
index 4ebb4040fcb..71555ff82cb 100644
--- a/crates/buzz-deletion/src/user.rs
+++ b/crates/buzz-deletion/src/user.rs
@@ -161,3 +161,6 @@ pub mod summary;
 mod transition;
 mod worker;
 mod workflow;
+
+/// Standalone owned-agent database cleanup with per-agent progress.
+pub mod agents;
diff --git a/crates/buzz-deletion/src/user/agents.rs b/crates/buzz-deletion/src/user/agents.rs
new file mode 100644
index 00000000000..8568f7c32b5
--- /dev/null
+++ b/crates/buzz-deletion/src/user/agents.rs
@@ -0,0 +1,91 @@
+//! Standalone agent database cleanup. The main worker/router does not invoke it.
+use anyhow::{bail, Context, Result};
+use buzz_core::CommunityId;
+use buzz_db::{
+    entity_deletion::{EntityLease, EntityStage},
+    user_deletion::agents::{AgentDbPhase, AgentDbStatus, AgentsDbBatch},
+    Db,
+};
+
+/// Run one bounded operation and persist per-agent progress on the current stage.
+///
+/// Repeat while `exhausted` is false, renewing the request lease between calls.
+/// Agents run in approved inventory order. Failed agents stop the stage and
+/// resume in place after the normal retry deadline; completed agents are skipped.
+/// Caller must gate/drain writers. Media-bearing events fail closed. Completion
+/// means supported agent DB content only: profiles, memberships, credentials,
+/// owner configuration, shared artifacts, media and hosted resources remain.
+pub async fn handle(
+    db: &Db,
+    lease: &EntityLease,
+    stage: &EntityStage,
+    relay_pubkey: &[u8; 32],
+    batch_size: u16,
+) -> Result<AgentsDbBatch> {
+    if stage.request_id != lease.request_id || stage.stage != "purge_agents_db" {
+        bail!("expected this request's purge_agents_db stage");
+    }
+    let community = CommunityId::from_uuid(
+        stage
+            .target_community_id
+            .context("missing agent community")?,
+    );
+    let store = db.entity_deletion_store();
+    let stage = store.start_stage(lease, stage.id).await?;
+    let agents = db.user_deletion_store();
+    let result = agents
+        .purge_agents_db_batch(
+            lease,
+            stage.id,
+            community,
+            relay_pubkey,
+            batch_size,
+            |agent, now| {
+                // Business phase order lives here; database operations commit this
+                // policy result in the same transaction as the completed batch.
+                agent.phase = match agent.phase {
+                    AgentDbPhase::Messages => AgentDbPhase::PrivateState,
+                    AgentDbPhase::PrivateState => AgentDbPhase::JobRecords,
+                    AgentDbPhase::JobRecords => AgentDbPhase::OwnedWorkflows,
+                    AgentDbPhase::OwnedWorkflows => AgentDbPhase::DeliveryState,
+                    AgentDbPhase::DeliveryState => AgentDbPhase::VerifyDbContent,
+                    AgentDbPhase::VerifyDbContent => {
+                        agent.status = AgentDbStatus::Completed;
+                        agent.completed_at = Some(now);
+                        AgentDbPhase::VerifyDbContent
+                    }
+                };
+            },
+        )
+        .await;
+    match result {
+        Ok(batch) => {
+            if batch.exhausted {
+                store
+                    .save_stage_success(
+                        lease,
+                        stage.id,
+                        serde_json::json!({
+                            "agent_db_content_exhausted": true,
+                            "agents_completed": batch.progress.agents.len(),
+                        }),
+                    )
+                    .await?;
+            }
+            Ok(batch)
+        }
+        Err(error) => {
+            if let Err(save_error) = agents
+                .fail_agents_db_stage(lease, stage.id, &error.to_string())
+                .await
+            {
+                return Err(anyhow::Error::new(error)
+                    .context(format!("also could not save agent retry: {save_error}")));
+            }
+            Err(error.into())
+        }
+    }
+}
+
+#[cfg(test)]
+mod postgres_tests;
diff --git a/crates/buzz-deletion/src/user/agents/README.md b/crates/buzz-deletion/src/user/agents/README.md
new file mode 100644
index 00000000000..535c52991df
--- /dev/null
+++ b/crates/buzz-deletion/src/user/agents/README.md
@@ -0,0 +1,156 @@
+# Standalone agent database cleanup
+
+`buzz_deletion::user::agents::handle` executes one bounded operation for a
+`purge_agents_db` stage in one approved community. The main worker, transition
+rules and CLI do not call it yet. No new tables or migrations are required.
+
+One stage handles all agents in that community. Agent progress lives in
+`entity_deletion_stage_history.metadata_json.progress`; the request row still
+owns the worker lease, generation, heartbeat and retry deadline. Agents do not
+have separate leases or workflow stage rows.
+
+## Approved scope and prerequisites
+
+Real inventory generation includes `agents: [{"pubkey": "<hex>"}]` on each
+discovered community, using `users.agent_owner_pubkey` in the same SQL snapshot
+as community discovery. The digest and approval cover this exact agent list.
+Limits are 1,024 agents per community and 4,096 per request; exceeding either
+fails rather than returning partial scope. Keys must be canonical and unique.
+
+Existing inventories remain unchanged and readable. An absent `agents` field
+does not authorize agent deletion; `agents: []` explicitly authorizes empty
+scope. Demo inventories/execution are unchanged. Discovery still uses current
+relay/channel memberships: this change does not discover every historical
+community or enable owned-community execution. There is no command to refresh
+and reapprove an existing saved inventory in this PR.
+
+The caller supplies the approved real-user request, current stage, live lease,
+trusted relay signing key and batch size (1–100). It must enforce gating and
+drain affected writers/workflows/deliveries before calling. Fence storage alone
+does not enforce serving paths. The handler requires this request's global
+owner fence and each agent's community fence with matching owner provenance.
+
+Mismatched live ownership, newly owned agents outside the approved list, invalid
+saved progress and the relay signing identity are refused. A missing profile
+does not erase approved scope: recovery uses saved inventory and fence evidence.
+Scope changes require reconciliation, not silently dropping or adopting agents.
+
+## One handler call
+
+1. Reload/start the current stage through the existing stage API; check the
+   request lease, approval and retry deadline.
+2. Take the community and request locks. Validate exact saved inputs and
+   fences, and reject active external-write leases.
+3. Read progress and choose the first unfinished agent in inventory order.
+4. Perform one bounded operation in that agent's current phase. When exhausted,
+   the policy in `buzz-deletion` selects the next internal phase. SQL effects
+   and the progress update commit in the same transaction.
+5. After final verification, mark that agent completed with a database-clock
+   timestamp. A subsequent call starts the next agent.
+6. When every agent is complete, save the stage result through the existing
+   completion API. Only the workflow transition function advances the stage.
+
+The result is `AgentsDbBatch { progress, exhausted }`. The caller repeats while
+not exhausted, heartbeats between calls and owns lease release/transitions.
+There is no new executor or unbounded loop inside the handler.
+
+| Internal phase | Work |
+| --- | --- |
+| `messages` | Shared user message cleanup: authored/trusted-relay-attributed messages, dependent reactions/edits/votes, delivery records, terminal workflow copies and thread-count repair. |
+| `private_state` | Authored lists/settings (3, 10000–10003, 30000, 30003, 30315, 30078), engrams (30174), metrics (44200), private reminders (30300), personal read frontiers/accounts. |
+| `job_records` | Agent-authored job protocol events 43001–43006. Other participants' signed events remain. |
+| `owned_workflows` | Owned workflow definitions, scheduled fires, approvals and terminal runs, then authored definition events (30620). Active runs block deletion. |
+| `delivery_state` | Agent push outbox/leases, listener outbox where the agent is sender/recipient, owned subscriptions/delivery logs, authored lease events (30350). Live delivery claims block deletion. |
+| `verify_db_content` | Read-only absence checks; remaining supported data prevents completion. |
+
+Non-message events reuse the same dependent-record, media and thread-repair
+checks. Workflow/subscription children are removed before their parent, leaving
+the parent as a recovery anchor between batches. Owned-workflow cleanup removes
+scheduled-fire claims; removing message copies from somebody else's workflow
+preserves that workflow and its scheduled-fire claim.
+
+## Progress and retry
+
+Illustrative progress (real keys have 64 hex characters):
+
+```json
+{
+  "version": 1,
+  "current_agent": "<agent-b>",
+  "agents": {
+    "<agent-a>": {
+      "status": "completed",
+      "phase": "verify_db_content",
+      "completed_at": "2026-10-30T20:00:00Z",
+      "last_error": null
+    },
+    "<agent-b>": {
+      "status": "failed",
+      "phase": "owned_workflows",
+      "completed_at": null,
+      "last_error": "agent-owned workflow runs must stop before deletion"
+    }
+  }
+}
+```
+
+`inputs`, predecessor information and other stage metadata are preserved.
+`current_agent` can be null between completed agents; approved order determines
+the next unfinished entry. Completed agents form a prefix and are skipped.
+
+- Batch error: roll back data and progress together. A separate transaction
+  marks the current agent/stage failed and the request `retry_wait`, with a
+  30-second deadline and bounded error. The error still propagates.
+- Retry: reload the same stage and failed phase, scanning remaining rows.
+  There is no row-offset cursor. Earlier completed agents stay completed.
+- Crash after commit: PostgreSQL retains the checkpoint. A replacement worker
+  resumes it, including when the same worker ID gets a newer lease generation.
+  Stale-generation writes are rejected.
+- Lease expiry during a batch: rollback. An expired worker cannot save failure;
+  the durable current stage is available for another worker to reclaim.
+- Final completion-save failure: all agent checkpoints remain complete. Retry
+  only needs to save stage completion, without repeating those agents.
+- Invalid inventory/progress or unavailable PostgreSQL may prevent recording
+  failure too. Both errors propagate; they never become successful empty scope.
+
+## Retained data and integration limits
+
+Profiles/profile events, owner-authored managed configuration, credentials,
+memberships, shared emoji/artifacts, audit/moderation records, global listener
+registrations, media and hosted resources remain for other stages. Global
+registrations may span communities. Completion here does not erase the identity.
+
+Attachment/link-bearing events still block deletion, including private/job
+events. Workflow definitions with links also block removal. The conservative
+check includes ordinary URLs. There is no media bypass: preserving references
+for later media cleanup remains an integration requirement. A verification
+failure from an ungated writer requires fixing that writer and reconciling
+progress; retrying verification cannot rerun an earlier finished phase. No
+administrative reset is added here.
+
+The 1–100 limit bounds primary rows in each operation; dependent cleanup and
+verification scans can be larger. Existing message dependency/thread caps,
+two-second lock timeout, ten-second statement timeout and final lease-expiry
+check apply. Statement timeouts are not whole-batch deadlines. Heartbeats cannot
+renew while the request row is locked. Measure large-account scans and the
+transaction lease budget before wiring this into serving workflows.
+
+## Code and tests
+
+- `buzz-deletion/src/user/agents.rs`: phase order, outcomes, final completion.
+- `buzz-db/src/store/user_deletion/agents/`: scoped SQL, progress validation,
+  atomic checkpoints and failure updates.
+- `buzz-db/src/store/user_deletion/execution.rs`: shared admission/lease checks;
+  `messages/`: shared message/dependent cleanup.
+
+```sh
+scripts/postgres-test-run.sh -p buzz-deletion -p buzz-db --lib \
+  -E 'test(user::agents::) | test(user::messages::) | test(user_deletion::)'
+```
+
+Tests use real inventories, approvals, fences and leases in isolated databases.
+They cover multiple agents, restart, completed-agent preservation, same-owner
+reclaim, transactional rollback, expiry at checkpoint, completion-save recovery,
+missing profiles, scope/fence/progress refusal, active workflow/delivery blockers
+and final verification. Existing message tests cover shared thread/attribution
+behavior. There is no operator CLI path for this standalone handler yet.
diff --git a/crates/buzz-deletion/src/user/agents/postgres_tests.rs b/crates/buzz-deletion/src/user/agents/postgres_tests.rs
new file mode 100644
index 00000000000..be373d7a696
--- /dev/null
+++ b/crates/buzz-deletion/src/user/agents/postgres_tests.rs
@@ -0,0 +1,403 @@
+use super::*;
+use buzz_db::entity_deletion::{NewEntityStage, NextStage, StageTransition};
+use buzz_db::user_deletion::{agents::AgentsDbProgress, UserDeletionInventory};
+use buzz_db::user_deletion_fences::{self, FenceIdentity};
+use nostr::{Event, EventBuilder, Keys, Kind, Tag};
+use serde_json::Value;
+use sqlx::PgPool;
+use std::time::Duration;
+use uuid::Uuid;
+
+mod recovery_postgres_tests;
+mod safety_postgres_tests;
+
+struct Fixture {
+    db: Db,
+    community: CommunityId,
+    owner: Keys,
+    agents: Vec<Keys>,
+    other: Keys,
+    relay: Keys,
+    lease: EntityLease,
+    stage: EntityStage,
+}
+
+fn database_url() -> String {
+    std::env::var("BUZZ_TEST_DATABASE_URL")
+        .or_else(|_| std::env::var("DATABASE_URL"))
+        .unwrap()
+}
+
+async fn fixture(count: usize) -> Fixture {
+    let db = Db::from_pool(PgPool::connect(&database_url()).await.unwrap());
+    let community = db
+        .ensure_configured_community(&format!("agents-{}.example", Uuid::new_v4()))
+        .await
+        .unwrap()
+        .id;
+    let owner = Keys::generate();
+    let other = Keys::generate();
+    let relay = Keys::generate();
+    let mut agents: Vec<_> = (0..count).map(|_| Keys::generate()).collect();
+    agents.sort_by_key(|key| key.public_key().to_hex());
+    for key in [&owner, &other].into_iter().chain(agents.iter()) {
+        sqlx::query("INSERT INTO users (community_id,pubkey,display_name) VALUES ($1,$2,'retained profile')")
+            .bind(community.as_uuid()).bind(key.public_key().to_bytes().as_slice()).execute(db.pool()).await.unwrap();
+    }
+    for agent in &agents {
+        sqlx::query("UPDATE users SET agent_owner_pubkey=$3 WHERE community_id=$1 AND pubkey=$2")
+            .bind(community.as_uuid())
+            .bind(agent.public_key().to_bytes().as_slice())
+            .bind(owner.public_key().to_bytes().as_slice())
+            .execute(db.pool())
+            .await
+            .unwrap();
+    }
+    sqlx::query("INSERT INTO relay_members (community_id,pubkey,role) VALUES ($1,$2,'member')")
+        .bind(community.as_uuid())
+        .bind(owner.public_key().to_hex())
+        .execute(db.pool())
+        .await
+        .unwrap();
+    let users = db.user_deletion_store();
+    let request = users
+        .submit(&owner.public_key().to_hex(), "submitter", None)
+        .await
+        .unwrap();
+    let inventory = UserDeletionInventory::read(&request).unwrap();
+    assert_eq!(
+        inventory
+            .agents(*community.as_uuid(), &request.target)
+            .unwrap(),
+        agents
+            .iter()
+            .map(|key| key.public_key().to_hex())
+            .collect::<Vec<_>>()
+    );
+    users
+        .approve(
+            request.id,
+            "approver",
+            request.inventory_digest.as_deref().unwrap(),
+        )
+        .await
+        .unwrap();
+    let store = db.entity_deletion_store();
+    let lease = store
+        .claim(request.id, "agent-worker", Duration::from_secs(600))
+        .await
+        .unwrap()
+        .unwrap();
+    let NextStage::Stage(stage) = store
+        .transition_stage(&lease, None, |_, _| {
+            Ok(StageTransition::Initialize(NewEntityStage {
+                stage: "purge_agents_db".into(),
+                community: Some(*community.as_uuid()),
+                inputs: inventory.inputs(&request.target, Some(*community.as_uuid()))?,
+            }))
+        })
+        .await
+        .unwrap()
+    else {
+        panic!("expected stage")
+    };
+    let mut tx = db.pool().begin().await.unwrap();
+    user_deletion_fences::insert(
+        &mut tx,
+        FenceIdentity::user(owner.public_key()),
+        &lease,
+        None,
+    )
+    .await
+    .unwrap();
+    for agent in &agents {
+        user_deletion_fences::insert(
+            &mut tx,
+            FenceIdentity::agent(community, agent.public_key()),
+            &lease,
+            Some(owner.public_key()),
+        )
+        .await
+        .unwrap();
+    }
+    tx.commit().await.unwrap();
+    Fixture {
+        db,
+        community,
+        owner,
+        agents,
+        other,
+        relay,
+        lease,
+        stage,
+    }
+}
+
+impl Fixture {
+    async fn event(&self, key: &Keys, kind: u16, content: &str, tags: Vec<Tag>) -> Event {
+        let event = EventBuilder::new(Kind::Custom(kind), content)
+            .tags(tags)
+            .sign_with_keys(key)
+            .unwrap();
+        self.db
+            .insert_event(self.community, &event, None)
+            .await
+            .unwrap();
+        event
+    }
+    async fn exists(&self, event: &Event) -> bool {
+        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2)")
+            .bind(self.community.as_uuid())
+            .bind(event.id.as_bytes().as_slice())
+            .fetch_one(self.db.pool())
+            .await
+            .unwrap()
+    }
+    async fn batch(&self) -> Result<AgentsDbBatch> {
+        handle(
+            &self.db,
+            &self.lease,
+            &self.stage,
+            &self.relay.public_key().to_bytes(),
+            1,
+        )
+        .await
+    }
+    async fn progress(&self) -> AgentsDbProgress {
+        let value: Value = sqlx::query_scalar(
+            "SELECT metadata_json->'progress' FROM entity_deletion_stage_history WHERE id=$1",
+        )
+        .bind(self.stage.id)
+        .fetch_one(self.db.pool())
+        .await
+        .unwrap();
+        serde_json::from_value(value).unwrap()
+    }
+    async fn finish(&self) -> AgentsDbBatch {
+        for _ in 0..150 {
+            let result = self.batch().await.unwrap();
+            if result.exhausted {
+                return result;
+            }
+        }
+        panic!("agent fixture did not finish")
+    }
+    async fn due(&self) {
+        sqlx::query(
+            "UPDATE entity_deletion_requests SET next_attempt_at=clock_timestamp() WHERE id=$1",
+        )
+        .bind(self.lease.request_id)
+        .execute(self.db.pool())
+        .await
+        .unwrap();
+    }
+    async fn reclaim(&mut self) {
+        sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at=clock_timestamp()-interval '1 second',next_attempt_at=NULL WHERE id=$1")
+            .bind(self.lease.request_id).execute(self.db.pool()).await.unwrap();
+        self.lease = self
+            .db
+            .entity_deletion_store()
+            .claim(
+                self.lease.request_id,
+                "agent-worker",
+                Duration::from_secs(600),
+            )
+            .await
+            .unwrap()
+            .unwrap();
+    }
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn cleans_two_agents_and_tracks_completion_without_removing_profiles_or_neighbors() {
+    let f = fixture(2).await;
+    let mut deleted = Vec::new();
+    let mut kept = Vec::new();
+    for key in &f.agents {
+        for kind in [9, 30174, 44200, 43004, 30078, 30300, 30350, 30620] {
+            deleted.push(f.event(key, kind, "agent payload", vec![]).await);
+        }
+        for kind in [0, 10100, 30030] {
+            kept.push(f.event(key, kind, "profile or shared emoji", vec![]).await);
+        }
+        sqlx::query("INSERT INTO personal_read_accounts (community_id,actor) VALUES ($1,$2)")
+            .bind(f.community.as_uuid())
+            .bind(key.public_key().to_bytes().as_slice())
+            .execute(f.db.pool())
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO subscriptions (community_id,id,owner_pubkey,delivery_url) VALUES ($1,$2,$3,'https://example.test/keep-until-drained')")
+            .bind(f.community.as_uuid()).bind(key.public_key().to_hex()).bind(key.public_key().to_bytes().as_slice())
+            .execute(f.db.pool()).await.unwrap();
+        sqlx::query(
+            "INSERT INTO delivery_log (community_id,subscription_id,event_id) VALUES ($1,$2,$3)",
+        )
+        .bind(f.community.as_uuid())
+        .bind(key.public_key().to_hex())
+        .bind([77u8; 32].as_slice())
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    }
+    kept.push(f.event(&f.owner, 9, "owner message", vec![]).await);
+    kept.push(f.event(&f.other, 9, "unrelated agent", vec![]).await);
+    kept.push(
+        f.event(
+            &f.owner,
+            30179,
+            "private agent config",
+            vec![Tag::parse(["d", &f.agents[0].public_key().to_hex()]).unwrap()],
+        )
+        .await,
+    );
+    let other_community =
+        f.db.ensure_configured_community("agent-other.example")
+            .await
+            .unwrap()
+            .id;
+    f.db.insert_event(other_community, &deleted[0], None)
+        .await
+        .unwrap();
+    let done = f.finish().await;
+    assert_eq!(done.progress.agents.len(), 2);
+    for agent in done.progress.agents.values() {
+        assert_eq!(agent.status, AgentDbStatus::Completed);
+        assert!(agent.completed_at.is_some());
+        assert_eq!(agent.phase, AgentDbPhase::VerifyDbContent);
+    }
+    for event in deleted {
+        assert!(!f.exists(&event).await);
+    }
+    for event in kept {
+        assert!(f.exists(&event).await);
+    }
+    for (table, expected) in [
+        ("users", 4),
+        ("personal_read_accounts", 0),
+        ("subscriptions", 0),
+        ("delivery_log", 0),
+    ] {
+        let count: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
+            "SELECT count(*) FROM {table} WHERE community_id=$1"
+        )))
+        .bind(f.community.as_uuid())
+        .fetch_one(f.db.pool())
+        .await
+        .unwrap();
+        assert_eq!(count, expected, "{table}");
+    }
+    let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1")
+        .bind(other_community.as_uuid())
+        .fetch_one(f.db.pool())
+        .await
+        .unwrap();
+    assert_eq!(count, 1);
+    let (status,current,result):(String,bool,Value)=sqlx::query_as("SELECT status,is_current,metadata_json->'result' FROM entity_deletion_stage_history WHERE id=$1")
+        .bind(f.stage.id).fetch_one(f.db.pool()).await.unwrap();
+    assert_eq!(status, "completed");
+    assert!(current);
+    assert_eq!(result["agents_completed"], 2);
+    assert_ne!(
+        f.db.entity_deletion_store()
+            .get(f.lease.request_id)
+            .await
+            .unwrap()
+            .status,
+        "completed"
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn failed_second_agent_preserves_first_completion_and_resumes_after_same_owner_reclaim() {
+    let mut f = fixture(2).await;
+    let first = f.agents[0].public_key().to_hex();
+    let second = f.agents[1].public_key().to_hex();
+    let media = f
+        .event(&f.agents[1], 9, "https://example.test/image", vec![])
+        .await;
+    for _ in 0..40 {
+        if f.batch().await.is_err() {
+            break;
+        }
+    }
+    let progress = f.progress().await;
+    assert_eq!(progress.agents[&first].status, AgentDbStatus::Completed);
+    assert_eq!(progress.agents[&second].status, AgentDbStatus::Failed);
+    assert_eq!(progress.current_agent, Some(second.clone()));
+    assert!(
+        progress.agents[&second]
+            .last_error
+            .as_deref()
+            .unwrap()
+            .contains("file")
+            || progress.agents[&second]
+                .last_error
+                .as_deref()
+                .unwrap()
+                .contains("attachment")
+    );
+    assert!(f.exists(&media).await);
+    assert!(f.batch().await.unwrap_err().to_string().contains("not due"));
+    let old = f.lease.clone();
+    let completed_at = progress.agents[&first].completed_at;
+    // Simulate the separate media stage resolving this blocker, not the handler bypassing it.
+    sqlx::query("DELETE FROM events WHERE community_id=$1 AND id=$2")
+        .bind(f.community.as_uuid())
+        .bind(media.id.as_bytes().as_slice())
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    f.reclaim().await;
+    assert!(f.lease.generation > old.generation);
+    // Keep every other guard valid: same owner, live expiry, approved scope and
+    // a running current stage. Only the old generation may reject these writes.
+    f.db.entity_deletion_store()
+        .start_stage(&f.lease, f.stage.id)
+        .await
+        .unwrap();
+    assert!(
+        handle(&f.db, &old, &f.stage, &f.relay.public_key().to_bytes(), 1)
+            .await
+            .is_err()
+    );
+    assert!(f
+        .db
+        .user_deletion_store()
+        .fail_agents_db_stage(&old, f.stage.id, "stale error")
+        .await
+        .is_err());
+    assert!(f
+        .db
+        .user_deletion_store()
+        .purge_agents_db_batch(
+            &old,
+            f.stage.id,
+            f.community,
+            &f.relay.public_key().to_bytes(),
+            1,
+            |_, _| {},
+        )
+        .await
+        .is_err());
+    f.db = Db::from_pool(PgPool::connect(&database_url()).await.unwrap());
+    let done = f.finish().await;
+    assert_eq!(done.progress.agents[&first].completed_at, completed_at);
+    assert_eq!(
+        done.progress.agents[&second].status,
+        AgentDbStatus::Completed
+    );
+    assert!(done.progress.agents[&second].last_error.is_none());
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn empty_approved_agent_list_finishes_without_changing_other_data() {
+    let f = fixture(0).await;
+    let event = f.event(&f.owner, 30174, "owner memory", vec![]).await;
+    let result = f.batch().await.unwrap();
+    assert!(result.exhausted);
+    assert!(result.progress.agents.is_empty());
+    assert!(f.exists(&event).await);
+}
diff --git a/crates/buzz-deletion/src/user/agents/postgres_tests/recovery_postgres_tests.rs b/crates/buzz-deletion/src/user/agents/postgres_tests/recovery_postgres_tests.rs
new file mode 100644
index 00000000000..660c908611b
--- /dev/null
+++ b/crates/buzz-deletion/src/user/agents/postgres_tests/recovery_postgres_tests.rs
@@ -0,0 +1,124 @@
+use super::*;
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn lease_expiring_at_checkpoint_rolls_back_agent_deletion_and_progress() {
+    let f = fixture(1).await;
+    let event = f
+        .event(&f.agents[0], 9, "must survive expiry", vec![])
+        .await;
+    // Test-only fault injection on the real final UPDATE; no production seam.
+    sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
+        "CREATE FUNCTION expire_agent_checkpoint() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN
+         UPDATE entity_deletion_requests SET lease_expires_at=clock_timestamp()-interval '1 second'
+         WHERE id='{}'; RETURN NEW; END $$;
+         CREATE TRIGGER expire_agent_checkpoint AFTER UPDATE ON entity_deletion_stage_history
+         FOR EACH ROW WHEN (NEW.id='{}' AND NEW.metadata_json->'progress' IS NOT NULL AND NEW.status='running')
+         EXECUTE FUNCTION expire_agent_checkpoint();", f.lease.request_id, f.stage.id
+    ))).execute(f.db.pool()).await.unwrap();
+    let outcome = f.batch().await;
+    sqlx::raw_sql("DROP TRIGGER expire_agent_checkpoint ON entity_deletion_stage_history; DROP FUNCTION expire_agent_checkpoint();")
+        .execute(f.db.pool()).await.unwrap();
+    assert!(outcome
+        .unwrap_err()
+        .to_string()
+        .contains("lease expired during cleanup batch"));
+    assert!(f.exists(&event).await);
+    let item = &f.progress().await.agents[&f.agents[0].public_key().to_hex()];
+    assert_eq!(item.phase, AgentDbPhase::Messages);
+    assert_eq!(item.status, AgentDbStatus::Failed);
+    f.due().await;
+    f.finish().await;
+    assert!(!f.exists(&event).await);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn restart_within_agent_uses_saved_scope_even_after_profile_disappears() {
+    let mut f = fixture(1).await;
+    let key = f.agents[0].public_key().to_hex();
+    let one = f.event(&f.agents[0], 9, "first", vec![]).await;
+    let two = f.event(&f.agents[0], 9, "second", vec![]).await;
+    let batch = f.batch().await.unwrap();
+    assert!(!batch.exhausted);
+    assert_eq!(batch.progress.current_agent.as_deref(), Some(key.as_str()));
+    assert_eq!(batch.progress.agents[&key].phase, AgentDbPhase::Messages);
+    assert_ne!(f.exists(&one).await, f.exists(&two).await);
+    // The handler must recover from approved inventory and fence provenance,
+    // not rediscover agents from a profile row which might already be absent.
+    sqlx::query("DELETE FROM users WHERE community_id=$1 AND pubkey=decode($2,'hex')")
+        .bind(f.community.as_uuid())
+        .bind(&key)
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    f.reclaim().await;
+    f.db = Db::from_pool(PgPool::connect(&database_url()).await.unwrap());
+    f.finish().await;
+    assert!(!f.exists(&one).await);
+    assert!(!f.exists(&two).await);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn completion_write_can_retry_after_all_agent_checkpoints_commit() {
+    let mut f = fixture(1).await;
+    sqlx::query(sqlx::AssertSqlSafe(format!(
+        "ALTER TABLE entity_deletion_stage_history ADD CONSTRAINT fail_agent_completion
+         CHECK (id <> '{}' OR status <> 'completed')",
+        f.stage.id
+    )))
+    .execute(f.db.pool())
+    .await
+    .unwrap();
+    let mut failed = false;
+    for _ in 0..20 {
+        if f.batch().await.is_err() {
+            failed = true;
+            break;
+        }
+    }
+    assert!(failed);
+    let before = f.progress().await;
+    let key = f.agents[0].public_key().to_hex();
+    assert_eq!(before.agents[&key].status, AgentDbStatus::Completed);
+    sqlx::query("ALTER TABLE entity_deletion_stage_history DROP CONSTRAINT fail_agent_completion")
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    f.reclaim().await;
+    let done = f.batch().await.unwrap();
+    assert!(done.exhausted);
+    assert_eq!(
+        done.progress.agents[&key].completed_at,
+        before.agents[&key].completed_at
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn verification_refuses_remaining_content_instead_of_marking_agent_complete() {
+    let f = fixture(1).await;
+    let key = f.agents[0].public_key().to_hex();
+    for _ in 0..20 {
+        let batch = f.batch().await.unwrap();
+        if batch.progress.agents[&key].phase == AgentDbPhase::VerifyDbContent {
+            break;
+        }
+        assert!(!batch.exhausted);
+    }
+    // Simulate an ungated writer after the private-state phase. This is not a
+    // supported serving path, but completion must still refuse the residual data.
+    let leftover = f.event(&f.agents[0], 30174, "late memory", vec![]).await;
+    assert!(f
+        .batch()
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("verification"));
+    assert!(f.exists(&leftover).await);
+    assert_eq!(
+        f.progress().await.agents[&key].status,
+        AgentDbStatus::Failed
+    );
+}
diff --git a/crates/buzz-deletion/src/user/agents/postgres_tests/safety_postgres_tests.rs b/crates/buzz-deletion/src/user/agents/postgres_tests/safety_postgres_tests.rs
new file mode 100644
index 00000000000..bc8612cbc36
--- /dev/null
+++ b/crates/buzz-deletion/src/user/agents/postgres_tests/safety_postgres_tests.rs
@@ -0,0 +1,250 @@
+use super::*;
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn private_and_job_media_references_are_preserved_in_the_failed_phase() {
+    for (kind, phase) in [
+        (30174, AgentDbPhase::PrivateState),
+        (43004, AgentDbPhase::JobRecords),
+    ] {
+        let f = fixture(1).await;
+        let event = f
+            .event(
+                &f.agents[0],
+                kind,
+                "https://example.test/shared-file",
+                vec![],
+            )
+            .await;
+        let mut failed = false;
+        for _ in 0..20 {
+            if f.batch().await.is_err() {
+                failed = true;
+                break;
+            }
+        }
+        assert!(failed);
+        assert!(f.exists(&event).await);
+        let progress = f.progress().await;
+        let item = &progress.agents[&f.agents[0].public_key().to_hex()];
+        assert_eq!(item.status, AgentDbStatus::Failed);
+        assert_eq!(item.phase, phase);
+    }
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn oversized_agent_discovery_fails_without_returning_partial_inventory() {
+    let f = fixture(0).await;
+    sqlx::query(
+        "INSERT INTO users (community_id,pubkey,agent_owner_pubkey)
+        SELECT $1,decode(lpad(to_hex(n),64,'0'),'hex'),$2 FROM generate_series(1,1025) n",
+    )
+    .bind(f.community.as_uuid())
+    .bind(f.owner.public_key().to_bytes().as_slice())
+    .execute(f.db.pool())
+    .await
+    .unwrap();
+    let error =
+        f.db.user_deletion_store()
+            .generate_inventory(&f.owner.public_key().to_hex())
+            .await
+            .unwrap_err();
+    assert!(error.to_string().contains("bounded scope"));
+    // This is preview, not permission to rewrite the already approved scope.
+    assert!(UserDeletionInventory::read(
+        &f.db
+            .entity_deletion_store()
+            .get(f.lease.request_id)
+            .await
+            .unwrap()
+    )
+    .unwrap()
+    .agents(*f.community.as_uuid(), &f.owner.public_key().to_hex())
+    .unwrap()
+    .is_empty());
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn missing_or_wrong_agent_fence_and_live_owner_mismatch_refuse_cleanup() {
+    let f = fixture(1).await;
+    let event = f.event(&f.agents[0], 9, "keep", vec![]).await;
+    let agent = f.agents[0].public_key().to_hex();
+    sqlx::query("UPDATE user_deletion_fences SET source_owner_pubkey=$3 WHERE pubkey=$1 AND scope_community_id=$2")
+        .bind(&agent).bind(f.community.as_uuid()).bind(f.other.public_key().to_hex()).execute(f.db.pool()).await.unwrap();
+    assert!(f
+        .batch()
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("provenance"));
+    assert!(f.exists(&event).await);
+    sqlx::query("UPDATE user_deletion_fences SET source_owner_pubkey=$3 WHERE pubkey=$1 AND scope_community_id=$2")
+        .bind(&agent).bind(f.community.as_uuid()).bind(f.owner.public_key().to_hex()).execute(f.db.pool()).await.unwrap();
+    sqlx::query(
+        "UPDATE users SET agent_owner_pubkey=$3 WHERE community_id=$1 AND pubkey=decode($2,'hex')",
+    )
+    .bind(f.community.as_uuid())
+    .bind(&agent)
+    .bind(f.other.public_key().to_bytes().as_slice())
+    .execute(f.db.pool())
+    .await
+    .unwrap();
+    f.due().await;
+    assert!(f
+        .batch()
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("ownership differs"));
+    assert!(f.exists(&event).await);
+    sqlx::query("DELETE FROM user_deletion_fences WHERE pubkey=$1 AND scope_community_id=$2")
+        .bind(&agent)
+        .bind(f.community.as_uuid())
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    f.due().await;
+    assert!(f.batch().await.unwrap_err().to_string().contains("fence"));
+    assert!(f.exists(&event).await);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn unexpected_new_agent_and_corrupt_progress_are_not_silently_skipped() {
+    let f = fixture(1).await;
+    let event = f.event(&f.agents[0], 9, "keep", vec![]).await;
+    sqlx::query("UPDATE users SET agent_owner_pubkey=$3 WHERE community_id=$1 AND pubkey=$2")
+        .bind(f.community.as_uuid())
+        .bind(f.other.public_key().to_bytes().as_slice())
+        .bind(f.owner.public_key().to_bytes().as_slice())
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    assert!(f
+        .batch()
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("changed since approval"));
+    assert!(f.exists(&event).await);
+    sqlx::query("UPDATE users SET agent_owner_pubkey=NULL WHERE community_id=$1 AND pubkey=$2")
+        .bind(f.community.as_uuid())
+        .bind(f.other.public_key().to_bytes().as_slice())
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    sqlx::query("UPDATE entity_deletion_stage_history SET metadata_json=jsonb_set(metadata_json,'{progress,agents}','{}') WHERE id=$1")
+        .bind(f.stage.id).execute(f.db.pool()).await.unwrap();
+    f.due().await;
+    assert!(f
+        .batch()
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("saved agent progress differs"));
+    assert!(f.exists(&event).await);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn batch_and_progress_rollback_together_on_database_error() {
+    let f = fixture(1).await;
+    let event = f.event(&f.agents[0], 9, "atomic deletion", vec![]).await;
+    // Force only successful progress writes to fail. Failure recording remains available.
+    // No test-only production hook: the constraint exercises the real final UPDATE.
+    sqlx::query(sqlx::AssertSqlSafe(format!(
+        "ALTER TABLE entity_deletion_stage_history ADD CONSTRAINT fail_agent_progress CHECK
+         (id <> '{}' OR status <> 'running' OR metadata_json->'progress' IS NULL)",
+        f.stage.id
+    )))
+    .execute(f.db.pool())
+    .await
+    .unwrap();
+    assert!(f.batch().await.is_err());
+    assert!(f.exists(&event).await);
+    assert_eq!(
+        f.progress().await.agents[&f.agents[0].public_key().to_hex()].status,
+        AgentDbStatus::Failed
+    );
+    sqlx::query("ALTER TABLE entity_deletion_stage_history DROP CONSTRAINT fail_agent_progress")
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    f.due().await;
+    f.finish().await;
+    assert!(!f.exists(&event).await);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn active_workflow_and_recipient_delivery_block_their_phases() {
+    let f = fixture(1).await;
+    let agent = f.agents[0].public_key().to_bytes();
+    let workflow:Uuid=sqlx::query_scalar("INSERT INTO workflows (community_id,name,owner_pubkey,definition,definition_hash) VALUES ($1,'owned',$2,'{}','\\x01') RETURNING id")
+        .bind(f.community.as_uuid()).bind(agent.as_slice()).fetch_one(f.db.pool()).await.unwrap();
+    let run:Uuid=sqlx::query_scalar("INSERT INTO workflow_runs (community_id,workflow_id,status) VALUES ($1,$2,'running') RETURNING id")
+        .bind(f.community.as_uuid()).bind(workflow).fetch_one(f.db.pool()).await.unwrap();
+    for _ in 0..20 {
+        if f.batch().await.is_err() {
+            break;
+        }
+    }
+    assert_eq!(
+        f.progress().await.agents[&hex::encode(agent)].phase,
+        AgentDbPhase::OwnedWorkflows
+    );
+    sqlx::query("UPDATE workflow_runs SET status='completed' WHERE community_id=$1 AND id=$2")
+        .bind(f.community.as_uuid())
+        .bind(run)
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    sqlx::query("INSERT INTO workflow_approvals (community_id,token,workflow_id,run_id,step_id,step_index,approver_spec,expires_at) VALUES ($1,'\\x01',$2,$3,'step',0,'owner',now())")
+        .bind(f.community.as_uuid()).bind(workflow).bind(run).execute(f.db.pool()).await.unwrap();
+    sqlx::query("INSERT INTO scheduled_workflow_fires (community_id,workflow_id,scheduled_for,workflow_run_id) VALUES ($1,$2,now(),$3)")
+        .bind(f.community.as_uuid()).bind(workflow).bind(run).execute(f.db.pool()).await.unwrap();
+    sqlx::query("INSERT INTO operator_listener_outbox (listener_pubkey,target_pubkey,community_id,event_id,event_kind,event_created_at,state,lease_until) VALUES ($1,$2,$3,$4,9,now(),'sending',now()+interval '1 hour')")
+        .bind(f.other.public_key().to_bytes().as_slice()).bind(agent.as_slice()).bind(f.community.as_uuid())
+        .bind([78u8;32].as_slice()).execute(f.db.pool()).await.unwrap();
+    f.due().await;
+    for _ in 0..30 {
+        if f.batch().await.is_err() {
+            break;
+        }
+    }
+    let progress = f.progress().await;
+    assert_eq!(
+        progress.agents[&hex::encode(agent)].phase,
+        AgentDbPhase::DeliveryState
+    );
+    assert!(progress.agents[&hex::encode(agent)]
+        .last_error
+        .as_deref()
+        .unwrap()
+        .contains("drain"));
+    sqlx::query("UPDATE operator_listener_outbox SET lease_until=NULL WHERE community_id=$1")
+        .bind(f.community.as_uuid())
+        .execute(f.db.pool())
+        .await
+        .unwrap();
+    f.due().await;
+    f.finish().await;
+    for table in [
+        "workflows",
+        "workflow_runs",
+        "workflow_approvals",
+        "scheduled_workflow_fires",
+        "operator_listener_outbox",
+    ] {
+        let count: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
+            "SELECT count(*) FROM {table} WHERE community_id=$1"
+        )))
+        .bind(f.community.as_uuid())
+        .fetch_one(f.db.pool())
+        .await
+        .unwrap();
+        assert_eq!(count, 0, "{table}");
+    }
+}
diff --git a/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs b/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs
index abbdf9a39ee..c72c3e279b4 100644
--- a/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs
+++ b/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs
@@ -189,13 +189,13 @@ async fn expiry_after_admission_rolls_back_deletions_and_count_repairs() {
     assert!(
         error
             .to_string()
-            .contains("lease expired during message batch"),
+            .contains("lease expired during cleanup batch"),
         "{error}"
     );
     assert_preserved(&f, &root, &source, &reaction).await;
     // The injected deadline is transactional too. Rollback restores the live
     // lease, so this case can assert durable failure recording independently.
-    assert_retry(&f, "lease expired during message batch").await;
+    assert_retry(&f, "lease expired during cleanup batch").await;
 }
 
 #[tokio::test]
diff --git a/docs/operator-user-deletion-inventory.md b/docs/operator-user-deletion-inventory.md
index 0f1dc2c7a99..6423cd92003 100644
--- a/docs/operator-user-deletion-inventory.md
+++ b/docs/operator-user-deletion-inventory.md
@@ -24,7 +24,7 @@ cargo build -p buzz-admin
 ```
 
 Set `DATABASE_URL` explicitly. Preview needs SELECT on `communities`,
-`relay_members`, and `channel_members`. The complete workflow also needs write
+`relay_members`, `channel_members`, and `users`. The complete workflow also needs write
 access to the `entity_deletion_*` tables. A production read-only account cannot
 submit, approve, or checkpoint a run.
 
@@ -71,8 +71,12 @@ Preview the inventory:
 ./target/debug/buzz-admin user-deletions inventory --pubkey "$SUBJECT"
 ```
 
-The default is demo if `--pubkey` is omitted. Its two fixed communities have the
-same shape as real discovery:
+Real discovery also includes `agents: [{"pubkey": "<hex>"}]` on each community,
+capturing currently owned agents in the approved inventory. Existing saved
+inventories are not rewritten. The standalone agent DB handler requires this
+explicit list; it is not wired into the dummy workflow.
+
+The default is demo if `--pubkey` is omitted. Its two fixed communities are:
 
 ```json
 {

From 205cca5adf81c5187e49a49c7c7c7b82a73161d0 Mon Sep 17 00:00:00 2001
From: Ravneet Arora <rarora@squareup.com>
Date: Mon, 2 Nov 2026 16:29:27 -0500
Subject: [PATCH 2/2] Cover agent ownership in the read-only inventory CLI test

Signed-off-by: Ravneet Arora <rarora@squareup.com>
---
 .../tests/postgres_user_inventory.rs          | 27 +++++++++++++++----
 1 file changed, 22 insertions(+), 5 deletions(-)

diff --git a/crates/buzz-admin/tests/postgres_user_inventory.rs b/crates/buzz-admin/tests/postgres_user_inventory.rs
index 2c53cb053db..691d58a2921 100644
--- a/crates/buzz-admin/tests/postgres_user_inventory.rs
+++ b/crates/buzz-admin/tests/postgres_user_inventory.rs
@@ -16,7 +16,7 @@ fn admin_binary() -> std::ffi::OsString {
 
 #[tokio::test]
 #[ignore = "requires Postgres"]
-async fn cluster_global_inventory_cli_uses_only_existing_membership_tables() {
+async fn cluster_global_inventory_cli_reads_memberships_and_agents_without_workflow_tables() {
     let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
         .or_else(|_| std::env::var("DATABASE_URL"))
         .unwrap();
@@ -46,6 +46,8 @@ async fn cluster_global_inventory_cli_uses_only_existing_membership_tables() {
     for schema in [
         include_str!("../../../schema/tables/public/channels.sql"),
         include_str!("../../../schema/tables/public/channel_members.sql"),
+        include_str!("../../../schema/types/channel_add_policy.sql"),
+        include_str!("../../../schema/tables/public/users.sql"),
     ] {
         sqlx::raw_sql(schema).execute(&mut *tx).await.unwrap();
     }
@@ -62,6 +64,15 @@ async fn cluster_global_inventory_cli_uses_only_existing_membership_tables() {
         .execute(&mut *tx)
         .await
         .unwrap();
+    let agent = "ab".repeat(32);
+    sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, decode($2, 'hex'))")
+        .bind(community)
+        .bind(pubkey)
+        .execute(&mut *tx)
+        .await
+        .unwrap();
+    sqlx::query("INSERT INTO users (community_id, pubkey, agent_owner_pubkey) VALUES ($1, decode($2, 'hex'), decode($3, 'hex'))")
+        .bind(community).bind(&agent).bind(pubkey).execute(&mut *tx).await.unwrap();
     let participant_community = Uuid::new_v4();
     sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'channel-only-cli.example')")
         .bind(participant_community)
@@ -167,16 +178,22 @@ async fn cluster_global_inventory_cli_uses_only_existing_membership_tables() {
         .unwrap();
     assert_eq!(explicit_member["label"], "inventory-cli.example");
     assert_eq!(explicit_member["relationship"], "participant");
+    assert_eq!(
+        explicit_member["agents"],
+        serde_json::json!([{ "pubkey": agent }])
+    );
     let participant = communities
         .iter()
         .find(|item| item["community_id"] == participant_community.to_string())
         .unwrap();
     assert_eq!(participant["label"], "channel-only-cli.example");
     assert_eq!(participant["relationship"], "participant");
-    for output in [&demo, inventory] {
-        for row in output["communities"].as_array().unwrap() {
-            assert_eq!(row.as_object().unwrap().len(), 3);
-        }
+    assert_eq!(participant["agents"], serde_json::json!([]));
+    for row in communities {
+        assert_eq!(row.as_object().unwrap().len(), 4);
+    }
+    for row in demo["communities"].as_array().unwrap() {
+        assert_eq!(row.as_object().unwrap().len(), 3);
     }
 }