diff --git a/cli/azd/docs/concurrency-model.md b/cli/azd/docs/concurrency-model.md
index 8be3b8dc942..ae20a060ac1 100644
--- a/cli/azd/docs/concurrency-model.md
+++ b/cli/azd/docs/concurrency-model.md
@@ -128,6 +128,7 @@ endpoint URL.
 | Lock                     | Protects                                          | Acquired by                                                                |
 |--------------------------|---------------------------------------------------|----------------------------------------------------------------------------|
 | `mu sync.RWMutex`        | `dotenv map[string]string`, `deletedKeys`         | `Getenv`, `LookupEnv`, `Dotenv`, `DotenvSet`, `DotenvDelete`, `Reload`, all helpers |
+| `Config` (own `sync.RWMutex`) | the config tree behind `Environment.Config`  | every `config.Config` method, plus `config.Clone`, `config.ApplyDelta`, `config.Replace` |
 
 **Contract**: All readers acquire `mu.RLock()`; all writers acquire `mu.Lock()`.
 Iteration over the underlying map (e.g. snapshotting for a hook) must hold
@@ -139,6 +140,21 @@ steps, parallel service deploy steps, and pre/post-provision/-deploy hooks.
 A second goroutine reading `dotenv` while another writes it is a data race
 and Go's runtime will panic on a concurrent map write.
 
+**`Environment.Config` is also internally synchronized.** Parallel provision
+layers share a single `Environment`, and providers write config directly during
+deployment (Bicep persists prompted parameters under `infra.parameters.*`), so
+the default `config.Config` implementation guards its own map.
+
+Two rules follow:
+
+1. **Never assign the `Config` field on a live `Environment`.** Reassignment
+   races with every concurrent reader of `env.Config` and strands goroutines
+   holding the previous value. Data stores call `env.replaceConfig`, which
+   swaps the contents in place via `config.Replace`.
+2. **`Config.Raw()` returns the live backing map, not a copy.** Mutating it, or
+   reading it while another goroutine may write, bypasses the lock. Inside
+   `pkg/config` use `snapshotRaw` (as `manager.Save` does when marshalling).
+
 ---
 
 ## `pkg/environment.Manager`
diff --git a/cli/azd/internal/cmd/provision_graph.go b/cli/azd/internal/cmd/provision_graph.go
index 71bbd1ebc04..91fe2ba4eb4 100644
--- a/cli/azd/internal/cmd/provision_graph.go
+++ b/cli/azd/internal/cmd/provision_graph.go
@@ -820,10 +820,8 @@ func provisionSingleLayer(
 	return err
 }
 
-// runProvisionSingleLayer provisions a single infrastructure layer. It creates
-// an isolated environment clone so that parallel layers don't interfere with
-// each other's parameter resolution, then merges outputs back into the shared
-// env.
+// runProvisionSingleLayer provisions a single infrastructure layer against the
+// shared environment, then merges its outputs back into that environment.
 //
 // The lifecycle matches the sequential path in [ProvisionAction]:
 //
@@ -837,23 +835,17 @@ func provisionSingleLayer(
 //  8. Final reload of deps.env from disk (capture hook/event subprocess writes)
 //
 // Steps 1-2 and 5-7 are serialized via hookMu to protect non-threadsafe
-// handlers. envMu (separate from hookMu) protects deps.env reads/writes in
-// steps 0, 4, and 8. Hook subprocesses in steps 1, 5, 6, 7 may write to the
-// dotenv file on disk via their own envManager; deps.env in this process
-// is intentionally NOT kept live during that window — step 8's reload is
-// the single point at which we re-converge with disk before returning.
-// Concurrent sibling layers (no dependsOn edge) running their own steps 0/4
-// will therefore not observe this layer's mid-flight hook writes, which is
-// the correct behavior — sibling layers without an explicit dependency on
-// us are by definition not allowed to read our hook-mediated values.
+// handlers. envMu (separate from hookMu) serializes the reload-modify-save
+// cycles against disk in steps 4 and 8; in-memory access to deps.env needs no
+// coordination here because [environment.Environment] and its [config.Config]
+// are internally synchronized.
 //
-// Cross-layer ordering contract: when the dependency graph contains an edge
-// `B → A` (either detected by [bicep.AnalyzeLayerDependencies] or declared
-// via `infra.layers[].dependsOn`), the scheduler treats this entire function
-// invocation as a single graph node. Layer B's node is only scheduled after
-// layer A's node returns, which by construction means after step 8
-// completes. Therefore B's clone of deps.env at the start of its own
-// invocation observes:
+// Parallel layers share deps.env, so a layer may observe a sibling's writes.
+// Cross-layer ordering is a scheduling concern, not an isolation one: when the
+// graph contains an edge `B → A` (detected by [bicep.AnalyzeLayerDependencies]
+// or declared via `infra.layers[].dependsOn`), B's node is only scheduled after
+// A's node returns, which by construction means after step 8 completes.
+// Therefore B observes:
 //
 //   - all of A's deployment outputs (merged in step 4), AND
 //   - any env mutations performed by A's hooks or event handlers via
@@ -862,7 +854,7 @@ func provisionSingleLayer(
 // The latter is the "hook-mediated edge" case the static analyzer is blind
 // to. Authors who need this guarantee must declare the edge explicitly via
 // `infra.layers[].dependsOn`; without an explicit edge, A and B may run in
-// parallel and B's clone may pre-date A's reload.
+// parallel and B may read deps.env before A's reload.
 //
 // Returns the raw [provisioning.DeployResult] so callers can record skip
 // semantics; on [provisioning.ProvisionValidationCanceledSkipped] it returns
@@ -876,23 +868,11 @@ func runProvisionSingleLayer(
 	console input.Console,
 	envMu *sync.Mutex,
 ) (*provisioning.DeployResult, error) {
-	// Snapshot the shared environment so this layer resolves parameters
-	// from current values (including outputs from prior phases).
-	envMu.Lock()
-	layerEnv := environment.NewWithValues(
-		deps.env.Name(), deps.env.Dotenv(),
-	)
-	envMu.Unlock()
-
-	// Use a noop-save env manager for the per-layer manager. Saves happen
-	// against the shared environment after outputs are merged.
-	noopMgr := &noopSaveEnvManager{Manager: deps.envManager}
-
 	mgr := provisioning.NewManager(
 		deps.serviceLocator,
 		deps.defaultProvider,
-		noopMgr,
-		layerEnv,
+		deps.envManager,
+		deps.env,
 		console,
 		deps.alphaFeatureManager,
 		deps.fileShareService,
@@ -915,8 +895,8 @@ func runProvisionSingleLayer(
 			Cwd: layerPath, ProjectDir: deps.projectPath,
 		}, deps.commandRunner)
 		hooksRunner = ext.NewHooksRunner(
-			hooksManager, deps.commandRunner, noopMgr, console,
-			layerPath, layer.Hooks, layerEnv, deps.serviceLocator,
+			hooksManager, deps.commandRunner, deps.envManager, console,
+			layerPath, layer.Hooks, deps.env, deps.serviceLocator,
 		)
 
 		// Validate layer hooks and warn about issues (mirrors sequential path).
@@ -1061,7 +1041,7 @@ func runProvisionSingleLayer(
 		}
 	}
 
-	// ── Step 8: Final reload of shared env ──
+	// ── Step 8: Reconcile layer-local saves and reload shared env ──
 	//
 	// Hooks (steps 1, 7) and event handlers (steps 2, 5, 6) may invoke
 	// `azd env set` in a subprocess to write values that downstream layers
@@ -1069,13 +1049,8 @@ func runProvisionSingleLayer(
 	// the static analyzer cannot infer. Each subprocess writes to disk via
 	// its own envManager but does not touch the parent process's in-memory
 	// deps.env. Without a final reload here, the next layer in topological
-	// order would clone from a stale deps.env and miss those values, making
+	// order would read a stale deps.env and miss those values, making
 	// `dependsOn: ["this-layer"]` declarations silently incomplete.
-	//
-	// Reload is idempotent: if no out-of-band writes happened, deps.env
-	// already matches disk (we Save'd into it in step 4) and Reload is a
-	// no-op. The cost is one stat + one read of the .env file per layer,
-	// which is negligible compared to a provisioning round-trip.
 	if err := reloadSharedEnvLocked(ctx, deps, envMu); err != nil {
 		return deployResult, fmt.Errorf(
 			"reloading shared env after layer %s: %w", stepName, err,
@@ -1122,12 +1097,8 @@ func mergeLayerOutputsLocked(
 	return provisioning.UpdateEnvironment(ctx, outputs, deps.env, deps.envManager)
 }
 
-// reloadSharedEnvLocked acquires envMu and reloads deps.env from disk,
-// capturing any out-of-band writes performed by hook / event subprocesses.
-// Required to make `dependsOn` ordering semantically complete for
-// hook-mediated env values: without it, the in-memory deps.env stays stale
-// and the next layer's clone of deps.env.Dotenv() misses subprocess writes
-// even though disk has them.
+// reloadSharedEnvLocked refreshes deps.env from disk under envMu so writes made
+// by hook and event-handler subprocesses are visible to subsequent layers.
 func reloadSharedEnvLocked(
 	ctx context.Context,
 	deps *provisionLayerDeps,
@@ -1135,7 +1106,11 @@ func reloadSharedEnvLocked(
 ) error {
 	envMu.Lock()
 	defer envMu.Unlock()
-	return deps.envManager.Reload(ctx, deps.env)
+
+	if err := deps.envManager.Reload(ctx, deps.env); err != nil {
+		return fmt.Errorf("reloading shared env: %w", err)
+	}
+	return nil
 }
 
 // resolveOutputString converts a provisioning output parameter to its string
@@ -1220,24 +1195,3 @@ func (c *syncConsole) EnsureBlankLine(ctx context.Context) {
 	defer c.mu.Unlock()
 	c.Console.EnsureBlankLine(ctx)
 }
-
-// noopSaveEnvManager wraps an [environment.Manager], suppressing Save and
-// SaveWithOptions. Per-layer managers use this to avoid partial environment
-// writes; the authoritative save happens through the shared environment.
-type noopSaveEnvManager struct {
-	environment.Manager
-}
-
-func (*noopSaveEnvManager) Save(
-	_ context.Context, _ *environment.Environment,
-) error {
-	return nil
-}
-
-func (*noopSaveEnvManager) SaveWithOptions(
-	_ context.Context,
-	_ *environment.Environment,
-	_ *environment.SaveOptions,
-) error {
-	return nil
-}
diff --git a/cli/azd/internal/cmd/provision_graph_test.go b/cli/azd/internal/cmd/provision_graph_test.go
index 8a52a7d6c6e..651da5afb44 100644
--- a/cli/azd/internal/cmd/provision_graph_test.go
+++ b/cli/azd/internal/cmd/provision_graph_test.go
@@ -19,32 +19,11 @@ import (
 	"github.com/azure/azure-dev/cli/azd/pkg/infra/provisioning"
 	"github.com/azure/azure-dev/cli/azd/pkg/infra/provisioning/bicep"
 	"github.com/azure/azure-dev/cli/azd/test/mocks"
-	"github.com/azure/azure-dev/cli/azd/test/mocks/mockenv"
 	"github.com/azure/azure-dev/cli/azd/test/mocks/mockinput"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
-func TestNoopSaveEnvManager(t *testing.T) {
-	t.Parallel()
-
-	inner := &mockenv.MockEnvManager{}
-	noop := &noopSaveEnvManager{Manager: inner}
-
-	env := environment.NewWithValues("test", nil)
-
-	// Save and SaveWithOptions must be no-ops — the inner mock should
-	// never be called for these methods.
-	require.NoError(t, noop.Save(t.Context(), env))
-	require.NoError(t, noop.SaveWithOptions(t.Context(), env, nil))
-
-	// Non-save methods delegate to the inner manager.
-	inner.On("Reload", mock.Anything, env).Return(nil)
-	require.NoError(t, noop.Reload(t.Context(), env))
-	inner.AssertCalled(t, "Reload", mock.Anything, env)
-}
-
 func TestSyncConsole_SerializesMessages(t *testing.T) {
 	t.Parallel()
 
@@ -350,13 +329,8 @@ func TestMergeLayerOutputsLocked_PreservesSubprocessWrites(t *testing.T) {
 	assert.Equal(t, "deploy-value", dotenv["DEPLOY_KEY"])
 }
 
-// TestReloadSharedEnvLocked_RefreshesDepsEnvFromDisk asserts the
-// behavioral primitive that the hook-mediated propagation contract on
-// [runProvisionSingleLayer] step 8 stands on: after a subprocess writes
-// a key directly to the dotenv file on disk, calling
-// [reloadSharedEnvLocked] must make that key visible in the in-memory
-// deps.env (and therefore in any subsequent
-// `environment.NewWithValues(name, deps.env.Dotenv())` clone).
+// TestReconcileLayerEnvironmentLocked_RefreshesDepsEnvWithoutDelta asserts
+// that step 8 still reloads out-of-process changes when no provider called Save.
 //
 // This is intentionally a unit test of the helper, not of the full
 // runProvisionSingleLayer lifecycle: the lifecycle test would need to
@@ -364,7 +338,7 @@ func TestMergeLayerOutputsLocked_PreservesSubprocessWrites(t *testing.T) {
 // which are mocked here. The full-lifecycle assertion is enforced by
 // inspection — runProvisionSingleLayer's step 8 is the only call site,
 // and the docstring above runProvisionSingleLayer pins the contract.
-func TestReloadSharedEnvLocked_RefreshesDepsEnvFromDisk(t *testing.T) {
+func TestReloadSharedEnvLocked_RefreshesDepsEnv(t *testing.T) {
 	t.Parallel()
 
 	deps, envMu, envPath := newPropagationTestDeps(t)
@@ -386,17 +360,12 @@ func TestReloadSharedEnvLocked_RefreshesDepsEnvFromDisk(t *testing.T) {
 		"precondition: in-memory deps.env should not see subprocess write before reload",
 	)
 
-	// The contract: reloadSharedEnvLocked makes the subprocess write
-	// visible in deps.env — and therefore to any downstream layer that
-	// clones from deps.env.Dotenv() at its own step 0.
+	// The contract: step 8's reload makes the subprocess write visible in
+	// deps.env — and therefore to every downstream layer, which now reads
+	// deps.env directly rather than cloning it.
 	require.NoError(t, reloadSharedEnvLocked(t.Context(), deps, envMu))
 
-	// Downstream layer's clone (this is exactly what runProvisionSingleLayer
-	// does at the start of B's invocation, line ~847).
-	downstreamLayerEnv := environment.NewWithValues(
-		deps.env.Name(), deps.env.Dotenv(),
-	)
-	assert.Equal(t, "from-a", downstreamLayerEnv.Dotenv()["HOOK_VAL"],
+	assert.Equal(t, "from-a", deps.env.Dotenv()["HOOK_VAL"],
 		"downstream layer did not see layer-A's hook-mediated env write — "+
 			"dependsOn ordering is silently incomplete",
 	)
@@ -445,6 +414,52 @@ func TestMergeLayerOutputsLocked_ConcurrentMergesConverge(t *testing.T) {
 	assert.Contains(t, disk, "FROM_B=\"b-value\"", "layer-b output clobbered by layer-a merge")
 }
 
+// TestSharedLayerEnvironment_ConcurrentProviderWritesAreSafe pins the contract that
+// replaced per-layer environment clones. Parallel layers now write directly to the
+// shared environment, so Environment and its Config must tolerate concurrent access
+// while another layer persists. Run under -race; without synchronized config this
+// panics with "concurrent map writes".
+func TestSharedLayerEnvironment_ConcurrentProviderWritesAreSafe(t *testing.T) {
+	t.Parallel()
+
+	deps, _, _ := newPropagationTestDeps(t)
+	layers := []string{"a", "b", "c", "d"}
+
+	var wg sync.WaitGroup
+	for _, name := range layers {
+		// Mirrors bicep's mustSetParamAsConfig + dotenv writes during Deploy.
+		wg.Go(func() {
+			for i := range 50 {
+				deps.env.DotenvSet("FROM_"+name, name)
+				assert.NoError(t, deps.env.Config.Set("infra.parameters."+name, i))
+				_, _ = deps.env.Config.Get("infra.parameters." + name)
+				_ = deps.env.Dotenv()
+			}
+		})
+	}
+
+	// A sibling layer persisting mid-flight marshals the same config.
+	wg.Go(func() {
+		for range 10 {
+			assert.NoError(t, deps.envManager.Save(t.Context(), deps.env))
+		}
+	})
+	wg.Wait()
+
+	for _, name := range layers {
+		assert.Equal(t, name, deps.env.Getenv("FROM_"+name))
+		// Saves round-trip through JSON, so the numeric type is not preserved.
+		assert.EqualValues(t, 49, valueAtConfigPath(t, deps.env.Config, "infra.parameters."+name))
+	}
+}
+
+func valueAtConfigPath(t *testing.T, source config.Config, path string) any {
+	t.Helper()
+	value, has := source.Get(path)
+	require.True(t, has)
+	return value
+}
+
 // newPropagationTestDeps builds a minimal provisionLayerDeps backed by a
 // real filesystem-backed envManager so tests can exercise the actual
 // reload / save semantics that the production code depends on.
diff --git a/cli/azd/pkg/config/config.go b/cli/azd/pkg/config/config.go
index 3d96a17cac5..2a17a853563 100644
--- a/cli/azd/pkg/config/config.go
+++ b/cli/azd/pkg/config/config.go
@@ -11,8 +11,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"path/filepath"
+	"reflect"
 	"regexp"
 	"strings"
+	"sync"
 
 	"github.com/google/uuid"
 )
@@ -66,8 +68,161 @@ func NewConfig(data map[string]any) Config {
 	}
 }
 
+// Clone returns an independent copy of source, including its local vault state.
+// Vault references remain references; secret values are not materialized in the raw data.
+func Clone(source Config) Config {
+	if source == nil {
+		return NewEmptyConfig()
+	}
+
+	sourceConfig, ok := source.(*config)
+	if !ok {
+		return &config{data: cloneMap(source.Raw())}
+	}
+
+	sourceConfig.mu.RLock()
+	defer sourceConfig.mu.RUnlock()
+
+	cloned := &config{data: cloneMap(sourceConfig.data), vaultId: sourceConfig.vaultId}
+	if sourceConfig.vault != nil {
+		cloned.vault = Clone(sourceConfig.vault)
+	}
+	return cloned
+}
+
+// Replace swaps destination's contents for source's in place, so existing holders of
+// destination observe the new state without the pointer swap that reassigning a
+// Config field would require. Reports whether the replacement was performed.
+func Replace(destination, source Config) bool {
+	destinationConfig, ok := destination.(*config)
+	if !ok {
+		return false
+	}
+
+	data := snapshotRaw(source)
+
+	var (
+		vaultId string
+		vault   Config
+	)
+	if sourceConfig, sourceOK := source.(*config); sourceOK {
+		sourceConfig.mu.RLock()
+		vaultId = sourceConfig.vaultId
+		vault = sourceConfig.vault
+		sourceConfig.mu.RUnlock()
+	}
+
+	destinationConfig.mu.Lock()
+	defer destinationConfig.mu.Unlock()
+
+	destinationConfig.data = data
+	destinationConfig.vaultId = vaultId
+	destinationConfig.vault = vault
+	return true
+}
+
+// ApplyDelta applies changes made between initial and updated to destination.
+// Unchanged values do not replace values written to destination after initial was captured.
+//
+// Only destination is locked: initial and updated are caller-owned snapshots.
+func ApplyDelta(destination, initial, updated Config) {
+	destinationConfig, destinationOK := destination.(*config)
+	if destinationOK {
+		destinationConfig.mu.Lock()
+		defer destinationConfig.mu.Unlock()
+	}
+
+	applyMapDelta(rawData(destination), rawData(initial), rawData(updated))
+
+	initialConfig, initialOK := initial.(*config)
+	updatedConfig, updatedOK := updated.(*config)
+	if !destinationOK || !initialOK || !updatedOK || updatedConfig.vault == nil {
+		return
+	}
+	if destinationConfig.vault == nil {
+		destinationConfig.vault = NewEmptyConfig()
+	}
+	if initialConfig.vault == nil {
+		initialConfig.vault = NewEmptyConfig()
+	}
+	ApplyDelta(destinationConfig.vault, initialConfig.vault, updatedConfig.vault)
+	destinationConfig.vaultId = updatedConfig.vaultId
+}
+
+func applyMapDelta(destination, initial, updated map[string]any) {
+	for key, initialValue := range initial {
+		updatedValue, hasUpdated := updated[key]
+		if !hasUpdated {
+			delete(destination, key)
+			continue
+		}
+
+		initialMap, initialIsMap := initialValue.(map[string]any)
+		updatedMap, updatedIsMap := updatedValue.(map[string]any)
+		if initialIsMap && updatedIsMap {
+			if reflect.DeepEqual(initialMap, updatedMap) {
+				continue
+			}
+			destinationMap, destinationIsMap := destination[key].(map[string]any)
+			if !destinationIsMap {
+				destinationMap = map[string]any{}
+				destination[key] = destinationMap
+			}
+			applyMapDelta(destinationMap, initialMap, updatedMap)
+			continue
+		}
+		if !reflect.DeepEqual(initialValue, updatedValue) {
+			destination[key] = cloneValue(updatedValue)
+		}
+	}
+
+	for key, updatedValue := range updated {
+		if _, existed := initial[key]; existed {
+			continue
+		}
+		updatedMap, updatedIsMap := updatedValue.(map[string]any)
+		if !updatedIsMap {
+			destination[key] = cloneValue(updatedValue)
+			continue
+		}
+		destinationMap, destinationIsMap := destination[key].(map[string]any)
+		if !destinationIsMap {
+			destinationMap = map[string]any{}
+			destination[key] = destinationMap
+		}
+		applyMapDelta(destinationMap, map[string]any{}, updatedMap)
+	}
+}
+
+func cloneMap(source map[string]any) map[string]any {
+	cloned := make(map[string]any, len(source))
+	for key, value := range source {
+		cloned[key] = cloneValue(value)
+	}
+	return cloned
+}
+
+func cloneValue(value any) any {
+	switch value := value.(type) {
+	case map[string]any:
+		return cloneMap(value)
+	case []any:
+		cloned := make([]any, len(value))
+		for i, item := range value {
+			cloned[i] = cloneValue(item)
+		}
+		return cloned
+	default:
+		return value
+	}
+}
+
 // Top level AZD configuration
+//
+// Exported methods are safe for concurrent use. A single Config is shared across
+// parallel provision layers and deploy steps via environment.Environment.
 type config struct {
+	mu      sync.RWMutex
 	vaultId string
 	vault   Config
 	data    map[string]any
@@ -75,18 +230,49 @@ type config struct {
 
 // Returns a value indicating whether the configuration is empty
 func (c *config) IsEmpty() bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	return len(c.data) == 0
 }
 
-// Gets the raw values stored in the configuration as a Go map
+// Gets the raw values stored in the configuration as a Go map.
+//
+// The returned map is the live backing store, not a copy. Callers that mutate it, or
+// that read it while another goroutine may be writing, must synchronize externally.
+// Within this package use rawData or snapshotRaw instead.
 func (c *config) Raw() map[string]any {
 	return c.data
 }
 
+// rawData returns the live backing map without acquiring any lock.
+func rawData(source Config) map[string]any {
+	if sourceConfig, ok := source.(*config); ok {
+		return sourceConfig.data
+	}
+
+	return source.Raw()
+}
+
+// snapshotRaw returns a deep copy of the raw data taken under the read lock.
+func snapshotRaw(source Config) map[string]any {
+	if sourceConfig, ok := source.(*config); ok {
+		sourceConfig.mu.RLock()
+		defer sourceConfig.mu.RUnlock()
+
+		return cloneMap(sourceConfig.data)
+	}
+
+	return source.Raw()
+}
+
 const vaultKeyName = "vault"
 
 // Gets the raw values stored in the configuration and resolve any vault references
 func (c *config) ResolvedRaw() map[string]any {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	resolvedRaw := &config{
 		data: map[string]any{},
 	}
@@ -99,7 +285,7 @@ func (c *config) ResolvedRaw() map[string]any {
 			continue
 		}
 		// get will always return true (no need to check) because the path was gotten from the raw config
-		value, _ := c.Get(path)
+		value, _ := c.get(path)
 		if err := resolvedRaw.Set(path, value); err != nil {
 			panic(fmt.Errorf("failed setting resolved raw value: %w", err))
 		}
@@ -126,10 +312,13 @@ func paths(start map[string]any) []string {
 
 // SetSecret stores the secrets at the specified path within a local user vault
 func (c *config) SetSecret(path string, value string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	if c.vaultId == "" {
 		c.vault = NewConfig(nil)
 		c.vaultId = uuid.New().String()
-		if err := c.Set(vaultKeyName, c.vaultId); err != nil {
+		if err := c.set(vaultKeyName, c.vaultId); err != nil {
 			return fmt.Errorf("failed setting vault id: %w", err)
 		}
 	}
@@ -140,11 +329,18 @@ func (c *config) SetSecret(path string, value string) error {
 		return fmt.Errorf("failed setting secret value: %w", err)
 	}
 
-	return c.Set(path, vaultRef)
+	return c.set(path, vaultRef)
 }
 
 // Sets a value at the specified location
 func (c *config) Set(path string, value any) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.set(path, value)
+}
+
+func (c *config) set(path string, value any) error {
 	depth := 1
 	currentNode := c.data
 	parts := strings.Split(path, ".")
@@ -178,6 +374,13 @@ func (c *config) Set(path string, value any) error {
 // When the path location is an object will remove the whole node
 // When the path does not exist, will return a `nil` value
 func (c *config) Unset(path string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.unset(path)
+}
+
+func (c *config) unset(path string) error {
 	depth := 1
 	currentNode := c.data
 	parts := strings.Split(path, ".")
@@ -210,6 +413,13 @@ func (c *config) Unset(path string) error {
 // Gets the value stored at the specified location
 // Returns the value if exists, otherwise returns nil & a value indicating if the value existing
 func (c *config) Get(path string) (any, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	return c.get(path)
+}
+
+func (c *config) get(path string) (any, bool) {
 	depth := 1
 	currentNode := c.data
 	parts := strings.Split(path, ".")
@@ -243,7 +453,10 @@ func (c *config) Get(path string) (any, bool) {
 
 // GetMap retrieves the map stored at the specified path
 func (c *config) GetMap(path string) (map[string]any, bool) {
-	value, ok := c.Get(path)
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	value, ok := c.get(path)
 	if !ok {
 		return nil, false
 	}
@@ -254,7 +467,10 @@ func (c *config) GetMap(path string) (map[string]any, bool) {
 
 // GetSlice retrieves the slice stored at the specified path
 func (c *config) GetSlice(path string) ([]any, bool) {
-	value, ok := c.Get(path)
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	value, ok := c.get(path)
 	if !ok {
 		return nil, false
 	}
@@ -265,7 +481,10 @@ func (c *config) GetSlice(path string) ([]any, bool) {
 
 // Gets the value stored at the specified location as a string
 func (c *config) GetString(path string) (string, bool) {
-	value, ok := c.Get(path)
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	value, ok := c.get(path)
 	if !ok {
 		return "", false
 	}
@@ -275,7 +494,10 @@ func (c *config) GetString(path string) (string, bool) {
 }
 
 func (c *config) GetSection(path string, section any) (bool, error) {
-	sectionConfig, ok := c.Get(path)
+	c.mu.RLock()
+	sectionConfig, ok := c.get(path)
+	c.mu.RUnlock()
+
 	if !ok {
 		return false, nil
 	}
diff --git a/cli/azd/pkg/config/config_test.go b/cli/azd/pkg/config/config_test.go
index dc62274f27a..254baf8fa33 100644
--- a/cli/azd/pkg/config/config_test.go
+++ b/cli/azd/pkg/config/config_test.go
@@ -10,6 +10,62 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestCloneAndApplyDelta(t *testing.T) {
+	destination := NewConfig(map[string]any{
+		"shared": map[string]any{"unchanged": "initial", "removed": "old"},
+	})
+	initial := Clone(destination)
+	updated := Clone(initial)
+
+	require.NoError(t, updated.Set("shared.unchanged", "provider"))
+	require.NoError(t, updated.Unset("shared.removed"))
+	require.NoError(t, updated.Set("shared.added", "new"))
+	require.NoError(t, destination.Set("concurrent", "preserved"))
+
+	ApplyDelta(destination, initial, updated)
+
+	require.Equal(t, "provider", valueAt(t, destination, "shared.unchanged"))
+	_, hasRemoved := destination.Get("shared.removed")
+	require.False(t, hasRemoved)
+	require.Equal(t, "new", valueAt(t, destination, "shared.added"))
+	require.Equal(t, "preserved", valueAt(t, destination, "concurrent"))
+}
+
+func TestApplyDeltaPreservesConcurrentNestedMapReplacement(t *testing.T) {
+	destination := NewConfig(map[string]any{
+		"shared": map[string]any{"unchanged": "initial"},
+	})
+	initial := Clone(destination)
+	updated := Clone(initial)
+	require.NoError(t, destination.Set("shared", "concurrent"))
+
+	ApplyDelta(destination, initial, updated)
+
+	require.Equal(t, "concurrent", valueAt(t, destination, "shared"))
+}
+
+func TestApplyDeltaMergesConcurrentNewNestedValues(t *testing.T) {
+	destination := NewEmptyConfig()
+	initial := Clone(destination)
+	first := Clone(initial)
+	second := Clone(initial)
+	require.NoError(t, first.Set("provider.first", "one"))
+	require.NoError(t, second.Set("provider.second", "two"))
+
+	ApplyDelta(destination, initial, first)
+	ApplyDelta(destination, initial, second)
+
+	require.Equal(t, "one", valueAt(t, destination, "provider.first"))
+	require.Equal(t, "two", valueAt(t, destination, "provider.second"))
+}
+
+func valueAt(t *testing.T, source Config, path string) any {
+	t.Helper()
+	value, has := source.Get(path)
+	require.True(t, has)
+	return value
+}
+
 func Test_SetGetUnsetWithValue(t *testing.T) {
 	tests := []struct {
 		name  string
diff --git a/cli/azd/pkg/config/manager.go b/cli/azd/pkg/config/manager.go
index 2e0a91b5795..0531fddc4fc 100644
--- a/cli/azd/pkg/config/manager.go
+++ b/cli/azd/pkg/config/manager.go
@@ -30,7 +30,8 @@ func NewManager() Manager {
 
 // Saves the azd configuration to the specified file path
 func (c *manager) Save(config Config, writer io.Writer) error {
-	configJson, err := json.MarshalIndent(config.Raw(), "", "  ")
+	// Snapshot under the read lock: a parallel layer may be writing while we marshal.
+	configJson, err := json.MarshalIndent(snapshotRaw(config), "", "  ")
 	if err != nil {
 		return fmt.Errorf("failed marshalling config JSON: %w", err)
 	}
diff --git a/cli/azd/pkg/environment/environment.go b/cli/azd/pkg/environment/environment.go
index 7e0ba86c5e9..a978c52a523 100644
--- a/cli/azd/pkg/environment/environment.go
+++ b/cli/azd/pkg/environment/environment.go
@@ -261,6 +261,15 @@ func (e *Environment) replaceState(dotenv map[string]string, deletedKeys map[str
 	e.deletedKeys = deletedKeys
 }
 
+// replaceConfig replaces the configuration contents in place. Used by data stores
+// during Reload: assigning the exported Config field instead would race with every
+// concurrent reader of env.Config, and would strand holders of the previous value.
+func (e *Environment) replaceConfig(cfg config.Config) {
+	if e.Config == nil || !config.Replace(e.Config, cfg) {
+		e.Config = cfg
+	}
+}
+
 // Name gets the name of the environment
 // If empty will fallback to the value of the AZURE_ENV_NAME environment variable
 func (e *Environment) Name() string {
diff --git a/cli/azd/pkg/environment/local_file_data_store.go b/cli/azd/pkg/environment/local_file_data_store.go
index 03bb0edbde9..a0ccb55f8da 100644
--- a/cli/azd/pkg/environment/local_file_data_store.go
+++ b/cli/azd/pkg/environment/local_file_data_store.go
@@ -171,7 +171,26 @@ func (fs *LocalFileDataStore) Reload(ctx context.Context, env *Environment) erro
 // reloadLocked performs the actual reload work. Caller MUST hold the env
 // file lock.
 func (fs *LocalFileDataStore) reloadLocked(ctx context.Context, env *Environment) error {
-	// Reload env values
+	if err := fs.reloadDotenvLocked(env); err != nil {
+		return err
+	}
+
+	// Reload env config
+	if cfg, err := fs.configManager.Load(fs.ConfigPath(env)); errors.Is(err, os.ErrNotExist) {
+		env.replaceConfig(config.NewEmptyConfig())
+	} else if err != nil {
+		return fmt.Errorf("loading config: %w", err)
+	} else {
+		env.replaceConfig(cfg)
+	}
+
+	fs.recordEnvTracing(env)
+
+	return nil
+}
+
+// reloadDotenvLocked reloads only the .env values. Caller MUST hold the env file lock.
+func (fs *LocalFileDataStore) reloadDotenvLocked(env *Environment) error {
 	var newDotenv map[string]string
 	if envMap, err := godotenv.Read(fs.EnvPath(env)); errors.Is(err, os.ErrNotExist) {
 		newDotenv = make(map[string]string)
@@ -182,15 +201,10 @@ func (fs *LocalFileDataStore) reloadLocked(ctx context.Context, env *Environment
 	}
 	env.replaceState(newDotenv, make(map[string]struct{}))
 
-	// Reload env config
-	if cfg, err := fs.configManager.Load(fs.ConfigPath(env)); errors.Is(err, os.ErrNotExist) {
-		env.Config = config.NewEmptyConfig()
-	} else if err != nil {
-		return fmt.Errorf("loading config: %w", err)
-	} else {
-		env.Config = cfg
-	}
+	return nil
+}
 
+func (fs *LocalFileDataStore) recordEnvTracing(env *Environment) {
 	if env.Name() != "" {
 		tracing.SetUsageAttributes(fields.StringHashed(fields.EnvNameKey, env.Name()))
 	}
@@ -200,8 +214,6 @@ func (fs *LocalFileDataStore) reloadLocked(ctx context.Context, env *Environment
 	} else {
 		tracing.SetGlobalAttributes(fields.StringHashed(fields.SubscriptionIdKey, env.GetSubscriptionId()))
 	}
-
-	return nil
 }
 
 // Save saves the environment to the persistent data store
@@ -239,9 +251,13 @@ func (fs *LocalFileDataStore) Save(ctx context.Context, env *Environment, option
 	deletedValues := maps.Clone(env.deletedKeys)
 	env.mu.RUnlock()
 
-	// reloadLocked replaces env.dotenv via replaceState (acquires env.mu
+	// reloadDotenvLocked replaces env.dotenv via replaceState (acquires env.mu
 	// internally) — we must NOT hold env.mu here or we deadlock.
-	if err := fs.reloadLocked(ctx, env); err != nil {
+	//
+	// Only .env is reloaded: config was written above under this same lock, so
+	// reading it back would discard config writes made by a parallel layer since
+	// that write, and would coerce values through JSON on every save.
+	if err := fs.reloadDotenvLocked(env); err != nil {
 		return fmt.Errorf("failed reloading env vars, %w", err)
 	}
 
diff --git a/cli/azd/pkg/environment/storage_blob_data_store.go b/cli/azd/pkg/environment/storage_blob_data_store.go
index a4a41357a19..cd634b9fc27 100644
--- a/cli/azd/pkg/environment/storage_blob_data_store.go
+++ b/cli/azd/pkg/environment/storage_blob_data_store.go
@@ -172,11 +172,11 @@ func (sbd *StorageBlobDataStore) Reload(ctx context.Context, env *Environment) e
 	defer configBuffer.Close()
 
 	if cfg, err := sbd.configManager.Load(configBuffer); errors.Is(err, os.ErrNotExist) {
-		env.Config = config.NewEmptyConfig()
+		env.replaceConfig(config.NewEmptyConfig())
 	} else if err != nil {
 		return fmt.Errorf("loading config: %w", err)
 	} else {
-		env.Config = cfg
+		env.replaceConfig(cfg)
 	}
 
 	if env.Name() != "" {