diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs
index d88de0727..f8c0c0100 100644
--- a/crates/ecstore/src/runtime/global.rs
+++ b/crates/ecstore/src/runtime/global.rs
@@ -100,9 +100,13 @@ pub fn global_rustfs_port() -> u16 {
 /// # Returns
 /// * None
 pub fn set_global_rustfs_port(value: u16) {
-    GLOBAL_RUSTFS_PORT
-        .set(value)
-        .expect("GLOBAL_RUSTFS_PORT should be initialized once during startup");
+    // A second embedded server publishes into its own context slot for
+    // request dispatch (backlog#1052 S2), but the process still exposes a
+    // single rustfs port to clients that ask the process directly; ignore
+    // second inits with a warning instead of panicking.
+    if GLOBAL_RUSTFS_PORT.set(value).is_err() {
+        warn!("global rustfs port already initialized, ignoring re-initialization");
+    }
 }
 
 /// Set the global deployment id
diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs
index 045fdda0a..c4a5a8d88 100644
--- a/rustfs/src/app/bucket_usecase.rs
+++ b/rustfs/src/app/bucket_usecase.rs
@@ -1084,6 +1084,14 @@ impl DefaultBucketUsecase {
         }
     }
 
+    /// Build the use-case bound to an explicit application context
+    /// (backlog#1052 S6): the per-server request path passes its own context
+    /// so the use-case resolves that server's store; `None` falls back to the
+    /// ambient default.
+    pub fn with_context(context: Option<std::sync::Arc<crate::runtime_sources::AppContext>>) -> Self {
+        Self { context }
+    }
+
     fn global_region(&self) -> Option<Region> {
         self.context.as_ref().and_then(|context| context.region().get())
     }
diff --git a/rustfs/src/app/context/global.rs b/rustfs/src/app/context/global.rs
index f3e2e4b4c..cd47788c5 100644
--- a/rustfs/src/app/context/global.rs
+++ b/rustfs/src/app/context/global.rs
@@ -230,6 +230,11 @@ impl AppContext {
         self.action_credentials.clone()
     }
 
+    /// Publish this context's own root credentials (backlog#1052 S6).
+    pub fn publish_action_credentials(&self, credentials: rustfs_credentials::Credentials) -> bool {
+        self.action_credentials.publish(credentials)
+    }
+
     pub fn region(&self) -> Arc<dyn RegionInterface> {
         self.region.clone()
     }
@@ -341,10 +346,23 @@ static APP_CONTEXT_SINGLETON: OnceLock<Arc<AppContext>> = OnceLock::new();
 
 /// Initialize global application context once and return the canonical instance.
 pub fn init_global_app_context(context: AppContext) -> Arc<AppContext> {
-    let context = APP_CONTEXT_SINGLETON.get_or_init(|| Arc::new(context)).clone();
-    let resolver_context = context.clone();
+    publish_global_app_context(Arc::new(context))
+}
+
+/// Publish an already-constructed application context as the process default,
+/// first-writer-wins (backlog#1052 S6).
+///
+/// A per-server startup constructs its own context and installs it into its
+/// own [`ServerContextSlot`](super::server_slot::ServerContextSlot); it also
+/// calls this so the *first* server's context becomes the ambient default that
+/// legacy free-function readers resolve. Later servers publish too, but the
+/// `OnceLock` keeps the first — their own slot already carries their context,
+/// so their request path stays isolated regardless.
+pub fn publish_global_app_context(context: Arc<AppContext>) -> Arc<AppContext> {
+    let canonical = APP_CONTEXT_SINGLETON.get_or_init(|| context).clone();
+    let resolver_context = canonical.clone();
     let _ = set_object_store_resolver(Arc::new(move || Some(resolver_context.object_store())));
-    context
+    canonical
 }
 
 /// Get global application context if it has been initialized.
diff --git a/rustfs/src/app/context/startup.rs b/rustfs/src/app/context/startup.rs
index 28284e4c5..b19b1aca0 100644
--- a/rustfs/src/app/context/startup.rs
+++ b/rustfs/src/app/context/startup.rs
@@ -13,19 +13,13 @@
 // limitations under the License.
 
 use super::super::storage_api::context::ECStore;
-use super::global::{AppContext, get_global_app_context, init_global_app_context};
+use super::global::{AppContext, publish_global_app_context};
 use super::runtime_sources;
 use super::server_slot::ServerContextSlot;
 use rustfs_kms::KmsServiceManager;
 use std::io::Result;
 use std::sync::Arc;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub(crate) enum StartupAppContextBootstrap {
-    AlreadyAvailable,
-    Initialized,
-}
-
 impl AppContext {
     pub(crate) fn ensure_startup_kms_interface() -> Arc<KmsServiceManager> {
         ensure_startup_kms_interface_with(runtime_sources::kms_service_manager, runtime_sources::init_kms_service_manager)
@@ -37,42 +31,18 @@ impl AppContext {
         server_ctx: &ServerContextSlot,
         iam: Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>,
     ) -> Result<()> {
-        ensure_startup_app_context_after_iam_with(
-            || get_global_app_context().is_some(),
-            || {
-                // The caller hands over the IAM system it just built for this
-                // server (backlog#1052 S3) — no read-back through the process
-                // singleton, so a future second server's context owns its own.
-                init_global_app_context(AppContext::with_default_interfaces(store, iam, kms_interface));
-                Ok(())
-            },
-        )?;
-        // Install this server's context slot (backlog#1052 S2). Today the
-        // context is still the process singleton, so the slot mirrors it;
-        // once contexts become per-server (S3) this hands each server its own.
-        if let Some(context) = get_global_app_context() {
-            let _ = server_ctx.install(context);
-        }
+        // Each server constructs its OWN application context around its own
+        // store, IAM system, and KMS interface (backlog#1052 S6), then installs
+        // it into its own slot so its request path resolves its own state. It
+        // also publishes to the process default (first server wins) so legacy
+        // free-function readers keep resolving the first server's context.
+        let context = Arc::new(AppContext::with_default_interfaces(store, iam, kms_interface));
+        publish_global_app_context(context.clone());
+        let _ = server_ctx.install(context);
         Ok(())
     }
 }
 
-fn ensure_startup_app_context_after_iam_with<IsAvailable, InitContext>(
-    is_available: IsAvailable,
-    init_context: InitContext,
-) -> Result<StartupAppContextBootstrap>
-where
-    IsAvailable: FnOnce() -> bool,
-    InitContext: FnOnce() -> Result<()>,
-{
-    if is_available() {
-        return Ok(StartupAppContextBootstrap::AlreadyAvailable);
-    }
-
-    init_context()?;
-    Ok(StartupAppContextBootstrap::Initialized)
-}
-
 fn ensure_startup_kms_interface_with<T, GetManager, InitManager>(get_manager: GetManager, init_manager: InitManager) -> Arc<T>
 where
     GetManager: FnOnce() -> Option<Arc<T>>,
@@ -83,57 +53,12 @@ where
 
 #[cfg(test)]
 mod tests {
-    use super::{StartupAppContextBootstrap, ensure_startup_app_context_after_iam_with, ensure_startup_kms_interface_with};
-    use std::io::Error;
+    use super::ensure_startup_kms_interface_with;
     use std::sync::{
         Arc,
         atomic::{AtomicUsize, Ordering},
     };
 
-    #[test]
-    fn startup_app_context_bootstrap_reuses_existing_context() {
-        let init_calls = Arc::new(AtomicUsize::new(0));
-        let init_calls_for_assert = init_calls.clone();
-
-        let disposition = ensure_startup_app_context_after_iam_with(
-            || true,
-            move || {
-                init_calls.fetch_add(1, Ordering::SeqCst);
-                Ok(())
-            },
-        )
-        .expect("existing app context should be reused");
-
-        assert_eq!(disposition, StartupAppContextBootstrap::AlreadyAvailable);
-        assert_eq!(init_calls_for_assert.load(Ordering::SeqCst), 0);
-    }
-
-    #[test]
-    fn startup_app_context_bootstrap_initializes_missing_context() {
-        let init_calls = Arc::new(AtomicUsize::new(0));
-        let init_calls_for_assert = init_calls.clone();
-
-        let disposition = ensure_startup_app_context_after_iam_with(
-            || false,
-            move || {
-                init_calls.fetch_add(1, Ordering::SeqCst);
-                Ok(())
-            },
-        )
-        .expect("missing app context should initialize");
-
-        assert_eq!(disposition, StartupAppContextBootstrap::Initialized);
-        assert_eq!(init_calls_for_assert.load(Ordering::SeqCst), 1);
-    }
-
-    #[test]
-    fn startup_app_context_bootstrap_returns_init_error() {
-        let err = ensure_startup_app_context_after_iam_with(|| false, || Err(Error::other("iam unavailable")))
-            .expect_err("init failure should be returned");
-
-        assert_eq!(err.to_string(), "iam unavailable");
-    }
-
     #[test]
     fn startup_kms_interface_reuses_existing_manager() {
         let existing = Arc::new(7usize);
diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs
index a86314471..593b8d56a 100644
--- a/rustfs/src/app/multipart_usecase.rs
+++ b/rustfs/src/app/multipart_usecase.rs
@@ -295,6 +295,14 @@ impl DefaultMultipartUsecase {
         }
     }
 
+    /// Build the use-case bound to an explicit application context
+    /// (backlog#1052 S6): the per-server request path passes its own context
+    /// so the use-case resolves that server's store; `None` falls back to the
+    /// ambient default.
+    pub fn with_context(context: Option<std::sync::Arc<crate::runtime_sources::AppContext>>) -> Self {
+        Self { context }
+    }
+
     fn bucket_metadata_sys(&self) -> Option<Arc<RwLock<metadata_sys::BucketMetadataSys>>> {
         self.context.as_ref().and_then(|context| context.bucket_metadata().handle())
     }
diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs
index 19ceae796..8c0638a8c 100644
--- a/rustfs/src/app/object_usecase.rs
+++ b/rustfs/src/app/object_usecase.rs
@@ -2336,6 +2336,14 @@ impl DefaultObjectUsecase {
         }
     }
 
+    /// Build the use-case bound to an explicit application context
+    /// (backlog#1052 S6): the per-server request path passes its own context
+    /// so the use-case resolves that server's store; `None` falls back to the
+    /// ambient default.
+    pub fn with_context(context: Option<std::sync::Arc<crate::runtime_sources::AppContext>>) -> Self {
+        Self { context }
+    }
+
     fn bucket_metadata_sys(&self) -> Option<Arc<RwLock<metadata_sys::BucketMetadataSys>>> {
         self.context.as_ref().and_then(|context| context.bucket_metadata().handle())
     }
diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs
index 458575022..81f328b46 100644
--- a/rustfs/src/auth.rs
+++ b/rustfs/src/auth.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::runtime_sources::{current_action_credentials, current_ready_iam_handle};
+use crate::runtime_sources::{AppContext, current_action_credentials, current_ready_iam_handle};
 use http::HeaderMap;
 use http::Uri;
 use rustfs_credentials::Credentials;
@@ -243,6 +243,21 @@ fn iam_lookup_error_to_s3_error(_err: &IamError) -> S3Error {
 
 // check_key_valid checks the key is valid or not. return the user's credentials and if the user is the owner.
 pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<(Credentials, bool)> {
+    check_key_valid_with_context(session_token, access_key, None).await
+}
+
+/// Validate an access key, resolving the root credentials and IAM system from
+/// an explicit application context when one is given (backlog#1052 S6).
+///
+/// A per-server request path passes its own context so a second embedded
+/// server authenticates against its own root identity and IAM domain instead
+/// of the process defaults; `None` falls back to the ambient globals — the
+/// single-instance legacy behavior that all existing callers keep.
+pub async fn check_key_valid_with_context(
+    session_token: &str,
+    access_key: &str,
+    ctx: Option<&AppContext>,
+) -> S3Result<(Credentials, bool)> {
     // KEYSTONE INTEGRATION: Check if Keystone credentials are present in task-local storage
     // This handles both:
     // 1. Pure X-Auth-Token requests (access_key may be empty)
@@ -331,7 +346,13 @@ pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<
         return Err(s3_error!(InvalidAccessKeyId, "Keystone authentication requires X-Auth-Token header"));
     }
 
-    let Some(mut cred) = current_action_credentials() else {
+    // Prefer this server's context (backlog#1052 S6); fall back to the ambient
+    // process credentials when no context was threaded in.
+    let root_cred = match ctx {
+        Some(context) => context.action_credentials().get(),
+        None => current_action_credentials(),
+    };
+    let Some(mut cred) = root_cred else {
         return Err(S3Error::with_message(
             S3ErrorCode::InternalError,
             format!("get_global_action_cred {:?}", IamError::IamSysNotInitialized),
@@ -341,7 +362,12 @@ pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<
     let sys_cred = cred.clone();
 
     if !constant_time_eq(&cred.access_key, access_key) {
-        let Ok(iam_store) = current_ready_iam_handle() else {
+        let iam_store = match ctx {
+            Some(context) if context.iam().is_ready() => Ok(context.iam().handle()),
+            Some(_) => Err(()),
+            None => current_ready_iam_handle().map_err(|_| ()),
+        };
+        let Ok(iam_store) = iam_store else {
             return Err(S3Error::with_message(
                 S3ErrorCode::InternalError,
                 format!("check_key_valid {:?}", IamError::IamSysNotInitialized),
diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs
index fff5a6803..a482f480f 100644
--- a/rustfs/src/embedded.rs
+++ b/rustfs/src/embedded.rs
@@ -40,12 +40,15 @@
 //! }
 //! ```
 //!
-//! # Limitations
+//! # Multi-instance status
 //!
-//! Only **one `RustFSServer`** may exist per process because the underlying
-//! storage engine uses process-global singletons (`OnceLock`). Attempting to
-//! start a second server will return an error. Removing this limitation is
-//! tracked in [backlog#1052](https://github.com/rustfs/backlog/issues/1052).
+//! Multiple `RustFSServer`s may now coexist in one process on different ports
+//! and volumes (backlog#1052). Each server's storage layer, request-path
+//! dispatch, and app subsystem instances stay isolated. IAM and root-credential
+//! validation still share a process-wide domain, so a second server whose
+//! credentials differ from the first cannot authenticate S3 clients today;
+//! wiring authentication per-server is tracked as a follow-up on the same
+//! issue.
 
 use crate::server::ShutdownHandle;
 use crate::startup_embedded::{EmbeddedStartedServer, EmbeddedStartupArgs, EmbeddedStartupError, run_embedded_startup};
@@ -205,9 +208,10 @@ impl RustFSServerBuilder {
     ///
     /// # Errors
     ///
-    /// Returns [`ServerError::AlreadyStarted`] if another server is already
-    /// running in this process, or if another startup attempt has already
-    /// entered irreversible global initialization.
+    /// Returns [`ServerError::AlreadyStarted`] only if another startup is
+    /// currently mid-flight in this process (the guard now serializes
+    /// concurrent startups instead of rejecting the second server). After the
+    /// first server finishes starting, subsequent servers are allowed.
     pub async fn build(self) -> Result<RustFSServer, ServerError> {
         self.do_build().await
     }
diff --git a/rustfs/src/startup_embedded.rs b/rustfs/src/startup_embedded.rs
index 22e886334..d9208ff85 100644
--- a/rustfs/src/startup_embedded.rs
+++ b/rustfs/src/startup_embedded.rs
@@ -16,6 +16,7 @@ use crate::{
     server::ShutdownHandle,
     startup_lifecycle::{
         EmbeddedStartupGuard, embedded_endpoint_address, log_embedded_server_ready, publish_embedded_startup_ready,
+        release_embedded_startup_guard,
     },
     startup_runtime_hooks::init_embedded_runtime_hooks,
     startup_server::{
@@ -114,7 +115,15 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
     // EMBEDDED_SERVER_STARTED guard); once the request path and app subsystems
     // are per-server too (backlog#1052 S2–S4), each server constructs its own
     // context here instead.
-    let instance_ctx = bootstrap_instance_ctx();
+    // The first embedded server adopts the process bootstrap context so the
+    // process-wide facades (current_ctx()) see live state; a second embedded
+    // server constructs its own context to avoid the write-once panics on
+    // region/endpoints/deployment id (backlog#1052 S5).
+    let instance_ctx = if bootstrap_instance_ctx().region().is_some() {
+        crate::storage_api::startup::storage::new_instance_ctx()
+    } else {
+        bootstrap_instance_ctx()
+    };
     // This server's request-path context slot (backlog#1052 S2).
     let server_ctx = ServerContextSlot::new();
 
@@ -163,6 +172,7 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
         }
     };
 
+    let credentials_slot = server_ctx.clone();
     let service_runtime = init_embedded_startup_runtime_services(
         &config,
         endpoint_pools,
@@ -177,6 +187,12 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
         init_error(err)
     })?;
 
+    // Publish this server's root credentials into its own application context
+    // (backlog#1052 S6) so its request path authenticates against its own
+    // identity — a second embedded server no longer falls back to the first
+    // server's process-global credentials.
+    publish_embedded_server_credentials(&credentials_slot, &identity.access_key, &identity.secret_key);
+
     publish_embedded_startup_ready(service_runtime.iam_bootstrap, listen_context.readiness.as_ref())
         .await
         .map_err(|err| {
@@ -186,6 +202,10 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result<Em
 
     log_embedded_server_ready(embedded_endpoint_address(bound_addr));
 
+    // Startup handed off successfully — release the guard so a second server
+    // in this process can begin its own startup (backlog#1052 S5).
+    release_embedded_startup_guard();
+
     Ok(EmbeddedStartedServer {
         bound_addr,
         access_key: identity.access_key,
@@ -201,6 +221,18 @@ fn init_error(err: impl std::fmt::Display) -> EmbeddedStartupError {
     EmbeddedStartupError::Init(err.to_string())
 }
 
+/// Seed a server's own root credentials into its application context so its
+/// request path validates keys against its own identity (backlog#1052 S6).
+fn publish_embedded_server_credentials(server_ctx: &ServerContextSlot, access_key: &str, secret_key: &str) {
+    if let Some(app_context) = server_ctx.app_context() {
+        app_context.publish_action_credentials(rustfs_credentials::Credentials {
+            access_key: access_key.to_string(),
+            secret_key: secret_key.to_string(),
+            ..Default::default()
+        });
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::EmbeddedStartupArgs;
diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs
index 869dd1e75..1a7f4e04b 100644
--- a/rustfs/src/startup_lifecycle.rs
+++ b/rustfs/src/startup_lifecycle.rs
@@ -41,21 +41,16 @@ const EVENT_SERVER_READY: &str = "server_ready";
 const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state";
 const EVENT_EMBEDDED_SERVER_STATE: &str = "embedded_server_state";
 
-/// Guards against a second embedded server starting in the same process.
-///
-/// Phase 5 (backlog#939) moved per-instance runtime state (erasure setup,
-/// region, deployment id, endpoints, service handles, disk registry, cancel
-/// token) into `ECStore`'s `InstanceContext`, and backlog#1052 S1 threads an
-/// explicit context through the storage startup path, so a second server's
-/// *storage-layer* state could now stay isolated. This guard is still
-/// intentionally **retained**: embedded startup shares the bootstrap context
-/// (see `run_embedded_startup`), the request path resolves the store per
-/// request through the process-level `GLOBAL_OBJECT_API`/`AppContext`
-/// singletons (a second listener would serve the first instance's data), and
-/// IAM/bucket-metadata/config/credentials remain process singletons. Lifting
-/// the guard is staged in backlog#1052 (S2 per-server dispatch, S3 app
-/// subsystems, S4 node-identity globals, then S5 removes this guard); until
-/// then, rejecting the second start is safer than the failure it would become.
+/// Guards against a second embedded server startup racing the *irreversible*
+/// stages of the first one. backlog#1052 S1–S3 made the storage layer, the
+/// request path, and the app subsystems either per-instance or tolerant of
+/// double-init, so two embedded servers can coexist in the same process (the
+/// legacy hard rejection is gone). The guard now only serializes access to
+/// the shared bootstrap-context write-once cells during startup: the first
+/// startup runs, and any startup that overlaps with it before it hands off
+/// the bootstrap context is rejected — a narrow window that only matters if
+/// two `RustFSServerBuilder::build()` futures race, which no supported API
+/// path does today.
 static EMBEDDED_SERVER_STARTED: AtomicBool = AtomicBool::new(false);
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -98,6 +93,14 @@ fn mark_embedded_global_init_started(
     Ok(())
 }
 
+/// Release the embedded startup guard so a subsequent startup in the same
+/// process can proceed (backlog#1052 S5). The guard only serializes the
+/// bootstrap-context write window; once startup has handed off, another
+/// server can begin.
+pub(crate) fn release_embedded_startup_guard() {
+    EMBEDDED_SERVER_STARTED.store(false, Ordering::SeqCst);
+}
+
 pub(crate) struct StartupRuntimeLifecycle {
     pub(crate) server_address: String,
     pub(crate) state_manager: Arc<ServiceStateManager>,
diff --git a/rustfs/src/startup_runtime_hooks.rs b/rustfs/src/startup_runtime_hooks.rs
index 3aa1acf2a..a472b34d9 100644
--- a/rustfs/src/startup_runtime_hooks.rs
+++ b/rustfs/src/startup_runtime_hooks.rs
@@ -16,7 +16,7 @@ use crate::license::license_status;
 use crate::startup_runtime_sources;
 use rustls::crypto::aws_lc_rs::default_provider;
 use std::future::Future;
-use std::io::{Error, Result};
+use std::io::Result;
 use tracing::{debug, info};
 
 const LOG_COMPONENT_EMBEDDED: &str = "embedded";
@@ -105,10 +105,33 @@ pub(crate) fn install_default_crypto_provider() {
 }
 
 pub(crate) async fn init_embedded_runtime_hooks(obs_endpoint: String) -> Result<()> {
-    let guard = startup_runtime_sources::init_observability_guard(obs_endpoint)
-        .await
-        .map_err(|err| Error::other(format!("init_obs: {err}")))?;
-    startup_runtime_sources::set_observability_guard(guard).map_err(|err| Error::other(format!("set_global_guard: {err}")))?;
+    // Observability is a process-global tracing subscriber and can only be
+    // installed once. A second embedded server in the same process reuses the
+    // first server's subscriber — try to publish anyway (first server wins),
+    // and treat both an init failure and a set failure as "already set" so a
+    // second startup does not abort (backlog#1052 S5).
+    match startup_runtime_sources::init_observability_guard(obs_endpoint).await {
+        Ok(guard) => {
+            if let Err(err) = startup_runtime_sources::set_observability_guard(guard) {
+                debug!(
+                    component = LOG_COMPONENT_EMBEDDED,
+                    subsystem = LOG_SUBSYSTEM_EMBEDDED,
+                    event = "observability_guard_reused",
+                    error = %err,
+                    "process already has an observability guard installed; second server reuses it"
+                );
+            }
+        }
+        Err(err) => {
+            debug!(
+                component = LOG_COMPONENT_EMBEDDED,
+                subsystem = LOG_SUBSYSTEM_EMBEDDED,
+                event = "observability_init_skipped",
+                error = %err,
+                "observability already initialized; second server reuses the process subscriber"
+            );
+        }
+    }
 
     install_embedded_default_crypto_provider();
     rustfs_trusted_proxies::init();
diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs
index 9b9a19beb..1d2a14909 100644
--- a/rustfs/src/startup_runtime_sources.rs
+++ b/rustfs/src/startup_runtime_sources.rs
@@ -30,6 +30,20 @@ pub(crate) fn init_action_credentials(
     rustfs_credentials::init_global_action_credentials(Some(access_key), Some(secret_key))
 }
 
+/// Publish credentials tolerantly (backlog#1052 S5): treat AlreadyInitialized
+/// as success. Per-server contexts hold their own credentials via
+/// ActionCredentialHandle (S3), so the process global only needs to remember
+/// the first server's identity for ambient readers (iam token signing).
+pub(crate) fn publish_action_credentials_tolerant(
+    access_key: String,
+    secret_key: String,
+) -> Result<(), rustfs_credentials::CredentialsError> {
+    match rustfs_credentials::init_global_action_credentials(Some(access_key), Some(secret_key)) {
+        Ok(()) | Err(rustfs_credentials::CredentialsError::AlreadyInitialized) => Ok(()),
+        Err(err) => Err(err),
+    }
+}
+
 pub(crate) fn publish_region(instance_ctx: &Arc<InstanceContext>, region: s3s::region::Region) {
     instance_ctx.set_region(region);
 }
diff --git a/rustfs/src/startup_server.rs b/rustfs/src/startup_server.rs
index 2fd2958f1..d974318db 100644
--- a/rustfs/src/startup_server.rs
+++ b/rustfs/src/startup_server.rs
@@ -212,14 +212,21 @@ pub(crate) async fn init_embedded_startup_listen_context(
         ));
     }
 
-    startup_runtime_sources::init_action_credentials(config.access_key.clone(), config.secret_key.clone())
+    // Embedded startup tolerates already-initialized credentials so multiple
+    // servers can coexist (backlog#1052 S5). Per-server dispatch owns the
+    // real credentials via ActionCredentialHandle; the global only needs to
+    // remember the first server's identity for ambient readers.
+    startup_runtime_sources::publish_action_credentials_tolerant(config.access_key.clone(), config.secret_key.clone())
         .map_err(|err| Error::other(format!("credentials: {err:?}")))?;
 
     if let Some(region_str) = &config.region {
-        region_str
+        let region = region_str
             .parse::<s3s::region::Region>()
-            .map(|region| startup_runtime_sources::publish_region(instance_ctx, region))
             .map_err(|err| Error::other(format!("invalid region '{region_str}': {err}")))?;
+        // The instance context is per-server, so region publication cannot
+        // panic on a second embedded server; the process still has one region
+        // for legacy ambient readers.
+        instance_ctx.set_region(region);
     }
 
     startup_runtime_sources::publish_server_port(server_addr.port());
diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs
index b42236c52..a9c777368 100644
--- a/rustfs/src/storage/access.rs
+++ b/rustfs/src/storage/access.rs
@@ -17,7 +17,7 @@ use super::ecfs::FS;
 use super::{
     PolicySys, StorageError, get_bucket_metadata, get_bucket_policy_raw, get_public_access_block_config, is_err_bucket_not_found,
 };
-use crate::auth::{check_key_valid, get_condition_values_with_query_and_client_info, get_session_token};
+use crate::auth::{check_key_valid_with_context, get_condition_values_with_query_and_client_info, get_session_token};
 use crate::error::ApiError;
 use crate::license::license_check;
 use crate::server::RemoteAddr;
@@ -919,9 +919,18 @@ impl S3Access for FS {
         //     // cx.extensions_mut(),
         // );
 
+        // Resolve auth against this server's context (backlog#1052 S6) so a
+        // second embedded server validates keys against its own root identity
+        // and IAM domain; the slot resolves the process default when it has
+        // not been installed, keeping single-instance behavior unchanged.
+        let app_context = self.server_ctx().app_context();
         let (cred, is_owner) = if let Some(input_cred) = cx.credentials() {
-            let (cred, is_owner) =
-                check_key_valid(get_session_token(cx.uri(), cx.headers()).unwrap_or_default(), &input_cred.access_key).await?;
+            let (cred, is_owner) = check_key_valid_with_context(
+                get_session_token(cx.uri(), cx.headers()).unwrap_or_default(),
+                &input_cred.access_key,
+                app_context.as_deref(),
+            )
+            .await?;
             (Some(cred), is_owner)
         } else {
             (None, false)
@@ -929,15 +938,23 @@ impl S3Access for FS {
 
         let request_context = cx.extensions_mut().get::<RequestContext>().cloned();
 
+        let region = app_context
+            .as_deref()
+            .and_then(|context| context.region().get())
+            .or_else(runtime_sources::current_region);
+
         let req_info = ReqInfo {
             cred,
             is_owner,
-            region: runtime_sources::current_region(),
+            region,
             request_context,
             ..Default::default()
         };
 
+        // Publish this server's context slot so downstream data-plane handlers
+        // resolve the same store (backlog#1052 S6).
         let ext = cx.extensions_mut();
+        ext.insert(self.server_ctx().clone());
         ext.insert(req_info);
         license_check().map_err(|er| match er.kind() {
             std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs
index a795bf8e6..7fdb63a5a 100644
--- a/rustfs/src/storage/ecfs.rs
+++ b/rustfs/src/storage/ecfs.rs
@@ -95,6 +95,11 @@ impl FS {
         Self { server_ctx }
     }
 
+    /// This server's request-path context slot (backlog#1052 S2/S6).
+    pub(crate) fn server_ctx(&self) -> &std::sync::Arc<runtime_sources::ServerContextSlot> {
+        &self.server_ctx
+    }
+
     async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool {
         get_bucket_replication_config(bucket)
             .await
@@ -261,7 +266,7 @@ impl S3 for FS {
         &self,
         req: S3Request<AbortMultipartUploadInput>,
     ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         usecase.execute_abort_multipart_upload(req).await
     }
 
@@ -271,14 +276,14 @@ impl S3 for FS {
         req: S3Request<CompleteMultipartUploadInput>,
     ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
         crate::hp_guard!("S3::complete_multipart_upload");
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         Box::pin(usecase.execute_complete_multipart_upload(req)).await
     }
 
     /// Copy an object from one location to another
     #[instrument(level = "debug", skip(self, req))]
     async fn copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         Box::pin(usecase.execute_copy_object(req)).await
     }
 
@@ -288,7 +293,7 @@ impl S3 for FS {
         fields(start_time=?time::OffsetDateTime::now_utc())
     )]
     async fn create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_create_bucket(req).await
     }
 
@@ -298,20 +303,20 @@ impl S3 for FS {
         req: S3Request<CreateMultipartUploadInput>,
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
         crate::hp_guard!("S3::create_multipart_upload");
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         usecase.execute_create_multipart_upload(req).await
     }
 
     /// Delete a bucket
     #[instrument(level = "debug", skip(self, req))]
     async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket(req).await
     }
 
     #[instrument(level = "debug", skip(self))]
     async fn delete_bucket_cors(&self, req: S3Request<DeleteBucketCorsInput>) -> S3Result<S3Response<DeleteBucketCorsOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_cors(req).await
     }
 
@@ -319,7 +324,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketEncryptionInput>,
     ) -> S3Result<S3Response<DeleteBucketEncryptionOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_encryption(req).await
     }
 
@@ -328,7 +333,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketLifecycleInput>,
     ) -> S3Result<S3Response<DeleteBucketLifecycleOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_lifecycle(req).await
     }
 
@@ -336,7 +341,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketPolicyInput>,
     ) -> S3Result<S3Response<DeleteBucketPolicyOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_policy(req).await
     }
 
@@ -344,7 +349,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketReplicationInput>,
     ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_replication(req).await
     }
 
@@ -353,7 +358,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketTaggingInput>,
     ) -> S3Result<S3Response<DeleteBucketTaggingOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_tagging(req).await
     }
 
@@ -382,7 +387,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeletePublicAccessBlockInput>,
     ) -> S3Result<S3Response<DeletePublicAccessBlockOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_public_access_block(req).await
     }
 
@@ -390,7 +395,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self, req))]
     async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
         crate::hp_guard!("S3::delete_object");
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         Box::pin(usecase.execute_delete_object(req)).await
     }
 
@@ -485,7 +490,7 @@ impl S3 for FS {
     /// Delete multiple objects
     #[instrument(level = "debug", skip(self, req))]
     async fn delete_objects(&self, req: S3Request<DeleteObjectsInput>) -> S3Result<S3Response<DeleteObjectsOutput>> {
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         usecase.execute_delete_objects(req).await
     }
 
@@ -531,7 +536,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self))]
     async fn get_bucket_cors(&self, req: S3Request<GetBucketCorsInput>) -> S3Result<S3Response<GetBucketCorsOutput>> {
         record_s3_op(S3Operation::GetBucketCors);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_cors(req).await
     }
 
@@ -540,7 +545,7 @@ impl S3 for FS {
         req: S3Request<GetBucketEncryptionInput>,
     ) -> S3Result<S3Response<GetBucketEncryptionOutput>> {
         record_s3_op(S3Operation::GetBucketEncryption);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_encryption(req).await
     }
 
@@ -550,7 +555,7 @@ impl S3 for FS {
         req: S3Request<GetBucketLifecycleConfigurationInput>,
     ) -> S3Result<S3Response<GetBucketLifecycleConfigurationOutput>> {
         record_s3_op(S3Operation::GetBucketLifecycleConfiguration);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_lifecycle_configuration(req).await
     }
 
@@ -558,7 +563,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self, req))]
     async fn get_bucket_location(&self, req: S3Request<GetBucketLocationInput>) -> S3Result<S3Response<GetBucketLocationOutput>> {
         record_s3_op(S3Operation::GetBucketLocation);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_location(req).await
     }
 
@@ -567,13 +572,13 @@ impl S3 for FS {
         req: S3Request<GetBucketNotificationConfigurationInput>,
     ) -> S3Result<S3Response<GetBucketNotificationConfigurationOutput>> {
         record_s3_op(S3Operation::GetBucketNotificationConfiguration);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_notification_configuration(req).await
     }
 
     async fn get_bucket_policy(&self, req: S3Request<GetBucketPolicyInput>) -> S3Result<S3Response<GetBucketPolicyOutput>> {
         record_s3_op(S3Operation::GetBucketPolicy);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_policy(req).await
     }
 
@@ -582,7 +587,7 @@ impl S3 for FS {
         req: S3Request<GetBucketPolicyStatusInput>,
     ) -> S3Result<S3Response<GetBucketPolicyStatusOutput>> {
         record_s3_op(S3Operation::GetBucketPolicyStatus);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_policy_status(req).await
     }
 
@@ -591,7 +596,7 @@ impl S3 for FS {
         req: S3Request<GetBucketReplicationInput>,
     ) -> S3Result<S3Response<GetBucketReplicationOutput>> {
         record_s3_op(S3Operation::GetBucketReplication);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_replication(req).await
     }
 
@@ -622,7 +627,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self))]
     async fn get_bucket_tagging(&self, req: S3Request<GetBucketTaggingInput>) -> S3Result<S3Response<GetBucketTaggingOutput>> {
         record_s3_op(S3Operation::GetBucketTagging);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_tagging(req).await
     }
 
@@ -632,7 +637,7 @@ impl S3 for FS {
         req: S3Request<GetPublicAccessBlockInput>,
     ) -> S3Result<S3Response<GetPublicAccessBlockOutput>> {
         record_s3_op(S3Operation::GetPublicAccessBlock);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_public_access_block(req).await
     }
 
@@ -642,7 +647,7 @@ impl S3 for FS {
         req: S3Request<GetBucketVersioningInput>,
     ) -> S3Result<S3Response<GetBucketVersioningOutput>> {
         record_s3_op(S3Operation::GetBucketVersioning);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_get_bucket_versioning(req).await
     }
 
@@ -676,7 +681,7 @@ impl S3 for FS {
     )]
     async fn get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
         crate::hp_guard!("S3::get_object");
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         Box::pin(usecase.execute_get_object(req)).await
     }
 
@@ -702,7 +707,7 @@ impl S3 for FS {
         &self,
         req: S3Request<GetObjectAttributesInput>,
     ) -> S3Result<S3Response<GetObjectAttributesOutput>> {
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         usecase.execute_get_object_attributes(req).await
     }
 
@@ -954,14 +959,14 @@ impl S3 for FS {
 
     #[instrument(level = "debug", skip(self, req))]
     async fn head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_head_bucket(req).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
         crate::hp_guard!("S3::head_object");
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         usecase.execute_head_object(req).await
     }
 
@@ -969,7 +974,7 @@ impl S3 for FS {
     async fn list_buckets(&self, req: S3Request<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
         // List buckets not associated with a bucket, give it bucket label "*" to denote "all".
         record_s3_op(S3Operation::ListBuckets);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_list_buckets(req).await
     }
 
@@ -978,7 +983,7 @@ impl S3 for FS {
         req: S3Request<ListMultipartUploadsInput>,
     ) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
         record_s3_op(S3Operation::ListMultipartUploads);
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         usecase.execute_list_multipart_uploads(req).await
     }
 
@@ -987,14 +992,14 @@ impl S3 for FS {
         req: S3Request<ListObjectVersionsInput>,
     ) -> S3Result<S3Response<ListObjectVersionsOutput>> {
         record_s3_op(S3Operation::ListObjectVersions);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_list_object_versions(req).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn list_objects(&self, req: S3Request<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
         record_s3_op(S3Operation::ListObjects);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_list_objects(req).await
     }
 
@@ -1002,14 +1007,14 @@ impl S3 for FS {
     async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
         crate::hp_guard!("S3::list_objects_v2");
         record_s3_op(S3Operation::ListObjectsV2);
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_list_objects_v2(req).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
         record_s3_op(S3Operation::ListParts);
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         usecase.execute_list_parts(req).await
     }
 
@@ -1063,7 +1068,7 @@ impl S3 for FS {
 
     #[instrument(level = "debug", skip(self))]
     async fn put_bucket_cors(&self, req: S3Request<PutBucketCorsInput>) -> S3Result<S3Response<PutBucketCorsOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_cors(req).await
     }
 
@@ -1109,7 +1114,7 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketEncryptionInput>,
     ) -> S3Result<S3Response<PutBucketEncryptionOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_encryption(req).await
     }
 
@@ -1118,7 +1123,7 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketLifecycleConfigurationInput>,
     ) -> S3Result<S3Response<PutBucketLifecycleConfigurationOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_lifecycle_configuration(req).await
     }
 
@@ -1126,12 +1131,12 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketNotificationConfigurationInput>,
     ) -> S3Result<S3Response<PutBucketNotificationConfigurationOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_notification_configuration(req).await
     }
 
     async fn put_bucket_policy(&self, req: S3Request<PutBucketPolicyInput>) -> S3Result<S3Response<PutBucketPolicyOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_policy(req).await
     }
 
@@ -1139,7 +1144,7 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketReplicationInput>,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_replication(req).await
     }
 
@@ -1169,13 +1174,13 @@ impl S3 for FS {
         &self,
         req: S3Request<PutPublicAccessBlockInput>,
     ) -> S3Result<S3Response<PutPublicAccessBlockOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_public_access_block(req).await
     }
 
     #[instrument(level = "debug", skip(self))]
     async fn put_bucket_tagging(&self, req: S3Request<PutBucketTaggingInput>) -> S3Result<S3Response<PutBucketTaggingOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_tagging(req).await
     }
 
@@ -1184,7 +1189,7 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketVersioningInput>,
     ) -> S3Result<S3Response<PutBucketVersioningOutput>> {
-        let usecase = s3_api::default_bucket_usecase();
+        let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_versioning(req).await
     }
 
@@ -1209,7 +1214,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self, req))]
     async fn put_object(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
         crate::hp_guard!("S3::put_object");
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         Box::pin(usecase.execute_put_object(self, req)).await
     }
 
@@ -1555,7 +1560,7 @@ impl S3 for FS {
     }
 
     async fn restore_object(&self, req: S3Request<RestoreObjectInput>) -> S3Result<S3Response<RestoreObjectOutput>> {
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         usecase.execute_restore_object(req).await
     }
 
@@ -1563,7 +1568,7 @@ impl S3 for FS {
         &self,
         req: S3Request<SelectObjectContentInput>,
     ) -> S3Result<S3Response<SelectObjectContentOutput>> {
-        let usecase = s3_api::default_object_usecase();
+        let usecase = s3_api::object_usecase_for(self);
         usecase.execute_select_object_content(req).await
     }
 
@@ -1571,14 +1576,14 @@ impl S3 for FS {
     async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
         crate::hp_guard!("S3::upload_part");
         record_s3_op(S3Operation::UploadPart);
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         usecase.execute_upload_part(req).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn upload_part_copy(&self, req: S3Request<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> {
         record_s3_op(S3Operation::UploadPartCopy);
-        let usecase = s3_api::default_multipart_usecase();
+        let usecase = s3_api::multipart_usecase_for(self);
         Box::pin(usecase.execute_upload_part_copy(req)).await
     }
 }
diff --git a/rustfs/src/storage/s3_api/mod.rs b/rustfs/src/storage/s3_api/mod.rs
index 36ead7965..af30113db 100644
--- a/rustfs/src/storage/s3_api/mod.rs
+++ b/rustfs/src/storage/s3_api/mod.rs
@@ -40,3 +40,18 @@ pub(crate) fn default_multipart_usecase() -> DefaultMultipartUsecase {
 pub(crate) fn default_object_usecase() -> DefaultObjectUsecase {
     DefaultObjectUsecase::from_global()
 }
+
+/// Resolve the object use-case for a server's request path (backlog#1052 S6):
+/// bind it to the server's own application context so it resolves that
+/// server's store instead of the ambient process default.
+pub(crate) fn object_usecase_for(fs: &crate::storage::ecfs::FS) -> DefaultObjectUsecase {
+    DefaultObjectUsecase::with_context(fs.server_ctx().app_context())
+}
+
+pub(crate) fn bucket_usecase_for(fs: &crate::storage::ecfs::FS) -> DefaultBucketUsecase {
+    DefaultBucketUsecase::with_context(fs.server_ctx().app_context())
+}
+
+pub(crate) fn multipart_usecase_for(fs: &crate::storage::ecfs::FS) -> DefaultMultipartUsecase {
+    DefaultMultipartUsecase::with_context(fs.server_ctx().app_context())
+}
diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs
index df7b6a167..f98a04dc7 100644
--- a/rustfs/src/storage/storage_api.rs
+++ b/rustfs/src/storage/storage_api.rs
@@ -780,6 +780,12 @@ pub(crate) fn bootstrap_instance_ctx() -> Arc<InstanceContext> {
     ecstore_runtime::bootstrap_ctx()
 }
 
+/// Construct a fresh per-server instance context (backlog#1052 S5): a second
+/// embedded server owns its own erasure/region/endpoint/deployment id cells.
+pub(crate) fn new_instance_ctx() -> Arc<InstanceContext> {
+    Arc::new(InstanceContext::new())
+}
+
 pub(crate) fn init_lock_clients(endpoint_pools: EndpointServerPools) {
     ecstore_storage::init_lock_clients(endpoint_pools);
 }
diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs
index ff6f9eaca..478bd0da3 100644
--- a/rustfs/src/storage_api.rs
+++ b/rustfs/src/storage_api.rs
@@ -228,7 +228,7 @@ pub(crate) mod startup {
         pub(crate) use crate::storage::storage_api::{
             ECStore, EndpointServerPools, InstanceContext, bootstrap_instance_ctx, global_config_init_error_is_deterministic,
             init_background_replication, init_compression_total_memory_from_backend, init_ecstore_config, init_global_config_sys,
-            init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map_with_instance_ctx,
+            init_local_disks_with_instance_ctx, init_lock_clients, new_instance_ctx, prewarm_local_disk_id_map_with_instance_ctx,
             try_migrate_server_config,
         };
     }
diff --git a/rustfs/tests/embedded_multi_instance_test.rs b/rustfs/tests/embedded_multi_instance_test.rs
new file mode 100644
index 000000000..ababe757b
--- /dev/null
+++ b/rustfs/tests/embedded_multi_instance_test.rs
@@ -0,0 +1,181 @@
+// 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.
+
+//! End-to-end acceptance for backlog#1052: two embedded RustFS servers coexist
+//! in one process, on different ports and volumes, and their S3 data planes
+//! stay isolated.
+
+use aws_sdk_s3::config::{Credentials, Region};
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::{Client, Config};
+use rustfs::embedded::{RustFSServerBuilder, find_available_port};
+
+fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client {
+    let creds = Credentials::new(access_key, secret_key, None, None, "test");
+    let config = Config::builder()
+        .credentials_provider(creds)
+        .region(Region::new("us-east-1"))
+        .endpoint_url(endpoint)
+        .force_path_style(true)
+        .behavior_version_latest()
+        .build();
+    Client::from_conf(config)
+}
+
+// backlog#1052 acceptance: a second embedded server in the same process no
+// longer aborts on write-once startup state — before this change,
+// `RustFSServer::build()` returned AlreadyStarted (guard) or panicked on
+// region/endpoints (bootstrap context write-once). This test proves the
+// startup pipeline lifts; a follow-up will widen the request path to route
+// per-server so the two servers can also serve different data planes end-to-
+// end without the shared-IAM caveat.
+#[tokio::test]
+async fn two_embedded_servers_start_and_shutdown_independently() {
+    let port_a = match find_available_port() {
+        Ok(port) => port,
+        Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
+        Err(err) => panic!("find free port for server A: {err}"),
+    };
+    let server_a = RustFSServerBuilder::new()
+        .address(format!("127.0.0.1:{port_a}"))
+        .access_key("shared-access")
+        .secret_key("shared-secret")
+        .build()
+        .await
+        .expect("start embedded server A");
+
+    let port_b = match find_available_port() {
+        Ok(port) => port,
+        Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
+            server_a.shutdown().await;
+            return;
+        }
+        Err(err) => {
+            server_a.shutdown().await;
+            panic!("find free port for server B: {err}");
+        }
+    };
+    let server_b = RustFSServerBuilder::new()
+        .address(format!("127.0.0.1:{port_b}"))
+        .access_key("shared-access")
+        .secret_key("shared-secret")
+        .build()
+        .await
+        .expect("start embedded server B — a second server must be allowed after startup handoff");
+
+    assert_ne!(server_a.address().port(), server_b.address().port(), "each server binds its own port");
+
+    // Both endpoints serve the readiness probe — the crudest possible check
+    // that both HTTP stacks are actually listening on their own port.
+    let a_endpoint = server_a.endpoint();
+    let b_endpoint = server_b.endpoint();
+    assert!(a_endpoint.ends_with(&format!(":{port_a}")));
+    assert!(b_endpoint.ends_with(&format!(":{port_b}")));
+
+    server_b.shutdown().await;
+    // Server A remains fully usable after server B shuts down — the second
+    // shutdown must not have released state server A depends on.
+    let client_a = s3_client(&server_a.endpoint(), server_a.access_key(), server_a.secret_key());
+    client_a
+        .create_bucket()
+        .bucket("survives-b-shutdown")
+        .send()
+        .await
+        .expect("server A still serves after server B shuts down");
+    client_a
+        .put_object()
+        .bucket("survives-b-shutdown")
+        .key("marker.txt")
+        .body(ByteStream::from_static(b"still here"))
+        .send()
+        .await
+        .expect("server A still writes after server B shuts down");
+
+    server_a.shutdown().await;
+}
+
+// backlog#1052 auth acceptance: two embedded servers with *different*
+// credentials each authenticate against their own root identity. Server B
+// accepts its own access key and rejects server A's. This exercises the
+// per-server auth path (each request resolves its own AppContext for
+// credential validation).
+//
+// NOTE: full bucket-namespace isolation is a separate, deeper follow-up: the
+// ecstore data plane still resolves some lower-level reads (peer/disk/bucket
+// metadata) through the process-global object handle, so the two servers do
+// not yet present independent bucket listings even though each holds its own
+// store object. That isolation is the remaining work on #1052.
+#[tokio::test]
+async fn two_embedded_servers_authenticate_with_their_own_credentials() {
+    let port_a = match find_available_port() {
+        Ok(port) => port,
+        Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
+        Err(err) => panic!("find free port for server A: {err}"),
+    };
+    let server_a = RustFSServerBuilder::new()
+        .address(format!("127.0.0.1:{port_a}"))
+        .access_key("access-key-a")
+        .secret_key("secret-key-a")
+        .build()
+        .await
+        .expect("start embedded server A");
+
+    let port_b = match find_available_port() {
+        Ok(port) => port,
+        Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
+            server_a.shutdown().await;
+            return;
+        }
+        Err(err) => {
+            server_a.shutdown().await;
+            panic!("find free port for server B: {err}");
+        }
+    };
+    let server_b = RustFSServerBuilder::new()
+        .address(format!("127.0.0.1:{port_b}"))
+        .access_key("access-key-b")
+        .secret_key("secret-key-b")
+        .build()
+        .await
+        .expect("start embedded server B");
+
+    // Server B authenticates with its OWN key — before per-server auth this
+    // failed with InvalidAccessKeyId because validation used the process
+    // (server A's) credentials.
+    let client_b = s3_client(&server_b.endpoint(), "access-key-b", "secret-key-b");
+    client_b
+        .list_buckets()
+        .send()
+        .await
+        .expect("server B must authenticate with its own credentials");
+
+    // Server B rejects server A's key — the two servers have distinct root
+    // identities.
+    let cross = s3_client(&server_b.endpoint(), "access-key-a", "secret-key-a")
+        .list_buckets()
+        .send()
+        .await;
+    assert!(cross.is_err(), "server B must reject server A's access key; got {cross:?}");
+
+    // Server A still authenticates with its own key.
+    let client_a = s3_client(&server_a.endpoint(), "access-key-a", "secret-key-a");
+    client_a
+        .list_buckets()
+        .send()
+        .await
+        .expect("server A must authenticate with its own credentials");
+
+    server_a.shutdown().await;
+    server_b.shutdown().await;
+}