diff --git a/crates/buzz-audit/src/entry.rs b/crates/buzz-audit/src/entry.rs
index 33b51f8cf3e..fa6561c4263 100644
--- a/crates/buzz-audit/src/entry.rs
+++ b/crates/buzz-audit/src/entry.rs
@@ -43,11 +43,11 @@ pub struct AuditEntry {
 /// `TenantContext`), never a client-supplied value — the same provenance rule
 /// the whole multi-tenant model rests on.
 ///
-/// Not `Serialize`/`Deserialize`: this is an in-process input struct (consumed
-/// by `AuditService::log`, threaded through the in-memory audit sink), never
-/// crossing a wire or DB boundary as a whole. Keeping it non-deserializable
-/// reinforces the fence — there is no path by which a client-supplied blob
-/// becomes a `NewAuditEntry` (and thus a `CommunityId`).
+/// Not `Serialize`/`Deserialize`: this is an in-process input struct consumed by
+/// `AuditService`; durable outbox columns are bound individually, never decoded
+/// from a client-supplied blob. Keeping it non-deserializable reinforces the
+/// fence — there is no path by which client JSON becomes a `NewAuditEntry` (and
+/// thus a `CommunityId`).
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct NewAuditEntry {
     /// Server-resolved community this entry belongs to. Typed as [`CommunityId`]
diff --git a/crates/buzz-audit/src/lib.rs b/crates/buzz-audit/src/lib.rs
index 0248a7dfd3f..8aca3e1e0d5 100644
--- a/crates/buzz-audit/src/lib.rs
+++ b/crates/buzz-audit/src/lib.rs
@@ -32,4 +32,4 @@ pub use action::AuditAction;
 pub use entry::{AuditEntry, NewAuditEntry};
 pub use error::AuditError;
 pub use hash::{compute_hash, GENESIS_HASH};
-pub use service::AuditService;
+pub use service::{AuditService, PendingAuditResult};
diff --git a/crates/buzz-audit/src/service.rs b/crates/buzz-audit/src/service.rs
index 9ae1d168590..e2de8301c55 100644
--- a/crates/buzz-audit/src/service.rs
+++ b/crates/buzz-audit/src/service.rs
@@ -1,6 +1,6 @@
 use chrono::{DateTime, Utc};
 use futures_util::FutureExt as _;
-use sqlx::{Acquire, PgPool, Row};
+use sqlx::{Acquire, PgPool, Postgres, Row, Transaction};
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -26,9 +26,35 @@ fn log_timestamp() -> DateTime<Utc> {
 /// Per-community advisory lock key. Derived in Postgres from the community UUID
 /// so two communities never serialize each other's audit writes (which would be
 /// both a throughput bottleneck and a cross-tenant timing oracle). The lock is
-/// taken with `pg_advisory_lock(hashtextextended(...))` — see [`AuditService::log`].
+/// taken with `pg_advisory_lock` for direct appends and
+/// `pg_advisory_xact_lock` for durable outbox delivery.
 const AUDIT_LOCK_NAMESPACE: &str = "buzz_audit:";
 
+/// Outcome of one durable audit-outbox delivery attempt.
+#[derive(Debug)]
+pub enum PendingAuditResult {
+    /// No community currently has a due head entry.
+    Idle,
+    /// One pending entry was appended and removed from the outbox atomically.
+    Appended,
+    /// A timeout deferred the entry without blocking other communities.
+    Deferred {
+        /// Number of delivery attempts made for this entry.
+        attempt_count: i32,
+        /// Delay before the entry becomes eligible again.
+        retry_delay_ms: u64,
+        /// PostgreSQL SQLSTATE that made the attempt retryable.
+        sqlstate: String,
+    },
+}
+
+struct ClaimedAuditEntry {
+    id: Uuid,
+    entry: NewAuditEntry,
+    enqueued_at: DateTime<Utc>,
+    attempt_count: i32,
+}
+
 /// Append-only, per-community hash-chain audit log backed by Postgres.
 ///
 /// Each community has an independent chain keyed `(community_id, seq)`. Writes
@@ -45,6 +71,172 @@ impl AuditService {
         Self { pool }
     }
 
+    /// Persist an audit intent before the producer reports completion.
+    ///
+    /// Delivery is asynchronous, but the intent survives relay shutdown and is
+    /// ordered by `enqueue_seq` within its community. A logical `dedupe_key`
+    /// suppresses retries after both pending and completed delivery.
+    pub async fn enqueue(
+        &self,
+        entry: NewAuditEntry,
+        dedupe_key: Option<&str>,
+    ) -> Result<(), AuditError> {
+        let mut tx = self.pool.begin().await?;
+        if let Some(dedupe_key) = dedupe_key {
+            let inserted = sqlx::query(
+                "INSERT INTO audit_delivery_keys (community_id, dedupe_key) \
+                 VALUES ($1, $2) \
+                 ON CONFLICT (community_id, dedupe_key) DO NOTHING",
+            )
+            .bind(entry.community_id.as_uuid())
+            .bind(dedupe_key)
+            .execute(&mut *tx)
+            .await?;
+            if inserted.rows_affected() == 0 {
+                tx.commit().await?;
+                return Ok(());
+            }
+        }
+        sqlx::query(
+            "INSERT INTO audit_outbox \
+                 (community_id, action, actor_pubkey, object_id, detail) \
+             VALUES ($1, $2, $3, $4, $5)",
+        )
+        .bind(entry.community_id.as_uuid())
+        .bind(entry.action.as_str())
+        .bind(entry.actor_pubkey.as_deref())
+        .bind(entry.object_id.as_deref())
+        .bind(&entry.detail)
+        .execute(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    /// Deliver one due per-community head from the durable outbox.
+    ///
+    /// A short claim lease lets multiple relay processes share the outbox.
+    /// Timeout failures are rescheduled with capped exponential backoff, so a
+    /// contended community cannot monopolize a worker. Appending the hash-chain
+    /// row and deleting its outbox row occur in the same transaction.
+    pub async fn deliver_next_pending(&self) -> Result<PendingAuditResult, AuditError> {
+        let Some(claimed) = self.claim_next_pending().await? else {
+            return Ok(PendingAuditResult::Idle);
+        };
+
+        match self.append_claimed(&claimed).await {
+            Ok(()) => Ok(PendingAuditResult::Appended),
+            Err(error) => {
+                let Some(sqlstate) = retryable_sqlstate(&error) else {
+                    return Err(error);
+                };
+                let retry_delay_ms = retry_delay_ms(claimed.attempt_count);
+                self.defer_claimed(&claimed, retry_delay_ms).await?;
+                Ok(PendingAuditResult::Deferred {
+                    attempt_count: claimed.attempt_count,
+                    retry_delay_ms,
+                    sqlstate,
+                })
+            }
+        }
+    }
+
+    async fn claim_next_pending(&self) -> Result<Option<ClaimedAuditEntry>, AuditError> {
+        let row = sqlx::query(
+            r#"
+            WITH candidate AS (
+                SELECT pending.community_id, pending.id
+                FROM audit_outbox AS pending
+                WHERE pending.next_attempt_at <= clock_timestamp()
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM audit_outbox AS older
+                      WHERE older.community_id = pending.community_id
+                        AND older.enqueue_seq < pending.enqueue_seq
+                  )
+                ORDER BY pending.next_attempt_at, pending.enqueue_seq
+                FOR UPDATE OF pending SKIP LOCKED
+                LIMIT 1
+            )
+            UPDATE audit_outbox AS pending
+            SET attempt_count = pending.attempt_count + 1,
+                next_attempt_at = clock_timestamp() + INTERVAL '5 seconds'
+            FROM candidate
+            WHERE pending.community_id = candidate.community_id
+              AND pending.id = candidate.id
+            RETURNING pending.id, pending.community_id, pending.action,
+                      pending.actor_pubkey, pending.object_id, pending.detail,
+                      pending.enqueued_at, pending.attempt_count
+            "#,
+        )
+        .fetch_optional(&self.pool)
+        .await?;
+
+        let Some(row) = row else {
+            return Ok(None);
+        };
+        let action_string: String = row.get("action");
+        let action = action_string
+            .parse()
+            .map_err(|_| AuditError::UnknownAction)?;
+        Ok(Some(ClaimedAuditEntry {
+            id: row.get("id"),
+            entry: NewAuditEntry {
+                community_id: CommunityId::from_uuid(row.get("community_id")),
+                action,
+                actor_pubkey: row.get("actor_pubkey"),
+                object_id: row.get("object_id"),
+                detail: row.get("detail"),
+            },
+            enqueued_at: row.get("enqueued_at"),
+            attempt_count: row.get("attempt_count"),
+        }))
+    }
+
+    async fn append_claimed(&self, claimed: &ClaimedAuditEntry) -> Result<(), AuditError> {
+        let mut tx = self.pool.begin().await?;
+        let lock_key = format!("{AUDIT_LOCK_NAMESPACE}{}", claimed.entry.community_id);
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+            .bind(&lock_key)
+            .execute(&mut *tx)
+            .await?;
+        Self::append_in_transaction(
+            &mut tx,
+            claimed.entry.clone(),
+            to_storage_precision(claimed.enqueued_at),
+        )
+        .await?;
+        let deleted = sqlx::query("DELETE FROM audit_outbox WHERE community_id = $1 AND id = $2")
+            .bind(claimed.entry.community_id.as_uuid())
+            .bind(claimed.id)
+            .execute(&mut *tx)
+            .await?;
+        if deleted.rows_affected() != 1 {
+            return Err(AuditError::Database(sqlx::Error::RowNotFound));
+        }
+        tx.commit().await?;
+        Ok(())
+    }
+
+    async fn defer_claimed(
+        &self,
+        claimed: &ClaimedAuditEntry,
+        retry_delay_ms: u64,
+    ) -> Result<(), AuditError> {
+        sqlx::query(
+            "UPDATE audit_outbox \
+             SET next_attempt_at = clock_timestamp() \
+                 + make_interval(secs => $3::double precision / 1000.0) \
+             WHERE community_id = $1 AND id = $2",
+        )
+        .bind(claimed.entry.community_id.as_uuid())
+        .bind(claimed.id)
+        .bind(retry_delay_ms as f64)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
     /// Append a new entry to the calling community's chain.
     ///
     /// Serialized per-community via `pg_advisory_lock`. Postgres advisory locks
@@ -91,6 +283,17 @@ impl AuditService {
     ) -> Result<AuditEntry, AuditError> {
         let mut tx = conn.begin().await?;
 
+        let audit_entry = Self::append_in_transaction(&mut tx, entry, log_timestamp()).await?;
+        tx.commit().await?;
+
+        Ok(audit_entry)
+    }
+
+    async fn append_in_transaction(
+        tx: &mut Transaction<'_, Postgres>,
+        entry: NewAuditEntry,
+        created_at: DateTime<Utc>,
+    ) -> Result<AuditEntry, AuditError> {
         // The stored row keys on the raw UUID; the typed `CommunityId` on the
         // input is the provenance fence, dereferenced here at the DB boundary.
         let community_id = *entry.community_id.as_uuid();
@@ -102,7 +305,7 @@ impl AuditService {
              ORDER BY seq DESC LIMIT 1",
         )
         .bind(community_id)
-        .fetch_optional(&mut *tx)
+        .fetch_optional(&mut **tx)
         .await?;
 
         let (prev_seq, prev_hash): (i64, Option<Vec<u8>>) = match head {
@@ -114,8 +317,6 @@ impl AuditService {
         };
         let seq = prev_seq + 1;
 
-        let created_at: DateTime<Utc> = log_timestamp();
-
         let mut audit_entry = AuditEntry {
             community_id,
             seq,
@@ -148,11 +349,9 @@ impl AuditService {
         .bind(audit_entry.object_id.as_deref())
         .bind(&audit_entry.detail)
         .bind(audit_entry.created_at)
-        .execute(&mut *tx)
+        .execute(&mut **tx)
         .await?;
 
-        tx.commit().await?;
-
         Ok(audit_entry)
     }
 
@@ -248,6 +447,19 @@ impl AuditService {
     }
 }
 
+fn retryable_sqlstate(error: &AuditError) -> Option<String> {
+    let AuditError::Database(sqlx::Error::Database(database_error)) = error else {
+        return None;
+    };
+    let code = database_error.code()?;
+    matches!(code.as_ref(), "55P03" | "57014").then(|| code.into_owned())
+}
+
+fn retry_delay_ms(attempt_count: i32) -> u64 {
+    let exponent = attempt_count.saturating_sub(1).clamp(0, 5) as u32;
+    (50u64.saturating_mul(1u64 << exponent)).min(1_000)
+}
+
 fn row_to_audit_entry(row: &sqlx::postgres::PgRow) -> Result<AuditEntry, AuditError> {
     let action_str: String = row.get("action");
     let action: AuditAction = action_str.parse().map_err(|_| {
diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs
index 00cd81c6940..82d6104734d 100644
--- a/crates/buzz-db/src/runtime/migration.rs
+++ b/crates/buzz-db/src/runtime/migration.rs
@@ -699,7 +699,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 40);
+        assert_eq!(migrations.len(), 41);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1338,6 +1338,26 @@ mod tests {
         assert!(!desired_schema.contains("NEW.kind IN (7, 9, 1059, 40007, 46010)"));
     }
 
+    #[test]
+    fn durable_audit_outbox_is_additive_deduplicated_and_fenced() {
+        let migration = MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 41)
+            .expect("durable audit outbox migration");
+        let sql = migration.sql.as_str();
+        assert!(sql.contains("CREATE TABLE audit_outbox"));
+        assert!(sql.contains("CREATE TABLE audit_delivery_keys"));
+        assert!(sql.contains("GENERATED ALWAYS AS IDENTITY"));
+        assert!(sql.contains("attach_community_write_fence('audit_outbox')"));
+        assert!(sql.contains("attach_community_write_fence('audit_delivery_keys')"));
+
+        let desired_schema = include_str!("../../../../schema/schema.sql");
+        assert!(desired_schema.contains("CREATE TABLE audit_outbox"));
+        assert!(desired_schema.contains("CREATE TABLE audit_delivery_keys"));
+        assert!(desired_schema.contains("attach_community_write_fence('audit_outbox')"));
+        assert!(desired_schema.contains("attach_community_write_fence('audit_delivery_keys')"));
+    }
+
     #[test]
     fn migration_lint_detects_tables_missing_community_id_by_default() {
         let sql = r#"
@@ -1776,6 +1796,9 @@ mod tests {
         let mut expected_fences = migration.fence_attachments.clone();
         expected_fences.remove("product_feedback");
         expected_fences.remove("rate_limit_violations");
+        for additive in MIGRATOR.iter().filter(|migration| migration.version > 29) {
+            expected_fences.extend(surface(additive.sql.as_ref()).fence_attachments);
+        }
         assert_eq!(
             expected_fences, schema.fence_attachments,
             "write-fence attachment targets differ after recovery policy"
@@ -2425,6 +2448,8 @@ mod tests {
             "channels",
             "scheduled_workflow_fires",
             "audit_log",
+            "audit_outbox",
+            "audit_delivery_keys",
         ] {
             let exists = sqlx::query_scalar::<_, bool>(
                 "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1)",
diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs
index 8349fb53874..4083cca3089 100644
--- a/crates/buzz-db/src/runtime/tests.rs
+++ b/crates/buzz-db/src/runtime/tests.rs
@@ -2389,6 +2389,64 @@ async fn session_timeouts_install_through_db_new_and_bound_lock_waits() {
     drop_scratch_db(&admin, db.pool.clone(), &name).await;
 }
 
+/// The idle-transaction timeout must terminate the wedged backend, and SQLx
+/// must replace it with a freshly configured writer connection.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn idle_transaction_timeout_reaps_backend_and_pool_replaces_connection() {
+    let admin = PgPool::connect(&admin_url().await)
+        .await
+        .expect("connect admin");
+    let (seed_pool, name) = create_scratch_db(&admin, "idle_txn_reap").await;
+    seed_pool.close().await;
+
+    let base = admin_url().await;
+    let idx = base.rfind('/').expect("db url has a path segment");
+    let scratch_url = format!("{}/{}", &base[..idx], name);
+    let db = Db::new(&DbConfig {
+        database_url: scratch_url,
+        max_connections: 1,
+        min_connections: 0,
+        idle_txn_timeout_ms: 100,
+        ..DbConfig::default()
+    })
+    .await
+    .expect("connect Db with idle transaction timeout");
+
+    let mut wedged = db.pool.acquire().await.expect("acquire writer connection");
+    let old_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+        .fetch_one(&mut *wedged)
+        .await
+        .expect("read original backend pid");
+    sqlx::query("BEGIN")
+        .execute(&mut *wedged)
+        .await
+        .expect("begin transaction");
+    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    sqlx::query("SELECT 1")
+        .execute(&mut *wedged)
+        .await
+        .expect_err("idle transaction backend must be terminated");
+    drop(wedged);
+
+    let mut replacement =
+        tokio::time::timeout(std::time::Duration::from_secs(3), db.pool.acquire())
+            .await
+            .expect("pool did not replace terminated backend")
+            .expect("acquire replacement writer connection");
+    let (new_pid, timeout): (i32, String) = sqlx::query_as(
+        "SELECT pg_backend_pid(), current_setting('idle_in_transaction_session_timeout')",
+    )
+    .fetch_one(&mut *replacement)
+    .await
+    .expect("inspect replacement backend");
+    assert_ne!(new_pid, old_pid, "pool must replace the terminated backend");
+    assert_eq!(timeout, "100ms", "replacement must reinstall the timeout");
+    drop(replacement);
+
+    drop_scratch_db(&admin, db.pool.clone(), &name).await;
+}
+
 /// The armed writer pool (`Db::new`) must enforce the floor end-to-end
 /// through the public insert APIs, and the session GUC must be verifiably
 /// set on pooled connections.
diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs
index c7fcdc09f66..5357868e6e6 100644
--- a/crates/buzz-db/src/store/deletion.rs
+++ b/crates/buzz-db/src/store/deletion.rs
@@ -58,6 +58,8 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
     "api_tokens",
     "archived_identities",
     "audit_log",
+    "audit_outbox",
+    "audit_delivery_keys",
     "channel_members",
     "channels",
     "community_bans",
@@ -109,6 +111,8 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[
     "parameterized_event_watermarks",
     "git_repo_names",
     "archived_identities",
+    "audit_outbox",
+    "audit_delivery_keys",
     "audit_log",
     "community_bans",
     "pubkey_allowlist",
diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs
index 780532ec5d0..96f8dfb832e 100644
--- a/crates/buzz-relay/src/api/media.rs
+++ b/crates/buzz-relay/src/api/media.rs
@@ -456,32 +456,41 @@ pub async fn upload_blob(
     )
     .increment(1);
 
-    // Audit via bounded channel — same pattern as event audit.
-    if let Some(audit_tx) = &state.audit_tx {
+    // Persist the audit intent before returning — same pattern as event audit.
+    if let Some(audit_queue) = &state.audit_queue {
         let desc = descriptor.clone();
-        if let Err(e) = audit_tx
-            .send(NewAuditEntry {
-                community_id: auth.tenant.community(),
-                action: AuditAction::MediaUploaded,
-                actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
-                object_id: Some(desc.sha256.clone()),
-                detail: serde_json::json!({
-                    "sha256": desc.sha256,
-                    "size": desc.size,
-                    "mime": desc.mime_type,
-                }),
-            })
+        let dedupe_key = media_audit_dedupe_key(&auth.auth_event);
+        audit_queue
+            .send(
+                NewAuditEntry {
+                    community_id: auth.tenant.community(),
+                    action: AuditAction::MediaUploaded,
+                    actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
+                    object_id: Some(desc.sha256.clone()),
+                    detail: serde_json::json!({
+                        "sha256": desc.sha256,
+                        "size": desc.size,
+                        "mime": desc.mime_type,
+                    }),
+                },
+                Some(&dedupe_key),
+            )
             .await
-        {
-            tracing::error!("Media audit channel closed — entry lost: {e}");
-            metrics::counter!("buzz_audit_send_errors_total").increment(1);
-        }
+            .map_err(|e| {
+                tracing::error!("Media audit intent could not be persisted: {e}");
+                metrics::counter!("buzz_audit_send_errors_total").increment(1);
+                MediaError::Internal
+            })?;
     }
 
     serving_write.finish().await.map_err(serving_lease_lost)?;
     Ok(Json(descriptor))
 }
 
+fn media_audit_dedupe_key(auth_event: &nostr::Event) -> String {
+    format!("media_uploaded:{}", auth_event.id)
+}
+
 pub(crate) fn media_base_url_for_tenant(config_relay_url: &str, tenant_host: &str) -> String {
     let scheme = if config_relay_url.trim_start().starts_with("wss://")
         || config_relay_url.trim_start().starts_with("https://")
@@ -1023,6 +1032,27 @@ mod tests {
 
     const VALID_HASH: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
 
+    #[test]
+    fn media_audit_dedupe_key_tracks_the_signed_upload_action() {
+        let first = EventBuilder::new(Kind::from(24242), "Upload media")
+            .sign_with_keys(&Keys::generate())
+            .expect("sign first upload authorization");
+        let second = EventBuilder::new(Kind::from(24242), "Upload media")
+            .sign_with_keys(&Keys::generate())
+            .expect("sign second upload authorization");
+
+        assert_eq!(
+            media_audit_dedupe_key(&first),
+            media_audit_dedupe_key(&first),
+            "a retry with the same signed authorization must dedupe",
+        );
+        assert_ne!(
+            media_audit_dedupe_key(&first),
+            media_audit_dedupe_key(&second),
+            "distinct signed uploads of identical bytes must remain distinct",
+        );
+    }
+
     #[test]
     fn serving_write_error_taxonomy_separates_fence_from_backend_failure() {
         let fenced = anyhow::Error::from(buzz_db::DbError::AccessDenied("fenced".to_string()));
diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs
index ccba40f3282..a4df61aa659 100644
--- a/crates/buzz-relay/src/handlers/event.rs
+++ b/crates/buzz-relay/src/handlers/event.rs
@@ -339,11 +339,11 @@ pub async fn fan_out_pubsub_event(state: &Arc<AppState>, channel_event: buzz_pub
 
 /// Schedule post-commit delivery/side effects for a stored event.
 ///
-/// This intentionally returns after only the bounded audit enqueue has completed:
+/// This intentionally returns after only the durable audit enqueue has completed:
 /// NIP-01 `OK` means the event was durably accepted, not that Redis publish,
 /// local fan-out, or workflow triggering have completed. Keeping audit enqueue on
-/// the awaited path preserves the bounded-channel backpressure posture when the
-/// audit DB is overloaded; the spawned task still runs the same guarded fan-out
+/// the awaited path preserves audit intent before acceptance when the audit DB
+/// is overloaded; the spawned task still runs the same guarded fan-out
 /// path, Redis publish, `mark_local_event` echo dedupe, and delivery metrics as
 /// the former inline path.
 pub(crate) async fn dispatch_persistent_event(
@@ -353,17 +353,10 @@ pub(crate) async fn dispatch_persistent_event(
     kind_u32: u32,
     actor_pubkey_hex: &str,
     threaded_visibility: Option<crate::state::ThreadedChannelVisibility>,
-) -> usize {
+) -> Result<usize, buzz_audit::AuditError> {
+    enqueue_persistent_event_audit(tenant, state, stored_event, kind_u32, actor_pubkey_hex).await?;
+
     let event_id_hex = stored_event.event.id.to_hex();
-    enqueue_event_created_audit(
-        tenant,
-        state,
-        stored_event,
-        kind_u32,
-        actor_pubkey_hex,
-        &event_id_hex,
-    )
-    .await;
 
     let tenant = tenant.clone();
     let state = Arc::clone(state);
@@ -389,7 +382,31 @@ pub(crate) async fn dispatch_persistent_event(
         );
     });
 
-    0
+    Ok(0)
+}
+
+/// Repair only the durable audit intent for an already-persisted event.
+///
+/// Internal producers use this when an earlier attempt committed the event but
+/// failed to enqueue its audit intent. Their duplicate retry must not repeat
+/// fan-out or workflows, but it must repair the missing audit handoff.
+pub(crate) async fn enqueue_persistent_event_audit(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
+    stored_event: &StoredEvent,
+    kind_u32: u32,
+    actor_pubkey_hex: &str,
+) -> Result<(), buzz_audit::AuditError> {
+    let event_id_hex = stored_event.event.id.to_hex();
+    enqueue_event_created_audit(
+        tenant,
+        state,
+        stored_event,
+        kind_u32,
+        actor_pubkey_hex,
+        &event_id_hex,
+    )
+    .await
 }
 
 /// Run post-commit delivery/side effects for a stored event.
@@ -506,7 +523,7 @@ async fn dispatch_persistent_event_inner(
     // `search_index_tx` mpsc are gone with the Typesense backend.
 
     if enqueue_audit {
-        enqueue_event_created_audit(
+        if let Err(error) = enqueue_event_created_audit(
             tenant,
             state,
             stored_event,
@@ -514,7 +531,10 @@ async fn dispatch_persistent_event_inner(
             actor_pubkey_hex,
             &event_id_hex,
         )
-        .await;
+        .await
+        {
+            error!(event_id = %event_id_hex, %error, "Pubsub audit intent could not be persisted");
+        }
     }
 
     // Skip workflow triggering for workflow-execution kinds and relay-signed workflow messages.
@@ -567,17 +587,14 @@ async fn enqueue_event_created_audit(
     kind_u32: u32,
     actor_pubkey_hex: &str,
     event_id_hex: &str,
-) {
-    let Some(audit_tx) = &state.audit_tx else {
-        return;
+) -> Result<(), buzz_audit::AuditError> {
+    let Some(audit_queue) = &state.audit_queue else {
+        return Ok(());
     };
-    // Audit via bounded channel (capacity 1000). Uses .send().await so entries
-    // are never silently dropped — backpressure propagates to the event handler
-    // if the queue is full. This is intentional: the audit advisory lock already
-    // serializes writes (at most 1 in-flight), so a full queue means the audit
-    // DB is genuinely overloaded and the relay should slow down rather than
-    // accumulate unbounded in-memory state. DB write failures in the worker are
-    // logged but not retried (same as the previous per-event tokio::spawn).
+    // Persist the audit intent before returning the NIP-01 acceptance. Delivery
+    // is asynchronous and retries both lock and statement timeouts from the
+    // durable outbox; a contended community cannot block another community's
+    // chain, and restart resumes any pending entry.
     let audit_entry = buzz_audit::NewAuditEntry {
         community_id: tenant.community(),
         action: buzz_audit::AuditAction::EventCreated,
@@ -594,10 +611,13 @@ async fn enqueue_event_created_audit(
             "channel_id": stored_event.channel_id,
         }),
     };
-    if let Err(e) = audit_tx.send(audit_entry).await {
-        error!(event_id = %event_id_hex, "Audit channel closed — entry lost: {e}");
+    let dedupe_key = format!("event_created:{event_id_hex}");
+    if let Err(e) = audit_queue.send(audit_entry, Some(&dedupe_key)).await {
+        error!(event_id = %event_id_hex, "Audit intent could not be persisted: {e}");
         metrics::counter!("buzz_audit_send_errors_total").increment(1);
+        return Err(e);
     }
+    Ok(())
 }
 
 /// Handle an EVENT message from a WebSocket connection.
diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs
index dd2fa6e93e0..e2ca61dd096 100644
--- a/crates/buzz-relay/src/handlers/ingest.rs
+++ b/crates/buzz-relay/src/handlers/ingest.rs
@@ -3120,7 +3120,8 @@ async fn ingest_event_inner(
             &pubkey_hex,
             threaded_visibility.clone(),
         )
-        .await;
+        .await
+        .map_err(|e| IngestError::Internal(format!("audit persistence failed: {e}")))?;
 
         info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline");
         return Ok(IngestResult {
@@ -3263,7 +3264,8 @@ async fn ingest_event_inner(
         &pubkey_hex,
         threaded_visibility.clone(),
     )
-    .await;
+    .await
+    .map_err(|e| IngestError::Internal(format!("audit persistence failed: {e}")))?;
 
     info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline");
 
diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs
index e5e4b9c3e3e..24d422a8717 100644
--- a/crates/buzz-relay/src/handlers/moderation_notices.rs
+++ b/crates/buzz-relay/src/handlers/moderation_notices.rs
@@ -28,7 +28,7 @@ use uuid::Uuid;
 use buzz_core::kind::{event_kind_u32, KIND_STREAM_MESSAGE};
 use buzz_core::tenant::TenantContext;
 
-use super::event::dispatch_persistent_event;
+use super::event::{dispatch_persistent_event, enqueue_persistent_event_audit};
 use super::side_effects::emit_group_discovery_events;
 use crate::state::AppState;
 
@@ -180,7 +180,7 @@ pub async fn send_moderation_notice(
         .await?;
 
     let kind_u32 = event_kind_u32(&stored.event);
-    dispatch_persistent_event(tenant, state, &stored, kind_u32, &relay_pubkey_hex, None).await;
+    dispatch_persistent_event(tenant, state, &stored, kind_u32, &relay_pubkey_hex, None).await?;
 
     Ok(())
 }
@@ -212,7 +212,10 @@ async fn publish_moderation_profile(
         .await?;
     if was_inserted {
         let kind_u32 = event_kind_u32(&stored.event);
-        dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await;
+        dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await?;
+    } else {
+        let kind_u32 = event_kind_u32(&stored.event);
+        enqueue_persistent_event_audit(tenant, state, &stored, kind_u32, relay_pubkey_hex).await?;
     }
     Ok(())
 }
diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs
index d3416d673c5..d0fdcb1cc50 100644
--- a/crates/buzz-relay/src/handlers/side_effects.rs
+++ b/crates/buzz-relay/src/handlers/side_effects.rs
@@ -16,7 +16,7 @@ use buzz_core::kind::{
 use buzz_core::StoredEvent;
 use buzz_db::channel::{MemberRecord, MemberRole};
 
-use super::event::dispatch_persistent_event;
+use super::event::{dispatch_persistent_event, enqueue_persistent_event_audit};
 use crate::protocol::RelayMessage;
 use crate::state::AppState;
 use buzz_core::tenant::TenantContext;
@@ -1042,7 +1042,10 @@ async fn emit_addressable_discovery_event(
         .await?;
     if was_inserted {
         let kind_u32 = event_kind_u32(&stored.event);
-        dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await;
+        dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await?;
+    } else {
+        let kind_u32 = event_kind_u32(&stored.event);
+        enqueue_persistent_event_audit(tenant, state, &stored, kind_u32, relay_pubkey_hex).await?;
     }
     Ok(())
 }
@@ -1064,7 +1067,7 @@ async fn store_group_members_event(
     state: &Arc<AppState>,
     channel_id: Uuid,
     member_snapshot: &mut buzz_db::channel::LockedMemberSnapshot,
-) -> anyhow::Result<Option<buzz_core::StoredEvent>> {
+) -> anyhow::Result<(buzz_core::StoredEvent, bool)> {
     let group_id = channel_id.to_string();
     let tags = group_members_tags(&group_id, &member_snapshot.members)?;
     let relay_pubkey = state.relay_keypair.public_key().to_bytes();
@@ -1094,16 +1097,17 @@ async fn store_group_members_event(
     let (stored, inserted) = member_snapshot
         .replace_member_event(tenant.community(), channel_id, &event)
         .await?;
-    Ok(inserted.then_some(stored))
+    Ok((stored, inserted))
 }
 
 async fn dispatch_group_members_event(
     tenant: &TenantContext,
     state: &Arc<AppState>,
-    stored: Option<buzz_core::StoredEvent>,
+    stored: (buzz_core::StoredEvent, bool),
     relay_pubkey_hex: &str,
-) {
-    if let Some(stored) = stored {
+) -> anyhow::Result<()> {
+    let (stored, inserted) = stored;
+    if inserted {
         dispatch_persistent_event(
             tenant,
             state,
@@ -1112,8 +1116,18 @@ async fn dispatch_group_members_event(
             relay_pubkey_hex,
             None,
         )
-        .await;
+        .await?;
+    } else {
+        enqueue_persistent_event_audit(
+            tenant,
+            state,
+            &stored,
+            KIND_NIP29_GROUP_MEMBERS,
+            relay_pubkey_hex,
+        )
+        .await?;
     }
+    Ok(())
 }
 
 /// Emit NIP-29 group discovery events (39000, 39001, 39002) signed by the relay keypair.
@@ -1229,7 +1243,7 @@ pub async fn emit_group_discovery_events(
     let stored_members =
         store_group_members_event(tenant, state, channel_id, &mut member_snapshot).await?;
     member_snapshot.release().await?;
-    dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await;
+    dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await?;
 
     Ok(())
 }
@@ -3124,7 +3138,16 @@ async fn publish_nip43_membership_list_inner(
             &relay_pubkey_hex,
             None,
         )
-        .await;
+        .await?;
+    } else {
+        enqueue_persistent_event_audit(
+            tenant,
+            state,
+            &stored,
+            KIND_NIP43_MEMBERSHIP_LIST,
+            &relay_pubkey_hex,
+        )
+        .await?;
     }
 
     info!(member_count, "NIP-43 membership list published");
@@ -3231,7 +3254,7 @@ pub async fn reconcile_large_channel_member_snapshots(
             let stored_members =
                 store_group_members_event(&tenant, state, channel_id, &mut member_snapshot).await?;
             member_snapshot.release().await?;
-            dispatch_group_members_event(&tenant, state, stored_members, &relay_pubkey_hex).await;
+            dispatch_group_members_event(&tenant, state, stored_members, &relay_pubkey_hex).await?;
             Ok::<bool, anyhow::Error>(true)
         }
         .await;
@@ -3438,6 +3461,14 @@ pub async fn publish_nipia_archival_list(
             .replace_addressable_event(tenant.community(), &event, None)
             .await?;
         if !was_inserted {
+            enqueue_persistent_event_audit(
+                tenant,
+                state,
+                &stored,
+                KIND_IA_ARCHIVED_LIST,
+                &relay_pubkey_hex,
+            )
+            .await?;
             continue;
         }
 
@@ -3461,7 +3492,7 @@ pub async fn publish_nipia_archival_list(
             &relay_pubkey_hex,
             None,
         )
-        .await;
+        .await?;
         info!(
             archived_count = archived.len(),
             "NIP-IA archived identities list published"
@@ -3551,7 +3582,16 @@ pub async fn publish_dm_visibility_snapshot(
             &relay_pubkey_hex,
             None,
         )
-        .await;
+        .await?;
+    } else {
+        enqueue_persistent_event_audit(
+            tenant,
+            state,
+            &stored,
+            KIND_DM_VISIBILITY,
+            &relay_pubkey_hex,
+        )
+        .await?;
     }
 
     info!(
@@ -3610,10 +3650,11 @@ async fn publish_nipia_delta(
         .insert_event(tenant.community(), &event, None)
         .await?;
     if !was_inserted {
+        enqueue_persistent_event_audit(tenant, state, &stored, kind, &relay_pubkey_hex).await?;
         return Ok(());
     }
 
-    dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex, None).await;
+    dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex, None).await?;
 
     info!(
         target = %target_pubkey_hex,
diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs
index d81602e2019..875ac005f25 100644
--- a/crates/buzz-relay/src/main.rs
+++ b/crates/buzz-relay/src/main.rs
@@ -1152,9 +1152,10 @@ async fn main() -> anyhow::Result<()> {
     serve(router, health_router, Arc::clone(&state)).await?;
     state.community_revalidator_cancel.cancel();
 
-    // Signal the audit worker to stop accepting, flush buffered entries, and
-    // exit. Uses a CancellationToken so it works regardless of how many
-    // Arc<AppState> clones are still alive in background tasks.
+    // Signal the audit worker to deliver ready entries and exit. Deferred
+    // entries remain in the durable outbox for another worker or restart. Uses
+    // a CancellationToken so it works regardless of how many Arc<AppState>
+    // clones are still alive in background tasks.
     audit_shutdown
         .drain(std::time::Duration::from_secs(5))
         .await;
diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs
index d1374e11b86..c4c79177126 100644
--- a/crates/buzz-relay/src/state.rs
+++ b/crates/buzz-relay/src/state.rs
@@ -10,7 +10,7 @@ use axum::body::Bytes;
 use axum::extract::ws::{Message as WsMessage, Utf8Bytes as WsUtf8Bytes};
 use dashmap::DashMap;
 use futures_util::future::join_all;
-use tokio::sync::{mpsc, watch, Semaphore};
+use tokio::sync::{mpsc, watch, Notify, Semaphore};
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
@@ -693,8 +693,8 @@ pub struct AppState {
     /// access check so open channels stay zero-cost. Invalidated on a flip.
     pub channel_visibility_cache: Arc<moka::sync::Cache<(CommunityId, Uuid), String>>,
 
-    /// Bounded channel for audit logging, absent when audit logging is disabled.
-    pub audit_tx: Option<mpsc::Sender<buzz_audit::NewAuditEntry>>,
+    /// Durable audit outbox producer, absent when audit logging is disabled.
+    pub audit_queue: Option<AuditQueue>,
     /// Media storage client (S3/MinIO).
     pub media_storage: Arc<MediaStorage>,
     /// Single-flight + cache state for the hourly S3 storage sweep. See
@@ -776,8 +776,9 @@ impl AppState {
     /// Constructs `AppState` from its component services.
     ///
     /// Returns `(state, audit_shutdown)`. The caller should call
-    /// `audit_shutdown.drain().await` during graceful shutdown so queued
-    /// audit entries are flushed before the process exits.
+    /// `audit_shutdown.drain().await` during graceful shutdown so ready audit
+    /// entries are delivered; deferred entries already live durably in the
+    /// outbox and resume on another worker or after restart.
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         config: Config,
@@ -796,43 +797,20 @@ impl AppState {
         let search_arc = Arc::new(search);
 
         let audit_arc = audit.into().map(Arc::new);
-        let (audit_tx, mut audit_rx) = mpsc::channel::<buzz_audit::NewAuditEntry>(1000);
-        let audit_for_worker = audit_arc.clone();
-        let audit_cancel = CancellationToken::new();
-        let audit_cancel_worker = audit_cancel.clone();
-        let audit_worker_handle = tokio::spawn(async move {
-            let Some(audit_for_worker) = audit_for_worker else {
-                audit_cancel_worker.cancelled().await;
-                return;
-            };
-            // Normal operation: process entries as they arrive.
-            loop {
-                tokio::select! {
-                    entry = audit_rx.recv() => {
-                        match entry {
-                            Some(entry) => log_audit_entry(&audit_for_worker, entry).await,
-                            None => break, // channel closed
-                        }
-                    }
-                    _ = audit_cancel_worker.cancelled() => {
-                        // Close the receiver: rejects future sends and lets us
-                        // drain everything already buffered without a race.
-                        audit_rx.close();
-                        break;
-                    }
-                }
+        let (audit_queue, audit_shutdown) = match audit_arc.clone() {
+            Some(audit) => {
+                let (queue, shutdown) = start_audit_worker(audit);
+                (Some(queue), shutdown)
             }
-            // Drain: recv() returns buffered entries, then None once empty.
-            let mut drained = 0u32;
-            while let Some(entry) = audit_rx.recv().await {
-                log_audit_entry(&audit_for_worker, entry).await;
-                drained += 1;
-            }
-            if drained > 0 {
-                tracing::info!(drained, "audit worker flushed remaining entries");
+            None => {
+                let cancel = CancellationToken::new();
+                let cancel_worker = cancel.clone();
+                let handle = tokio::spawn(async move {
+                    cancel_worker.cancelled().await;
+                });
+                (None, AuditShutdownHandle { cancel, handle })
             }
-            tracing::warn!("audit log worker exited (expected on shutdown)");
-        });
+        };
 
         let git_max_concurrent_ops = config.git_max_concurrent_ops;
         let media_max_concurrent_uploads = config.media_max_concurrent_uploads;
@@ -857,7 +835,6 @@ impl AppState {
             Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
         let gif_http_client = crate::api::gifs::build_gif_http_client();
         let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
-        let audit_enabled = audit_arc.is_some();
         let state = Self {
             config: Arc::new(config),
             db,
@@ -905,7 +882,7 @@ impl AppState {
                     .support_invalidation_closures()
                     .build(),
             ),
-            audit_tx: audit_enabled.then_some(audit_tx),
+            audit_queue,
             media_storage: Arc::new(media_storage),
             storage_sweep: Arc::new(tokio::sync::Mutex::new(
                 crate::storage_sweep::StorageSweepState::default(),
@@ -946,13 +923,7 @@ impl AppState {
             tracer: Arc::new(crate::conformance::NoopTracer),
             mesh: Arc::new(std::sync::OnceLock::new()),
         };
-        (
-            state,
-            AuditShutdownHandle {
-                cancel: audit_cancel,
-                handle: audit_worker_handle,
-            },
-        )
+        (state, audit_shutdown)
     }
 
     /// Inter-relay mesh handle. `None` ⇒ mesh-off / single-instance: callers
@@ -1319,61 +1290,133 @@ pub struct ThreadedChannelVisibility {
 
 /// Handle for graceful audit worker shutdown.
 ///
-/// Signals the worker to stop accepting new entries, drain its buffer,
-/// and exit. Independent of `Arc<AppState>` lifetime — works even when
-/// background tasks (reaper, pubsub, health) still hold state clones.
+/// Signals the worker to finish its current delivery pass and exit. Pending
+/// entries remain durable in PostgreSQL. Independent of `Arc<AppState>`
+/// lifetime — works even when background tasks (reaper, pubsub, health) still
+/// hold state clones.
 pub struct AuditShutdownHandle {
     cancel: CancellationToken,
     handle: JoinHandle<()>,
 }
 
 impl AuditShutdownHandle {
-    /// Signal the audit worker to drain and wait up to `timeout` for it to finish.
+    /// Stop the audit worker and wait up to `timeout` for it to finish.
+    ///
+    /// Producers persist entries before waking the worker, so shutdown does not
+    /// wait indefinitely on a contended community and pending rows resume after
+    /// restart.
     pub async fn drain(self, timeout: std::time::Duration) {
         self.cancel.cancel();
-        match tokio::time::timeout(timeout, self.handle).await {
+        let mut handle = self.handle;
+        match tokio::time::timeout(timeout, &mut handle).await {
             Ok(Ok(())) => tracing::info!("Audit worker drained cleanly"),
             Ok(Err(e)) => tracing::error!("Audit worker panicked: {e}"),
-            Err(_) => tracing::error!(
-                ?timeout,
-                "Audit worker did not drain in time — exiting anyway"
-            ),
+            Err(_) => {
+                tracing::error!(?timeout, "Audit worker did not drain in time — aborting it");
+                handle.abort();
+                if let Err(error) = handle.await {
+                    if !error.is_cancelled() {
+                        tracing::error!(%error, "Audit worker failed while being aborted");
+                    }
+                }
+            }
         }
     }
 }
 
-/// Log a single audit entry with metrics. Extracted so the normal loop
-/// and the post-cancel drain share the same logic.
-async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::NewAuditEntry) {
-    let t = std::time::Instant::now();
-    let mut retry_delay_ms = 50u64;
-    let mut retries = 0u64;
-    loop {
-        match audit.log(entry.clone()).await {
-            Ok(_) => {
-                metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64());
-                return;
+/// Durable audit producer backed by the shared PostgreSQL outbox.
+#[derive(Clone)]
+pub struct AuditQueue {
+    audit: Arc<AuditService>,
+    wake: Arc<Notify>,
+}
+
+impl AuditQueue {
+    /// Persist one entry and wake a worker after the transaction commits.
+    pub async fn send(
+        &self,
+        entry: buzz_audit::NewAuditEntry,
+        dedupe_key: Option<&str>,
+    ) -> Result<(), buzz_audit::AuditError> {
+        self.audit.enqueue(entry, dedupe_key).await?;
+        self.wake.notify_one();
+        Ok(())
+    }
+}
+
+fn start_audit_worker(audit: Arc<AuditService>) -> (AuditQueue, AuditShutdownHandle) {
+    let wake = Arc::new(Notify::new());
+    let queue = AuditQueue {
+        audit: Arc::clone(&audit),
+        wake: Arc::clone(&wake),
+    };
+    let cancel = CancellationToken::new();
+    let cancel_worker = cancel.clone();
+    let handle = tokio::spawn(async move {
+        let mut final_delivery_attempted = false;
+        loop {
+            // Allow one final delivery attempt after cancellation so a producer
+            // that just committed an intent can still observe graceful delivery.
+            // The second check bounds teardown even while the outbox stays busy.
+            if cancel_worker.is_cancelled() {
+                if final_delivery_attempted {
+                    break;
+                }
+                final_delivery_attempted = true;
             }
-            Err(buzz_audit::AuditError::Database(sqlx::Error::Database(database_error)))
-                if database_error.code().as_deref() == Some("55P03") =>
-            {
-                retries += 1;
-                metrics::counter!("buzz_audit_log_lock_retries_total").increment(1);
-                tracing::warn!(
-                    retries,
+            let started = std::time::Instant::now();
+            match audit.deliver_next_pending().await {
+                Ok(buzz_audit::PendingAuditResult::Appended) => {
+                    metrics::histogram!("buzz_audit_log_seconds")
+                        .record(started.elapsed().as_secs_f64());
+                }
+                Ok(buzz_audit::PendingAuditResult::Deferred {
+                    attempt_count,
                     retry_delay_ms,
-                    "Audit advisory lock timed out; preserving entry for retry"
-                );
-                tokio::time::sleep(std::time::Duration::from_millis(retry_delay_ms)).await;
-                retry_delay_ms = (retry_delay_ms * 2).min(1_000);
-            }
-            Err(error) => {
-                metrics::counter!("buzz_audit_log_errors_total").increment(1);
-                tracing::error!("Audit log failed: {error}");
-                return;
+                    sqlstate,
+                }) => {
+                    metrics::counter!(
+                        "buzz_audit_log_retries_total",
+                        "sqlstate" => sqlstate.clone()
+                    )
+                    .increment(1);
+                    if sqlstate == "55P03" {
+                        metrics::counter!("buzz_audit_log_lock_retries_total").increment(1);
+                    }
+                    tracing::warn!(
+                        attempt_count,
+                        retry_delay_ms,
+                        %sqlstate,
+                        "Audit delivery timed out; durable entry rescheduled"
+                    );
+                }
+                Ok(buzz_audit::PendingAuditResult::Idle) => {
+                    if cancel_worker.is_cancelled() {
+                        break;
+                    }
+                    tokio::select! {
+                        _ = cancel_worker.cancelled() => {},
+                        _ = wake.notified() => {},
+                        _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {},
+                    }
+                }
+                Err(error) => {
+                    metrics::counter!("buzz_audit_log_errors_total").increment(1);
+                    tracing::error!(%error, "Audit outbox delivery failed; entry remains durable");
+                    if cancel_worker.is_cancelled() {
+                        break;
+                    }
+                    tokio::select! {
+                        _ = cancel_worker.cancelled() => break,
+                        _ = wake.notified() => {},
+                        _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {},
+                    }
+                }
             }
         }
-    }
+        tracing::warn!("audit log worker exited (expected on shutdown)");
+    });
+    (queue, AuditShutdownHandle { cancel, handle })
 }
 
 impl std::fmt::Debug for AppState {
@@ -1472,26 +1515,16 @@ pub(crate) mod tests {
         let observer = sqlx::PgPool::connect(&database_url)
             .await
             .expect("connect observer pool");
-        let application_name = format!("audit-retry-test-{}", Uuid::new_v4());
-        let hook_application_name = application_name.clone();
-        let audit_pool = sqlx::postgres::PgPoolOptions::new()
-            .max_connections(1)
-            .after_connect(move |conn, _meta| {
-                let application_name = hook_application_name.clone();
-                Box::pin(async move {
-                    sqlx::query(
-                        "SELECT set_config('application_name', $1, false), \
-                                set_config('lock_timeout', '100', false)",
-                    )
-                    .bind(application_name)
-                    .execute(&mut *conn)
-                    .await?;
-                    Ok(())
-                })
-            })
-            .connect(&database_url)
-            .await
-            .expect("connect audit pool");
+        let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
+            database_url: database_url.clone(),
+            max_connections: 1,
+            min_connections: 0,
+            lock_timeout_ms: 100,
+            statement_timeout_ms: 0,
+            ..buzz_db::DbConfig::default()
+        })
+        .await
+        .expect("connect audit pool");
 
         let community_id = Uuid::new_v4();
         sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
@@ -1519,37 +1552,28 @@ pub(crate) mod tests {
             .expect("hold community audit lock");
 
         let audit = Arc::new(AuditService::new(audit_pool));
-        let worker = tokio::spawn({
-            let audit = Arc::clone(&audit);
-            async move { log_audit_entry(&audit, entry).await }
-        });
+        let (queue, shutdown) = start_audit_worker(Arc::clone(&audit));
+        queue
+            .send(entry, None)
+            .await
+            .expect("durably enqueue audit entry");
 
-        // Observe one timed-out advisory-lock attempt and then a second wait.
-        // Releasing during the first wait would not prove that the worker
-        // preserved and retried the original queue entry.
+        // Observe the durable row reach a second attempt. Releasing during the
+        // first wait would not prove that the worker preserved and retried it.
         tokio::time::timeout(std::time::Duration::from_secs(3), async {
-            let mut saw_first_wait = false;
-            let mut saw_retry_gap = false;
             loop {
-                let waiting: bool = sqlx::query_scalar(
-                    "SELECT EXISTS (\
-                         SELECT 1 FROM pg_stat_activity \
-                         WHERE application_name = $1 \
-                           AND query LIKE 'SELECT pg_advisory_lock%' \
-                           AND wait_event = 'advisory'\
-                     )",
+                let attempts: i32 = sqlx::query_scalar(
+                    "SELECT COALESCE(MAX(attempt_count), 0)::integer \
+                     FROM audit_outbox \
+                     WHERE community_id = $1 AND object_id = $2",
                 )
-                .bind(&application_name)
+                .bind(community_id)
+                .bind(&object_id)
                 .fetch_one(&observer)
                 .await
-                .expect("inspect audit lock waiter");
-                if waiting {
-                    if saw_retry_gap {
-                        break;
-                    }
-                    saw_first_wait = true;
-                } else if saw_first_wait {
-                    saw_retry_gap = true;
+                .expect("inspect durable retry count");
+                if attempts >= 2 {
+                    break;
                 }
                 tokio::time::sleep(std::time::Duration::from_millis(5)).await;
             }
@@ -1562,10 +1586,142 @@ pub(crate) mod tests {
             .execute(&mut *holder)
             .await
             .expect("release community audit lock");
-        tokio::time::timeout(std::time::Duration::from_secs(3), worker)
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let rows: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
+                )
+                .bind(community_id)
+                .bind(&object_id)
+                .fetch_one(&observer)
+                .await
+                .expect("count retried audit rows");
+                if rows == 1 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("audit worker did not finish after lock release");
+        shutdown.drain(std::time::Duration::from_secs(1)).await;
+
+        let rows: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
+        )
+        .bind(community_id)
+        .bind(&object_id)
+        .fetch_one(&observer)
+        .await
+        .expect("count retried audit rows");
+        assert_eq!(rows, 1, "the preserved entry must be appended exactly once");
+
+        sqlx::query("DELETE FROM audit_log WHERE community_id = $1 AND object_id = $2")
+            .bind(community_id)
+            .bind(&object_id)
+            .execute(&observer)
+            .await
+            .expect("remove test audit row");
+        sqlx::query("DELETE FROM communities WHERE id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove test community");
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn audit_worker_retries_statement_timeout_until_original_entry_is_appended_once() {
+        let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
+        let observer = sqlx::PgPool::connect(&database_url)
+            .await
+            .expect("connect observer pool");
+        let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
+            database_url: database_url.clone(),
+            max_connections: 1,
+            min_connections: 0,
+            lock_timeout_ms: 1_000,
+            statement_timeout_ms: 100,
+            ..buzz_db::DbConfig::default()
+        })
+        .await
+        .expect("connect audit pool");
+
+        let community_id = Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community_id)
+            .bind(format!("audit-statement-retry-{community_id}.example"))
+            .execute(&observer)
+            .await
+            .expect("insert test community");
+        let object_id = format!("audit-statement-retry-object-{}", Uuid::new_v4());
+        let entry = buzz_audit::NewAuditEntry {
+            community_id: CommunityId::from_uuid(community_id),
+            action: buzz_audit::AuditAction::EventCreated,
+            actor_pubkey: Some(vec![0xcd; 32]),
+            object_id: Some(object_id.clone()),
+            detail: serde_json::json!({"test": "statement-timeout-retry"}),
+        };
+
+        let lock_key = format!("buzz_audit:{community_id}");
+        let mut holder = observer.acquire().await.expect("acquire lock holder");
+        sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))")
+            .bind(&lock_key)
+            .execute(&mut *holder)
+            .await
+            .expect("hold community audit lock");
+
+        let audit = Arc::new(AuditService::new(audit_pool));
+        let (queue, shutdown) = start_audit_worker(Arc::clone(&audit));
+        queue
+            .send(entry, None)
+            .await
+            .expect("durably enqueue audit entry");
+
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let attempts: i32 = sqlx::query_scalar(
+                    "SELECT COALESCE(MAX(attempt_count), 0)::integer \
+                     FROM audit_outbox \
+                     WHERE community_id = $1 AND object_id = $2",
+                )
+                .bind(community_id)
+                .bind(&object_id)
+                .fetch_one(&observer)
+                .await
+                .expect("inspect durable statement-timeout retry count");
+                if attempts >= 2 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("audit worker did not exercise statement-timeout retry");
+        sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))")
+            .bind(&lock_key)
+            .execute(&mut *holder)
             .await
-            .expect("audit worker did not finish after lock release")
-            .expect("audit worker task panicked");
+            .expect("release community audit lock");
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let rows: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
+                )
+                .bind(community_id)
+                .bind(&object_id)
+                .fetch_one(&observer)
+                .await
+                .expect("count retried audit rows");
+                if rows == 1 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("audit worker did not finish after statement timeout");
+        shutdown.drain(std::time::Duration::from_secs(1)).await;
 
         let rows: i64 = sqlx::query_scalar(
             "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
@@ -1590,6 +1746,386 @@ pub(crate) mod tests {
             .expect("remove test community");
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn audit_queue_dedupes_logical_event_while_pending_and_after_delivery() {
+        let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
+        let observer = sqlx::PgPool::connect(&database_url)
+            .await
+            .expect("connect observer pool");
+        let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
+            database_url,
+            max_connections: 2,
+            min_connections: 0,
+            lock_timeout_ms: 100,
+            statement_timeout_ms: 0,
+            ..buzz_db::DbConfig::default()
+        })
+        .await
+        .expect("connect audit pool");
+
+        let community_id = Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community_id)
+            .bind(format!("audit-dedupe-{community_id}.example"))
+            .execute(&observer)
+            .await
+            .expect("insert test community");
+        let object_id = format!("audit-dedupe-object-{}", Uuid::new_v4());
+        let dedupe_key = format!("event_created:{object_id}");
+        let entry = buzz_audit::NewAuditEntry {
+            community_id: CommunityId::from_uuid(community_id),
+            action: buzz_audit::AuditAction::EventCreated,
+            actor_pubkey: Some(vec![0xde; 32]),
+            object_id: Some(object_id.clone()),
+            detail: serde_json::json!({"test": "logical-dedupe"}),
+        };
+        let audit = Arc::new(AuditService::new(audit_pool));
+        let (queue, shutdown) = start_audit_worker(audit);
+
+        queue
+            .send(entry.clone(), Some(&dedupe_key))
+            .await
+            .expect("enqueue original audit intent");
+        queue
+            .send(entry.clone(), Some(&dedupe_key))
+            .await
+            .expect("dedupe pending audit retry");
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let rows: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
+                )
+                .bind(community_id)
+                .bind(&object_id)
+                .fetch_one(&observer)
+                .await
+                .expect("count delivered audit rows");
+                if rows == 1 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("original audit intent was not delivered");
+
+        queue
+            .send(entry, Some(&dedupe_key))
+            .await
+            .expect("dedupe delivered audit retry");
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        let (audit_rows, pending_rows, key_rows): (i64, i64, i64) = sqlx::query_as(
+            "SELECT \
+                 (SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2), \
+                 (SELECT count(*) FROM audit_outbox WHERE community_id = $1), \
+                 (SELECT count(*) FROM audit_delivery_keys \
+                  WHERE community_id = $1 AND dedupe_key = $3)",
+        )
+        .bind(community_id)
+        .bind(&object_id)
+        .bind(&dedupe_key)
+        .fetch_one(&observer)
+        .await
+        .expect("inspect logical audit dedupe state");
+        assert_eq!(audit_rows, 1, "logical event must be audited exactly once");
+        assert_eq!(
+            pending_rows, 0,
+            "duplicate retry must not recreate outbox row"
+        );
+        assert_eq!(key_rows, 1, "delivery key must survive completed delivery");
+        shutdown.drain(std::time::Duration::from_secs(1)).await;
+
+        sqlx::query("DELETE FROM audit_log WHERE community_id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove audit rows");
+        sqlx::query("DELETE FROM audit_delivery_keys WHERE community_id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove audit delivery key");
+        sqlx::query("DELETE FROM communities WHERE id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove test community");
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn durable_audit_queue_survives_saturation_and_contended_tenant_teardown() {
+        let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
+        let observer = sqlx::PgPool::connect(&database_url)
+            .await
+            .expect("connect observer pool");
+        let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
+            database_url: database_url.clone(),
+            max_connections: 4,
+            min_connections: 0,
+            lock_timeout_ms: 100,
+            statement_timeout_ms: 0,
+            ..buzz_db::DbConfig::default()
+        })
+        .await
+        .expect("connect audit pool");
+
+        let community_a = Uuid::new_v4();
+        let community_b = Uuid::new_v4();
+        for (id, label) in [(community_a, "a"), (community_b, "b")] {
+            sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+                .bind(id)
+                .bind(format!("audit-fairness-{label}-{id}.example"))
+                .execute(&observer)
+                .await
+                .expect("insert test community");
+        }
+
+        let lock_key_a = format!("buzz_audit:{community_a}");
+        let mut holder = observer.acquire().await.expect("acquire lock holder");
+        sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))")
+            .bind(&lock_key_a)
+            .execute(&mut *holder)
+            .await
+            .expect("hold community A audit lock");
+
+        let audit = Arc::new(AuditService::new(audit_pool));
+        let (queue, shutdown) = start_audit_worker(Arc::clone(&audit));
+        for index in 0..=1000 {
+            queue
+                .send(
+                    buzz_audit::NewAuditEntry {
+                        community_id: CommunityId::from_uuid(community_a),
+                        action: buzz_audit::AuditAction::EventCreated,
+                        actor_pubkey: Some(vec![0xaa; 32]),
+                        object_id: Some(format!("audit-saturated-a-{index}")),
+                        detail: serde_json::json!({"index": index}),
+                    },
+                    None,
+                )
+                .await
+                .expect("durably enqueue community A audit entry");
+        }
+        let object_b = format!("audit-progress-b-{}", Uuid::new_v4());
+        queue
+            .send(
+                buzz_audit::NewAuditEntry {
+                    community_id: CommunityId::from_uuid(community_b),
+                    action: buzz_audit::AuditAction::EventCreated,
+                    actor_pubkey: Some(vec![0xbb; 32]),
+                    object_id: Some(object_b.clone()),
+                    detail: serde_json::json!({"test": "cross-community-progress"}),
+                },
+                None,
+            )
+            .await
+            .expect("durably enqueue community B audit entry");
+
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let rows: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
+                )
+                .bind(community_b)
+                .bind(&object_b)
+                .fetch_one(&observer)
+                .await
+                .expect("count community B audit rows");
+                if rows == 1 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("contended community A blocked community B progress");
+
+        let pending_a: i64 =
+            sqlx::query_scalar("SELECT count(*) FROM audit_outbox WHERE community_id = $1")
+                .bind(community_a)
+                .fetch_one(&observer)
+                .await
+                .expect("count durable community A audit intents");
+        assert_eq!(pending_a, 1001, "saturated entries must remain durable");
+
+        let shutdown_started = std::time::Instant::now();
+        shutdown.drain(std::time::Duration::from_secs(1)).await;
+        assert!(
+            shutdown_started.elapsed() < std::time::Duration::from_secs(1),
+            "durable queue teardown must not wait on a contended tenant"
+        );
+
+        sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))")
+            .bind(&lock_key_a)
+            .execute(&mut *holder)
+            .await
+            .expect("release community A audit lock");
+        let (_queue_after_restart, shutdown_after_restart) = start_audit_worker(Arc::clone(&audit));
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let rows: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM audit_log \
+                     WHERE community_id = $1 AND object_id = 'audit-saturated-a-0'",
+                )
+                .bind(community_a)
+                .fetch_one(&observer)
+                .await
+                .expect("count recovered community A audit row");
+                if rows == 1 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("restart did not resume the durable community A intent");
+        shutdown_after_restart
+            .drain(std::time::Duration::from_secs(1))
+            .await;
+
+        sqlx::query("DELETE FROM audit_outbox WHERE community_id IN ($1, $2)")
+            .bind(community_a)
+            .bind(community_b)
+            .execute(&observer)
+            .await
+            .expect("remove pending audit test rows");
+        sqlx::query("DELETE FROM audit_log WHERE community_id IN ($1, $2)")
+            .bind(community_a)
+            .bind(community_b)
+            .execute(&observer)
+            .await
+            .expect("remove audit test rows");
+        sqlx::query("DELETE FROM communities WHERE id IN ($1, $2)")
+            .bind(community_a)
+            .bind(community_b)
+            .execute(&observer)
+            .await
+            .expect("remove test communities");
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn audit_worker_cancellation_stops_a_continuously_ready_backlog() {
+        let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
+        let observer = sqlx::PgPool::connect(&database_url)
+            .await
+            .expect("connect observer pool");
+        let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
+            database_url,
+            max_connections: 4,
+            min_connections: 0,
+            lock_timeout_ms: 100,
+            statement_timeout_ms: 0,
+            ..buzz_db::DbConfig::default()
+        })
+        .await
+        .expect("connect audit pool");
+
+        let community_id = Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community_id)
+            .bind(format!("audit-ready-backlog-{community_id}.example"))
+            .execute(&observer)
+            .await
+            .expect("insert test community");
+        sqlx::query(
+            "INSERT INTO audit_outbox (community_id, action, object_id, detail) \
+             SELECT $1, 'event_created', 'audit-ready-' || item::text, \
+                    jsonb_build_object('item', item) \
+             FROM generate_series(1, 10000) AS item",
+        )
+        .bind(community_id)
+        .execute(&observer)
+        .await
+        .expect("seed continuously ready audit backlog");
+
+        let audit = Arc::new(AuditService::new(audit_pool));
+        let (_queue, shutdown) = start_audit_worker(audit);
+        tokio::time::timeout(std::time::Duration::from_secs(3), async {
+            loop {
+                let appended: i64 =
+                    sqlx::query_scalar("SELECT count(*) FROM audit_log WHERE community_id = $1")
+                        .bind(community_id)
+                        .fetch_one(&observer)
+                        .await
+                        .expect("count appended audit rows");
+                if appended > 0 {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("audit worker did not begin draining ready backlog");
+
+        shutdown.drain(std::time::Duration::from_secs(1)).await;
+        let appended_after_shutdown: i64 =
+            sqlx::query_scalar("SELECT count(*) FROM audit_log WHERE community_id = $1")
+                .bind(community_id)
+                .fetch_one(&observer)
+                .await
+                .expect("count rows after shutdown");
+        let pending_after_shutdown: i64 =
+            sqlx::query_scalar("SELECT count(*) FROM audit_outbox WHERE community_id = $1")
+                .bind(community_id)
+                .fetch_one(&observer)
+                .await
+                .expect("count durable rows after shutdown");
+        assert!(
+            pending_after_shutdown > 0,
+            "shutdown must not drain the entire continuously ready backlog"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        let appended_later: i64 =
+            sqlx::query_scalar("SELECT count(*) FROM audit_log WHERE community_id = $1")
+                .bind(community_id)
+                .fetch_one(&observer)
+                .await
+                .expect("count rows after worker should have stopped");
+        assert_eq!(
+            appended_later, appended_after_shutdown,
+            "audit worker must not remain detached after shutdown"
+        );
+
+        sqlx::query("DELETE FROM audit_outbox WHERE community_id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove pending audit rows");
+        sqlx::query("DELETE FROM audit_log WHERE community_id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove appended audit rows");
+        sqlx::query("DELETE FROM communities WHERE id = $1")
+            .bind(community_id)
+            .execute(&observer)
+            .await
+            .expect("remove test community");
+    }
+
+    #[tokio::test]
+    async fn audit_shutdown_aborts_a_worker_that_exceeds_its_timeout() {
+        let completed = Arc::new(AtomicBool::new(false));
+        let completed_worker = Arc::clone(&completed);
+        let handle = tokio::spawn(async move {
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+            completed_worker.store(true, Ordering::SeqCst);
+        });
+        let shutdown = AuditShutdownHandle {
+            cancel: CancellationToken::new(),
+            handle,
+        };
+
+        shutdown.drain(std::time::Duration::from_millis(5)).await;
+        tokio::time::sleep(std::time::Duration::from_millis(125)).await;
+        assert!(
+            !completed.load(Ordering::SeqCst),
+            "timed-out worker must be aborted instead of detached"
+        );
+    }
+
     #[test]
     fn send_to_resets_grace_counter_on_success() {
         let (mgr, id, _rx, _ctrl_rx, _cancel, bp) = setup_conn(16);
diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs
index 4ceb3b39308..ef60aead8b1 100644
--- a/crates/buzz-relay/src/workflow_sink.rs
+++ b/crates/buzz-relay/src/workflow_sink.rs
@@ -16,7 +16,7 @@ use nostr::{EventBuilder, Kind, Tag};
 use tracing::info;
 use uuid::Uuid;
 
-use crate::handlers::event::dispatch_persistent_event;
+use crate::handlers::event::{dispatch_persistent_event, enqueue_persistent_event_audit};
 use crate::state::AppState;
 
 /// Resolves `@Name` mentions in workflow message text to the pubkeys of the
@@ -445,7 +445,7 @@ impl ActionSink for RelayActionSink {
             // 5. Post-persist side effects (fan-out, search, audit)
             //    Only if actually inserted (idempotency guard).
             if was_inserted {
-                let _ = dispatch_persistent_event(
+                dispatch_persistent_event(
                     &tenant,
                     &state,
                     &stored_event,
@@ -453,8 +453,21 @@ impl ActionSink for RelayActionSink {
                     &author_pubkey_hex,
                     None,
                 )
-                .await;
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+            } else {
+                enqueue_persistent_event_audit(
+                    &tenant,
+                    &state,
+                    &stored_event,
+                    kind_u32,
+                    &author_pubkey_hex,
+                )
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+            }
 
+            if was_inserted {
                 // A threaded reply changed its thread's counters — push a fresh
                 // relay-signed kind:39005 so subscribed clients update badge
                 // counts without refetching the head window, exactly as the
diff --git a/migrations/0041_durable_audit_outbox.sql b/migrations/0041_durable_audit_outbox.sql
new file mode 100644
index 00000000000..81db0017564
--- /dev/null
+++ b/migrations/0041_durable_audit_outbox.sql
@@ -0,0 +1,37 @@
+-- Durable, fair handoff between accepted relay operations and the per-community
+-- audit hash chain. Producers commit an intent here before replying; workers
+-- claim only the oldest due row in each community, so one contended chain does
+-- not block unrelated tenants. The worker appends audit_log and deletes this
+-- row in one transaction.
+
+-- Logical producer keys survive outbox delivery. This makes retry repair
+-- idempotent both while an intent is pending and after it has been appended.
+CREATE TABLE audit_delivery_keys (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    dedupe_key   TEXT NOT NULL,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    PRIMARY KEY (community_id, dedupe_key)
+);
+
+CREATE TABLE audit_outbox (
+    community_id   UUID NOT NULL REFERENCES communities(id),
+    id             UUID NOT NULL DEFAULT gen_random_uuid(),
+    enqueue_seq    BIGINT GENERATED ALWAYS AS IDENTITY,
+    action         VARCHAR(64) NOT NULL,
+    actor_pubkey   BYTEA,
+    object_id      TEXT,
+    detail         JSONB NOT NULL,
+    enqueued_at    TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    attempt_count  INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
+    PRIMARY KEY (community_id, id)
+);
+
+CREATE INDEX idx_audit_outbox_due
+    ON audit_outbox (next_attempt_at, enqueue_seq);
+
+CREATE INDEX idx_audit_outbox_community_order
+    ON audit_outbox (community_id, enqueue_seq);
+
+SELECT attach_community_write_fence('audit_outbox');
+SELECT attach_community_write_fence('audit_delivery_keys');
diff --git a/schema/schema.sql b/schema/schema.sql
index 54566103335..9bc1bb98ac9 100644
--- a/schema/schema.sql
+++ b/schema/schema.sql
@@ -657,6 +657,32 @@ CREATE TABLE audit_log (
 
 CREATE UNIQUE INDEX idx_audit_log_hash ON audit_log (community_id, hash);
 
+CREATE TABLE audit_delivery_keys (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    dedupe_key   TEXT NOT NULL,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    PRIMARY KEY (community_id, dedupe_key)
+);
+
+CREATE TABLE audit_outbox (
+    community_id    UUID NOT NULL REFERENCES communities(id),
+    id              UUID NOT NULL DEFAULT gen_random_uuid(),
+    enqueue_seq     BIGINT GENERATED ALWAYS AS IDENTITY,
+    action          VARCHAR(64) NOT NULL,
+    actor_pubkey    BYTEA,
+    object_id       TEXT,
+    detail          JSONB NOT NULL,
+    enqueued_at     TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    attempt_count   INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
+    PRIMARY KEY (community_id, id)
+);
+
+CREATE INDEX idx_audit_outbox_due
+    ON audit_outbox (next_attempt_at, enqueue_seq);
+CREATE INDEX idx_audit_outbox_community_order
+    ON audit_outbox (community_id, enqueue_seq);
+
 -- ── NIP-56 reports (kind:1984 ingest) ─────────────────────────────────────────
 -- One row per accepted report event. Reports are signals, never triggers:
 -- nothing auto-actions on them (NIP-56). Reporter identity is visible to
@@ -1731,6 +1757,8 @@ $$;
 SELECT attach_community_write_fence('api_tokens');
 SELECT attach_community_write_fence('archived_identities');
 SELECT attach_community_write_fence('audit_log');
+SELECT attach_community_write_fence('audit_outbox');
+SELECT attach_community_write_fence('audit_delivery_keys');
 SELECT attach_community_write_fence('channel_members');
 SELECT attach_community_write_fence('channels');
 SELECT attach_community_write_fence('community_bans');