diff --git a/.config/nextest.toml b/.config/nextest.toml
index f14ba3abd..f264f7559 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -173,7 +173,7 @@ test-group = 'ecstore-serial-flaky'
 # the nightly profile derives its set as "the replication module MINUS this
 # allowlist", so any new replication test lands in nightly by default (never
 # silently unrun) until it is explicitly blessed as fast here. Keep the two
-# regexes byte-identical. Count invariant: 20 here + 27 nightly = 47 total
+# regexes byte-identical. Count invariant: 20 here + 28 nightly = 48 total
 # (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md).
 # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane
 # (#4724) because they set a loopback (127.0.0.1) replication target that the
@@ -228,7 +228,7 @@ fail-fast = false
 #     and poll until source and target converge; two replicate over HTTPS, two
 #     pin active SSE failure contracts, and one guards event/history observers.
 #     The SSE-S3 contract remains ignored under backlog#1291.
-#   * 11 `_real_dual_node` site-replication tests — each spawns TWO full rustfs
+#   * 12 `_real_dual_node` site-replication tests — each spawns TWO full rustfs
 #     servers and drives the cross-process site-replication control plane.
 #   * 1 `_real_three_node` site-replication test.
 #   * 1 `_real_single_node` service-account round-trip test.
diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs
index bf61f2c3c..8c6ff2a04 100644
--- a/crates/e2e_test/src/replication_extension_test.rs
+++ b/crates/e2e_test/src/replication_extension_test.rs
@@ -4803,6 +4803,153 @@ async fn test_site_replication_replicates_object_with_bucket_versioning_real_dua
     Ok(())
 }
 
+/// Re-applying a site's own replication config must not disable the peer's reverse direction.
+///
+/// `PutBucketReplication` broadcasts the config to every peer — the console's replication
+/// Save button, `mc replicate import`, and a bucket-metadata import all go through it. The
+/// receiver used to overwrite its rules with the sender's, whose destination ARN names the
+/// receiver itself. No bucket target can satisfy that ARN, so every object written on the
+/// receiver was dropped with only a debug line, while `replicate status` still reported
+/// "1/1 Buckets in sync" because both configs were byte-identical.
+#[tokio::test]
+#[serial]
+async fn test_site_replication_config_broadcast_keeps_reverse_direction_real_dual_node() -> TestResult {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env
+        .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env
+        .start_rustfs_server_without_cleanup_with_env(LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+    let bucket = "site-repl-config-broadcast";
+
+    let add_status = site_replication_add(
+        &source_env,
+        &[
+            PeerSite {
+                name: "broadcast-source".to_string(),
+                endpoint: source_env.url.clone(),
+                access_key: source_env.access_key.clone(),
+                secret_key: source_env.secret_key.clone(),
+                ..Default::default()
+            },
+            PeerSite {
+                name: "broadcast-target".to_string(),
+                endpoint: target_env.url.clone(),
+                access_key: target_env.access_key.clone(),
+                secret_key: target_env.secret_key.clone(),
+                ..Default::default()
+            },
+        ],
+    )
+    .await?;
+    assert!(add_status.success, "unexpected site add result: {add_status:?}");
+    wait_for_site_replication_enabled(&source_env, 2).await?;
+    wait_for_site_replication_enabled(&target_env, 2).await?;
+
+    source_client.create_bucket().bucket(bucket).send().await?;
+    wait_for_bucket_on_target(&target_client, bucket).await?;
+
+    // Both directions work before the broadcast.
+    source_client
+        .put_object()
+        .bucket(bucket)
+        .key("from-source.txt")
+        .body(ByteStream::from_static(b"written on the initiating site"))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_object_on_target(&target_client, bucket, "from-source.txt").await?,
+        b"written on the initiating site".to_vec(),
+    );
+    target_client
+        .put_object()
+        .bucket(bucket)
+        .key("from-target.txt")
+        .body(ByteStream::from_static(b"written on the joined site"))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_object_on_target(&source_client, bucket, "from-target.txt").await?,
+        b"written on the joined site".to_vec(),
+    );
+
+    // Round-trip the source's own config through PutBucketReplication, exactly what the
+    // console does when an operator opens the bucket's replication page and saves it.
+    let source_config = source_client
+        .get_bucket_replication()
+        .bucket(bucket)
+        .send()
+        .await?
+        .replication_configuration
+        .ok_or("source bucket has no replication configuration")?;
+    source_client
+        .put_bucket_replication()
+        .bucket(bucket)
+        .replication_configuration(source_config)
+        .send()
+        .await?;
+
+    let target_config = wait_for_site_replication_rule(&target_client, bucket).await?;
+    let target_deployment_id = site_replication_info(&target_env)
+        .await?
+        .sites
+        .iter()
+        .find(|peer| peer.endpoint == target_env.url)
+        .map(|peer| peer.deployment_id.clone())
+        .ok_or("joined site missing from its own replication info")?;
+    for rule in &target_config.rules {
+        let destination = rule
+            .destination
+            .as_ref()
+            .map(|destination| destination.bucket.as_str())
+            .unwrap_or_default();
+        assert!(
+            !destination.contains(&target_deployment_id),
+            "joined site adopted a rule pointing at itself: {destination}"
+        );
+    }
+
+    target_client
+        .put_object()
+        .bucket(bucket)
+        .key("from-target-after-broadcast.txt")
+        .body(ByteStream::from_static(b"written after the config broadcast"))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_object_on_target(&source_client, bucket, "from-target-after-broadcast.txt").await?,
+        b"written after the config broadcast".to_vec(),
+        "config broadcast made replication one-directional"
+    );
+
+    Ok(())
+}
+
+async fn wait_for_site_replication_rule(
+    client: &aws_sdk_s3::Client,
+    bucket: &str,
+) -> Result<aws_sdk_s3::types::ReplicationConfiguration, Box<dyn Error + Send + Sync>> {
+    for _ in 0..40 {
+        if let Ok(response) = client.get_bucket_replication().bucket(bucket).send().await
+            && let Some(config) = response.replication_configuration
+            && !config.rules.is_empty()
+        {
+            return Ok(config);
+        }
+        sleep(Duration::from_millis(250)).await;
+    }
+
+    Err(format!("bucket {bucket} never reported a replication rule").into())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_site_replication_active_active_converges_without_loops_real_dual_node() -> TestResult {
diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs
index 9c16ae413..cf2a3e1ec 100644
--- a/crates/ecstore/src/bucket/bucket_target_sys.rs
+++ b/crates/ecstore/src/bucket/bucket_target_sys.rs
@@ -807,15 +807,29 @@ impl BucketTargetSys {
             && !new_targets.is_empty()
         {
             for target in &new_targets.targets {
-                if let Ok(client) = self.get_remote_target_client_internal(target).await {
-                    arn_remotes_map.insert(
-                        target.arn.clone(),
-                        ArnTarget {
-                            client: Some(Arc::new(client)),
-                            last_refresh: OffsetDateTime::now_utc(),
-                        },
-                    );
-                    self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
+                match self.get_remote_target_client_internal(target).await {
+                    Ok(client) => {
+                        arn_remotes_map.insert(
+                            target.arn.clone(),
+                            ArnTarget {
+                                client: Some(Arc::new(client)),
+                                last_refresh: OffsetDateTime::now_utc(),
+                            },
+                        );
+                        self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
+                    }
+                    // The target stays in `targets_map`, so it keeps showing up in
+                    // `bucket remote ls` while no client exists to replicate through it —
+                    // replication then drops every object for this ARN. Without this the
+                    // rejection (loopback endpoint, bad CA, unparseable URL) left no trace
+                    // anywhere.
+                    Err(err) => warn!(
+                        bucket = %bucket,
+                        arn = %target.arn,
+                        endpoint = %target.endpoint,
+                        error = %err,
+                        "replication target client unavailable; objects for this ARN will not replicate"
+                    ),
                 }
             }
             targets_map.insert(bucket.to_string(), new_targets.targets.clone());
diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs
index fb2f85ce5..e8a2f3472 100644
--- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs
+++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs
@@ -2098,14 +2098,20 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
 
     for arn in tgt_arns {
         let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &arn).await else {
+            // Deliberately debug: this fires once per object per ARN, so a target that
+            // stays unreachable would flood the log from the replication hot path. The
+            // condition is reported once per pass by the site-replication reconciler and
+            // once per rebuild by `update_all_targets`, which is where an operator can act
+            // on it; the per-object event below still records each dropped object.
             debug!(
                 event = EVENT_RESYNC_RUNTIME_SKIPPED,
                 component = LOG_COMPONENT_ECSTORE,
                 subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
                 bucket = %bucket,
+                object = %object,
                 arn = %arn,
                 reason = "target_client_missing",
-                "Skipping replication object target"
+                "Replication rule has no bucket target for its destination ARN; object not replicated"
             );
             send_local_event(EventArgs {
                 event_name: EventName::ObjectReplicationNotTracked.to_string(),
diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs
index aa83a1b80..feb53fbe3 100644
--- a/crates/iam/src/manager.rs
+++ b/crates/iam/src/manager.rs
@@ -822,6 +822,15 @@ where
             cr.secret_key = secret;
         }
 
+        if let Some(parent_user) = opts.parent_user {
+            // Same gate as the account itself: a rebind grants the account whatever the new
+            // parent can do, so only the site-replication repair path may ask for one.
+            if parent_user.is_empty() || !opts.allow_site_replicator_account || name != SITE_REPLICATOR_SERVICE_ACCOUNT {
+                return Err(Error::IAMActionNotAllowed);
+            }
+            cr.parent_user = parent_user;
+        }
+
         if opts.name.is_some() {
             cr.name = opts.name;
         }
@@ -889,6 +898,10 @@ where
         if name == SITE_REPLICATOR_SERVICE_ACCOUNT && opts.allow_site_replicator_account {
             m.insert(SITE_REPLICATOR_CLAIM.to_owned(), Value::Bool(true));
         }
+        // The parent lives in the token as well, and `prepare_service_account_auth` denies the
+        // request when the two disagree — a rebind that updated only the credential would
+        // lock the account out.
+        m.insert("parent".to_owned(), Value::String(cr.parent_user.clone()));
 
         cr.session_token = jwt_sign(&m, &cr.secret_key)?;
 
@@ -2735,6 +2748,7 @@ mod tests {
                             description: Some("new".to_string()),
                             expiration: None,
                             status: None,
+                            parent_user: None,
                             allow_site_replicator_account: false,
                         },
                     )
@@ -3097,6 +3111,7 @@ mod tests {
             description: Some("Updated service account".to_string()),
             expiration: None,
             session_policy: Some(policy),
+            parent_user: None,
             allow_site_replicator_account: false,
         };
 
diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs
index 47b58add2..608632b20 100644
--- a/crates/iam/src/sys.rs
+++ b/crates/iam/src/sys.rs
@@ -1634,6 +1634,13 @@ pub struct UpdateServiceAccountOpts {
     pub description: Option<String>,
     pub expiration: Option<OffsetDateTime>,
     pub status: Option<String>,
+    /// Rebind the account to a different parent.
+    ///
+    /// Only site replication sets this, and only to repair an account left pointing at a
+    /// parent that a root-credential change invalidated. Rebinding an arbitrary service
+    /// account would let it inherit another user's policies, so it is gated behind
+    /// `allow_site_replicator_account` in the same way the account itself is.
+    pub parent_user: Option<String>,
     pub allow_site_replicator_account: bool,
 }
 
@@ -2061,6 +2068,7 @@ mod tests {
                     description: None,
                     expiration: None,
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2090,6 +2098,7 @@ mod tests {
                     description: None,
                     expiration: None,
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2286,6 +2295,7 @@ mod tests {
                     description: None,
                     expiration: Some(updated_expiration),
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2332,6 +2342,7 @@ mod tests {
                     description: None,
                     expiration: Some(updated_expiration),
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2383,6 +2394,7 @@ mod tests {
                         description: None,
                         expiration: None,
                         status: Some(STATUS_ENABLED.to_string()),
+                        parent_user: None,
                         allow_site_replicator_account: false,
                     },
                 )
@@ -2401,6 +2413,7 @@ mod tests {
                     description: None,
                     expiration: None,
                     status: Some(STATUS_ENABLED.to_string()),
+                    parent_user: None,
                     allow_site_replicator_account: true,
                 },
             )
@@ -2416,6 +2429,116 @@ mod tests {
         );
     }
 
+    /// A root-credential change can leave `site-replicator-0` bound to a parent that no
+    /// longer exists, and the repair must rebind in place: deleting first would leave the
+    /// site with no replication account at all if the recreate failed. The parent also lives
+    /// in the session token, so both copies have to move together or authorization denies
+    /// the account.
+    #[tokio::test]
+    async fn test_site_replicator_parent_rebind_updates_credential_and_claim() {
+        ensure_test_global_credentials();
+
+        let store = StsTestMockStore::new(false);
+        let cache_manager = IamCache::new(store).await.unwrap();
+        let iam_sys = IamSys::new(cache_manager);
+
+        let (cred, _) = iam_sys
+            .new_service_account(
+                "stale-parent-user",
+                None,
+                NewServiceAccountOpts {
+                    access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                    secret_key: "siteReplicatorSecretKeyForTest1234567890".to_string(),
+                    allow_site_replicator_account: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("site replicator account should be created");
+
+        iam_sys
+            .update_service_account(
+                &cred.access_key,
+                UpdateServiceAccountOpts {
+                    session_policy: None,
+                    secret_key: None,
+                    name: None,
+                    description: None,
+                    expiration: None,
+                    status: None,
+                    parent_user: Some("current-parent-user".to_string()),
+                    allow_site_replicator_account: true,
+                },
+            )
+            .await
+            .expect("site replication repair may rebind the parent");
+
+        let (identity, claims) = iam_sys
+            .get_account_with_claims_allow_missing_exp(&cred.access_key)
+            .await
+            .expect("rebound account should still be readable");
+        assert_eq!(identity.credentials.parent_user, "current-parent-user");
+        assert_eq!(
+            claims.get("parent").and_then(Value::as_str),
+            Some("current-parent-user"),
+            "the token claim must follow the credential or authorization denies the account"
+        );
+        assert_eq!(
+            iam_sys
+                .get_site_replicator_service_account_secret(&cred.access_key)
+                .await
+                .expect("secret survives a rebind"),
+            cred.secret_key,
+            "peers keep using the same secret, so a rebind must not rotate it"
+        );
+    }
+
+    /// The rebind is a site-replication repair primitive, not a general capability: letting
+    /// any caller re-parent a service account would let it inherit another user's policies.
+    #[tokio::test]
+    async fn test_parent_rebind_is_rejected_for_ordinary_service_accounts() {
+        ensure_test_global_credentials();
+
+        let store = StsTestMockStore::new(false);
+        let cache_manager = IamCache::new(store).await.unwrap();
+        let iam_sys = IamSys::new(cache_manager);
+
+        let (cred, _) = iam_sys
+            .new_service_account(
+                "ordinary-parent",
+                None,
+                NewServiceAccountOpts {
+                    access_key: "ordinary-service-account".to_string(),
+                    secret_key: "ordinaryServiceAccountSecret1234567890".to_string(),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("ordinary service account should be created");
+
+        for allow in [false, true] {
+            assert!(
+                iam_sys
+                    .update_service_account(
+                        &cred.access_key,
+                        UpdateServiceAccountOpts {
+                            session_policy: None,
+                            secret_key: None,
+                            name: None,
+                            description: None,
+                            expiration: None,
+                            status: None,
+                            parent_user: Some("victim-user".to_string()),
+                            allow_site_replicator_account: allow,
+                        },
+                    )
+                    .await
+                    .is_err(),
+                "re-parenting an ordinary service account must be rejected (allow={allow})"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_created_access_token_authorizes_with_parent_policy() {
         ensure_test_global_credentials();
diff --git a/crates/madmin/src/site_replication.rs b/crates/madmin/src/site_replication.rs
index e131d2ed2..a8555c904 100644
--- a/crates/madmin/src/site_replication.rs
+++ b/crates/madmin/src/site_replication.rs
@@ -509,6 +509,13 @@ pub struct SRBucketInfo {
     pub cors_config_updated_at: Option<OffsetDateTime>,
     #[serde(default, skip_serializing_if = "String::is_empty")]
     pub location: String,
+    /// Whether every `site-repl-*` rule on the reporting site resolves to a usable bucket
+    /// target. A rule without one silently drops every object, and the rule set alone cannot
+    /// reveal it — the config is well-formed, the endpoint behind it is not reachable.
+    ///
+    /// `None` means the peer predates this field: treat it as "unknown", never as a fault.
+    #[serde(rename = "replicationTargetsOnline", default, skip_serializing_if = "Option::is_none")]
+    pub replication_targets_online: Option<bool>,
     #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")]
     pub api_version: Option<String>,
 }
diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md
index e02d48774..ec71b2729 100644
--- a/docs/testing/e2e-suite-inventory.md
+++ b/docs/testing/e2e-suite-inventory.md
@@ -73,7 +73,7 @@
 | quota_test | 14 |  |
 | reliability_disk_fault_test | 3 |  |
 | reliant | 10 | 5 ✅ |
-| replication_extension_test | 47 | 20 ✅ +27 🌙 |
+| replication_extension_test | 48 | 20 ✅ +28 🌙 |
 | security_boundary_test | 4 |  |
 | ssec_copy_test | 2 | ✅ |
 | server_startup_failfast_test | 1 |  |
@@ -87,4 +87,4 @@
 
 `notification_webhook_test` also has 1 ignored store-and-forward regression tracked by rustfs#4852; ignored tests are excluded from the active counts above.
 
-**Total listed: 485 tests across 67 modules · PR smoke subset: 129 tests / 32 modules** (30 full modules + 5 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 27 tests** · generated 2026-07-24.
+**Total listed: 486 tests across 67 modules · PR smoke subset: 129 tests / 32 modules** (30 full modules + 5 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 28 tests** · generated 2026-07-26.
diff --git a/rustfs/src/admin/handlers/service_account.rs b/rustfs/src/admin/handlers/service_account.rs
index 7919c5c66..63a11dc61 100644
--- a/rustfs/src/admin/handlers/service_account.rs
+++ b/rustfs/src/admin/handlers/service_account.rs
@@ -652,6 +652,7 @@ impl Operation for UpdateServiceAccount {
             description: new_description.clone(),
             expiration: new_expiration,
             session_policy: sp,
+            parent_user: None,
             allow_site_replicator_account: false,
         };
 
diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs
index 0c7e7489b..c94ca6f75 100644
--- a/rustfs/src/admin/handlers/site_replication.rs
+++ b/rustfs/src/admin/handlers/site_replication.rs
@@ -149,6 +149,7 @@ const SITE_REPLICATION_PEER_TLS_CAPABILITY_PATH: &str =
 const SITE_REPLICATION_PEER_EDIT_REFRESH_PATH: &str = "/rustfs/admin/v3/site-replication/peer/edit?refresh-targets=true";
 const SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH: &str = "internal:endpoint-target-refresh";
 const SITE_REPLICATION_PEER_REMOVE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/remove";
+const SITE_REPLICATION_DEVNULL_PATH: &str = "/rustfs/admin/v3/site-replication/devnull";
 const RUSTFS_ADMIN_V3_PREFIX: &str = "/rustfs/admin/v3";
 const MINIO_ADMIN_V3_PREFIX: &str = "/minio/admin/v3";
 const MINIO_SITE_REPLICATION_JOIN_PATH: &str = "/minio/admin/v3/site-replication/join";
@@ -369,6 +370,17 @@ impl SiteReplicationLifecycleGuard {
             _guard: SITE_REPLICATION_LIFECYCLE_LOCK.lock().await,
         }
     }
+
+    /// Non-blocking variant for background work that must never interleave with an
+    /// add/remove/endpoint-refresh: those run in phases, and rebuilding rules between two of
+    /// them would resurrect exactly what the operation just tore down. Skipping a round is
+    /// free — the next tick picks it up.
+    fn try_acquire() -> Option<Self> {
+        SITE_REPLICATION_LIFECYCLE_LOCK
+            .try_lock()
+            .ok()
+            .map(|guard| Self { _guard: guard })
+    }
 }
 
 impl SiteReplicationAddInProgressGuard {
@@ -779,6 +791,11 @@ impl SRStatusOptions {
 }
 
 pub fn register_site_replication_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    // Hand the reconciler to the infra-layer scheduler here rather than letting startup call
+    // into this module: startup sits below this layer and must not depend upwards. The admin
+    // router is built before startup reconciles, so the hook is always installed in time.
+    crate::site_replication_reconcile::register_site_replication_reconciler(reconcile_site_replication_wiring);
+
     for (method, path, operation) in [
         (Method::PUT, "/v3/site-replication/add", AdminOperation(&SiteReplicationAddHandler {})),
         (
@@ -2468,6 +2485,7 @@ async fn set_site_replicator_service_account_secret(parent_user: &str, secret_ke
                     description: None,
                     expiration: None,
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: true,
                 },
             )
@@ -2518,6 +2536,347 @@ async fn ensure_site_replicator_service_account(parent_user: &str, rotate_secret
     Ok((access_key, secret_key))
 }
 
+/// Whether a bucket target is one this site's own peer topology produced.
+///
+/// Bucket targets are writable by anyone holding `admin:SetBucketTarget`, so a target that
+/// merely carries the `site-replicator-0` access key proves nothing: an attacker with only
+/// that permission could plant a secret of their choosing and have reconciliation recreate
+/// the broadly privileged replication account with it. Require the target to name a peer in
+/// the persisted state *and* to point at that peer's recorded endpoint, which is state only
+/// `replicate add`/`edit` can write.
+fn bucket_target_matches_configured_peer(target: &BucketTarget, state: &SiteReplicationState) -> bool {
+    let Some(deployment_id) = bucket_target_deployment_id(target) else {
+        return false;
+    };
+    state
+        .peers
+        .get(&deployment_id)
+        .is_some_and(|peer| bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint))
+}
+
+/// Recover the shared site-replication secret from a bucket target belonging to a configured
+/// peer.
+///
+/// Every site-replication bucket target stores the `site-replicator-0` credentials in
+/// `BUCKET_TARGETS_FILE`, which is not encrypted with the root credentials. That makes it the
+/// one local copy that survives a root-credential change, so it can reseed the IAM account
+/// when the IAM record itself became unreadable.
+///
+/// Returns `None` unless every matching target agrees on the secret: disagreement means at
+/// least one was written by something other than this site's own reconciliation, and picking
+/// either would be a guess.
+async fn site_replicator_secret_from_bucket_targets(access_key: &str, state: &SiteReplicationState) -> Option<String> {
+    let store = current_object_store_handle()?;
+    let buckets = store.list_bucket(&BucketOptions::default()).await.ok()?;
+    let mut recovered: Option<String> = None;
+
+    for bucket in buckets {
+        let Ok(targets) = metadata_sys::list_bucket_targets(&bucket.name).await else {
+            continue;
+        };
+        for target in targets.targets {
+            if target.target_type != BucketTargetType::ReplicationService
+                || !bucket_target_matches_configured_peer(&target, state)
+            {
+                continue;
+            }
+            let Some(credentials) = target.credentials.as_ref() else {
+                continue;
+            };
+            if credentials.access_key != access_key || credentials.secret_key.is_empty() {
+                continue;
+            }
+            match &recovered {
+                Some(seen) if seen != &credentials.secret_key => return None,
+                Some(_) => {}
+                None => recovered = Some(credentials.secret_key.clone()),
+            }
+        }
+    }
+
+    recovered
+}
+
+/// Whether the parent recorded in site-replication state can actually back a service account.
+///
+/// Repairing against a parent that does not exist would produce an account no policy path can
+/// resolve, so a missing parent means "leave the current binding alone and report it".
+async fn site_replicator_parent_is_usable(parent: &str) -> bool {
+    if rustfs_iam::is_root_access_key(parent) {
+        return true;
+    }
+    match current_iam_handle() {
+        Some(iam_sys) => iam_sys.get_user_info(parent).await.is_ok(),
+        None => false,
+    }
+}
+
+/// Whether an IAM lookup failure means the account is absent or unreadable, as opposed to a
+/// transient store failure. Only the former may trigger a reseed from bucket targets.
+fn is_missing_service_account_error(err: &rustfs_iam::error::Error) -> bool {
+    matches!(
+        err,
+        rustfs_iam::error::Error::NoSuchAccount(_)
+            | rustfs_iam::error::Error::NoSuchServiceAccount(_)
+            | rustfs_iam::error::Error::NoSuchUser(_)
+            | rustfs_iam::error::Error::ConfigNotFound
+    )
+}
+
+/// Reconcile the local `site-replicator-0` account against the persisted site-replication
+/// state, repairing the two drifts that no other code path can undo.
+///
+/// `update_service_account` cannot rewrite `parent_user`, so once the account is bound to a
+/// parent that a root-credential change invalidated, every later `replicate add` takes the
+/// update branch and preserves the stale binding forever. Worse, IAM records encrypted with
+/// the previous root secret fail to decrypt and surface as "no such account", which silently
+/// disables every control-plane push while `replicate info` still reports the site enabled.
+/// Both used to require deleting and recreating the account by hand.
+async fn reconcile_site_replicator_service_account() -> S3Result<()> {
+    let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await;
+    let state = load_site_replication_state().await?;
+    if !state.enabled() || state.service_account_access_key != SITE_REPLICATOR_SERVICE_ACCOUNT {
+        return Ok(());
+    }
+
+    let Some(iam_sys) = current_iam_handle() else {
+        return Err(s3_error!(InvalidRequest, "iam not init"));
+    };
+
+    let parent_user = state.service_account_parent.clone();
+    if parent_user.is_empty() {
+        warn!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "service_account_parent_unknown",
+            "admin site replication state"
+        );
+        return Ok(());
+    }
+
+    let access_key = SITE_REPLICATOR_SERVICE_ACCOUNT;
+    let session_policy = site_replicator_service_account_policy()?;
+    let reason = match iam_sys.get_site_replicator_service_account_secret(access_key).await {
+        Ok(_) => {
+            // Never rebuild on an unread parent: `unwrap_or_default` here would compare an
+            // empty string against a real parent and repair a healthy account on every boot.
+            let Ok((credentials, _)) = iam_sys.get_service_account(access_key).await else {
+                return Ok(());
+            };
+            if credentials.parent_user == parent_user {
+                return Ok(());
+            }
+
+            if !site_replicator_parent_is_usable(&parent_user).await {
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "service_account_parent_missing",
+                    reason = "stale_parent",
+                    parent = %parent_user,
+                    "admin site replication state"
+                );
+                return Ok(());
+            }
+
+            // The record is intact and may be authenticating replication traffic right now,
+            // so rebind it in place. Deleting first would open a window — however brief —
+            // where a crash or a storage error leaves the site with no replication account
+            // at all, which is worse than the stale binding being repaired.
+            iam_sys
+                .update_service_account(
+                    access_key,
+                    UpdateServiceAccountOpts {
+                        session_policy: Some(session_policy),
+                        secret_key: None,
+                        name: None,
+                        description: None,
+                        expiration: None,
+                        status: None,
+                        parent_user: Some(parent_user.clone()),
+                        allow_site_replicator_account: true,
+                    },
+                )
+                .await
+                .map_err(ApiError::from)?;
+            "stale_parent"
+        }
+        // Only a genuinely absent or unreadable account may be reseeded from a bucket
+        // target. A transient store error must not trigger a rewrite of a live account.
+        Err(err) if is_missing_service_account_error(&err) => {
+            let Some(secret) = site_replicator_secret_from_bucket_targets(access_key, &state).await else {
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "service_account_unrecoverable",
+                    error = ?err,
+                    "admin site replication state"
+                );
+                return Ok(());
+            };
+
+            if !site_replicator_parent_is_usable(&parent_user).await {
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "service_account_parent_missing",
+                    reason = "account_unreadable",
+                    parent = %parent_user,
+                    "admin site replication state"
+                );
+                return Ok(());
+            }
+
+            // Nothing readable to preserve, so creation is the whole repair — there is no
+            // delete to leave a gap behind.
+            iam_sys
+                .new_service_account(
+                    &parent_user,
+                    None,
+                    NewServiceAccountOpts {
+                        session_policy: Some(session_policy),
+                        access_key: access_key.to_string(),
+                        secret_key: secret,
+                        name: None,
+                        description: None,
+                        expiration: None,
+                        allow_site_replicator_account: true,
+                        claims: None,
+                    },
+                )
+                .await
+                .map_err(ApiError::from)?;
+            "account_unreadable"
+        }
+        Err(err) => return Err(ApiError::from(err).into()),
+    };
+
+    warn!(
+        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+        component = LOG_COMPONENT_ADMIN,
+        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+        result = "service_account_repaired",
+        reason,
+        parent = %parent_user,
+        "admin site replication state"
+    );
+
+    Ok(())
+}
+
+/// Rebuild every replicated bucket's outbound rules and targets from the current peer set.
+///
+/// A bucket whose `site-repl-*` rule was overwritten by a peer's config points at this very
+/// deployment and replicates nothing. Nothing else revisits an existing bucket — the rule
+/// builders only run on bucket creation, peer bucket-ops and metadata pushes — so without a
+/// pass here an upgraded site keeps the broken rules until someone recreates the bucket.
+/// Reconciliation is a no-op write-wise when the rules already match.
+async fn reconcile_site_replication_buckets() -> S3Result<()> {
+    let Some(runtime) = runtime_site_replication_targets().await? else {
+        return Ok(());
+    };
+    let Some(store) = current_object_store_handle() else {
+        return Ok(());
+    };
+    let buckets = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?;
+
+    for bucket in buckets {
+        if let Err(err) = ensure_site_replication_bucket_setup_with_runtime(&bucket.name, &runtime).await {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                bucket = %bucket.name,
+                result = "bucket_reconcile_failed",
+                error = ?err,
+                "admin site replication state"
+            );
+            continue;
+        }
+
+        // Once per bucket per pass, so an operator sees a rule that resolves to nothing
+        // without the replication hot path logging it for every object. Reconciliation
+        // cannot fix this case: the rules are right and the peer endpoint is not reachable.
+        if let Ok(metadata) = metadata_sys::get(&bucket.name).await
+            && !site_replication_targets_online(&bucket.name, &metadata.replication_config_xml).await
+        {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                bucket = %bucket.name,
+                result = "replication_target_offline",
+                "a site replication rule has no usable remote target; objects for it are not replicating"
+            );
+        }
+    }
+
+    Ok(())
+}
+
+/// Repair drifted site-replication wiring: the service account first, since the bucket pass
+/// signs its targets with that account's secret.
+///
+/// Registered into the infra-layer scheduler (`site_replication_reconcile`) rather than
+/// called from it, so startup never has to reach up into this layer.
+///
+/// Gives up the whole round rather than racing a multi-phase operation. Two mechanisms are
+/// needed: the lifecycle lock covers add and remove, while an endpoint refresh commits
+/// bucket targets and peer state in separate steps *without* holding that lock
+/// (`SiteReplicationEditHandler`), so a tick landing between them would rewrite the targets
+/// from the stale endpoint. The pending marker in the persisted state closes that window.
+/// Skipping costs nothing — the timer comes back.
+fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
+    Box::pin(async {
+        // The scheduler starts before IAM and the object store are guaranteed ready (IAM
+        // bootstrap may still be recovering), so an early tick returns quietly instead of
+        // logging a failure for every reconciler.
+        if current_iam_handle().is_none() || current_object_store_handle().is_none() {
+            return;
+        }
+
+        let Some(_lifecycle) = SiteReplicationLifecycleGuard::try_acquire() else {
+            return;
+        };
+
+        match load_site_replication_state().await {
+            Ok(state) => {
+                if state.pending_endpoint_refresh.is_some() || state.pending_remove.is_some() || state.pending_rotation.is_some()
+                {
+                    return;
+                }
+            }
+            // Unreadable state is reported by the reconcilers below; do not double-log here.
+            Err(_) => return,
+        }
+
+        if let Err(err) = reconcile_site_replicator_service_account().await {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                result = "service_account_reconcile_failed",
+                error = ?err,
+                "admin site replication state"
+            );
+        }
+        if let Err(err) = reconcile_site_replication_buckets().await {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                result = "bucket_reconcile_failed",
+                error = ?err,
+                "admin site replication state"
+            );
+        }
+    })
+}
+
 fn site_replication_peer_wire_path(path: &str) -> String {
     let (path_only, query) = path
         .split_once('?')
@@ -3839,6 +4198,8 @@ async fn build_sr_info(state: &SiteReplicationState, local_peer: &PeerInfo) -> S
             entry.quota_config_updated_at = maybe_time(metadata.quota_config_updated_at);
             entry.expiry_lc_config_updated_at = maybe_time(metadata.lifecycle_config_updated_at);
             entry.cors_config_updated_at = maybe_time(metadata.cors_config_updated_at);
+            entry.replication_targets_online =
+                Some(site_replication_targets_online(&bucket.name, &metadata.replication_config_xml).await);
         }
 
         info.buckets.insert(bucket.name, entry);
@@ -4134,7 +4495,7 @@ fn canonical_status_json(value: &Value) -> Value {
     }
 }
 
-fn site_replication_rule_complete(rule: &ReplicationRule) -> bool {
+fn site_replication_rule_complete(rule: &ReplicationRule, owner_deployment_id: &str) -> bool {
     let delete_marker_enabled = rule
         .delete_marker_replication
         .as_ref()
@@ -4155,16 +4516,29 @@ fn site_replication_rule_complete(rule: &ReplicationRule) -> bool {
             replica_modifications.status == ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED)
         });
 
+    // A rule whose destination ARN names the site that holds it can never replicate:
+    // `reconcile_site_replication_bucket_targets` skips the local peer, so no bucket
+    // target backs that ARN and every object is dropped. Two sites holding byte-identical
+    // configs used to satisfy this check while exactly one of them could push.
+    let points_at_remote_site = replication_target_arn_deployment_id(&rule.destination.bucket)
+        .is_some_and(|deployment_id| deployment_id != owner_deployment_id);
+
     rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-"))
         && rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED)
+        && points_at_remote_site
         && delete_marker_enabled
         && delete_enabled
         && existing_object_enabled
         && replica_modifications_enabled
 }
 
-fn site_replication_config_mismatch<'a>(values: impl Iterator<Item = Option<&'a String>>, total_sites: usize) -> (usize, bool) {
-    let values = values.flatten().collect::<Vec<_>>();
+fn site_replication_config_mismatch<'a>(
+    values: impl Iterator<Item = (&'a str, Option<&'a String>)>,
+    total_sites: usize,
+) -> (usize, bool) {
+    let values = values
+        .filter_map(|(deployment_id, value)| value.map(|value| (deployment_id, value)))
+        .collect::<Vec<_>>();
     let present = values.len();
     if present == 0 {
         return (0, false);
@@ -4174,15 +4548,20 @@ fn site_replication_config_mismatch<'a>(values: impl Iterator<Item = Option<&'a
     }
 
     let expected_rules = total_sites.saturating_sub(1);
-    let replicated = values.iter().all(|raw| {
+    let replicated = values.iter().all(|(deployment_id, raw)| {
         // `raw` is the wire form produced by build_sr_info, i.e. base64-encoded XML
         // (raw_config_to_base64). Decode it before XML-parsing — parsing the base64 text
         // directly always fails, which would falsely report every replicated bucket as
         // out-of-sync ("0/N Buckets in sync"). decode_bucket_meta_wire_value falls back to
         // the raw bytes when the value is not base64, so plain-XML callers still work.
         let xml = decode_bucket_meta_wire_value(raw);
-        deserialize::<ReplicationConfiguration>(&xml)
-            .is_ok_and(|config| config.rules.len() == expected_rules && config.rules.iter().all(site_replication_rule_complete))
+        deserialize::<ReplicationConfiguration>(&xml).is_ok_and(|config| {
+            config.rules.len() == expected_rules
+                && config
+                    .rules
+                    .iter()
+                    .all(|rule| site_replication_rule_complete(rule, deployment_id))
+        })
     });
 
     (present, !replicated)
@@ -4233,14 +4612,27 @@ fn merge_bucket_status_info(status: &mut SRStatusInfo, site_infos: &BTreeMap<Str
                 .map(|info| info.buckets.get(bucket_name).and_then(|bucket| bucket.versioning.as_ref())),
             total_sites,
         );
-        let (_, replication_mismatch) = site_replication_config_mismatch(
-            site_infos.values().map(|info| {
-                info.buckets
-                    .get(bucket_name)
-                    .and_then(|bucket| bucket.replication_config.as_ref())
+        let (_, rules_mismatch) = site_replication_config_mismatch(
+            site_infos.iter().map(|(deployment_id, info)| {
+                (
+                    deployment_id.as_str(),
+                    info.buckets
+                        .get(bucket_name)
+                        .and_then(|bucket| bucket.replication_config.as_ref()),
+                )
             }),
             total_sites,
         );
+        // Well-formed rules on a site that cannot reach the peer behind them replicate
+        // nothing, so a site reporting an offline target is out of sync regardless of how
+        // its rule set reads. Peers that do not report the field are left out of the verdict.
+        let targets_offline = site_infos.values().any(|info| {
+            info.buckets
+                .get(bucket_name)
+                .and_then(|bucket| bucket.replication_targets_online)
+                == Some(false)
+        });
+        let replication_mismatch = rules_mismatch || targets_offline;
         let (quota_count, quota_mismatch) = string_config_mismatch(
             site_infos
                 .values()
@@ -6070,6 +6462,81 @@ fn prune_removed_site_replication_bucket_targets(
     (BucketTargets { targets }, removed)
 }
 
+fn is_site_replication_rule(rule: &ReplicationRule) -> bool {
+    rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-"))
+}
+
+/// Whether every `site-repl-*` rule on this bucket resolves to a live remote target.
+///
+/// The rule set alone cannot answer this: a rule can be perfectly formed while the endpoint
+/// recorded for its peer is one this site cannot reach, so `update_all_targets` never built
+/// a client for it and `replicate_object` drops every object against that ARN. Reads the
+/// already-resolved client map rather than rebuilding clients, so it stays cheap enough for
+/// the status path.
+async fn site_replication_targets_online(bucket: &str, replication_config_xml: &[u8]) -> bool {
+    let Ok(config) = deserialize::<ReplicationConfiguration>(replication_config_xml) else {
+        return true;
+    };
+
+    for rule in config.rules.iter().filter(|rule| is_site_replication_rule(rule)) {
+        if BucketTargetSys::get()
+            .get_remote_target_client_by_arn(bucket, &rule.destination.bucket)
+            .await
+            .is_none()
+        {
+            return false;
+        }
+    }
+
+    true
+}
+
+/// Merge a peer's replication config into the local one.
+///
+/// `site-repl-*` rules encode the *sender's* outbound direction — their destination ARN
+/// names the receiver — so applying a peer's rule set verbatim replaces the receiver's
+/// reverse rule with one pointing at itself. No bucket target can satisfy that ARN
+/// (`reconcile_site_replication_bucket_targets` skips the local peer), so the receiver
+/// silently stops replicating back: the one-directional symptom. Only operator-authored
+/// rules travel between sites; each site owns its own `site-repl-*` rules.
+fn merge_incoming_replication_config(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+) -> Option<ReplicationConfiguration> {
+    let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default();
+    // Operator rules first, then the local site rules — the same order
+    // `ensure_site_replication_bucket_replication_config_with_runtime` produces, so its
+    // no-op check matches and the bucket metadata is written once per broadcast, not twice.
+    let mut rules: Vec<ReplicationRule> = incoming
+        .into_iter()
+        .flat_map(|config| config.rules)
+        .filter(|rule| !is_site_replication_rule(rule))
+        .collect();
+    rules.extend(
+        local
+            .into_iter()
+            .flat_map(|config| config.rules)
+            .filter(is_site_replication_rule),
+    );
+
+    if rules.is_empty() {
+        return None;
+    }
+
+    for (index, rule) in rules.iter_mut().enumerate() {
+        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
+    }
+
+    // A site-replication ARN in `role` is the sender's, and `site_replication_target_arns_by_peer`
+    // reads it — carrying it over would pin the receiver's targets to the sender's identity.
+    let role = match replication_target_arn_deployment_id(&incoming_role) {
+        Some(_) => String::new(),
+        None => incoming_role,
+    };
+
+    Some(ReplicationConfiguration { role, rules })
+}
+
 fn replication_rule_deployment_id(rule: &ReplicationRule) -> Option<String> {
     if let Some(rule_id) = rule.id.as_deref() {
         if let Some(deployment_id) = rule_id.strip_prefix("site-repl-")
@@ -6155,14 +6622,26 @@ fn build_site_replication_config(
     state: &SiteReplicationState,
     local_peer: &PeerInfo,
     service_account_secret_key: &str,
+    existing: Option<&ReplicationConfiguration>,
 ) -> S3Result<Option<ReplicationConfiguration>> {
+    // Reuse the ARN already recorded for a peer so the rule keeps pointing at the same
+    // bucket target `reconcile_site_replication_bucket_targets` keys off (a MinIO-era
+    // `arn:minio:...` target would otherwise be orphaned by a freshly minted ARN).
+    let configured_arns = site_replication_target_arns_by_peer(existing);
     let mut rules = Vec::new();
     for peer in state.peers.values() {
         if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
             continue;
         }
 
-        let Some(target) = site_replication_bucket_target_for_peer(bucket, state, peer, service_account_secret_key, None)? else {
+        let Some(target) = site_replication_bucket_target_for_peer(
+            bucket,
+            state,
+            peer,
+            service_account_secret_key,
+            configured_arns.get(&peer.deployment_id).cloned(),
+        )?
+        else {
             continue;
         };
         rules.push(build_site_replication_rule(
@@ -6194,6 +6673,8 @@ async fn ensure_site_replication_bucket_targets_with_runtime(
         Err(StorageError::ConfigNotFound) => BucketTargets::default(),
         Err(err) => return Err(ApiError::from(err).into()),
     };
+    let existing_json = serde_json::to_vec(&existing)
+        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize bucket targets failed: {e}")))?;
 
     let updated =
         reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config, service_account_secret_key)?;
@@ -6203,6 +6684,12 @@ async fn ensure_site_replication_bucket_targets_with_runtime(
 
     let json_targets = serde_json::to_vec(&updated)
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize bucket targets failed: {e}")))?;
+    // Rewriting identical targets would churn bucket metadata and rebuild every remote S3
+    // client — noticeable now that startup reconciles all buckets, not just the one bucket
+    // an operation touched.
+    if json_targets == existing_json {
+        return Ok(());
+    }
     metadata_sys::update(bucket, BUCKET_TARGETS_FILE, json_targets)
         .await
         .map_err(ApiError::from)?;
@@ -6241,54 +6728,48 @@ async fn ensure_site_replication_bucket_replication_config_with_runtime(
     local_peer: &PeerInfo,
     service_account_secret_key: &str,
 ) -> S3Result<()> {
-    // Fix 6: reconcile rather than early-returning when any config already exists.
-    // The old code bailed on Ok(_), so the second site joined a replicated bucket and
-    // ended up with NO rule pointing back to the first site. Objects on that site could
-    // never travel back, producing the "one-directional" replication symptom.
-    let Some(desired) = build_site_replication_config(bucket, state, local_peer, service_account_secret_key)? else {
-        return Ok(());
-    };
-
-    // Load the existing rules (may be empty if never configured).
-    let mut existing_rules = match metadata_sys::get_replication_config(bucket).await {
-        Ok((existing, _)) => existing.rules,
-        Err(StorageError::ConfigNotFound) => Vec::new(),
+    let existing = match metadata_sys::get_replication_config(bucket).await {
+        Ok((existing, _)) => Some(existing),
+        Err(StorageError::ConfigNotFound) => None,
         Err(err) => return Err(ApiError::from(err).into()),
     };
 
-    // Collect the IDs of existing site-repl-* rules so we don't duplicate them.
-    let existing_rule_ids: HashSet<String> = existing_rules
-        .iter()
-        .filter_map(|r| r.id.as_deref())
-        .filter(|id| id.starts_with("site-repl-"))
-        .map(String::from)
-        .collect();
+    let Some(desired) = build_site_replication_config(bucket, state, local_peer, service_account_secret_key, existing.as_ref())?
+    else {
+        return Ok(());
+    };
 
-    let mut added = false;
-    for rule in desired.rules {
-        let rule_id = rule.id.as_deref().unwrap_or("");
-        if !existing_rule_ids.contains(rule_id) {
-            existing_rules.push(rule);
-            added = true;
-        }
+    // `site-repl-*` rules are derived state owned by this site: rebuild them from the
+    // current peer set on every pass instead of preserving whatever is on disk. A rule
+    // left over from a removed peer — or one whose destination ARN names this very
+    // deployment, which no bucket target can ever satisfy — must not survive, otherwise
+    // objects are queued against an ARN that resolves to nothing.
+    let (existing_role, existing_rules) = existing
+        .map(|config| (config.role, config.rules))
+        .unwrap_or_else(|| (String::new(), Vec::new()));
+    let mut rules: Vec<ReplicationRule> = existing_rules
+        .iter()
+        .filter(|rule| !is_site_replication_rule(rule))
+        .cloned()
+        .collect();
+    rules.extend(desired.rules);
+    for (index, rule) in rules.iter_mut().enumerate() {
+        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
     }
 
-    if !added {
-        // All desired rules are already present — nothing to write.
+    // Only a site-replication ARN in `role` is ours to drop — an operator-authored role is
+    // part of the bucket's S3-visible configuration, and repairing a reverse rule must not
+    // quietly rewrite it. Same rule as `merge_incoming_replication_config`.
+    let role = match replication_target_arn_deployment_id(&existing_role) {
+        Some(_) => String::new(),
+        None => existing_role.clone(),
+    };
+
+    if rules == existing_rules && role == existing_role {
         return Ok(());
     }
 
-    // Re-assign contiguous priorities to avoid conflicts with any preserved rules.
-    for (i, rule) in existing_rules.iter_mut().enumerate() {
-        rule.priority = Some((i + 1) as i32);
-    }
-
-    let config = ReplicationConfiguration {
-        role: String::new(),
-        rules: existing_rules,
-    };
-
-    let data = serialize(&config)
+    let data = serialize(&ReplicationConfiguration { role, rules })
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}")))?;
     metadata_sys::update(bucket, BUCKET_REPLICATION_CONFIG, data)
         .await
@@ -6421,6 +6902,52 @@ pub async fn site_replication_peer_deployment_id_for_endpoint(endpoint: &str) ->
 /// kick a resync toward every remote peer so pre-existing objects back-fill. Returns a list of
 /// human-readable per-bucket failure messages (empty on full success) so the caller can surface
 /// them to the operator instead of silently reporting success; a failure never aborts the caller.
+/// Probe every remote peer from the joining site before reporting the join a success.
+///
+/// A peer's endpoint is whatever that peer derived from the `Host` header of the admin
+/// request that created the topology, so the initiator can record an address only it can
+/// reach — a console-port rewrite, a NAT address, a LAN-only host. The initiator's own
+/// probes all succeed in that case, and the reverse direction then fails silently forever
+/// because nothing else pushes from here until an object is written. Report it in the add
+/// response instead of rejecting the join: an operator may legitimately be opening the
+/// return path afterwards.
+async fn probe_reverse_peer_reachability(state: &SiteReplicationState, local_peer: &PeerInfo) -> SiteReplicationErrorSummary {
+    let mut errors = SiteReplicationErrorSummary::default();
+    let secret_key = match site_replicator_service_account_secret(&state.service_account_access_key).await {
+        Ok(secret) => secret,
+        Err(err) => {
+            errors.push(format!("reverse reachability probe skipped: {err}"));
+            return errors;
+        }
+    };
+
+    for peer in state.peers.values() {
+        if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
+            continue;
+        }
+        let connection = match runtime_peer_connection(peer) {
+            Ok(connection) => connection,
+            Err(err) => {
+                errors.push(format!("{} is not reachable from this site: {err}", peer.endpoint));
+                continue;
+            }
+        };
+        if let Err(err) = send_peer_admin_request(
+            &connection,
+            SITE_REPLICATION_DEVNULL_PATH,
+            &state.service_account_access_key,
+            &secret_key,
+            &serde_json::json!({}),
+        )
+        .await
+        {
+            errors.push(format!("{} is not reachable from this site: {err}", peer.endpoint));
+        }
+    }
+
+    errors
+}
+
 async fn backfill_existing_buckets_after_add(
     state: &SiteReplicationState,
     local_peer: &PeerInfo,
@@ -7045,16 +7572,40 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
         }
     }
 
-    if item.r#type == "replication-config" {
-        item.replication_config
-            .as_ref()
-            .map(|raw| {
-                let data = decode_bucket_meta_wire_value(raw);
-                deserialize::<s3s::dto::ReplicationConfiguration>(&data)
-            })
-            .transpose()
-            .map_err(|e| s3_error!(InvalidRequest, "invalid replication config: {e}"))?;
-    }
+    // Nothing to write and nothing on disk to clear: the common case for a site joining a
+    // replicated bucket, where every incoming rule is the sender's own. Skipping the write
+    // avoids stamping an empty config over a bucket that never had one;
+    // `ensure_site_replication_bucket_setup` below still installs this site's own rules.
+    let mut skip_config_write = false;
+    let merged_replication_config =
+        if item.r#type == "replication-config" {
+            let incoming = item
+                .replication_config
+                .as_ref()
+                .map(|raw| {
+                    let data = decode_bucket_meta_wire_value(raw);
+                    deserialize::<ReplicationConfiguration>(&data)
+                })
+                .transpose()
+                .map_err(|e| s3_error!(InvalidRequest, "invalid replication config: {e}"))?;
+            let local = match metadata_sys::get_replication_config(&item.bucket).await {
+                Ok((config, _)) => Some(config),
+                Err(StorageError::ConfigNotFound) => None,
+                Err(err) => return Err(ApiError::from(err).into()),
+            };
+            let local_absent = local.is_none();
+            match merge_incoming_replication_config(incoming, local) {
+                Some(config) => Some(serialize(&config).map_err(|e| {
+                    S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}"))
+                })?),
+                None => {
+                    skip_config_write = local_absent;
+                    None
+                }
+            }
+        } else {
+            None
+        };
 
     let data = match item.r#type.as_str() {
         "policy" => item
@@ -7071,25 +7622,29 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
         "version-config" => decode_bucket_meta_wire_option(item.versioning),
         "object-lock-config" => decode_bucket_meta_wire_option(item.object_lock_config),
         "sse-config" => decode_bucket_meta_wire_option(item.sse_config),
-        "replication-config" => decode_bucket_meta_wire_option(item.replication_config),
+        "replication-config" => merged_replication_config,
         "lc-config" => decode_bucket_meta_wire_option(item.expiry_lc_config),
         "cors-config" => decode_bucket_meta_wire_option(item.cors),
         _ => unreachable!(),
     };
 
-    if let Some(data) = data {
-        metadata_sys::update(&item.bucket, config_file, data)
-            .await
-            .map_err(ApiError::from)?;
-    } else {
-        metadata_sys::delete(&item.bucket, config_file)
-            .await
-            .map_err(ApiError::from)?;
+    if !skip_config_write {
+        if let Some(data) = data {
+            metadata_sys::update(&item.bucket, config_file, data)
+                .await
+                .map_err(ApiError::from)?;
+        } else {
+            metadata_sys::delete(&item.bucket, config_file)
+                .await
+                .map_err(ApiError::from)?;
+        }
     }
     drop(targets_guard);
 
     if item.r#type == "replication-config" {
-        ensure_site_replication_bucket_targets(&item.bucket).await?;
+        // Rebuild the local outbound rules too: a site that joined an already-replicated
+        // bucket receives this item before it has any `site-repl-*` rule of its own.
+        ensure_site_replication_bucket_setup(&item.bucket).await?;
     }
 
     if item.r#type == "version-config"
@@ -7369,6 +7924,7 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
                                     secret_key: Some(create.secret_key),
                                     expiration: create.expiration,
                                     status: (!create.status.is_empty()).then_some(create.status),
+                                    parent_user: None,
                                     allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
                                 },
                             )
@@ -7421,6 +7977,9 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
                             description: (!update.description.is_empty()).then_some(update.description),
                             expiration: update.expiration,
                             status: (!update.status.is_empty()).then_some(update.status),
+                            // Peers replicate credentials, never the local parent binding:
+                            // each site resolves its own parent from its own IAM.
+                            parent_user: None,
                             allow_site_replicator_account,
                         },
                     )
@@ -7890,6 +8449,7 @@ impl Operation for SRPeerJoinHandler {
                             description: None,
                             expiration: None,
                             status: None,
+                            parent_user: None,
                             allow_site_replicator_account: join_req.svc_acct_access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
                         },
                     )
@@ -7936,7 +8496,8 @@ impl Operation for SRPeerJoinHandler {
         // Fix 1 (receiving side): ensure the joining peer also sets up replication for any
         // buckets it already owns so the reverse direction works from the start. Per-bucket
         // failures are logged (BUG2) so a reverse-direction back-fill gap is observable.
-        let backfill_errors = backfill_existing_buckets_after_add(&state, &local_peer, bootstrap_token.as_deref()).await;
+        let mut backfill_errors = probe_reverse_peer_reachability(&state, &local_peer).await;
+        backfill_errors.extend(backfill_existing_buckets_after_add(&state, &local_peer, bootstrap_token.as_deref()).await);
         if !backfill_errors.is_empty() {
             warn!(
                 event = EVENT_ADMIN_SITE_REPLICATION_STATE,
@@ -10627,13 +11188,118 @@ mod tests {
         let site_b_xml = String::from_utf8(serialize(&site_b_config).expect("site replication XML should serialize"))
             .expect("site replication XML should be UTF-8");
 
-        assert!(site_replication_rule_complete(&site_a_config.rules[0]));
+        assert!(site_replication_rule_complete(&site_a_config.rules[0], "site-a"));
         assert_eq!(
-            site_replication_config_mismatch(vec![Some(&site_a_xml), Some(&site_b_xml)].into_iter(), 2),
+            site_replication_config_mismatch(vec![("site-a", Some(&site_a_xml)), ("site-b", Some(&site_b_xml))].into_iter(), 2),
             (2, false)
         );
     }
 
+    // A site whose rules are well-formed but whose peer endpoint it cannot reach builds no
+    // target client, so it replicates nothing while its rule set still reads as correct.
+    // Rule-shape checking alone cannot see that, so the reporting site says so directly.
+    #[test]
+    fn test_merge_bucket_status_reports_offline_targets_as_mismatch() {
+        // Both sites carry a correct, peer-specific rule set: rule-shape checking alone
+        // sees a healthy pair. Only the reported target health distinguishes them.
+        let site_info = |peer: &str, targets_online: Option<bool>| {
+            let xml = String::from_utf8(serialize(&site_repl_config(peer)).unwrap()).unwrap();
+            let mut info = SRInfo::default();
+            info.buckets.insert(
+                "photos".to_string(),
+                SRBucketInfo {
+                    bucket: "photos".to_string(),
+                    replication_config: Some(xml),
+                    replication_targets_online: targets_online,
+                    ..Default::default()
+                },
+            );
+            info
+        };
+
+        let mut status = SRStatusInfo::default();
+        let site_infos = BTreeMap::from([
+            ("site-a".to_string(), site_info("site-b", Some(true))),
+            ("site-b".to_string(), site_info("site-a", Some(false))),
+        ]);
+        merge_bucket_status_info(&mut status, &site_infos, &SRStatusOptions::default());
+
+        let summary = status
+            .bucket_stats
+            .get("photos")
+            .and_then(|per_site| per_site.get("site-a"))
+            .expect("bucket stats should carry a per-site summary");
+        assert!(
+            summary.replication_cfg_mismatch,
+            "a peer reporting an offline replication target must not read as in sync"
+        );
+    }
+
+    // A peer that predates the field reports nothing; that is unknown, not a fault, and must
+    // not flip every bucket to out-of-sync during a mixed-version upgrade.
+    #[test]
+    fn test_merge_bucket_status_treats_absent_target_health_as_unknown() {
+        let site_info = |peer: &str| {
+            let xml = String::from_utf8(serialize(&site_repl_config(peer)).unwrap()).unwrap();
+            let mut info = SRInfo::default();
+            info.buckets.insert(
+                "photos".to_string(),
+                SRBucketInfo {
+                    bucket: "photos".to_string(),
+                    replication_config: Some(xml),
+                    replication_targets_online: None,
+                    ..Default::default()
+                },
+            );
+            info
+        };
+
+        let mut status = SRStatusInfo::default();
+        let site_infos = BTreeMap::from([
+            ("site-a".to_string(), site_info("site-b")),
+            ("site-b".to_string(), site_info("site-a")),
+        ]);
+        merge_bucket_status_info(&mut status, &site_infos, &SRStatusOptions::default());
+
+        let summary = status
+            .bucket_stats
+            .get("photos")
+            .and_then(|per_site| per_site.get("site-a"))
+            .expect("bucket stats should carry a per-site summary");
+        assert!(
+            !summary.replication_cfg_mismatch,
+            "peers that do not report target health must not be treated as broken"
+        );
+    }
+
+    // The one-directional regression: a `replication-config` broadcast overwrote the receiver's
+    // rules with the sender's, leaving both sites holding byte-identical XML whose destination
+    // ARN names the receiver. Only one site could push, yet the status check counted rules and
+    // reported "in sync" — the operator's single health signal agreed with the broken state.
+    #[test]
+    fn test_site_replication_config_mismatch_rejects_rule_pointing_at_owning_site() {
+        let shared_config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![build_site_replication_rule(
+                "arn:rustfs:replication::site-b:test-replication",
+                1,
+                "site-repl-site-b",
+            )],
+        };
+        let shared_xml = String::from_utf8(serialize(&shared_config).expect("site replication XML should serialize"))
+            .expect("site replication XML should be UTF-8");
+
+        assert!(
+            !site_replication_rule_complete(&shared_config.rules[0], "site-b"),
+            "a rule whose destination ARN names its own site can never replicate"
+        );
+        assert_eq!(
+            site_replication_config_mismatch(vec![("site-a", Some(&shared_xml)), ("site-b", Some(&shared_xml))].into_iter(), 2),
+            (2, true),
+            "identical configs mean site-b points at itself and cannot push"
+        );
+    }
+
     #[test]
     fn test_status_policy_compare_ignores_string_array_order() {
         let site_a_policy = serde_json::json!({
@@ -12330,6 +12996,219 @@ mod tests {
         assert_eq!(deployment_id.as_deref(), Some("remote-dep"));
     }
 
+    fn site_repl_config(peer: &str) -> ReplicationConfiguration {
+        ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![build_site_replication_rule(
+                &format!("arn:rustfs:replication::{peer}:photos"),
+                1,
+                &format!("site-repl-{peer}"),
+            )],
+        }
+    }
+
+    fn replication_target(deployment_id: &str, endpoint: &str, secret: &str) -> BucketTarget {
+        BucketTarget {
+            source_bucket: "photos".to_string(),
+            target_bucket: "photos".to_string(),
+            endpoint: endpoint.to_string(),
+            deployment_id: deployment_id.to_string(),
+            arn: format!("arn:rustfs:replication::{deployment_id}:photos"),
+            target_type: BucketTargetType::ReplicationService,
+            credentials: Some(crate::admin::storage_api::bucket::target::Credentials {
+                access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                secret_key: secret.to_string(),
+                session_token: None,
+                expiration: None,
+            }),
+            ..Default::default()
+        }
+    }
+
+    fn state_with_peer(deployment_id: &str, endpoint: &str) -> SiteReplicationState {
+        let mut state = SiteReplicationState::default();
+        state.peers.insert(
+            deployment_id.to_string(),
+            PeerInfo {
+                deployment_id: deployment_id.to_string(),
+                ..peer(deployment_id, endpoint)
+            },
+        );
+        state
+    }
+
+    // Bucket targets are writable by anyone holding `admin:SetBucketTarget`. Recovering a
+    // secret from a target that merely carries the site-replicator access key would let such
+    // a principal choose the secret for the broadly privileged replication account.
+    #[test]
+    fn test_secret_recovery_rejects_target_outside_the_peer_topology() {
+        let state = state_with_peer("remote", "http://remote.example.com:9000");
+
+        assert!(
+            bucket_target_matches_configured_peer(
+                &replication_target("remote", "remote.example.com:9000", "shared-secret"),
+                &state
+            ),
+            "a target naming a configured peer at its recorded endpoint is ours"
+        );
+        assert!(
+            !bucket_target_matches_configured_peer(
+                &replication_target("attacker", "attacker.example.com:9000", "planted-secret"),
+                &state
+            ),
+            "a target naming an unknown deployment must never seed the replication account"
+        );
+        assert!(
+            !bucket_target_matches_configured_peer(
+                &replication_target("remote", "attacker.example.com:9000", "planted-secret"),
+                &state
+            ),
+            "a target reusing a peer id but pointing elsewhere must not seed the account"
+        );
+    }
+
+    // A transient store failure must not be read as "the account is gone" and trigger a
+    // reseed that overwrites a live account.
+    #[test]
+    fn test_only_missing_account_errors_allow_reseeding() {
+        use rustfs_iam::error::Error as IamError;
+
+        assert!(is_missing_service_account_error(&IamError::NoSuchAccount("x".into())));
+        assert!(is_missing_service_account_error(&IamError::NoSuchServiceAccount("x".into())));
+        assert!(is_missing_service_account_error(&IamError::ConfigNotFound));
+        assert!(
+            !is_missing_service_account_error(&IamError::IAMActionNotAllowed),
+            "a permission failure is not evidence that the account is absent"
+        );
+    }
+
+    fn operator_rule(id: &str) -> ReplicationRule {
+        ReplicationRule {
+            id: Some(id.to_string()),
+            ..build_site_replication_rule("arn:aws:s3:::backup", 1, id)
+        }
+    }
+
+    // The one-directional bug: the joined site applied the initiator's replication config
+    // verbatim, so its own `site-repl-<initiator>` rule was replaced by a rule pointing at
+    // itself. No bucket target backs that ARN, so every object was dropped without a log.
+    #[test]
+    fn test_merge_incoming_replication_config_keeps_local_reverse_rule() {
+        let merged = merge_incoming_replication_config(Some(site_repl_config("home")), Some(site_repl_config("office")))
+            .expect("merge should keep the local rule");
+
+        assert_eq!(merged.rules.len(), 1);
+        assert_eq!(merged.rules[0].id.as_deref(), Some("site-repl-office"));
+        assert_eq!(merged.rules[0].destination.bucket, "arn:rustfs:replication::office:photos");
+    }
+
+    // A peer deleting its replication config must not delete the receiver's reverse rule
+    // either — the delete travels as `replication-config` with no payload.
+    #[test]
+    fn test_merge_incoming_replication_config_survives_peer_delete() {
+        let merged = merge_incoming_replication_config(None, Some(site_repl_config("office")))
+            .expect("local site rules must survive a peer delete");
+
+        assert_eq!(merged.rules.len(), 1);
+        assert_eq!(merged.rules[0].id.as_deref(), Some("site-repl-office"));
+    }
+
+    #[test]
+    fn test_merge_incoming_replication_config_replicates_operator_rules() {
+        let mut incoming = site_repl_config("home");
+        incoming.rules.push(operator_rule("nightly-backup"));
+        incoming.role = "arn:rustfs:replication::home:photos".to_string();
+
+        let merged = merge_incoming_replication_config(Some(incoming), Some(site_repl_config("office")))
+            .expect("merge should produce rules");
+
+        let ids: Vec<_> = merged.rules.iter().filter_map(|rule| rule.id.as_deref()).collect();
+        assert_eq!(ids, vec!["nightly-backup", "site-repl-office"]);
+        assert_eq!(merged.rules[0].priority, Some(1));
+        assert_eq!(merged.rules[1].priority, Some(2));
+        assert!(
+            merged.role.is_empty(),
+            "a site-replication ARN in `role` belongs to the sender and must not be adopted"
+        );
+    }
+
+    #[test]
+    fn test_merge_incoming_replication_config_returns_none_when_nothing_remains() {
+        assert!(merge_incoming_replication_config(Some(site_repl_config("home")), None).is_none());
+    }
+
+    // `role` is part of the bucket's S3-visible configuration. Repairing a reverse rule must
+    // drop only a sender-owned site-replication ARN, never an operator's own role — the same
+    // rule the merge path applies, so both paths agree on what is ours to rewrite.
+    #[test]
+    fn test_replication_role_is_only_cleared_when_it_is_a_site_replication_arn() {
+        let operator_role = "arn:aws:iam::123456789012:role/replication";
+        assert!(
+            replication_target_arn_deployment_id(operator_role).is_none(),
+            "an operator IAM role is not a site-replication ARN and must be preserved"
+        );
+        assert_eq!(
+            replication_target_arn_deployment_id("arn:rustfs:replication::home:photos").as_deref(),
+            Some("home"),
+            "a site-replication ARN is sender-owned and gets cleared"
+        );
+
+        let mut incoming = site_repl_config("home");
+        incoming.role = operator_role.to_string();
+        let merged = merge_incoming_replication_config(Some(incoming), Some(site_repl_config("office")))
+            .expect("merge should produce rules");
+        assert_eq!(merged.role, operator_role, "operator role must survive the merge");
+    }
+
+    // Rules and targets are keyed off the same ARN. Minting a fresh one while
+    // `reconcile_site_replication_bucket_targets` preserves a MinIO-era `arn:minio:...`
+    // target would leave the rule pointing at an ARN no target satisfies.
+    #[test]
+    fn test_build_site_replication_config_reuses_configured_arn() {
+        let mut state = SiteReplicationState {
+            service_account_access_key: "site-replicator-0".to_string(),
+            ..Default::default()
+        };
+        state.peers.insert(
+            "local".to_string(),
+            PeerInfo {
+                deployment_id: "local".to_string(),
+                ..peer("local", "https://local.example.com")
+            },
+        );
+        state.peers.insert(
+            "remote".to_string(),
+            PeerInfo {
+                deployment_id: "remote".to_string(),
+                ..peer("remote", "http://remote.example.com:9000")
+            },
+        );
+        let existing = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![build_site_replication_rule(
+                "arn:minio:replication::remote:photos",
+                1,
+                "site-repl-remote",
+            )],
+        };
+
+        let config = build_site_replication_config(
+            "photos",
+            &state,
+            &PeerInfo {
+                deployment_id: "local".to_string(),
+                ..peer("local", "https://local.example.com")
+            },
+            "runtime-iam-secret",
+            Some(&existing),
+        )
+        .expect("build site replication config")
+        .expect("a remote peer yields one rule");
+
+        assert_eq!(config.rules.len(), 1);
+        assert_eq!(config.rules[0].destination.bucket, "arn:minio:replication::remote:photos");
+    }
+
     #[test]
     fn test_reconcile_site_replication_bucket_targets_upserts_remote_peer_targets() {
         let mut state = SiteReplicationState {
@@ -13303,31 +14182,32 @@ mod tests {
     fn test_derive_sync_state_from_replication_completeness() {
         // A peer that is reachable and has complete replication rules for all other peers
         // should be Enable; one that is reachable but has an incomplete config should be Disable.
-        let repl_complete_xml = {
+        let site_config_xml = |peer: &str| {
             let config = ReplicationConfiguration {
                 role: String::new(),
                 rules: vec![build_site_replication_rule(
-                    "arn:rustfs:replication::dep-b:bucket",
+                    &format!("arn:rustfs:replication::{peer}:bucket"),
                     1,
-                    "site-repl-dep-b",
+                    &format!("site-repl-{peer}"),
                 )],
             };
             String::from_utf8(serialize(&config).unwrap()).unwrap()
         };
+        let dep_a_xml = site_config_xml("dep-b");
+        let dep_b_xml = site_config_xml("dep-a");
 
         // Peer that has complete config for 2-site setup
-        assert!(site_replication_rule_complete(&build_site_replication_rule(
-            "arn:rustfs:replication::dep-b:bucket",
-            1,
-            "site-repl-dep-b"
-        )));
+        assert!(site_replication_rule_complete(
+            &build_site_replication_rule("arn:rustfs:replication::dep-b:bucket", 1, "site-repl-dep-b"),
+            "dep-a"
+        ));
         assert_eq!(
-            site_replication_config_mismatch(vec![Some(&repl_complete_xml), Some(&repl_complete_xml)].into_iter(), 2),
+            site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_xml)), ("dep-b", Some(&dep_b_xml))].into_iter(), 2),
             (2, false),
             "complete rules on both sites → no mismatch"
         );
         assert_eq!(
-            site_replication_config_mismatch(vec![Some(&repl_complete_xml)].into_iter(), 2),
+            site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_xml)), ("dep-b", None)].into_iter(), 2),
             (1, true),
             "config only on one of two sites → mismatch"
         );
@@ -13339,34 +14219,37 @@ mod tests {
     // bucket as out-of-sync ("0/N Buckets in sync"). This test feeds the real base64 wire form.
     #[test]
     fn test_site_replication_config_mismatch_accepts_base64_wire_form() {
-        let xml = {
+        let site_config_xml = |peer: &str| {
             let config = ReplicationConfiguration {
                 role: String::new(),
                 rules: vec![build_site_replication_rule(
-                    "arn:rustfs:replication::dep-b:bucket",
+                    &format!("arn:rustfs:replication::{peer}:bucket"),
                     1,
-                    "site-repl-dep-b",
+                    &format!("site-repl-{peer}"),
                 )],
             };
             String::from_utf8(serialize(&config).unwrap()).unwrap()
         };
-        let b64 = BASE64_STANDARD.encode(xml.as_bytes());
+        let dep_a_xml = site_config_xml("dep-b");
+        let dep_b_xml = site_config_xml("dep-a");
+        let dep_a_b64 = BASE64_STANDARD.encode(dep_a_xml.as_bytes());
+        let dep_b_b64 = BASE64_STANDARD.encode(dep_b_xml.as_bytes());
 
         // Both sites present the complete config in base64 wire form → NOT a mismatch.
         assert_eq!(
-            site_replication_config_mismatch(vec![Some(&b64), Some(&b64)].into_iter(), 2),
+            site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_b64)), ("dep-b", Some(&dep_b_b64))].into_iter(), 2),
             (2, false),
             "base64-encoded complete configs on both sites must not be reported as a mismatch"
         );
         // The tolerant decode keeps plain-XML callers working too.
         assert_eq!(
-            site_replication_config_mismatch(vec![Some(&xml), Some(&xml)].into_iter(), 2),
+            site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_xml)), ("dep-b", Some(&dep_b_xml))].into_iter(), 2),
             (2, false),
             "raw-XML wire form still parses via the base64 fallback"
         );
         // A base64 config present on only one of two sites is still a mismatch.
         assert_eq!(
-            site_replication_config_mismatch(vec![Some(&b64)].into_iter(), 2),
+            site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_b64)), ("dep-b", None)].into_iter(), 2),
             (1, true),
             "config present on only one site is a mismatch regardless of encoding"
         );
diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs
index 803e9e915..463cd02b8 100644
--- a/rustfs/src/admin/handlers/user.rs
+++ b/rustfs/src/admin/handlers/user.rs
@@ -1198,6 +1198,7 @@ impl Operation for ImportIam {
                                         description: None,
                                         expiration: None,
                                         status: Some(status),
+                                        parent_user: None,
                                         allow_site_replicator_account: false,
                                     },
                                 )
diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs
index 59ebed379..3a9034113 100644
--- a/rustfs/src/lib.rs
+++ b/rustfs/src/lib.rs
@@ -90,6 +90,7 @@ pub mod protocols;
 pub mod runtime_capabilities;
 pub(crate) mod runtime_sources;
 pub mod server;
+pub(crate) mod site_replication_reconcile;
 pub(crate) mod startup_audit;
 pub(crate) mod startup_auth;
 pub(crate) mod startup_background;
diff --git a/rustfs/src/site_replication_reconcile.rs b/rustfs/src/site_replication_reconcile.rs
new file mode 100644
index 000000000..e327e3669
--- /dev/null
+++ b/rustfs/src/site_replication_reconcile.rs
@@ -0,0 +1,81 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Startup and periodic reconciliation of site-replication wiring.
+//!
+//! The reconcilers themselves live in `admin::handlers::site_replication`, next to the
+//! site-replication state they repair. Startup runs in this (infra) layer and must not
+//! reach up into the admin (interface) layer, so the direction is inverted here: this
+//! module owns the schedule and the contract, and the admin layer registers its
+//! implementations into it from `register_site_replication_route`. That registration
+//! happens while the admin router is built, which `init_startup_http_servers` does before
+//! `init_startup_runtime_services` calls in — so the hooks are always installed by the
+//! time startup reconciles.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::OnceLock;
+use std::time::Duration;
+
+use tokio_util::sync::CancellationToken;
+use tracing::warn;
+
+const RECONCILE_INTERVAL: Duration = Duration::from_secs(600);
+
+/// A reconciler reports its own failures; the outcome carries no value because neither
+/// caller can act on one — a site that cannot repair its replication wiring still serves S3.
+type ReconcileHook = fn() -> Pin<Box<dyn Future<Output = ()> + Send>>;
+
+static RECONCILER: OnceLock<ReconcileHook> = OnceLock::new();
+
+/// Install the admin layer's reconciler. Idempotent: a second call is ignored, which keeps
+/// repeated router construction (tests, the embedded server) from panicking.
+pub(crate) fn register_site_replication_reconciler(reconcile: ReconcileHook) {
+    let _ = RECONCILER.set(reconcile);
+}
+
+/// Repair drifted site-replication wiring, immediately and then on a timer.
+///
+/// The first pass runs inside the spawned task rather than on the caller's path: it walks
+/// every bucket, so awaiting it would delay the rest of startup in proportion to bucket
+/// count. It is also spawned unconditionally, including when IAM bootstrap was deferred —
+/// a node that recovers IAM in the background would otherwise keep self-pointing rules
+/// until someone restarted it. The reconciler itself returns early while IAM or the object
+/// store are still unavailable, so an early tick is silent rather than noisy.
+///
+/// Later passes are no-ops once the wiring matches — the bucket pass compares the serialized
+/// targets and rule set before writing — so a healthy deployment reads and writes nothing.
+pub(crate) fn spawn_site_replication_reconcile_task(ctx: CancellationToken) {
+    if RECONCILER.get().is_none() {
+        warn!("site replication reconciler is not registered; periodic reconcile disabled");
+        return;
+    }
+
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+        loop {
+            tokio::select! {
+                _ = ctx.cancelled() => break,
+                // The first tick fires immediately, which is the startup repair pass.
+                _ = ticker.tick() => {
+                    if let Some(reconcile) = RECONCILER.get() {
+                        reconcile().await;
+                    }
+                }
+            }
+        }
+    });
+}
diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs
index 73988839b..7a587ddec 100644
--- a/rustfs/src/startup_services.rs
+++ b/rustfs/src/startup_services.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::site_replication_reconcile::spawn_site_replication_reconcile_task;
 use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot};
 use crate::{
     config::Config,
@@ -79,6 +80,11 @@ pub(crate) async fn init_startup_runtime_services(
 
     let buckets = init_bucket_metadata_runtime(store.clone(), ctx.clone()).await?;
     let iam_bootstrap = init_iam_runtime(store.clone(), ctx.clone(), readiness, state_manager, server_ctx).await?;
+    // Unconditional: deferred IAM recovers in the background and has no callback into this
+    // scheduler, so gating on the inline disposition would leave a recovered node with
+    // self-pointing replication rules until the next restart. The task waits for IAM and
+    // bucket metadata itself, and its first pass runs off this path.
+    spawn_site_replication_reconcile_task(ctx.clone());
     init_auth_integrations().await?;
     init_notification_runtime(endpoint_pools, buckets).await?;
     let enable_scanner = init_background_service_runtime(store.clone()).await?;