BUZZ-413: Add user deletion fence storage (#8632)

## Summary

Add storage and lock helpers for the first user-deletion fencing PR. No
serving handler or deletion worker calls these helpers yet, so this does
not block users or enable real deletion.

The operator-global `user_deletion_fences` table stores global user
restrictions and community-scoped agent restrictions. Community and
profile deletion retain these rows unchanged. `scope_community_id`
records a restriction's scope; it does not make the row community-owned.

The helpers use the caller's transaction, stable identity lock keys,
ordered shared/exclusive locks, and indexed lookups. Inserts preserve
the original request and owner on retry and reject conflicting
provenance. Inserts validate the current worker lease and approved
inventory through the existing entity-deletion guards. Callers must
still verify that the target identities and agent ownership belong to
the request.

This is an additive schema expansion with a five-second migration lock
timeout. It adds no database triggers, functions, foreign keys, or
backfills. Both schema sources and operator-global registrations are
updated. The unused request-ID index was removed. The existing
community-deletion catalog and purge logic are unchanged. Tests cover
live schema parity, concurrency, rollback, and retention through the
existing purge and verification functions.

### Related issue

Part of
[BUZZ-413](https://linear.app/squareup/issue/BUZZ-413/gate-user-and-owned-agent-activity).
This is the storage foundation; serving checks and real workflow
handlers follow separately. No duplicate fence-storage PR found.

### Testing

No manual app testing: these utilities have no application callers yet.
Local PostgreSQL validation uses an isolated test container and the
repository's per-test database runner.

Passed 84 selected PostgreSQL tests covering fence authorization,
concurrent inserts, schema parity, retention, community deletion,
entity-deletion storage, and user discovery. The authorization
regression also failed as expected when the insertion guard was
temporarily removed; the guard was restored before the final run.

Command: `scripts/postgres-test-run.sh -p buzz-db --lib -E
'test(/user_deletion_fences/) | test(/admin_schema_parity_between/) |
test(/^store::deletion::postgres_tests::/) |
test(/^store::entity_deletion::postgres_tests::/) |
test(/^store::user_deletion::/)'` with the local test PostgreSQL
connection and `PG_BIN_DIR` configured.

The full `just test` run had one failure:
`composition_tests::composition_log_does_not_leak_issuer_url` missed an
expected log message. The same compiled test passed when rerun alone. No
relay logging code is changed by this PR.

The full local `just ci` run and all normal pre-push hooks passed on the
updated branch.

Generated with Codex

---------

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

crates/buzz-db/src/lib.rs

@@ -69,7 +69,8 @@ pub use store::{
     channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation,
     nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction,
     read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
-    replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion, workflow,
+    replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion,
+    user_deletion_fences, workflow,
 };
 
 pub use allowlist::AllowlistEntry;

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

@@ -549,6 +549,7 @@ mod postgres_tests {
             "entity_deletion_approvals",
             "entity_deletion_requests",
             "entity_deletion_stage_history",
+            "user_deletion_fences",
             "community_deletion_approvals",
             "community_deletion_checkpoints",
             "community_deletion_manifest_keys",
@@ -775,7 +776,8 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 73);
+        assert_eq!(migrations.len(), 74);
+        assert_eq!(migrations[73].version, 74);
         assert_eq!(migrations[72].version, 73);
         assert_eq!(migrations[71].version, 72);
         assert_eq!(migrations[70].version, 71);
@@ -2905,6 +2907,7 @@ mod postgres_tests {
             "entity_deletion_approvals",
             "entity_deletion_requests",
             "entity_deletion_stage_history",
+            "user_deletion_fences",
         ] {
             assert_eq!(
                 columns(&desired, table).await,
@@ -2941,6 +2944,7 @@ mod postgres_tests {
         for table in [
             "entity_deletion_requests",
             "entity_deletion_stage_history",
+            "user_deletion_fences",
             "entity_deletion_approvals",
         ] {
             assert_eq!(

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

@@ -4400,6 +4400,116 @@ mod postgres_tests {
     use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
     use std::str::FromStr;
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn user_fences_survive_community_and_profile_deletion() {
+        assert_user_fence_retention().await;
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn migration_schema_user_fences_survive_community_and_profile_deletion() {
+        assert_user_fence_retention().await;
+    }
+
+    async fn assert_user_fence_retention() {
+        use crate::user_deletion_fences::{self as fences, FenceIdentity};
+        let (db, store) = store().await;
+        let (request, inventory) = inventoried_request(&db, &store).await;
+        let entity = db
+            .user_deletion_store()
+            .submit("demo:fence-retention", "operator", None)
+            .await
+            .unwrap();
+        let entities = db.entity_deletion_store();
+        entities
+            .approve(
+                entity.id,
+                "approver",
+                entity.inventory_digest.as_deref().unwrap(),
+            )
+            .await
+            .unwrap();
+        let entity_lease = entities
+            .claim(entity.id, "fence-worker", Duration::from_secs(300))
+            .await
+            .unwrap()
+            .unwrap();
+        let human_key = nostr::Keys::generate().public_key();
+        let agent_key = nostr::Keys::generate().public_key();
+        let human = FenceIdentity::user(human_key);
+        let agent = FenceIdentity::agent(request.community_id, agent_key);
+        sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)")
+            .bind(request.community_id.as_uuid())
+            .bind(human_key.to_bytes().to_vec())
+            .execute(&db.pool)
+            .await
+            .unwrap();
+        let mut tx = db.pool.begin().await.unwrap();
+        fences::lock_request(&mut tx, &entity_lease).await.unwrap();
+        fences::lock_exclusive(&mut tx, &[human, agent])
+            .await
+            .unwrap();
+        let global = fences::insert(&mut tx, human, &entity_lease, None)
+            .await
+            .unwrap();
+        let scoped = fences::insert(&mut tx, agent, &entity_lease, Some(human_key))
+            .await
+            .unwrap();
+        tx.commit().await.unwrap();
+        sqlx::query("DELETE FROM users WHERE community_id = $1 AND pubkey = $2")
+            .bind(request.community_id.as_uuid())
+            .bind(human_key.to_bytes().to_vec())
+            .execute(&db.pool)
+            .await
+            .unwrap();
+        let mut tx = db.pool.begin().await.unwrap();
+        assert_eq!(
+            fences::find(&mut tx, human).await.unwrap(),
+            Some(global.clone())
+        );
+        tx.commit().await.unwrap();
+
+        store.approve(request.id, "approver", None).await.unwrap();
+        let claim = store
+            .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION)
+            .await
+            .unwrap()
+            .unwrap();
+        store.begin_quiescing(&claim.lease).await.unwrap();
+        let generation = store.fence(&claim.lease).await.unwrap();
+        let token = LeaseToken {
+            fence_generation: Some(generation),
+            ..claim.lease
+        };
+        store
+            .freeze_destructive_storage_manifest(&token, &inventory.storage)
+            .await
+            .unwrap();
+        store.mark_drained(&token).await.unwrap();
+        store
+            .mark_bindings_removed(&token, serde_json::json!({"keys": 0}))
+            .await
+            .unwrap();
+        store.purge_postgres(&token).await.unwrap();
+        // This database-only fixture models the external cache adapter's
+        // completion before exercising PostgreSQL's logical-absence verifier.
+        store
+            .mark_cache_purged(&token, serde_json::json!({"test_cache": "empty"}))
+            .await
+            .unwrap();
+        store
+            .verify_postgres_logically_deleted(&token)
+            .await
+            .unwrap();
+        // Full row equality also proves that scoped restrictions were not
+        // widened to global ones by clearing scope_community_id.
+        let mut tx = db.pool.begin().await.unwrap();
+        assert_eq!(fences::find(&mut tx, human).await.unwrap(), Some(global));
+        assert_eq!(fences::find(&mut tx, agent).await.unwrap(), Some(scoped));
+        tx.commit().await.unwrap();
+    }
+
     async fn store() -> (Db, DeletionStore) {
         let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
             .or_else(|_| std::env::var("DATABASE_URL"))

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

@@ -140,7 +140,10 @@ impl EntityDeletionStore {
     }
 }
 
-pub(super) async fn lock_owned(tx: &mut Tx<'_>, lease: &EntityLease) -> Result<EntityRequest> {
+pub(in crate::store) async fn lock_owned(
+    tx: &mut Tx<'_>,
+    lease: &EntityLease,
+) -> Result<EntityRequest> {
     // Lock by id first, then check expiry after any wait for that lock. An old
     // snapshot/transaction start time must not extend a worker's authority.
     let request: EntityRequest = sqlx::query_as(sqlx::AssertSqlSafe(format!(
@@ -167,7 +170,7 @@ pub(super) async fn lock_owned(tx: &mut Tx<'_>, lease: &EntityLease) -> Result<E
     Ok(request)
 }
 
-pub(super) async fn validate_approved_inventory(
+pub(in crate::store) async fn validate_approved_inventory(
     tx: &mut Tx<'_>,
     request: &EntityRequest,
 ) -> Result<()> {

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

@@ -25,6 +25,7 @@ mod leases;
 pub(crate) mod requests;
 mod stages;
 pub use approvals::EntityApproval;
+pub(super) use leases::{lock_owned, validate_approved_inventory};
 pub use leases::{EntityExecution, EntityLease};
 pub use requests::{inventory_digest, EntityInspection, EntityRequest};
 use requests::{validate_operator, validated_inventory_digest};

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

@@ -28,6 +28,8 @@ pub(crate) mod event_follow_up;
 pub mod nip_fi_binding;
 /// User-specific deletion submission and inventory preparation.
 pub mod user_deletion;
+/// Operator-global user restrictions and identity-lock primitives.
+pub mod user_deletion_fences;
 
 #[cfg(test)]
 mod event_follow_up_postgres_tests;

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

@@ -0,0 +1,260 @@
+//! Operator-global restriction storage and identity locks; no serving or worker
+//! path calls these helpers yet. Installing fences does not itself enforce them.
+//!
+//! Callers use a writer-pool READ COMMITTED transaction. Acquire existing
+//! community guards first, then `lock_request` for the worker, then all human
+//! locks, then scoped agent locks.
+//! Within each group acquire the whole set in one call (sorted here), never
+//! upgrade shared locks to exclusive. Gate workers may discover agents after
+//! locking humans, then lock the complete agent set. Keep the transaction open
+//! through the protected write/checkpoint; commit or rollback releases the locks.
+
+use buzz_core::CommunityId;
+use chrono::{DateTime, Utc};
+use nostr::PublicKey;
+use sha2::{Digest, Sha256};
+use sqlx::{Postgres, Transaction};
+use uuid::Uuid;
+
+use super::entity_deletion::{lock_owned, validate_approved_inventory, EntityLease};
+use crate::{DbError, Result};
+
+/// A stable lock/restriction identity, independent of mutable profile rows.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct FenceIdentity {
+    pubkey: PublicKey,
+    scope: Option<CommunityId>,
+}
+
+impl FenceIdentity {
+    /// A human key restricted across the operator's hosted communities.
+    pub fn user(pubkey: PublicKey) -> Self {
+        Self {
+            pubkey,
+            scope: None,
+        }
+    }
+
+    /// An agent key restricted only within its ownership community.
+    pub fn agent(community: CommunityId, pubkey: PublicKey) -> Self {
+        Self {
+            pubkey,
+            scope: Some(community),
+        }
+    }
+
+    fn scope_id(self) -> Option<Uuid> {
+        self.scope.map(|scope| *scope.as_uuid())
+    }
+
+    fn lock_key(self) -> i64 {
+        let mut hash = Sha256::new();
+        hash.update(b"buzz:user-deletion-fence:v1\0");
+        hash.update([u8::from(self.scope.is_some())]);
+        if let Some(scope) = self.scope {
+            hash.update(scope.as_uuid().as_bytes());
+        }
+        hash.update(self.pubkey.to_bytes());
+        let digest = hash.finalize();
+        let mut bytes = [0; 8];
+        bytes.copy_from_slice(&digest[..8]);
+        i64::from_be_bytes(bytes)
+    }
+}
+
+/// Durable evidence of the first restriction on an identity and scope.
+#[derive(Clone, Debug, PartialEq, Eq, sqlx::FromRow)]
+pub struct UserDeletionFence {
+    /// Stable restriction identifier.
+    pub id: Uuid,
+    /// Canonical lowercase public key.
+    pub pubkey: String,
+    /// Agent scope, or NULL for a global human restriction.
+    pub scope_community_id: Option<Uuid>,
+    /// Request that first installed this restriction.
+    pub request_id: Uuid,
+    /// Verified owner at fencing time; absent for a human restriction.
+    pub source_owner_pubkey: Option<String>,
+    /// Time of the first insertion; retries preserve it.
+    pub blocked_at: DateTime<Utc>,
+}
+
+/// Acquire shared identity locks for normal writes. Does not check restrictions.
+pub async fn lock_shared(
+    tx: &mut Transaction<'_, Postgres>,
+    identities: &[FenceIdentity],
+) -> Result<()> {
+    lock(tx, identities, false).await
+}
+
+/// Acquire exclusive identity locks for fencing. Does not install restrictions.
+pub async fn lock_exclusive(
+    tx: &mut Transaction<'_, Postgres>,
+    identities: &[FenceIdentity],
+) -> Result<()> {
+    lock(tx, identities, true).await
+}
+
+async fn lock(
+    tx: &mut Transaction<'_, Postgres>,
+    identities: &[FenceIdentity],
+    exclusive: bool,
+) -> Result<()> {
+    if identities.len() > 4096 {
+        return Err(DbError::DeletionSafety(
+            "too many user-deletion identity locks".into(),
+        ));
+    }
+    let (isolation, timeout): (String, String) = sqlx::query_as(
+        "SELECT current_setting('transaction_isolation'), current_setting('lock_timeout')",
+    )
+    .fetch_one(&mut **tx)
+    .await?;
+    if isolation != "read committed" {
+        return Err(DbError::DeletionSafety(
+            "user-deletion locks require READ COMMITTED".into(),
+        ));
+    }
+    // Bound each lock wait without widening a caller's shorter timeout. Restore
+    // the original setting on success; a SQL error requires caller rollback.
+    sqlx::query(
+        "SELECT set_config('lock_timeout',
+         (EXTRACT(EPOCH FROM LEAST(NULLIF(current_setting('lock_timeout')::interval,
+          interval '0'), interval '5 seconds')) * 1000)::bigint::text, true)",
+    )
+    .execute(&mut **tx)
+    .await?;
+    let mut keys: Vec<_> = identities
+        .iter()
+        .map(|identity| (identity.scope.is_some(), identity.lock_key()))
+        .collect();
+    keys.sort_unstable();
+    keys.dedup();
+    for (_, key) in keys {
+        let query = if exclusive {
+            "SELECT pg_advisory_xact_lock($1)"
+        } else {
+            "SELECT pg_advisory_xact_lock_shared($1)"
+        };
+        sqlx::query(query).bind(key).execute(&mut **tx).await?;
+    }
+    sqlx::query("SELECT set_config('lock_timeout', $1, true)")
+        .bind(timeout)
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
+/// Lock and read the exact identity restriction in the caller's transaction.
+///
+/// For multiple identities, acquire the full shared set first. The caller must
+/// refuse a restricted operation and retain this transaction through any write.
+/// An agent lookup is scoped; it does not resolve or check the human owner.
+/// The planned agent-write path locks the verified owner and agent together,
+/// then reads the agent restriction only. This requires the gate to install the
+/// human and all owned-agent restrictions atomically, and creation/assignment
+/// paths to reject restricted owners. These callers are not wired in yet.
+pub async fn find(
+    tx: &mut Transaction<'_, Postgres>,
+    identity: FenceIdentity,
+) -> Result<Option<UserDeletionFence>> {
+    lock_shared(tx, &[identity]).await?;
+    read(tx, identity).await
+}
+
+async fn read(
+    tx: &mut Transaction<'_, Postgres>,
+    identity: FenceIdentity,
+) -> Result<Option<UserDeletionFence>> {
+    // Explicit predicates let PostgreSQL use the matching partial unique index.
+    let sql = if identity.scope.is_some() {
+        "SELECT id, pubkey, scope_community_id, request_id, source_owner_pubkey, blocked_at
+         FROM user_deletion_fences WHERE pubkey = $1 AND scope_community_id = $2"
+    } else {
+        "SELECT id, pubkey, scope_community_id, request_id, source_owner_pubkey, blocked_at
+         FROM user_deletion_fences WHERE pubkey = $1 AND scope_community_id IS NULL"
+    };
+    let query = sqlx::query_as(sql).bind(identity.pubkey.to_hex());
+    let query = if let Some(scope) = identity.scope_id() {
+        query.bind(scope)
+    } else {
+        query
+    };
+    Ok(query.fetch_optional(&mut **tx).await?)
+}
+
+/// Lock the parent request and validate the worker's lease and approved inventory.
+///
+/// Call before acquiring any identity locks in a fencing transaction. Each
+/// insertion rechecks this guard; holding the row lock prevents another worker
+/// from claiming the request until this transaction ends. The caller must still
+/// verify that the identities and agent ownership belong to this request.
+pub async fn lock_request(tx: &mut Transaction<'_, Postgres>, lease: &EntityLease) -> Result<()> {
+    let request = lock_owned(tx, lease).await?;
+    if request.entity_type != "user" {
+        return Err(DbError::DeletionSafety(
+            "fences require a user deletion request".into(),
+        ));
+    }
+    validate_approved_inventory(tx, &request).await
+}
+
+/// Insert an idempotent restriction under a valid worker lease and identity lock.
+///
+/// For multiple identities, first call `lock_request`, then acquire the complete exclusive
+/// identity set in order, verify agent ownership, and save its checkpoint in this
+/// same transaction. A retry preserves the first row; conflicting provenance
+/// errors. On any error roll back, including other fences/checkpoints.
+pub async fn insert(
+    tx: &mut Transaction<'_, Postgres>,
+    identity: FenceIdentity,
+    lease: &EntityLease,
+    source_owner: Option<PublicKey>,
+) -> Result<UserDeletionFence> {
+    if identity.scope.is_some() != source_owner.is_some() {
+        return Err(DbError::DeletionSafety(
+            "agent fences require an owner; user fences do not".into(),
+        ));
+    }
+    lock_request(tx, lease).await?;
+    let request_id = lease.request_id;
+    lock_exclusive(tx, &[identity]).await?;
+    let owner = source_owner.map(|key| key.to_hex());
+    sqlx::query(
+        "INSERT INTO user_deletion_fences (pubkey, scope_community_id, request_id, source_owner_pubkey)
+         VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
+    )
+    .bind(identity.pubkey.to_hex()).bind(identity.scope_id()).bind(request_id).bind(&owner)
+    .execute(&mut **tx).await?;
+    let saved = read(tx, identity)
+        .await?
+        .ok_or_else(|| DbError::DeletionSafety("fence insertion returned no row".into()))?;
+    if saved.request_id != request_id || saved.source_owner_pubkey != owner {
+        return Err(DbError::DeletionSafety(
+            "identity is already fenced by different provenance".into(),
+        ));
+    }
+    Ok(saved)
+}
+
+#[cfg(test)]
+#[path = "user_deletion_fences_postgres_tests.rs"]
+mod postgres_tests;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lock_mapping_is_stable_across_processes_and_releases() {
+        // A change to these values would split the lock domain during rollout.
+        let key =
+            PublicKey::from_hex("79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798")
+                .unwrap();
+        assert_eq!(FenceIdentity::user(key).lock_key(), -4153709860967480559);
+        assert_eq!(
+            FenceIdentity::agent(CommunityId::from_uuid(Uuid::from_u128(1)), key).lock_key(),
+            -3253902268089525902,
+        );
+    }
+}

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

@@ -0,0 +1,453 @@
+use super::*;
+use sqlx::PgPool;
+use std::time::Duration;
+
+async fn pool() -> PgPool {
+    PgPool::connect(&crate::test_support::database_url())
+        .await
+        .unwrap()
+}
+
+async fn request(pool: &PgPool) -> EntityLease {
+    let db = crate::Db::from_pool(pool.clone());
+    let request = db
+        .user_deletion_store()
+        .submit(
+            &format!("demo:{}", Uuid::new_v4().simple()),
+            "operator",
+            None,
+        )
+        .await
+        .unwrap();
+    let store = db.entity_deletion_store();
+    store
+        .approve(
+            request.id,
+            "approver",
+            request.inventory_digest.as_deref().unwrap(),
+        )
+        .await
+        .unwrap();
+    store
+        .claim(request.id, "worker", Duration::from_secs(300))
+        .await
+        .unwrap()
+        .unwrap()
+}
+
+fn user() -> FenceIdentity {
+    FenceIdentity::user(nostr::Keys::generate().public_key())
+}
+
+async fn assert_plain_schema(pool: &PgPool) {
+    let guards: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM pg_trigger WHERE tgrelid = 'user_deletion_fences'::regclass",
+    )
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    assert_eq!(guards, 0);
+    let fks: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM pg_constraint
+         WHERE conrelid = 'user_deletion_fences'::regclass AND contype = 'f'",
+    )
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    assert_eq!(fks, 0);
+    crate::Db::from_pool(pool.clone())
+        .validate_deletion_catalog()
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn desired_schema_is_global_and_has_no_cascades_or_triggers() {
+    assert_plain_schema(&pool().await).await;
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn migration_schema_is_global_and_has_no_cascades_or_triggers() {
+    let pool = pool().await;
+    crate::Db::from_pool(pool.clone()).migrate().await.unwrap();
+    assert_plain_schema(&pool).await;
+    let registered: bool = sqlx::query_scalar(
+        "SELECT EXISTS (SELECT 1 FROM _operator_global_tables WHERE table_name = 'user_deletion_fences')",
+    ).fetch_one(&pool).await.unwrap();
+    assert!(registered);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn fences_are_atomic_idempotent_scoped_and_preserve_provenance() {
+    let pool = pool().await;
+    let lease = request(&pool).await;
+    let human = user();
+    let agent_key = nostr::Keys::generate().public_key();
+    let a = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), agent_key);
+    let b = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), agent_key);
+    let mut tx = pool.begin().await.unwrap();
+    lock_request(&mut tx, &lease).await.unwrap();
+    lock_exclusive(&mut tx, &[a, human]).await.unwrap();
+    insert(&mut tx, human, &lease, None).await.unwrap();
+    insert(&mut tx, a, &lease, Some(human.pubkey))
+        .await
+        .unwrap();
+    tx.rollback().await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(find(&mut tx, human).await.unwrap().is_none());
+    assert!(find(&mut tx, a).await.unwrap().is_none());
+    tx.commit().await.unwrap();
+
+    let mut tx = pool.begin().await.unwrap();
+    lock_request(&mut tx, &lease).await.unwrap();
+    lock_exclusive(&mut tx, &[a, human]).await.unwrap();
+    let first = insert(&mut tx, human, &lease, None).await.unwrap();
+    let scoped = insert(&mut tx, a, &lease, Some(human.pubkey))
+        .await
+        .unwrap();
+    tx.commit().await.unwrap();
+    // A new connection/transaction reads durable state, not an in-memory cache.
+    let mut tx = pool.begin().await.unwrap();
+    assert_eq!(insert(&mut tx, human, &lease, None).await.unwrap(), first);
+    assert_eq!(
+        insert(&mut tx, a, &lease, Some(human.pubkey))
+            .await
+            .unwrap(),
+        scoped
+    );
+    tx.commit().await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    let count: i64 = sqlx::query_scalar("SELECT count(*) FROM user_deletion_fences")
+        .fetch_one(&mut *tx)
+        .await
+        .unwrap();
+    assert_eq!(
+        count, 2,
+        "retries must not create duplicate global or scoped rows"
+    );
+    assert!(find(&mut tx, b).await.unwrap().is_none());
+    assert!(find(&mut tx, FenceIdentity::user(agent_key))
+        .await
+        .unwrap()
+        .is_none());
+    tx.commit().await.unwrap();
+    let other_lease = request(&pool).await;
+    let mut tx = pool.begin().await.unwrap();
+    assert!(insert(&mut tx, human, &other_lease, None).await.is_err());
+    tx.rollback().await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(insert(&mut tx, a, &lease, Some(user().pubkey))
+        .await
+        .is_err());
+    tx.rollback().await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert_eq!(find(&mut tx, human).await.unwrap(), Some(first));
+    assert_eq!(find(&mut tx, a).await.unwrap(), Some(scoped));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn rejects_missing_requests_and_invalid_owner_shape() {
+    let pool = pool().await;
+    let lease = request(&pool).await;
+    let human = user();
+    let agent = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), user().pubkey);
+    let mut tx = pool.begin().await.unwrap();
+    let missing = EntityLease {
+        request_id: Uuid::new_v4(),
+        ..lease.clone()
+    };
+    assert!(insert(&mut tx, human, &missing, None).await.is_err());
+    assert!(insert(&mut tx, human, &lease, Some(user().pubkey))
+        .await
+        .is_err());
+    assert!(insert(&mut tx, agent, &lease, None).await.is_err());
+    let count: i64 = sqlx::query_scalar("SELECT count(*) FROM user_deletion_fences")
+        .fetch_one(&mut *tx)
+        .await
+        .unwrap();
+    assert_eq!(count, 0);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn identity_locks_share_conflict_and_release_at_transaction_end() {
+    let pool = pool().await;
+    let human = user();
+    let mut first = pool.begin().await.unwrap();
+    lock_shared(&mut first, &[human]).await.unwrap();
+    let mut second = pool.begin().await.unwrap();
+    lock_shared(&mut second, &[human]).await.unwrap();
+    second.commit().await.unwrap();
+    let mut second = pool.begin().await.unwrap();
+    sqlx::query("SET LOCAL lock_timeout = '100ms'")
+        .execute(&mut *second)
+        .await
+        .unwrap();
+    let error = lock_exclusive(&mut second, &[human]).await.unwrap_err();
+    assert!(error.to_string().contains("lock timeout"), "{error}");
+    second.rollback().await.unwrap();
+    first.commit().await.unwrap();
+
+    let mut gate = pool.begin().await.unwrap();
+    lock_exclusive(&mut gate, &[human]).await.unwrap();
+    let mut normal = pool.begin().await.unwrap();
+    // An unrelated identity is not blocked.
+    lock_shared(&mut normal, &[user()]).await.unwrap();
+    sqlx::query("SET LOCAL lock_timeout = '100ms'")
+        .execute(&mut *normal)
+        .await
+        .unwrap();
+    assert!(lock_shared(&mut normal, &[human]).await.is_err());
+    normal.rollback().await.unwrap();
+    gate.rollback().await.unwrap();
+    let mut normal = pool.begin().await.unwrap();
+    lock_shared(&mut normal, &[human]).await.unwrap();
+    let timeout: String = sqlx::query_scalar("SHOW lock_timeout")
+        .fetch_one(&mut *normal)
+        .await
+        .unwrap();
+    assert_eq!(timeout, "0");
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn read_waits_for_gate_then_observes_committed_restriction() {
+    let pool = pool().await;
+    let lease = request(&pool).await;
+    let human = user();
+    let mut gate = pool.begin().await.unwrap();
+    let saved = insert(&mut gate, human, &lease, None).await.unwrap();
+    let mut normal = pool.begin().await.unwrap();
+    let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+        .fetch_one(&mut *normal)
+        .await
+        .unwrap();
+    let reader = tokio::spawn(async move { find(&mut normal, human).await.unwrap() });
+    // Observe the actual lock wait before committing, rather than assuming a
+    // sleep was long enough for the competing transaction to reach the guard.
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let waiting: bool = sqlx::query_scalar(
+                "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE pid = $1 AND locktype = 'advisory' AND NOT granted)",
+            ).bind(pid).fetch_one(&pool).await.unwrap();
+            if waiting { break; }
+            tokio::task::yield_now().await;
+        }
+    }).await.unwrap();
+    gate.commit().await.unwrap();
+    assert_eq!(
+        tokio::time::timeout(Duration::from_secs(5), reader)
+            .await
+            .unwrap()
+            .unwrap(),
+        Some(saved)
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn batch_order_is_stable_and_agent_locks_are_community_scoped() {
+    let pool = pool().await;
+    let human = user();
+    let a = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), user().pubkey);
+    let b = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), a.pubkey);
+    let mut first = pool.begin().await.unwrap();
+    lock_exclusive(&mut first, &[a]).await.unwrap();
+    let mut second = pool.begin().await.unwrap();
+    lock_exclusive(&mut second, &[b]).await.unwrap();
+    second.commit().await.unwrap();
+    first.commit().await.unwrap();
+    let run = |identities: Vec<FenceIdentity>| {
+        let pool = pool.clone();
+        async move {
+            let mut tx = pool.begin().await.unwrap();
+            lock_exclusive(&mut tx, &identities).await.unwrap();
+            tx.commit().await.unwrap();
+        }
+    };
+    tokio::time::timeout(Duration::from_secs(10), async {
+        tokio::join!(run(vec![a, human, b, human]), run(vec![b, a, human]));
+    })
+    .await
+    .unwrap();
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn rejects_stale_snapshot_isolation_and_oversized_lock_sets() {
+    let pool = pool().await;
+    let mut tx = pool.begin().await.unwrap();
+    sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
+        .execute(&mut *tx)
+        .await
+        .unwrap();
+    assert!(find(&mut tx, user()).await.is_err());
+    tx.rollback().await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(lock_exclusive(&mut tx, &vec![user(); 4097]).await.is_err());
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn insert_requires_approved_inventory_and_current_unexpired_worker() {
+    let pool = pool().await;
+    let db = crate::Db::from_pool(pool.clone());
+    let store = db.entity_deletion_store();
+    let submitted = db
+        .user_deletion_store()
+        .submit("demo:unapproved", "operator", None)
+        .await
+        .unwrap();
+    assert!(store
+        .claim(submitted.id, "worker", Duration::from_secs(300))
+        .await
+        .unwrap()
+        .is_none());
+    let unapproved = EntityLease {
+        request_id: submitted.id,
+        owner: "worker".into(),
+        generation: 0,
+    };
+    let lease = request(&pool).await;
+    let wrong_worker = EntityLease {
+        owner: "other-worker".into(),
+        ..lease.clone()
+    };
+    let wrong_generation = EntityLease {
+        generation: lease.generation + 1,
+        ..lease.clone()
+    };
+    let identity = user();
+    for rejected in [&unapproved, &wrong_worker, &wrong_generation] {
+        let mut tx = pool.begin().await.unwrap();
+        assert!(matches!(
+            insert(&mut tx, identity, rejected, None).await,
+            Err(DbError::AccessDenied(_))
+        ));
+        tx.rollback().await.unwrap();
+    }
+    // Expire deterministically instead of racing a short wall-clock timeout.
+    sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at = clock_timestamp() - interval '1 second' WHERE id = $1")
+        .bind(lease.request_id).execute(&pool).await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(matches!(
+        insert(&mut tx, identity, &lease, None).await,
+        Err(DbError::AccessDenied(_))
+    ));
+    tx.rollback().await.unwrap();
+    let successor = store
+        .claim(lease.request_id, "successor", Duration::from_secs(300))
+        .await
+        .unwrap()
+        .unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(matches!(
+        insert(&mut tx, identity, &lease, None).await,
+        Err(DbError::AccessDenied(_))
+    ));
+    tx.rollback().await.unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(find(&mut tx, identity).await.unwrap().is_none());
+    tx.commit().await.unwrap();
+
+    let mut tx = pool.begin().await.unwrap();
+    let saved = insert(&mut tx, identity, &successor, None).await.unwrap();
+    tx.commit().await.unwrap();
+    // Exercise the insertion guard independently of claim-time validation.
+    sqlx::query("DELETE FROM entity_deletion_approvals WHERE request_id = $1")
+        .bind(successor.request_id)
+        .execute(&pool)
+        .await
+        .unwrap();
+    let mut tx = pool.begin().await.unwrap();
+    assert!(matches!(
+        insert(&mut tx, user(), &successor, None).await,
+        Err(DbError::DeletionSafety(_))
+    ));
+    tx.rollback().await.unwrap();
+    let rows: Vec<UserDeletionFence> = sqlx::query_as(
+        "SELECT id, pubkey, scope_community_id, request_id, source_owner_pubkey, blocked_at FROM user_deletion_fences"
+    ).fetch_all(&pool).await.unwrap();
+    assert_eq!(rows, vec![saved]);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn competing_inserts_wait_then_preserve_or_reject_original_provenance() {
+    let pool = pool().await;
+    let lease = request(&pool).await;
+    let other_lease = request(&pool).await;
+    for (scoped, conflict) in [(false, false), (false, true), (true, false), (true, true)] {
+        let owner = user().pubkey;
+        let identity = if scoped {
+            FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), user().pubkey)
+        } else {
+            user()
+        };
+        let source_owner = scoped.then_some(owner);
+        let mut first = pool.begin().await.unwrap();
+        let saved = insert(&mut first, identity, &lease, source_owner)
+            .await
+            .unwrap();
+        let next_lease = if conflict && !scoped {
+            other_lease.clone()
+        } else {
+            lease.clone()
+        };
+        let next_owner = if conflict && scoped {
+            Some(user().pubkey)
+        } else {
+            source_owner
+        };
+        let mut second = pool.begin().await.unwrap();
+        let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+            .fetch_one(&mut *second)
+            .await
+            .unwrap();
+        let writer = tokio::spawn(async move {
+            let result = insert(&mut second, identity, &next_lease, next_owner).await;
+            if result.is_ok() {
+                second.commit().await.unwrap();
+            } else {
+                second.rollback().await.unwrap();
+            }
+            result
+        });
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                let waiting: bool = sqlx::query_scalar(
+                    "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE pid = $1 AND NOT granted)",
+                )
+                .bind(pid)
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+                if waiting {
+                    break;
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap();
+        first.commit().await.unwrap();
+        let result = tokio::time::timeout(Duration::from_secs(5), writer)
+            .await
+            .unwrap()
+            .unwrap();
+        if conflict {
+            assert!(
+                matches!(result, Err(DbError::DeletionSafety(ref message)) if message.contains("different provenance"))
+            );
+        } else {
+            assert_eq!(result.unwrap(), saved);
+        }
+        let mut tx = pool.begin().await.unwrap();
+        assert_eq!(find(&mut tx, identity).await.unwrap(), Some(saved));
+        tx.commit().await.unwrap();
+    }
+}

migrations/0074_user_deletion_fences.sql

@@ -0,0 +1,23 @@
+SET LOCAL lock_timeout = '5s';
+
+-- Operator-global restrictions, deliberately retained after profile and community
+-- deletion. scope_community_id is the scope of an agent restriction, not ownership
+-- of this row. Never clear it during community deletion: NULL means a global user.
+CREATE TABLE user_deletion_fences (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    pubkey TEXT NOT NULL CHECK (pubkey ~ '^[0-9a-f]{64}$'),
+    scope_community_id UUID,
+    request_id UUID NOT NULL,
+    source_owner_pubkey TEXT CHECK (source_owner_pubkey ~ '^[0-9a-f]{64}$'),
+    blocked_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CHECK ((scope_community_id IS NULL) = (source_owner_pubkey IS NULL))
+);
+
+CREATE UNIQUE INDEX user_deletion_fences_global_key
+    ON user_deletion_fences (pubkey) WHERE scope_community_id IS NULL;
+CREATE UNIQUE INDEX user_deletion_fences_scoped_key
+    ON user_deletion_fences (pubkey, scope_community_id) WHERE scope_community_id IS NOT NULL;
+
+INSERT INTO _operator_global_tables (table_name, reason)
+VALUES ('user_deletion_fences', 'Operator-global user and agent restrictions retained after profile and community deletion')
+ON CONFLICT (table_name) DO NOTHING;

schema/schema.sql

@@ -125,6 +125,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 \i tables/public/entity_deletion_approvals.sql
 \i tables/public/entity_deletion_requests.sql
 \i tables/public/entity_deletion_stage_history.sql
+\i tables/public/user_deletion_fences.sql
 \i tables/public/community_deletion_approvals.sql
 \i tables/public/community_deletion_checkpoints.sql
 \i tables/public/community_deletion_manifest_keys.sql
@@ -200,6 +201,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 \i seeds/entity_deletion_approvals.sql
 \i seeds/entity_deletion_requests.sql
 \i seeds/entity_deletion_stage_history.sql
+\i seeds/user_deletion_fences.sql
 \i seeds/community_deletion_approvals.sql
 \i seeds/community_deletion_checkpoints.sql
 \i seeds/community_deletion_manifest_keys.sql

schema/seeds/user_deletion_fences.sql

@@ -0,0 +1,3 @@
+INSERT INTO _operator_global_tables (table_name, reason)
+VALUES ('user_deletion_fences', 'Operator-global user and agent restrictions retained after profile and community deletion')
+ON CONFLICT (table_name) DO NOTHING;

schema/tables/public/user_deletion_fences.sql

@@ -0,0 +1,17 @@
+-- Operator-global restrictions, deliberately retained after profile and community
+-- deletion. scope_community_id is the scope of an agent restriction, not ownership
+-- of this row. Never clear it during community deletion: NULL means a global user.
+CREATE TABLE user_deletion_fences (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    pubkey TEXT NOT NULL CHECK (pubkey ~ '^[0-9a-f]{64}$'),
+    scope_community_id UUID,
+    request_id UUID NOT NULL,
+    source_owner_pubkey TEXT CHECK (source_owner_pubkey ~ '^[0-9a-f]{64}$'),
+    blocked_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CHECK ((scope_community_id IS NULL) = (source_owner_pubkey IS NULL))
+);
+
+CREATE UNIQUE INDEX user_deletion_fences_global_key
+    ON user_deletion_fences (pubkey) WHERE scope_community_id IS NULL;
+CREATE UNIQUE INDEX user_deletion_fences_scoped_key
+    ON user_deletion_fences (pubkey, scope_community_id) WHERE scope_community_id IS NOT NULL;