diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs
index 02a25f55ad6..4ae2c9b11c9 100644
--- a/crates/buzz-db/src/runtime/migration.rs
+++ b/crates/buzz-db/src/runtime/migration.rs
@@ -890,7 +890,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 88);
+        assert_eq!(migrations.len(), 89);
         assert_eq!(migrations[87].version, 88);
         assert!(migrations[87]
             .sql
@@ -992,6 +992,34 @@ mod postgres_tests {
             .sql
             .as_str()
             .contains("CREATE FUNCTION profile_name_fields"));
+        // The app owns the NIP-RS watermark, retention-free purge and
+        // mention liveness, so 0089 drops the migration-only triggers that
+        // 0009, 0010, 0011 and 0019 installed. Those files stay frozen.
+        assert_eq!(migrations[88].version, 89);
+        let retire_nip_rs = strip_sql_comments(migrations[88].sql.as_str());
+        for statement in [
+            "SET LOCAL lock_timeout = '5s';",
+            "DROP TRIGGER trg_events_nip_rs_watermark ON events;",
+            "DROP TRIGGER trg_events_guard_nip_rs_hard_delete ON events;",
+            "DROP TRIGGER trg_events_purge_soft_deleted_nip_rs ON events;",
+            "DROP TRIGGER trg_events_purge_soft_deleted_buzz_mesh_status ON events;",
+            "DROP TRIGGER trg_event_mentions_require_live_event ON event_mentions;",
+            "DROP FUNCTION guard_nip_rs_watermark();",
+            "DROP FUNCTION guard_nip_rs_hard_delete();",
+            "DROP FUNCTION purge_soft_deleted_nip_rs();",
+            "DROP FUNCTION purge_soft_deleted_buzz_mesh_status();",
+            "DROP FUNCTION guard_event_mention_live();",
+        ] {
+            assert!(
+                retire_nip_rs.contains(statement),
+                "0089 must run {statement}"
+            );
+        }
+        assert!(
+            retire_nip_rs.find("SET LOCAL lock_timeout").unwrap()
+                < retire_nip_rs.find("DROP TRIGGER").unwrap(),
+            "0089 must bound its ACCESS EXCLUSIVE lock on events before the first DROP"
+        );
         assert_eq!(migrations[55].version, 56);
         assert!(migrations[55]
             .sql
@@ -2928,6 +2956,10 @@ mod postgres_tests {
     /// drift this test guards against. `indoption` (not just `indexdef` text) is
     /// asserted so a resurrected `NULLS FIRST` in a migration that `pgschema`
     /// cannot represent is caught.
+    ///
+    /// It also asserts the fully migrated `events` and
+    /// `event_mentions` row triggers match `schema/` exactly, so no
+    /// migration-only trigger survives on the event write path.
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn admin_schema_parity_between_desired_state_and_migrations() {
@@ -3007,6 +3039,25 @@ mod postgres_tests {
             .expect("read events key definitions")
         }
 
+        // Row triggers on a parent table (partition clones follow the parent):
+        // name, function, enable mode and the tgtype bitmask.
+        async fn trigger_shapes(pool: &PgPool, table: &str) -> Vec<(String, String, String, i16)> {
+            sqlx::query_as(
+                "SELECT trigger.tgname::text, procedure.proname::text, \
+                 trigger.tgenabled::text, trigger.tgtype \
+                 FROM pg_trigger trigger \
+                 JOIN pg_class c ON c.oid = trigger.tgrelid \
+                 JOIN pg_namespace n ON n.oid = c.relnamespace \
+                 JOIN pg_proc procedure ON procedure.oid = trigger.tgfoid \
+                 WHERE n.nspname = 'public' AND c.relname = $1 AND NOT trigger.tgisinternal \
+                 ORDER BY trigger.tgname",
+            )
+            .bind(table)
+            .fetch_all(pool)
+            .await
+            .expect("read trigger shapes")
+        }
+
         let base_url = std::env::var("BUZZ_TEST_DATABASE_URL")
             .or_else(|_| std::env::var("DATABASE_URL"))
             .unwrap_or_else(|_| TEST_DB_URL.to_owned());
@@ -3128,6 +3179,17 @@ mod postgres_tests {
                 "constraint parity mismatch for {table}"
             );
         }
+
+        // Every migration-only trigger on the event tables is retired, so a
+        // fully migrated database carries exactly the desired-state triggers.
+        for table in ["events", "event_mentions"] {
+            assert_eq!(
+                trigger_shapes(&desired, table).await,
+                trigger_shapes(&migrated, table).await,
+                "trigger parity mismatch for {table}: a migration installs a trigger that \
+                 schema/ does not declare (or the reverse)"
+            );
+        }
     }
 
     /// Re-applying `schema/schema.sql` to a database bootstrapped the way every
diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs
index 97f459240c7..c803c224ee8 100644
--- a/crates/buzz-db/src/runtime/tests.rs
+++ b/crates/buzz-db/src/runtime/tests.rs
@@ -430,316 +430,6 @@ async fn make_community(pool: &PgPool) -> Uuid {
     id
 }
 
-#[tokio::test]
-#[ignore = "requires Postgres"]
-async fn migration_schema_database_guard_covers_legacy_writer_and_nip09_deletion() {
-    use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
-
-    let db = setup_db().await;
-    let community = CommunityId::from_uuid(make_community(&db.pool).await);
-    let keys = Keys::generate();
-    let d_tag = format!("read-state:{}", "b".repeat(32));
-    let tags = vec![
-        Tag::parse(["d", d_tag.as_str()]).expect("d tag"),
-        Tag::parse(["t", "read-state"]).expect("t tag"),
-    ];
-    let base = Timestamp::now().as_secs();
-    let a = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "A")
-        .tags(tags.clone())
-        .custom_created_at(Timestamp::from(base))
-        .sign_with_keys(&keys)
-        .expect("sign A");
-    let x = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "X")
-        .tags(tags.clone())
-        .custom_created_at(Timestamp::from(base + 1))
-        .sign_with_keys(&keys)
-        .expect("sign X");
-    let b = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "B")
-        .tags(tags.clone())
-        .custom_created_at(Timestamp::from(base + 2))
-        .sign_with_keys(&keys)
-        .expect("sign B");
-    let c = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "C")
-        .tags(tags)
-        .custom_created_at(Timestamp::from(base + 3))
-        .sign_with_keys(&keys)
-        .expect("sign C");
-
-    async fn legacy_insert(
-        pool: &PgPool,
-        community: CommunityId,
-        event: &nostr::Event,
-        d_tag: &str,
-    ) -> std::result::Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
-        sqlx::query(
-                "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, d_tag) \
-                 VALUES ($1, $2, $3, to_timestamp($4), $5, $6, $7, $8, NOW(), $9) ON CONFLICT DO NOTHING",
-            )
-            .bind(community.as_uuid())
-            .bind(event.id.as_bytes().as_slice())
-            .bind(event.pubkey.to_bytes())
-            .bind(event.created_at.as_secs() as f64)
-            .bind(buzz_core::kind::KIND_READ_STATE as i32)
-            .bind(serde_json::to_value(&event.tags).expect("serialize tags"))
-            .bind(&event.content)
-            .bind(event.sig.serialize().as_slice())
-            .bind(d_tag)
-            .execute(pool)
-            .await
-    }
-
-    legacy_insert(&db.pool, community, &a, &d_tag)
-        .await
-        .expect("legacy insert A");
-    let duplicate = legacy_insert(&db.pool, community, &a, &d_tag)
-        .await
-        .expect("legacy duplicate A remains idempotent");
-    assert_eq!(duplicate.rows_affected(), 0);
-
-    sqlx::query(
-        "INSERT INTO event_mentions \
-                 (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078)",
-    )
-    .bind(community.as_uuid())
-    .bind("c".repeat(64))
-    .bind(a.id.as_bytes().as_slice())
-    .bind(a.created_at.as_secs() as f64)
-    .execute(&db.pool)
-    .await
-    .expect("insert live mention");
-
-    // Emulate the pre-PR replacement path after migration 0007: soft-delete
-    // the live row, then insert B without any application watermark write.
-    sqlx::query(
-            "UPDATE events SET deleted_at=NOW() \
-             WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(&d_tag)
-        .execute(&db.pool)
-        .await
-        .expect("legacy soft-delete A");
-    let mentions_after_delete: i64 = sqlx::query_scalar(
-        "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2",
-    )
-    .bind(community.as_uuid())
-    .bind(a.id.as_bytes().as_slice())
-    .fetch_one(&db.pool)
-    .await
-    .expect("count mentions after delete");
-    assert_eq!(mentions_after_delete, 0);
-
-    let stale_mention = sqlx::query(
-        "INSERT INTO event_mentions \
-                 (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078)",
-    )
-    .bind(community.as_uuid())
-    .bind("d".repeat(64))
-    .bind(a.id.as_bytes().as_slice())
-    .bind(a.created_at.as_secs() as f64)
-    .execute(&db.pool)
-    .await
-    .expect("stale post-commit mention is skipped");
-    assert_eq!(stale_mention.rows_affected(), 0);
-
-    legacy_insert(&db.pool, community, &b, &d_tag)
-        .await
-        .expect("legacy insert B");
-    let duplicate_b = legacy_insert(&db.pool, community, &b, &d_tag)
-        .await
-        .expect("live duplicate B is skipped");
-    assert_eq!(duplicate_b.rows_affected(), 0);
-
-    sqlx::query(
-        "INSERT INTO event_mentions \
-                 (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078)",
-    )
-    .bind(community.as_uuid())
-    .bind("e".repeat(64))
-    .bind(b.id.as_bytes().as_slice())
-    .bind(b.created_at.as_secs() as f64)
-    .execute(&db.pool)
-    .await
-    .expect("insert B mention");
-
-    // Exercise the new Rust hard-delete path independently. An in-flight
-    // mention holds KEY SHARE on B, so replacement by C must block, then
-    // complete after the mention commits and remove both B and its mention.
-    let mut rust_mention_tx = db
-        .pool
-        .begin()
-        .await
-        .expect("begin Rust mention transaction");
-    sqlx::query(
-        "INSERT INTO event_mentions \
-                 (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078) ON CONFLICT DO NOTHING",
-    )
-    .bind(community.as_uuid())
-    .bind("e".repeat(64))
-    .bind(b.id.as_bytes().as_slice())
-    .bind(b.created_at.as_secs() as f64)
-    .execute(&mut *rust_mention_tx)
-    .await
-    .expect("hold B live-event key-share lock");
-
-    let replace_db = db.clone();
-    let replace_d_tag = d_tag.clone();
-    let replace_c = c.clone();
-    let replace_task = tokio::spawn(async move {
-        replace_db
-            .replace_parameterized_event(community, &replace_c, &replace_d_tag, None)
-            .await
-    });
-    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-    assert!(
-        !replace_task.is_finished(),
-        "Rust hard delete should wait for mention lock"
-    );
-    rust_mention_tx
-        .commit()
-        .await
-        .expect("release Rust mention lock");
-    let replaced = tokio::time::timeout(std::time::Duration::from_secs(2), replace_task)
-        .await
-        .expect("Rust hard delete deadlocked with mention insert")
-        .expect("replacement task panicked")
-        .expect("replace B with C");
-    assert!(replaced.1, "C must replace B");
-    let b_mentions: i64 = sqlx::query_scalar(
-        "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2",
-    )
-    .bind(community.as_uuid())
-    .bind(b.id.as_bytes().as_slice())
-    .fetch_one(&db.pool)
-    .await
-    .expect("count B mentions after Rust replacement");
-    assert_eq!(b_mentions, 0);
-
-    sqlx::query(
-        "INSERT INTO event_mentions \
-                 (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078)",
-    )
-    .bind(community.as_uuid())
-    .bind("f".repeat(64))
-    .bind(c.id.as_bytes().as_slice())
-    .bind(c.created_at.as_secs() as f64)
-    .execute(&db.pool)
-    .await
-    .expect("insert C mention");
-
-    // Exercise legacy UPDATE-trigger deletion with the same barrier. While
-    // deletion waits on C's KEY SHARE lock, an exact replay must already be
-    // a zero-row trigger no-op; it must not wait for deletion or resurrect C.
-    let mut legacy_mention_tx = db
-        .pool
-        .begin()
-        .await
-        .expect("begin legacy mention transaction");
-    sqlx::query(
-        "INSERT INTO event_mentions \
-                 (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078) ON CONFLICT DO NOTHING",
-    )
-    .bind(community.as_uuid())
-    .bind("f".repeat(64))
-    .bind(c.id.as_bytes().as_slice())
-    .bind(c.created_at.as_secs() as f64)
-    .execute(&mut *legacy_mention_tx)
-    .await
-    .expect("hold C live-event key-share lock");
-
-    let delete_pool = db.pool.clone();
-    let delete_pubkey = keys.public_key().to_bytes();
-    let delete_d_tag = d_tag.clone();
-    let delete_task = tokio::spawn(async move {
-        sqlx::query(
-                "UPDATE events SET deleted_at=NOW() \
-                 WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
-            )
-            .bind(community.as_uuid())
-            .bind(delete_pubkey)
-            .bind(delete_d_tag)
-            .execute(&delete_pool)
-            .await
-    });
-    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-    assert!(
-        !delete_task.is_finished(),
-        "legacy delete should wait for mention lock"
-    );
-
-    let replay_while_delete_waits = legacy_insert(&db.pool, community, &c, &d_tag)
-        .await
-        .expect("concurrent exact C replay is skipped");
-    assert_eq!(replay_while_delete_waits.rows_affected(), 0);
-
-    legacy_mention_tx
-        .commit()
-        .await
-        .expect("release legacy mention lock");
-    tokio::time::timeout(std::time::Duration::from_secs(2), delete_task)
-        .await
-        .expect("legacy delete deadlocked with mention insert")
-        .expect("delete task panicked")
-        .expect("legacy NIP-09 delete C");
-
-    let payloads: i64 = sqlx::query_scalar(
-            "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(&d_tag)
-        .fetch_one(&db.pool)
-        .await
-        .expect("count retained payloads");
-    assert_eq!(
-        payloads, 0,
-        "legacy soft deletes must not retain NIP-RS payloads"
-    );
-
-    // Opposite commit order: deletion has committed before exact replay.
-    // Equality remains an observable zero-row no-op, never a resurrection.
-    let replay_c = legacy_insert(&db.pool, community, &c, &d_tag)
-        .await
-        .expect("post-delete exact C replay is skipped");
-    assert_eq!(replay_c.rows_affected(), 0);
-    let payloads_after_exact_replay: i64 = sqlx::query_scalar(
-            "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(&d_tag)
-        .fetch_one(&db.pool)
-        .await
-        .expect("count payloads after exact replay");
-    assert_eq!(payloads_after_exact_replay, 0);
-
-    let replay = legacy_insert(&db.pool, community, &x, &d_tag).await;
-    assert!(
-        replay.is_err(),
-        "database guard must reject A < X < C replay"
-    );
-
-    let watermark: (chrono::DateTime<chrono::Utc>, Vec<u8>) = sqlx::query_as(
-        "SELECT created_at, event_id FROM parameterized_event_watermarks \
-             WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3",
-    )
-    .bind(community.as_uuid())
-    .bind(keys.public_key().to_bytes())
-    .bind(&d_tag)
-    .fetch_one(&db.pool)
-    .await
-    .expect("read C watermark");
-    assert_eq!(watermark.0.timestamp(), base as i64 + 3);
-    assert_eq!(watermark.1, c.id.as_bytes().as_slice());
-}
-
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn migration_schema_operator_listener_outbox_is_not_tenant_scoped() {
diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs
index c38d86a2ddc..c9bc4b90f94 100644
--- a/crates/buzz-db/src/store/deletion.rs
+++ b/crates/buzz-db/src/store/deletion.rs
@@ -2311,12 +2311,6 @@ impl DeletionStore {
         validate_catalog_on(&mut tx).await?;
         verify_lease_and_fence(&mut tx, token, DeletionStage::BindingsRemoved, generation).await?;
         set_executor_gucs(&mut tx, token.community_id, generation).await?;
-        // Migration 0011 fences hard deletion of NIP-RS rows against legacy
-        // writers. Whole-community deletion is an intentional hard-delete path,
-        // and the transaction is already bound to an approved, fenced tenant.
-        sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
-            .execute(&mut *tx)
-            .await?;
 
         // Preserve deployment-global operator evidence while severing tenant provenance.
         for table in ["product_feedback", "rate_limit_violations"] {
diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs
index 62889e3befc..d6dd2f8ecae 100644
--- a/crates/buzz-db/src/store/event.rs
+++ b/crates/buzz-db/src/store/event.rs
@@ -1655,7 +1655,10 @@ fn count_events_sql(q: &EventQuery) -> QueryBuilder<sqlx::Postgres> {
 }
 
 // Preserve the public event-store deletion API after the focused extraction.
-pub(crate) use super::event_deletion::soft_delete_event_and_update_thread_in_tx;
+pub(crate) use super::event_deletion::{
+    is_nip_rs_event, remove_coordinate_versions_in_tx, soft_delete_event_and_update_thread_in_tx,
+    CoordinateRemoval,
+};
 pub use super::event_deletion::{
     soft_delete_by_coordinate, soft_delete_event_and_update_thread, DeletionTarget,
 };
diff --git a/crates/buzz-db/src/store/event_deletion.rs b/crates/buzz-db/src/store/event_deletion.rs
index 44f0dfee153..06a0a8d56e2 100644
--- a/crates/buzz-db/src/store/event_deletion.rs
+++ b/crates/buzz-db/src/store/event_deletion.rs
@@ -10,37 +10,88 @@ use crate::{
     AdmittedTx,
 };
 
+/// SQL predicate over an `events` row for a conforming NIP-RS read-state
+/// coordinate: kind 30078, a `read-state:<32 lowercase hex>` d-tag carried by
+/// exactly one `d` tag, and exactly one `["t", "read-state"]` tag.
+macro_rules! nip_rs_event_predicate {
+    () => {
+        "(kind = 30078 \
+          AND d_tag ~ '^read-state:[0-9a-f]{32}$' \
+          AND (SELECT count(*) \
+               FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
+               WHERE jsonb_typeof(tag) = 'array' AND tag->0 = '\"d\"'::jsonb) = 1 \
+          AND EXISTS (SELECT 1 \
+               FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
+               WHERE jsonb_typeof(tag) = 'array' AND jsonb_array_length(tag) >= 2 \
+                 AND jsonb_typeof(tag->1) = 'string' AND tag->>0 = 'd' AND tag->>1 = d_tag) \
+          AND (SELECT count(*) \
+               FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
+               WHERE tag = '[\"t\", \"read-state\"]'::jsonb) = 1)"
+    };
+}
+
+/// SQL predicate over an `events` row for a Buzz mesh heartbeat: kind 30003,
+/// a `buzz-mesh-member-status:` d-tag, and a tag containing both `k` and
+/// `buzz-mesh-status` (jsonb containment, so extra or reordered elements still
+/// match).
+macro_rules! mesh_status_event_predicate {
+    () => {
+        "(kind = 30003 \
+          AND d_tag LIKE 'buzz-mesh-member-status:%' \
+          AND tags @> '[[\"k\", \"buzz-mesh-status\"]]'::jsonb)"
+    };
+}
+
 /// SQL predicate over an `events` row for coordinates that carry no
 /// historical value: NIP-RS read state and Buzz mesh heartbeats. Their
 /// deletion is physical, including the mention index, so soft-deleted payloads
-/// never accumulate. This is exactly the classification used by the migration
-/// 0011 / 0019 purge triggers this path replaces.
+/// never accumulate. These predicates are the only definition of the class:
+/// every removal path decides row by row through
+/// [`remove_coordinate_versions_in_tx`] rather than re-implementing them in Rust.
 macro_rules! retention_free_event_predicate {
     () => {
-        "((kind = 30078 \
-           AND d_tag ~ '^read-state:[0-9a-f]{32}$' \
-           AND (SELECT count(*) \
-                FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
-                WHERE jsonb_typeof(tag) = 'array' AND tag->0 = '\"d\"'::jsonb) = 1 \
-           AND EXISTS (SELECT 1 \
-                FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
-                WHERE jsonb_typeof(tag) = 'array' AND jsonb_array_length(tag) >= 2 \
-                  AND jsonb_typeof(tag->1) = 'string' AND tag->>0 = 'd' AND tag->>1 = d_tag) \
-           AND (SELECT count(*) \
-                FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
-                WHERE tag = '[\"t\", \"read-state\"]'::jsonb) = 1) \
-          OR (kind = 30003 \
-              AND d_tag LIKE 'buzz-mesh-member-status:%' \
-              AND tags @> '[[\"k\", \"buzz-mesh-status\"]]'::jsonb))"
+        concat!(
+            "(",
+            nip_rs_event_predicate!(),
+            " OR ",
+            mesh_status_event_predicate!(),
+            ")"
+        )
     };
 }
 
 /// Kinds that can hold a retention-free coordinate. Cheap Rust gate so
-/// ordinary deletions skip the purge statement entirely.
+/// ordinary writes skip the classification and purge statements entirely.
 fn may_be_retention_free(kind: i32) -> bool {
     kind == 30078 || kind == 30003
 }
 
+/// Whether an event that has not been written yet is a conforming NIP-RS
+/// read-state coordinate, which advances the stale-write watermark. Uses the
+/// same SQL predicate as the purge and delete paths so the two cannot drift.
+pub(crate) async fn is_nip_rs_event(
+    conn: &mut sqlx::PgConnection,
+    kind: i32,
+    d_tag: &str,
+    tags: &serde_json::Value,
+) -> Result<bool> {
+    if kind != 30078 {
+        return Ok(false);
+    }
+    let nip_rs: bool = sqlx::query_scalar(concat!(
+        "SELECT COALESCE(",
+        nip_rs_event_predicate!(),
+        ", false) \
+         FROM (SELECT $1::integer AS kind, $2::text AS d_tag, $3::jsonb AS tags) event_row"
+    ))
+    .bind(kind)
+    .bind(d_tag)
+    .bind(tags)
+    .fetch_one(conn)
+    .await?;
+    Ok(nip_rs)
+}
+
 /// Soft-delete the live row for an addressable coordinate
 /// `(kind, pubkey, d_tag)` — the NIP-33 replacement key — provided it is not
 /// newer than the deletion request.
@@ -86,14 +137,55 @@ pub async fn soft_delete_by_coordinate(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
+    let deleted = remove_coordinate_versions_in_tx(
+        &mut tx,
+        kind,
+        pubkey,
+        d_tag,
+        deletion_created_at,
+        CoordinateRemoval::Deletion,
+    )
+    .await?;
+    tx.commit().await?;
+    Ok(deleted)
+}
+
+/// Why [`remove_coordinate_versions_in_tx`] retires a coordinate's versions.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum CoordinateRemoval {
+    /// NIP-09 coordinate deletion. Soft-deleted versions keep their mention
+    /// rows, like every other deletion path.
+    Deletion,
+    /// Parameterized replacement. Soft-deleted versions leave with their
+    /// mention rows, as `superseded_with_mentions_sql!` does for the other
+    /// replacement paths.
+    Supersede,
+}
+
+/// Remove every live version of `(kind, pubkey, d_tag)` created at or before
+/// `created_at_or_before`, in the caller's transaction. Each row is classified
+/// by its own stored shape: retention-free rows are purged physically with
+/// their mentions, every other row is soft-deleted. NIP-09 coordinate deletion
+/// and parameterized replacement both remove versions through this function.
+///
+/// Returns `true` if any row was removed.
+pub(crate) async fn remove_coordinate_versions_in_tx(
+    tx: &mut AdmittedTx,
+    kind: i32,
+    pubkey: &[u8],
+    d_tag: &str,
+    created_at_or_before: DateTime<Utc>,
+    removal: CoordinateRemoval,
+) -> Result<bool> {
+    let community_id = tx.community();
     let purged = if may_be_retention_free(kind) {
         purge_retention_free_events(
-            &mut tx,
+            tx,
             RetentionFreeTarget::Coordinate {
                 kind,
                 pubkey,
                 d_tag,
-                created_at_or_before: deletion_created_at,
+                created_at_or_before,
             },
         )
         .await?
@@ -104,26 +196,50 @@ pub async fn soft_delete_by_coordinate(
     // statement order: under READ COMMITTED this UPDATE takes a fresh snapshot,
     // so a retention-free head committed by a racing replacement after the
     // purge ran is spared instead of soft-deleted (the "deletion arrived
-    // first" outcome documented above). COALESCE keeps a NULL predicate
-    // (e.g. NULL `d_tag`) on the ordinary soft-delete path, as the triggers do.
-    let result = sqlx::query(concat!(
-        "UPDATE events SET deleted_at = NOW() \
-         WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
-         AND created_at <= $5 AND NOT COALESCE(",
-        retention_free_event_predicate!(),
-        ", false)"
-    ))
-    .bind(community_id.as_uuid())
-    .bind(kind)
-    .bind(pubkey)
-    .bind(d_tag)
-    .bind(deletion_created_at)
-    .execute(tx.conn())
-    .await?;
-
-    tx.commit().await?;
+    // first" outcome documented on [`soft_delete_by_coordinate`]). COALESCE
+    // keeps a NULL predicate (e.g. NULL `d_tag`) on the ordinary soft-delete
+    // path.
+    let soft_deleted: i64 = match removal {
+        CoordinateRemoval::Deletion => sqlx::query(concat!(
+            "UPDATE events SET deleted_at = NOW() \
+             WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
+             AND created_at <= $5 AND NOT COALESCE(",
+            retention_free_event_predicate!(),
+            ", false)"
+        ))
+        .bind(community_id.as_uuid())
+        .bind(kind)
+        .bind(pubkey)
+        .bind(d_tag)
+        .bind(created_at_or_before)
+        .execute(tx.conn())
+        .await?
+        .rows_affected() as i64,
+        CoordinateRemoval::Supersede => {
+            sqlx::query_scalar(concat!(
+                "WITH superseded AS ( \
+                     UPDATE events SET deleted_at = NOW() \
+                     WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
+                     AND created_at <= $5 AND NOT COALESCE(",
+                retention_free_event_predicate!(),
+                ", false) RETURNING id \
+                 ), superseded_mentions AS ( \
+                     DELETE FROM event_mentions mention USING superseded \
+                     WHERE mention.community_id = $1 AND mention.event_id = superseded.id \
+                 ) \
+                 SELECT count(*) FROM superseded"
+            ))
+            .bind(community_id.as_uuid())
+            .bind(kind)
+            .bind(pubkey)
+            .bind(d_tag)
+            .bind(created_at_or_before)
+            .fetch_one(tx.conn())
+            .await?
+        }
+    };
 
-    Ok(purged > 0 || result.rows_affected() > 0)
+    Ok(purged > 0 || soft_deleted > 0)
 }
 
 /// Purge live retention-free rows matching `$target`, plus their mention rows,
@@ -189,13 +305,6 @@ async fn purge_retention_free_events(
     target: RetentionFreeTarget<'_>,
 ) -> Result<u64> {
     let community_id = tx.community();
-    // Migration 0011 fences NIP-RS hard deletes behind a transaction-local
-    // opt-in. The fence and this opt-in are removed together once the
-    // migration-only triggers are dropped.
-    sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
-        .execute(tx.conn())
-        .await?;
-
     let purged: i64 = match target {
         RetentionFreeTarget::Event {
             event_id,
diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs
index 971e638bc50..ef8f78f9b1a 100644
--- a/crates/buzz-db/src/store/replaceable.rs
+++ b/crates/buzz-db/src/store/replaceable.rs
@@ -131,40 +131,8 @@ async fn replace_parameterized_event_in_transaction_impl(
     )
     .await?;
 
-    let d_tag_count = event
-        .tags
-        .iter()
-        .filter(|tag| tag.as_slice().first().is_some_and(|part| part == "d"))
-        .count();
-    let has_exact_d_tag = event.tags.iter().any(|tag| {
-        let parts = tag.as_slice();
-        parts.len() >= 2 && parts[0] == "d" && parts[1] == d_tag
-    });
-    let read_state_t_tag_count = event
-        .tags
-        .iter()
-        .filter(|tag| {
-            let parts = tag.as_slice();
-            parts.len() == 2 && parts[0] == "t" && parts[1] == "read-state"
-        })
-        .count();
-    let is_nip_rs = kind_i32 == buzz_core::kind::KIND_READ_STATE as i32
-        && d_tag_count == 1
-        && has_exact_d_tag
-        && d_tag.strip_prefix("read-state:").is_some_and(|slot| {
-            slot.len() == 32
-                && slot
-                    .bytes()
-                    .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
-        })
-        && read_state_t_tag_count == 1;
-    let is_buzz_mesh_status = kind_i32 == buzz_core::kind::KIND_BOOKMARK_SET as i32
-        && d_tag.starts_with("buzz-mesh-member-status:")
-        && event.tags.iter().any(|tag| {
-            let parts = tag.as_slice();
-            parts.len() == 2 && parts[0] == "k" && parts[1] == "buzz-mesh-status"
-        });
-    let hard_delete_superseded = is_nip_rs || is_buzz_mesh_status;
+    let tags_json = serde_json::to_value(&event.tags)?;
+    let is_nip_rs = crate::event::is_nip_rs_event(tx.conn(), kind_i32, d_tag, &tags_json).await?;
 
     let existing: Option<(DateTime<Utc>, Vec<u8>)> = sqlx::query_as(
         "SELECT created_at, id FROM events \
@@ -263,46 +231,20 @@ async fn replace_parameterized_event_in_transaction_impl(
         crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
             .await?;
         if existing.is_some() {
-            let previous_nip_rs_hard_delete: Option<String> = if is_nip_rs {
-                sqlx::query_scalar(
-                    "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')",
-                )
-                .fetch_one(tx.conn())
-                .await?
-            } else {
-                None
-            };
-            if is_nip_rs {
-                sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
-                    .execute(tx.conn())
-                    .await?;
-            }
-            let statement = if hard_delete_superseded {
-                superseded_with_mentions_sql!(
-                    "DELETE FROM events \
-                     WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
-                )
-            } else {
-                superseded_with_mentions_sql!(
-                    "UPDATE events SET deleted_at = NOW() \
-                     WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
-                )
-            };
-            sqlx::query(statement)
-                .bind(community_id.as_uuid())
-                .bind(kind_i32)
-                .bind(pubkey_bytes.as_slice())
-                .bind(d_tag)
-                .execute(tx.conn())
-                .await?;
-
-            if is_nip_rs {
-                let previous_value = previous_nip_rs_hard_delete.as_deref().unwrap_or_default();
-                sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', $1, true)")
-                    .bind(previous_value)
-                    .execute(tx.conn())
-                    .await?;
-            }
+            // Every live version is at or before the incoming event (dominance
+            // was checked above). Each is removed by its own stored shape, so
+            // a retention-free head superseded by a nonconforming event is
+            // still purged, and a nonconforming head keeps legacy retention.
+            // Superseded versions leave with their mention rows.
+            crate::event::remove_coordinate_versions_in_tx(
+                tx,
+                kind_i32,
+                pubkey_bytes.as_slice(),
+                d_tag,
+                created_at,
+                crate::event::CoordinateRemoval::Supersede,
+            )
+            .await?;
         }
 
         let sig_bytes = event.sig.serialize();
@@ -671,7 +613,7 @@ mod postgres_tests {
     use crate::test_support::ScratchDb;
     use crate::{event, migration, replaceable};
     use sqlx::postgres::PgPoolOptions;
-    use sqlx::{Acquire, PgPool};
+    use sqlx::PgPool;
     use std::time::Duration;
 
     const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials
@@ -1220,32 +1162,46 @@ mod postgres_tests {
         .expect("count NIP-RS rows");
         assert_eq!(rows, 1, "superseded payload must be physically deleted");
 
-        sqlx::query(
-            "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(&d_tag)
-        .execute(&db.pool)
-        .await
-        .expect("simulate NIP-09 coordinate deletion");
-
         assert!(
-            !db.replace_parameterized_event(community, &old, &d_tag, None)
-                .await
-                .expect("replay old")
-                .1
+            event::soft_delete_by_coordinate(
+                &db.pool,
+                community,
+                i32::from(buzz_core::kind::KIND_READ_STATE as u16),
+                &keys.public_key().to_bytes(),
+                &d_tag,
+                (base + 1) as i64,
+            )
+            .await
+            .expect("NIP-09 coordinate deletion"),
+            "coordinate deletion must remove the live head"
         );
-        let live: i64 = sqlx::query_scalar(
-            "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
+
+        // Neither the superseded head nor an exact replay of the deleted head
+        // may come back: the watermark holds the deleted head's (created_at, id).
+        for (label, replay) in [("old", &old), ("deleted head", &new)] {
+            assert!(
+                !db.replace_parameterized_event(community, replay, &d_tag, None)
+                    .await
+                    .expect("replay")
+                    .1,
+                "replay of {label} must be rejected"
+            );
+        }
+        let (rows, live): (i64, i64) = sqlx::query_as(
+            "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \
+             WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3",
         )
         .bind(community.as_uuid())
         .bind(keys.public_key().to_bytes())
         .bind(&d_tag)
         .fetch_one(&db.pool)
         .await
-        .expect("count live NIP-RS rows");
-        assert_eq!(live, 0, "watermark must block stale resurrection");
+        .expect("count NIP-RS rows");
+        assert_eq!(
+            (rows, live),
+            (0, 0),
+            "deletion is physical and the watermark blocks resurrection"
+        );
     }
 
     #[tokio::test]
@@ -1358,90 +1314,6 @@ mod postgres_tests {
         );
     }
 
-    #[tokio::test]
-    #[ignore = "requires Postgres"]
-    async fn migration_schema_nip_rs_transaction_operation_restores_hard_delete_opt_in() {
-        use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
-
-        let db = setup_db().await;
-        let community = CommunityId::from_uuid(make_community(&db.pool).await);
-        let keys = Keys::generate();
-        let base = Timestamp::now().as_secs();
-        let replace_d_tag = format!("read-state:{}", "b".repeat(32));
-        let victim_d_tag = format!("read-state:{}", "c".repeat(32));
-        let event = |d_tag: &str, content: &str, timestamp: u64| {
-            EventBuilder::new(
-                Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16),
-                content,
-            )
-            .tags(vec![
-                Tag::parse(["d", d_tag]).expect("d tag"),
-                Tag::parse(["t", "read-state"]).expect("t tag"),
-            ])
-            .custom_created_at(Timestamp::from(timestamp))
-            .sign_with_keys(&keys)
-            .expect("sign read state")
-        };
-        let old = event(&replace_d_tag, "old", base);
-        let new = event(&replace_d_tag, "new", base + 1);
-        let victim = event(&victim_d_tag, "victim", base);
-
-        assert!(
-            db.replace_parameterized_event(community, &old, &replace_d_tag, None)
-                .await
-                .expect("insert old head")
-                .1
-        );
-        assert!(
-            db.replace_parameterized_event(community, &victim, &victim_d_tag, None)
-                .await
-                .expect("insert victim head")
-                .1
-        );
-
-        let mut tx = db
-            .begin_event_write_transaction(community)
-            .await
-            .expect("begin caller transaction");
-        let result = db
-            .replace_parameterized_event_in_transaction(
-                &mut tx,
-                &new,
-                &replace_d_tag,
-                None,
-                replaceable::ParameterizedReplacePrecondition::Unconditional,
-            )
-            .await
-            .expect("replace inside caller transaction");
-        assert_eq!(
-            result.status,
-            replaceable::ParameterizedReplaceStatus::Inserted
-        );
-
-        let leaked: Option<String> = sqlx::query_scalar(
-            "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')",
-        )
-        .fetch_one(tx.conn())
-        .await
-        .expect("read hard-delete opt-in after replacement");
-        assert_ne!(leaked.as_deref(), Some("on"));
-
-        let unauthorized = sqlx::query(
-            "DELETE FROM events WHERE community_id=$1 AND kind=30078 \
-             AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(&victim_d_tag)
-        .execute(tx.conn())
-        .await;
-        assert!(
-            unauthorized.is_err(),
-            "replacement opt-in must not authorize later caller SQL"
-        );
-        tx.rollback().await.expect("roll back caller transaction");
-    }
-
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn parameterized_replacement_in_existing_transaction_honors_revision_and_rollback() {
@@ -1802,7 +1674,7 @@ mod postgres_tests {
 
     #[tokio::test]
     #[ignore = "requires Postgres"]
-    async fn migration_schema_mesh_status_replacement_keeps_one_physical_row() {
+    async fn mesh_status_replacement_keeps_one_physical_row() {
         use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
 
         let db = setup_db().await;
@@ -1842,31 +1714,6 @@ mod postgres_tests {
         .await
         .expect("count mesh status rows");
         assert_eq!((rows, live), (1, 1));
-
-        sqlx::query(
-            "UPDATE events SET deleted_at=NOW() \
-             WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(d_tag)
-        .execute(&db.pool)
-        .await
-        .expect("simulate old relay soft delete");
-        let rows_after_legacy_delete: i64 = sqlx::query_scalar(
-            "SELECT count(*) FROM events \
-             WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(d_tag)
-        .fetch_one(&db.pool)
-        .await
-        .expect("count rows after old relay soft delete");
-        assert_eq!(
-            rows_after_legacy_delete, 0,
-            "migration trigger must purge soft-deleted mesh status"
-        );
     }
 
     #[tokio::test]
@@ -1963,168 +1810,139 @@ mod postgres_tests {
         }
     }
 
+    /// Replacement removes each superseded row by its own stored shape, not by
+    /// the incoming event's: a conforming read-state head superseded by a
+    /// nonconforming event is still purged with its mentions, and a
+    /// nonconforming head superseded by a conforming event keeps legacy
+    /// retention (a soft-deleted row) but, like every superseded head, loses
+    /// its mention rows.
     #[tokio::test]
     #[ignore = "requires Postgres"]
-    async fn migration_schema_nip_rs_hard_delete_fence_fails_closed_and_scopes_opt_in_to_transaction(
-    ) {
+    async fn mixed_shape_replacement_classifies_each_superseded_row() {
         use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
 
         let db = setup_db().await;
         let community = CommunityId::from_uuid(make_community(&db.pool).await);
-        let keys = Keys::generate();
         let base = Timestamp::now().as_secs();
-        let conforming_d = format!("read-state:{}", "6".repeat(32));
-        let conforming = EventBuilder::new(
-            Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16),
-            "fenced-conforming",
-        )
-        .tags(vec![
-            Tag::parse(["d", conforming_d.as_str()]).expect("d tag"),
-            Tag::parse(["t", "read-state"]).expect("t tag"),
-        ])
-        .custom_created_at(Timestamp::from(base))
-        .sign_with_keys(&keys)
-        .expect("sign conforming event");
-        assert!(
-            db.replace_parameterized_event(community, &conforming, &conforming_d, None)
-                .await
-                .expect("insert conforming event")
-                .1
-        );
-        sqlx::query(
-            "INSERT INTO event_mentions \
-             (community_id, pubkey_hex, event_id, event_created_at, event_kind) \
-             VALUES ($1, $2, $3, to_timestamp($4), 30078)",
-        )
-        .bind(community.as_uuid())
-        .bind("6".repeat(64))
-        .bind(conforming.id.as_bytes().as_slice())
-        .bind(conforming.created_at.as_secs() as f64)
-        .execute(&db.pool)
-        .await
-        .expect("insert mention");
-
-        // Model ce10's first destructive statement. RAISE aborts the transaction,
-        // so its later mention delete and incoming insert can never commit.
-        let mut old_writer = db.pool.begin().await.expect("begin old-writer tx");
-        let rejected = sqlx::query(
-            "DELETE FROM events WHERE community_id=$1 AND kind=30078 \
-             AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
-        )
-        .bind(community.as_uuid())
-        .bind(keys.public_key().to_bytes())
-        .bind(&conforming_d)
-        .execute(&mut *old_writer)
-        .await;
-        assert!(rejected.is_err(), "old-writer hard delete must be rejected");
-        old_writer.rollback().await.expect("rollback rejected tx");
-        let preserved: (i64, i64) = sqlx::query_as(
-            "SELECT (SELECT count(*) FROM events WHERE community_id=$1 AND id=$2), \
-                    (SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2)",
-        )
-        .bind(community.as_uuid())
-        .bind(conforming.id.as_bytes().as_slice())
-        .fetch_one(&db.pool)
-        .await
-        .expect("count preserved payload and mention");
-        assert_eq!(preserved, (1, 1));
+        let conforming = |d_tag: &str| {
+            vec![
+                Tag::parse(["d", d_tag]).expect("d tag"),
+                Tag::parse(["t", "read-state"]).expect("t tag"),
+            ]
+        };
+        let duplicate_t = |d_tag: &str| {
+            vec![
+                Tag::parse(["d", d_tag]).expect("d tag"),
+                Tag::parse(["t", "read-state"]).expect("first t tag"),
+                Tag::parse(["t", "read-state"]).expect("second t tag"),
+            ]
+        };
 
-        let nonconforming_d = format!("read-state:{}", "7".repeat(32));
-        let nonconforming = EventBuilder::new(
-            Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16),
-            "fenced-nonconforming",
-        )
-        .tags(vec![
-            Tag::parse(["d", nonconforming_d.as_str()]).expect("first d tag"),
-            Tag::parse(["d", "other"]).expect("second d tag"),
-            Tag::parse(["t", "read-state"]).expect("t tag"),
-        ])
-        .custom_created_at(Timestamp::from(base + 1))
-        .sign_with_keys(&keys)
-        .expect("sign nonconforming event");
-        assert!(
-            db.replace_parameterized_event(community, &nonconforming, &nonconforming_d, None,)
-                .await
-                .expect("insert nonconforming event")
-                .1
-        );
-        let rejected_nonconforming = sqlx::query(
-            "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)",
-        )
-        .bind(community.as_uuid())
-        .bind(nonconforming.id.as_bytes().as_slice())
-        .bind(nonconforming.created_at.as_secs() as f64)
-        .execute(&db.pool)
-        .await;
-        assert!(
-            rejected_nonconforming.is_err(),
-            "fence must cover a nonconforming OLD row at a regex coordinate"
-        );
+        for (case, old_tags, new_tags, expected_old) in [
+            (
+                "conforming then nonconforming",
+                conforming as fn(&str) -> Vec<Tag>,
+                duplicate_t as fn(&str) -> Vec<Tag>,
+                (0, 0, 0),
+            ),
+            (
+                "nonconforming then conforming",
+                duplicate_t as fn(&str) -> Vec<Tag>,
+                conforming as fn(&str) -> Vec<Tag>,
+                (1, 0, 0),
+            ),
+        ] {
+            let keys = Keys::generate();
+            let d_tag = format!("read-state:{}", "f".repeat(32));
+            let sign = |content: &str, tags: Vec<Tag>, at: u64| {
+                EventBuilder::new(
+                    Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16),
+                    content,
+                )
+                .tags(tags)
+                .custom_created_at(Timestamp::from(at))
+                .sign_with_keys(&keys)
+                .expect("sign event")
+            };
+            let old = sign("old", old_tags(&d_tag), base);
+            let new = sign("new", new_tags(&d_tag), base + 1);
 
-        let unrelated_d = format!("read-state:{}", "8".repeat(32));
-        let unrelated = EventBuilder::new(Kind::Custom(30023), "unrelated")
-            .tags(vec![Tag::parse(["d", unrelated_d.as_str()]).expect("d tag")])
-            .custom_created_at(Timestamp::from(base + 2))
-            .sign_with_keys(&keys)
-            .expect("sign unrelated event");
-        assert!(
-            db.replace_parameterized_event(community, &unrelated, &unrelated_d, None)
-                .await
-                .expect("insert unrelated event")
-                .1
-        );
-        let unrelated_delete = sqlx::query(
-            "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)",
-        )
-        .bind(community.as_uuid())
-        .bind(unrelated.id.as_bytes().as_slice())
-        .bind(unrelated.created_at.as_secs() as f64)
-        .execute(&db.pool)
-        .await
-        .expect("delete unrelated event");
-        assert_eq!(unrelated_delete.rows_affected(), 1);
-
-        // Check both transaction exits on one physical session; pool selection
-        // cannot accidentally hide a leaked session-local authorization value.
-        let mut conn = db.pool.acquire().await.expect("acquire dedicated session");
-        for commit in [true, false] {
-            let mut tx = conn.begin().await.expect("begin GUC transaction");
-            let value: String =
-                sqlx::query_scalar("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
-                    .fetch_one(&mut *tx)
+            assert!(
+                db.replace_parameterized_event(community, &old, &d_tag, None)
                     .await
-                    .expect("set transaction-local GUC");
-            assert_eq!(value, "on");
-            if commit {
-                tx.commit().await.expect("commit GUC transaction");
-            } else {
-                tx.rollback().await.expect("rollback GUC transaction");
-            }
-            let leaked: Option<String> = sqlx::query_scalar(
-                "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')",
+                    .expect("insert old")
+                    .1,
+                "{case}: old head stored"
+            );
+            sqlx::query(
+                "INSERT INTO event_mentions \
+                 (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) \
+                 SELECT community_id, $3, id, created_at, NULL, kind \
+                 FROM events WHERE community_id = $1 AND id = $2",
             )
-            .fetch_one(&mut *conn)
+            .bind(community.as_uuid())
+            .bind(old.id.as_bytes().as_slice())
+            .bind("c".repeat(64))
+            .execute(&db.pool)
             .await
-            .expect("read GUC after transaction");
-            assert_ne!(leaked.as_deref(), Some("on"));
+            .expect("index mention");
+            assert!(
+                db.replace_parameterized_event(community, &new, &d_tag, None)
+                    .await
+                    .expect("replace")
+                    .1,
+                "{case}: newer head replaces"
+            );
+
+            let (rows, live): (i64, i64) = sqlx::query_as(
+                "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) \
+                 FROM events WHERE community_id=$1 AND id=$2",
+            )
+            .bind(community.as_uuid())
+            .bind(old.id.as_bytes().as_slice())
+            .fetch_one(&db.pool)
+            .await
+            .expect("count old rows");
+            let mentions: i64 = sqlx::query_scalar(
+                "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2",
+            )
+            .bind(community.as_uuid())
+            .bind(old.id.as_bytes().as_slice())
+            .fetch_one(&db.pool)
+            .await
+            .expect("count old mentions");
+            assert_eq!(
+                (rows, live, mentions),
+                expected_old,
+                "{case}: superseded row classified by its own shape"
+            );
+            let coordinate_live: i64 = sqlx::query_scalar(
+                "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 \
+                 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
+            )
+            .bind(community.as_uuid())
+            .bind(keys.public_key().to_bytes())
+            .bind(&d_tag)
+            .fetch_one(&db.pool)
+            .await
+            .expect("count live heads");
+            assert_eq!(coordinate_live, 1, "{case}: exactly the newer head is live");
         }
     }
 
-    /// NIP-09 deletion of a retention-free coordinate (NIP-RS read state, mesh
-    /// status) is physical in application code, so it holds on desired-state
-    /// databases that never installed the migration 0009/0019 purge triggers.
-    /// Ordinary addressable events, and look-alikes that fail classification,
-    /// keep the soft-delete contract.
+    /// NIP-09 deletion and replacement of a retention-free coordinate (NIP-RS
+    /// read state, mesh status) are physical in application code; no database
+    /// trigger takes part. Ordinary addressable events, and look-alikes that
+    /// fail classification, keep the soft-delete contract. Every removal path
+    /// classifies through the same SQL predicate, so one table pins all three.
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn nip09_deletion_purges_retention_free_rows_and_mentions() {
         assert_nip09_deletion_purges_retention_free_rows_and_mentions(false).await;
     }
 
-    /// The same contract on a migration-built database, where the 0009/0011/
-    /// 0019 purge and hard-delete fence triggers are still installed. Each row
-    /// shape is also soft-deleted the way a pre-0009 writer would, so the
-    /// table proves the app classification agrees with the triggers.
+    /// The same contract on a migration-built database, where migration 0089
+    /// dropped the 0009/0011/0019 purge and hard-delete fence triggers.
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn migration_schema_nip09_deletion_purges_retention_free_rows_and_mentions() {
@@ -2138,20 +1956,30 @@ mod postgres_tests {
         enum Delete {
             Coordinate,
             Id,
+            /// Supersede through `replace_parameterized_event` with a newer
+            /// event of the same shape at the same coordinate.
+            Replace,
         }
 
         let db = setup_db().await;
-        let triggers_installed: bool = sqlx::query_scalar(
-            "SELECT EXISTS(SELECT 1 FROM pg_trigger \
-             WHERE tgname = 'trg_events_purge_soft_deleted_nip_rs')",
-        )
-        .fetch_one(&db.pool)
-        .await
-        .expect("inspect purge trigger");
+        let migrated: bool =
+            sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
+                .fetch_one(&db.pool)
+                .await
+                .expect("inspect migration ledger");
         assert_eq!(
-            triggers_installed, migration,
+            migrated, migration,
             "schema mode must match the test variant (run through the Postgres wrapper)"
         );
+        let purge_triggers: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM pg_trigger \
+             WHERE tgname IN ('trg_events_purge_soft_deleted_nip_rs', \
+                              'trg_events_purge_soft_deleted_buzz_mesh_status')",
+        )
+        .fetch_one(&db.pool)
+        .await
+        .expect("inspect purge triggers");
+        assert_eq!(purge_triggers, 0, "the app alone classifies and purges");
         let community = CommunityId::from_uuid(make_community(&db.pool).await);
         let nip_rs = buzz_core::kind::KIND_READ_STATE as u16;
         let mesh = buzz_core::kind::KIND_BOOKMARK_SET as u16;
@@ -2237,7 +2065,7 @@ mod postgres_tests {
                 true,
             ),
             (
-                // Containment semantics (`tags @>`), shared by trigger and app.
+                // Containment semantics (`tags @>`) of the SQL predicate.
                 "mesh-status k tag with extra element",
                 mesh,
                 mesh_d.clone(),
@@ -2340,36 +2168,36 @@ mod postgres_tests {
                 (rows, live, mentions)
             }
         };
-        let expected = |purged: bool| -> (i64, i64, i64) {
-            if purged {
-                (0, 0, 0)
-            } else {
-                (1, 0, 1)
+        // A soft-deleted (non-purged) row keeps its mentions after a NIP-09
+        // deletion, but a superseded head leaves with its mention rows.
+        let expected = |purged: bool, mode: Delete| -> (i64, i64, i64) {
+            match (purged, mode) {
+                (true, _) => (0, 0, 0),
+                (false, Delete::Replace) => (1, 0, 0),
+                (false, _) => (1, 0, 1),
             }
         };
 
         for (label, kind, d_tag, tags, purged) in cases {
-            if migration {
-                // Trigger half of the parity table: a pre-0009 writer's plain
-                // soft delete, classified only by the 0009/0011/0019 triggers.
-                let (_, id, _) = store(label, kind, d_tag.clone(), tags.clone()).await;
-                sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id=$1 AND id=$2")
-                    .bind(community.as_uuid())
-                    .bind(&id)
-                    .execute(&db.pool)
-                    .await
-                    .expect("legacy soft delete");
-                assert_eq!(
-                    state(id).await,
-                    expected(purged),
-                    "{label}: trigger classification"
-                );
-            }
-
-            for mode in [Delete::Coordinate, Delete::Id] {
+            for mode in [Delete::Coordinate, Delete::Id, Delete::Replace] {
                 let (keys, id, created_at) = store(label, kind, d_tag.clone(), tags.clone()).await;
+                let newer = EventBuilder::new(Kind::Custom(kind), format!("{label}-newer"))
+                    .tags(
+                        tags.iter()
+                            .map(|tag| Tag::parse(tag.clone()).expect("tag"))
+                            .collect::<Vec<_>>(),
+                    )
+                    .custom_created_at(Timestamp::from(created_at + 1))
+                    .sign_with_keys(&keys)
+                    .expect("sign newer event");
                 let delete = || async {
                     match mode {
+                        Delete::Replace => {
+                            db.replace_parameterized_event(community, &newer, &d_tag, None)
+                                .await
+                                .expect("replace")
+                                .1
+                        }
                         Delete::Coordinate => event::soft_delete_by_coordinate(
                             &db.pool,
                             community,
@@ -2393,7 +2221,7 @@ mod postgres_tests {
                 );
                 assert_eq!(
                     state(id.clone()).await,
-                    expected(purged),
+                    expected(purged, mode),
                     "{label}/{mode:?}: app classification"
                 );
                 assert!(
@@ -2406,7 +2234,7 @@ mod postgres_tests {
         // A mesh-status-shaped row whose `d_tag` column was never backfilled:
         // `kind = 30003 AND NULL LIKE … AND tags @> …` makes the purge
         // predicate NULL (a read-state row cannot: its tag/column EXISTS is
-        // false), which the triggers treat as false. The id path must still
+        // false). The id path must still
         // soft-delete it (the `COALESCE` in the generic soft delete), not
         // silently skip it. Coordinate deletes key on `d_tag`, so only the id
         // path can reach this row.
@@ -2426,20 +2254,6 @@ mod postgres_tests {
                 .expect("clear d_tag column");
             id
         };
-        if migration {
-            let id = null_d_tag_row().await;
-            sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id=$1 AND id=$2")
-                .bind(community.as_uuid())
-                .bind(&id)
-                .execute(&db.pool)
-                .await
-                .expect("legacy soft delete");
-            assert_eq!(
-                state(id).await,
-                expected(false),
-                "NULL d_tag: trigger classification"
-            );
-        }
         let id = null_d_tag_row().await;
         assert!(
             event::soft_delete_event_and_update_thread(&db.pool, community, &id, None, None)
@@ -2449,7 +2263,7 @@ mod postgres_tests {
         );
         assert_eq!(
             state(id).await,
-            expected(false),
+            expected(false, Delete::Id),
             "NULL d_tag/Id: app classification"
         );
     }
@@ -2494,10 +2308,6 @@ mod postgres_tests {
         // The replacement: insert a newer head (still at or before the
         // tombstone), hard-delete the old head, and hold the transaction open.
         let mut replacement = db.pool.begin().await.expect("begin replacement");
-        sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
-            .execute(&mut *replacement)
-            .await
-            .expect("opt in to hard delete");
         sqlx::query(
             "INSERT INTO events \
              (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id, d_tag) \
diff --git a/migrations/0089_drop_nip_rs_mesh_migration_triggers.sql b/migrations/0089_drop_nip_rs_mesh_migration_triggers.sql
new file mode 100644
index 00000000000..cac8d46653f
--- /dev/null
+++ b/migrations/0089_drop_nip_rs_mesh_migration_triggers.sql
@@ -0,0 +1,31 @@
+-- Drop the NIP-RS, mention-liveness and mesh-status triggers that exist only
+-- in migration-managed databases. The application now owns each rule inside
+-- the event transaction (crates/buzz-db/src/store/replaceable.rs and
+-- crates/buzz-db/src/store/event.rs):
+--
+-- - it advances parameterized_event_watermarks and rejects stale NIP-RS writes;
+-- - it physically deletes superseded and NIP-09-deleted NIP-RS and mesh-status
+--   rows, with their mentions;
+-- - it indexes mentions in the same transaction as the event, so a mention
+--   can no longer race a hard delete.
+--
+-- schema/ never declared these objects, so desired-state databases already run
+-- without them. Dropping a trigger on the partitioned parent also drops its
+-- partition clones. Published migrations 0009, 0010, 0011 and 0019 stay
+-- checksum-frozen.
+--
+-- DROP TRIGGER takes ACCESS EXCLUSIVE on events and every partition; fail the
+-- deployment rather than queue relay reads and writes behind a long holder.
+SET LOCAL lock_timeout = '5s';
+
+DROP TRIGGER trg_events_nip_rs_watermark ON events;
+DROP TRIGGER trg_events_guard_nip_rs_hard_delete ON events;
+DROP TRIGGER trg_events_purge_soft_deleted_nip_rs ON events;
+DROP TRIGGER trg_events_purge_soft_deleted_buzz_mesh_status ON events;
+DROP TRIGGER trg_event_mentions_require_live_event ON event_mentions;
+
+DROP FUNCTION guard_nip_rs_watermark();
+DROP FUNCTION guard_nip_rs_hard_delete();
+DROP FUNCTION purge_soft_deleted_nip_rs();
+DROP FUNCTION purge_soft_deleted_buzz_mesh_status();
+DROP FUNCTION guard_event_mention_live();