Merge owner preparation authority repair into Desktop deletion draft

Signed-off-by: Codex <noreply@openai.com>

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

@@ -1419,6 +1419,28 @@ impl DeletionStore {
         let storage = serde_json::to_value(&inventory.storage)?;
         let frozen = serde_json::to_value(inventory)?;
         let mut tx = self.pool.begin().await?;
+
+        lock_community_deletion_shared(&mut tx, token.community_id).await?;
+        let community = sqlx::query(
+            "SELECT archived_at, deletion_state, deleted_at FROM communities \
+             WHERE id = $1 FOR UPDATE",
+        )
+        .bind(token.community_id.as_uuid())
+        .fetch_optional(&mut *tx)
+        .await?
+        .ok_or_else(|| {
+            DbError::DeletionSafety(format!(
+                "owner deletion {} community is missing before automatic approval",
+                token.request_id
+            ))
+        })?;
+        let current_owners: Vec<String> = sqlx::query_scalar(
+            "SELECT pubkey FROM relay_members \
+             WHERE community_id = $1 AND role = 'owner' ORDER BY pubkey FOR UPDATE",
+        )
+        .bind(token.community_id.as_uuid())
+        .fetch_all(&mut *tx)
+        .await?;
         let request_row =
             sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE")
                 .bind(token.request_id)
@@ -1446,6 +1468,22 @@ impl DeletionStore {
         if !lease_matches {
             return Err(stale_lease_error(token));
         }
+        let archived_at: Option<DateTime<Utc>> = community.try_get("archived_at")?;
+        let deletion_state: String = community.try_get("deletion_state")?;
+        let deleted_at: Option<DateTime<Utc>> = community.try_get("deleted_at")?;
+        let owner_authority_matches = request.owner_pubkey.as_ref().is_some_and(|owner| {
+            current_owners.len() == 1 && current_owners.first() == Some(owner)
+        });
+        if archived_at.is_none()
+            || deletion_state != "active"
+            || deleted_at.is_some()
+            || !owner_authority_matches
+        {
+            return Err(DbError::DeletionSafety(format!(
+                "owner deletion {} community is no longer archived under the admitted owner",
+                token.request_id
+            )));
+        }
         if request.stage == DeletionStage::Approved {
             let approval: Option<(Vec<u8>, String, String)> = sqlx::query_as(
                 "SELECT inventory_digest, approved_by, approval_origin \
@@ -5128,6 +5166,77 @@ mod postgres_tests {
         );
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn privileged_abort_fences_a_live_owner_preparation_lease() {
+        let (db, store) = store().await;
+        let (host, owner, community) = archived_owned_community(&db).await;
+        let operator = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+        let request_id = Uuid::new_v4();
+        store
+            .admit_owner_request(&host, &owner, operator, 1, request_id)
+            .await
+            .expect("admit owner request");
+        let claim = store
+            .claim_specific_owner_submission(request_id, "preparer", DEFAULT_LEASE_DURATION)
+            .await
+            .expect("claim owner request")
+            .expect("owner request is preparable");
+        let inventory = FrozenInventory {
+            schema: store
+                .inventory_schema(community)
+                .await
+                .expect("schema inventory"),
+            storage: empty_storage_manifest(community),
+        };
+        let (lease_owner, lease_generation, lease_is_live): (Option<String>, i64, bool) =
+            sqlx::query_as(
+                "SELECT lease_owner, lease_generation, lease_until >= now() \
+                 FROM community_deletion_requests WHERE id = $1",
+            )
+            .bind(request_id)
+            .fetch_one(&db.pool)
+            .await
+            .expect("read live preparation lease");
+        assert_eq!(lease_owner.as_deref(), Some("preparer"));
+        assert_eq!(lease_generation, claim.lease.generation);
+        assert!(lease_is_live, "preparation lease must be live before abort");
+
+        let aborted = store
+            .abort(request_id, "recovery-operator", "cancel live preparation")
+            .await
+            .expect("abort live preparation");
+        assert_eq!(aborted.stage, DeletionStage::Aborted);
+        assert_eq!(aborted.lease_generation, claim.lease.generation + 1);
+        assert!(aborted.lease_owner.is_none());
+        assert!(aborted.lease_until.is_none());
+
+        let heartbeat_error = store
+            .heartbeat_owner_submission(&claim.lease, "drain", DEFAULT_LEASE_DURATION, false)
+            .await
+            .expect_err("aborted preparation lease cannot heartbeat");
+        assert!(is_stale_deletion_lease(&heartbeat_error));
+        let completion_error = store
+            .complete_owner_preparation(&claim.lease, &inventory)
+            .await
+            .expect_err("aborted preparation lease cannot approve");
+        assert!(is_stale_deletion_lease(&completion_error));
+
+        let request = store.get(request_id).await.expect("load aborted request");
+        assert_eq!(request.stage, DeletionStage::Aborted);
+        assert!(request.inventory_digest.is_none());
+        assert_eq!(
+            sqlx::query_scalar::<_, i64>(
+                "SELECT count(*) FROM community_deletion_approvals WHERE request_id = $1",
+            )
+            .bind(request_id)
+            .fetch_one(&db.pool)
+            .await
+            .expect("count automatic approvals"),
+            0
+        );
+    }
+
     /// The reversible boundary extends through `fenced`. From `drained`
     /// onward, destruction may have begun, so abort must stay closed.
     #[tokio::test]
@@ -5504,6 +5613,92 @@ mod postgres_tests {
         );
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn owner_preparation_rechecks_archived_current_owner_before_automatic_approval() {
+        enum AuthorityDrift {
+            Unarchived,
+            OwnerChanged,
+        }
+
+        let mut failures = Vec::new();
+        for drift in [AuthorityDrift::Unarchived, AuthorityDrift::OwnerChanged] {
+            let label = match drift {
+                AuthorityDrift::Unarchived => "unarchived",
+                AuthorityDrift::OwnerChanged => "owner-changed",
+            };
+            let (db, store) = store().await;
+            let (host, owner, community) = archived_owned_community(&db).await;
+            let operator = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+            let request_id = Uuid::new_v4();
+            store
+                .admit_owner_request(&host, &owner, operator, 1, request_id)
+                .await
+                .expect("admit owner request");
+            let claim = store
+                .claim_specific_owner_submission(request_id, "preparer", DEFAULT_LEASE_DURATION)
+                .await
+                .expect("claim owner request")
+                .expect("owner request is preparable");
+            let inventory = FrozenInventory {
+                schema: store
+                    .inventory_schema(community)
+                    .await
+                    .expect("schema inventory"),
+                storage: empty_storage_manifest(community),
+            };
+
+            match drift {
+                AuthorityDrift::Unarchived => {
+                    sqlx::query("UPDATE communities SET archived_at = NULL WHERE id = $1")
+                        .bind(community.as_uuid())
+                        .execute(&db.pool)
+                        .await
+                        .expect("simulate stale archive authority");
+                }
+                AuthorityDrift::OwnerChanged => {
+                    sqlx::query(
+                        "UPDATE relay_members SET role = 'member' \
+                         WHERE community_id = $1 AND pubkey = $2 AND role = 'owner'",
+                    )
+                    .bind(community.as_uuid())
+                    .bind(&owner)
+                    .execute(&db.pool)
+                    .await
+                    .expect("simulate stale owner authority");
+                }
+            }
+
+            let completion = store
+                .complete_owner_preparation(&claim.lease, &inventory)
+                .await;
+            let request = store.get(request_id).await.expect("load request");
+            let approval_count = sqlx::query_scalar::<_, i64>(
+                "SELECT count(*) FROM community_deletion_approvals WHERE request_id = $1",
+            )
+            .bind(request_id)
+            .fetch_one(&db.pool)
+            .await
+            .expect("count automatic approvals");
+            if completion.is_ok()
+                || request.stage != DeletionStage::Submitted
+                || request.inventory_digest.is_some()
+                || approval_count != 0
+            {
+                failures.push(format!(
+                    "{label}: completion={completion:?}, stage={}, inventory_frozen={}, approvals={approval_count}",
+                    request.stage,
+                    request.inventory_digest.is_some(),
+                ));
+            }
+        }
+
+        assert!(
+            failures.is_empty(),
+            "stale owner authority reached automatic approval: {failures:#?}"
+        );
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn stale_owner_preparation_generation_cannot_freeze_or_approve() {

docs/operator-community-deletion.md

@@ -6,10 +6,11 @@ that command as a Kubernetes CronJob; it does not call relay HTTP and it does
 not add another queue or retry service.
 
 Postgres remains the handoff and source of truth. A run gives already-approved
-work priority. When none is ready, it may claim an operator-attested owner-origin
-request at `submitted`, build the existing bounded inventory, and atomically
-freeze that inventory with a digest-bound `owner_automatic` approval. The same
-lease then enters the unchanged executor and resumes from durable checkpoints.
+work priority. When none is ready, it may claim an operator-attested
+owner-origin request at `submitted`, build the existing bounded inventory, and
+atomically freeze that inventory with a digest-bound `owner_automatic`
+approval. The same lease then enters the unchanged executor and resumes from
+durable checkpoints.
 Operator-origin requests never auto-progress. `concurrencyPolicy: Forbid` prevents scheduled
 pod overlap, `backoffLimit: 0` prevents Kubernetes Job retries, and the deletion
 store remains authoritative when a pod exits, reaches its deadline, or is
@@ -144,8 +145,8 @@ handoff.
 
 Owner self-serve relay admission still records only a `submitted` row and does
 no inventory, approval, S3 work, or execution synchronously. A successful drain
-has no human approval step or cooling-off period: operator-attested owner intent is
-prepared automatically under privileged policy and becomes immediately
+has no human approval step or cooling-off period: operator-attested owner intent
+is prepared automatically under privileged policy and becomes immediately
 eligible for execution. Transient preparation failures use the existing retry
 schedule; permanent or exhausted failures block durably. Owner-facing
 admission has no cancellation endpoint.