diff --git a/cli/azd/cmd/telemetry_test.go b/cli/azd/cmd/telemetry_test.go
index ff38f49e496..0a7bdfbe3ea 100644
--- a/cli/azd/cmd/telemetry_test.go
+++ b/cli/azd/cmd/telemetry_test.go
@@ -102,7 +102,10 @@ func TestTelemetryFieldConstants(t *testing.T) {
 
 		measurementFields := []fields.AttributeKey{
 			fields.AgentFixAttempts,
+			fields.ExeGraphDeployConcurrencyKey,
 			fields.ExeGraphMaxConcurrencyKey,
+			fields.ExeGraphPackageConcurrencyKey,
+			fields.ExeGraphProvisionConcurrencyKey,
 			fields.ToolExitCode,
 		}
 		for _, field := range measurementFields {
@@ -112,6 +115,25 @@ func TestTelemetryFieldConstants(t *testing.T) {
 		require.False(t, fields.ServiceErrorCode.IsMeasurement)
 	})
 
+	t.Run("ExecutionGraphConcurrencyFields", func(t *testing.T) {
+		t.Parallel()
+
+		concurrencyFields := []struct {
+			field fields.AttributeKey
+			key   string
+		}{
+			{fields.ExeGraphPackageConcurrencyKey, "exegraph.package_concurrency"},
+			{fields.ExeGraphProvisionConcurrencyKey, "exegraph.provision_concurrency"},
+			{fields.ExeGraphDeployConcurrencyKey, "exegraph.deploy_concurrency"},
+		}
+		for _, tt := range concurrencyFields {
+			require.Equal(t, tt.key, string(tt.field.Key))
+			require.Equal(t, fields.SystemMetadata, tt.field.Classification)
+			require.Equal(t, fields.PerformanceAndHealth, tt.field.Purpose)
+			require.True(t, tt.field.IsMeasurement)
+		}
+	})
+
 	// Hooks command telemetry fields
 	t.Run("HooksFields", func(t *testing.T) {
 		t.Parallel()
diff --git a/cli/azd/docs/concurrency-model.md b/cli/azd/docs/concurrency-model.md
index af6e57dfe6f..e51a19f1c56 100644
--- a/cli/azd/docs/concurrency-model.md
+++ b/cli/azd/docs/concurrency-model.md
@@ -19,6 +19,36 @@ it protects are co-located by convention.
 
 ---
 
+## Scheduler limits and phase groups
+
+The graph scheduler applies a hard global ceiling and optional limits for named
+groups. It admits ready work in round-robin order across groups while preserving
+critical-path priority within each group. Limits are maxima, not reservations.
+When other groups have no ready work, one group can use every available global
+slot up to its own limit.
+
+The scheduler coordinator enforces all limits before dispatch. Workers never
+wait for group capacity, so package work cannot occupy every worker while it
+waits for another package step to finish. Active work is not preempted, but a
+continuously ready group cannot starve another ready group when slots become
+available.
+
+| Command | Hard global ceiling | Phase groups |
+|---------|---------------------|--------------|
+| `azd up` | `AZD_CONCURRENCY_MAX`, then `AZD_UP_CONCURRENCY`, then `AZD_DEPLOY_CONCURRENCY`, then the scheduler default | Package: `AZD_PACKAGE_CONCURRENCY`, then `AZD_UP_CONCURRENCY`; provision: `AZD_PROVISION_CONCURRENCY`, then `AZD_UP_CONCURRENCY`; publish and deploy: `AZD_DEPLOY_CONCURRENCY`, then `AZD_UP_CONCURRENCY` |
+| `azd deploy` | `AZD_CONCURRENCY_MAX`, then `AZD_DEPLOY_CONCURRENCY`, then the scheduler default | Package: `AZD_PACKAGE_CONCURRENCY`, then `AZD_DEPLOY_CONCURRENCY`; publish and deploy: `AZD_DEPLOY_CONCURRENCY` |
+| `azd provision` | `AZD_CONCURRENCY_MAX`, then `AZD_PROVISION_CONCURRENCY`, then the scheduler default | Provision: `AZD_PROVISION_CONCURRENCY` |
+
+Package and provision work in `azd up` can overlap while retaining independent
+limits. Publish and deploy share one budget because both are part of the
+deployment phase. Standalone `azd package` remains sequential.
+
+All configured values are positive integers clamped to `64`. An explicitly set
+invalid or non-positive value disables that limit and blocks fallback. Fallback
+occurs only when the higher-precedence variable is unset.
+
+---
+
 ## Service Deploy Ordering
 
 Service deployment uses a **sequential-by-default** model to preserve
diff --git a/cli/azd/docs/environment-variables.md b/cli/azd/docs/environment-variables.md
index 37bfa1ac9be..77b67d18715 100644
--- a/cli/azd/docs/environment-variables.md
+++ b/cli/azd/docs/environment-variables.md
@@ -52,14 +52,18 @@ integration.
 | `AZD_CONTAINER_RUNTIME` | The container runtime to use (e.g., `docker`, `podman`). |
 | `AZD_ALLOW_NON_EMPTY_FOLDER` | If set, allows `azd init` to run in a non-empty directory without prompting. |
 | `AZD_BUILDER_IMAGE` | The builder docker image used to perform Dockerfile-less builds. |
-| `AZD_DEPLOY_CONCURRENCY` | Maximum number of services to deploy in parallel during `azd deploy`. Only takes effect when at least one service declares `uses:` targeting another service; without `uses:` edges, services deploy sequentially in alphabetical order for backward compatibility (see [concurrency model](concurrency-model.md)). Parsed as a positive integer; clamped to a maximum of `64`. When unset, concurrency is unlimited (bounded only by the number of services). |
+| `AZD_CONCURRENCY_MAX` | Hard maximum number of graph steps that can run at once during `azd up`, `azd deploy`, or `azd provision`. Values saved in the active azd environment take precedence over process environment values. When unset, the command-specific concurrency variable is the hard maximum: `AZD_UP_CONCURRENCY` (then `AZD_DEPLOY_CONCURRENCY`) for `azd up`, `AZD_DEPLOY_CONCURRENCY` for `azd deploy`, or `AZD_PROVISION_CONCURRENCY` for `azd provision`. When all are unset, the scheduler uses `min(stepCount, GOMAXPROCS*2)`. |
+| `AZD_PACKAGE_CONCURRENCY` | Maximum number of service package steps that can run at once during `azd up` or `azd deploy`. Falls back to `AZD_UP_CONCURRENCY` for `azd up` and `AZD_DEPLOY_CONCURRENCY` for `azd deploy`. Standalone `azd package` remains sequential. |
+| `AZD_PROVISION_CONCURRENCY` | Maximum number of infrastructure layer provision steps that can run at once during `azd provision` or `azd up`. Falls back to `AZD_UP_CONCURRENCY` during `azd up`. During `azd provision`, it is also the hard maximum when `AZD_CONCURRENCY_MAX` is unset. |
+| `AZD_DEPLOY_CONCURRENCY` | Maximum combined number of service publish and deploy steps that can run at once during `azd deploy` or `azd up`. It is also the package-step fallback and hard maximum for `azd deploy`. During `azd up`, it falls back to `AZD_UP_CONCURRENCY` for the group limit, and it remains the last hard-maximum fallback when `AZD_CONCURRENCY_MAX` and `AZD_UP_CONCURRENCY` are both unset. Without service `uses:` edges, deploy steps remain sequential in alphabetical order, but publish steps can still run in parallel (see [concurrency model](concurrency-model.md)). |
+| `AZD_UP_CONCURRENCY` | Fallback maximum for each package, provision, and combined publish/deploy phase during `azd up`. It is also the hard maximum for the full `azd up` graph when `AZD_CONCURRENCY_MAX` is unset. |
 | `AZD_DEPLOY_TIMEOUT` | Timeout for deployment operations, parsed as an integer number of seconds (for example, `1200`). Defaults to `1200` seconds (20 minutes). |
-| `AZD_PROVISION_CONCURRENCY` | Maximum number of infrastructure layers to provision in parallel during `azd provision`. Parsed as a positive integer; clamped to a maximum of `64`. When unset, concurrency is unlimited (bounded only by the dependency graph). |
 | `AZD_DEPLOYMENT_ID_FILE` | Absolute path of a file where `azd` writes ARM deployment IDs in NDJSON format (one JSON line per layer) during `azd provision` or `azd up`. The file is truncated at the start of each provisioning run, and each infrastructure layer appends one line as its ARM deployment starts. Each line has the shape `{"deploymentId":"/subscriptions/.../deployments/<name>","layer":"<layer-name>"}` — the `layer` field is empty for non-layered (single-module) provisioning. Consumers should tail/watch the file and parse each line independently; unknown fields must be ignored for forward compatibility. The path must be absolute (relative paths are ignored); the containing directory must already exist and be writable. Lines are only appended when an ARM deployment is actually started — runs short-circuited by the deployment-state cache or canceled by provision validation do not produce output. A process-wide mutex serializes writes so each line is always complete. If the file cannot be written (for example, the parent directory does not exist, the path is not writable, or the path points to a directory rather than a file), provisioning continues and the failure is recorded via the standard log; that output is only visible when `--debug` or `AZD_DEBUG_LOG` is enabled. On Windows, consumers should use a file-watcher pattern that does not keep a read handle open, otherwise new appends may fail. Only Bicep deployments are supported. |
-| `AZD_UP_CONCURRENCY` | Maximum number of steps to run in parallel during `azd up`. Parsed as a positive integer; clamped to a maximum of `64`. Falls back to `AZD_DEPLOY_CONCURRENCY` when unset. When both are unset, concurrency is unlimited. |
 | `AZD_DEPLOY_{SERVICE}_SLOT_NAME` | Sets the App Service deployment slot target for a service. Replace `{SERVICE}` with the uppercase service name (hyphens become underscores). Set to `production` to deploy to the main app, or a slot name (e.g., `staging`). When slots exist and this is not set, `--no-prompt` mode fails with an error listing available targets. Applies to `host: appservice` only; Function Apps always deploy to the main site. |
 | `AZD_DEPLOY_{SERVICE}_SKIP_STATUS_CHECK` | If `true`, skips deployment status tracking for the named Linux App Service after the zip deployment request is accepted. By default, azd waits up to five minutes without a deployment status change. Each new status resets the five-minute wait. If the status remains unchanged, azd completes deployment with a warning. Useful when the target web app is intentionally stopped. Parsed as a boolean (`true`/`false`/`1`/`0`). `{SERVICE}` follows the same naming rules as `AZD_DEPLOY_{SERVICE}_SLOT_NAME`. |
 
+All concurrency variables are parsed as positive integers and clamped to `64`. If a variable is explicitly set to an invalid or non-positive value, its limit is disabled and azd does not consult that variable's fallback. Fallback occurs only when the higher-precedence variable is unset.
+
 ## azd exec
 
 The `azd exec` command runs commands and scripts with the active azd environment loaded into the child
diff --git a/cli/azd/internal/cmd/concurrency.go b/cli/azd/internal/cmd/concurrency.go
new file mode 100644
index 00000000000..2a579bf9dd1
--- /dev/null
+++ b/cli/azd/internal/cmd/concurrency.go
@@ -0,0 +1,121 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package cmd
+
+import (
+	"log"
+	"strconv"
+	"strings"
+)
+
+const (
+	concurrencyMaxEnvVar       = "AZD_CONCURRENCY_MAX"
+	packageConcurrencyEnvVar   = "AZD_PACKAGE_CONCURRENCY"
+	provisionConcurrencyEnvVar = "AZD_PROVISION_CONCURRENCY"
+	deployConcurrencyEnvVar    = "AZD_DEPLOY_CONCURRENCY"
+	upConcurrencyEnvVar        = "AZD_UP_CONCURRENCY"
+
+	packageConcurrencyGroup   = "package"
+	provisionConcurrencyGroup = "provision"
+	deployConcurrencyGroup    = "deploy"
+
+	maxConfiguredConcurrency = 64
+)
+
+type environmentLookup func(string) (string, bool)
+
+type concurrencySetting struct {
+	value int
+	set   bool
+}
+
+type graphConcurrencyOptions struct {
+	max    int
+	groups map[string]int
+}
+
+func resolveConcurrencySetting(lookup environmentLookup, envName string) concurrencySetting {
+	envValue, ok := lookup(envName)
+	if !ok {
+		return concurrencySetting{}
+	}
+
+	setting := concurrencySetting{set: true}
+	value, err := strconv.Atoi(envValue)
+	if err != nil {
+		log.Printf("warning: ignoring invalid %s=%q: %v", envName, envValue, err)
+		return setting
+	}
+	if value <= 0 {
+		log.Printf(
+			"warning: ignoring invalid %s=%q: value must be greater than zero; "+
+				"lower-precedence concurrency settings will not apply",
+			envName,
+			envValue,
+		)
+		return setting
+	}
+
+	setting.value = min(value, maxConfiguredConcurrency)
+	if setting.value < value {
+		label := strings.ToLower(strings.ReplaceAll(strings.TrimPrefix(envName, "AZD_"), "_", " "))
+		log.Printf("clamping %s from %d to %d", label, value, setting.value)
+	}
+	return setting
+}
+
+func firstConcurrency(settings ...concurrencySetting) int {
+	for _, setting := range settings {
+		if setting.set {
+			return setting.value
+		}
+	}
+	return 0
+}
+
+func resolveUpGraphConcurrency(lookup environmentLookup) graphConcurrencyOptions {
+	maxSetting := resolveConcurrencySetting(lookup, concurrencyMaxEnvVar)
+	upSetting := resolveConcurrencySetting(lookup, upConcurrencyEnvVar)
+	packageSetting := resolveConcurrencySetting(lookup, packageConcurrencyEnvVar)
+	provisionSetting := resolveConcurrencySetting(lookup, provisionConcurrencyEnvVar)
+	deploySetting := resolveConcurrencySetting(lookup, deployConcurrencyEnvVar)
+
+	// AZD_DEPLOY_CONCURRENCY remains the last hard-ceiling fallback so users who
+	// tuned `azd deploy` parallelism before `azd up` gained its own variable do
+	// not silently get the unbounded scheduler default for the whole graph.
+	return graphConcurrencyOptions{
+		max: firstConcurrency(maxSetting, upSetting, deploySetting),
+		groups: map[string]int{
+			packageConcurrencyGroup:   firstConcurrency(packageSetting, upSetting),
+			provisionConcurrencyGroup: firstConcurrency(provisionSetting, upSetting),
+			deployConcurrencyGroup:    firstConcurrency(deploySetting, upSetting),
+		},
+	}
+}
+
+func resolveDeployGraphConcurrency(lookup environmentLookup) graphConcurrencyOptions {
+	maxSetting := resolveConcurrencySetting(lookup, concurrencyMaxEnvVar)
+	packageSetting := resolveConcurrencySetting(lookup, packageConcurrencyEnvVar)
+	deploySetting := resolveConcurrencySetting(lookup, deployConcurrencyEnvVar)
+
+	return graphConcurrencyOptions{
+		max: firstConcurrency(maxSetting, deploySetting),
+		groups: map[string]int{
+			packageConcurrencyGroup: firstConcurrency(packageSetting, deploySetting),
+			deployConcurrencyGroup:  deploySetting.value,
+		},
+	}
+}
+
+func resolveProvisionGraphConcurrency(lookup environmentLookup) graphConcurrencyOptions {
+	maxSetting := resolveConcurrencySetting(lookup, concurrencyMaxEnvVar)
+	provisionSetting := resolveConcurrencySetting(lookup, provisionConcurrencyEnvVar)
+
+	return graphConcurrencyOptions{
+		max: firstConcurrency(maxSetting, provisionSetting),
+		groups: map[string]int{
+			provisionConcurrencyGroup: provisionSetting.value,
+		},
+	}
+}
diff --git a/cli/azd/internal/cmd/concurrency_feedback_test.go b/cli/azd/internal/cmd/concurrency_feedback_test.go
new file mode 100644
index 00000000000..3a2f778c678
--- /dev/null
+++ b/cli/azd/internal/cmd/concurrency_feedback_test.go
@@ -0,0 +1,50 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package cmd
+
+import (
+	"bytes"
+	"log"
+	"testing"
+
+	"github.com/azure/azure-dev/cli/azd/pkg/environment"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestResolveConcurrencySettingWarnsForNonPositiveValues(t *testing.T) {
+	var output bytes.Buffer
+	originalWriter := log.Writer()
+	log.SetOutput(&output)
+	t.Cleanup(func() {
+		log.SetOutput(originalWriter)
+	})
+
+	for _, value := range []string{"0", "-1"} {
+		output.Reset()
+		setting := resolveConcurrencySetting(lookupEnvironment(map[string]string{
+			packageConcurrencyEnvVar: value,
+		}), packageConcurrencyEnvVar)
+
+		assert.True(t, setting.set)
+		assert.Zero(t, setting.value)
+		assert.Contains(t, output.String(), "value must be greater than zero")
+		assert.Contains(t, output.String(), "lower-precedence concurrency settings will not apply")
+	}
+}
+
+func TestUpGraphRunOptionsUsesActiveEnvironment(t *testing.T) {
+	t.Setenv(packageConcurrencyEnvVar, "1")
+	t.Setenv(upConcurrencyEnvVar, "2")
+	env := environment.NewWithValues("test", map[string]string{
+		packageConcurrencyEnvVar: "3",
+		upConcurrencyEnvVar:      "4",
+	})
+
+	opts := (&UpGraphAction{env: env}).runOptions()
+
+	assert.Equal(t, 4, opts.MaxConcurrency)
+	assert.Equal(t, 3, opts.GroupConcurrency[packageConcurrencyGroup])
+	assert.Equal(t, 4, opts.GroupConcurrency[provisionConcurrencyGroup])
+	assert.Equal(t, 4, opts.GroupConcurrency[deployConcurrencyGroup])
+}
diff --git a/cli/azd/internal/cmd/concurrency_test.go b/cli/azd/internal/cmd/concurrency_test.go
new file mode 100644
index 00000000000..31a598f9c04
--- /dev/null
+++ b/cli/azd/internal/cmd/concurrency_test.go
@@ -0,0 +1,226 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package cmd
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func lookupEnvironment(values map[string]string) environmentLookup {
+	return func(name string) (string, bool) {
+		value, ok := values[name]
+		return value, ok
+	}
+}
+
+func TestResolveConcurrencySetting(t *testing.T) {
+	tests := []struct {
+		name   string
+		values map[string]string
+		want   concurrencySetting
+	}{
+		{"unset", nil, concurrencySetting{}},
+		{"valid", map[string]string{packageConcurrencyEnvVar: "4"}, concurrencySetting{value: 4, set: true}},
+		{"clamped", map[string]string{packageConcurrencyEnvVar: "100"}, concurrencySetting{value: 64, set: true}},
+		{"maximum", map[string]string{packageConcurrencyEnvVar: "64"}, concurrencySetting{value: 64, set: true}},
+		{"invalid", map[string]string{packageConcurrencyEnvVar: "abc"}, concurrencySetting{set: true}},
+		{"zero", map[string]string{packageConcurrencyEnvVar: "0"}, concurrencySetting{set: true}},
+		{"negative", map[string]string{packageConcurrencyEnvVar: "-1"}, concurrencySetting{set: true}},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := resolveConcurrencySetting(lookupEnvironment(test.values), packageConcurrencyEnvVar)
+			assert.Equal(t, test.want, got)
+		})
+	}
+}
+
+func TestResolveUpGraphConcurrency(t *testing.T) {
+	tests := []struct {
+		name   string
+		values map[string]string
+		max    int
+		groups map[string]int
+	}{
+		{
+			name: "unset",
+			groups: map[string]int{
+				packageConcurrencyGroup:   0,
+				provisionConcurrencyGroup: 0,
+				deployConcurrencyGroup:    0,
+			},
+		},
+		{
+			name:   "up fallback",
+			values: map[string]string{upConcurrencyEnvVar: "4"},
+			max:    4,
+			groups: map[string]int{
+				packageConcurrencyGroup:   4,
+				provisionConcurrencyGroup: 4,
+				deployConcurrencyGroup:    4,
+			},
+		},
+		{
+			name: "specific phase limits",
+			values: map[string]string{
+				upConcurrencyEnvVar:        "10",
+				packageConcurrencyEnvVar:   "2",
+				provisionConcurrencyEnvVar: "8",
+				deployConcurrencyEnvVar:    "6",
+			},
+			max: 10,
+			groups: map[string]int{
+				packageConcurrencyGroup:   2,
+				provisionConcurrencyGroup: 8,
+				deployConcurrencyGroup:    6,
+			},
+		},
+		{
+			name: "explicit global maximum",
+			values: map[string]string{
+				concurrencyMaxEnvVar: "3",
+				upConcurrencyEnvVar:  "10",
+			},
+			max: 3,
+			groups: map[string]int{
+				packageConcurrencyGroup:   10,
+				provisionConcurrencyGroup: 10,
+				deployConcurrencyGroup:    10,
+			},
+		},
+		{
+			name: "invalid phase override blocks fallback",
+			values: map[string]string{
+				upConcurrencyEnvVar:      "10",
+				packageConcurrencyEnvVar: "invalid",
+			},
+			max: 10,
+			groups: map[string]int{
+				packageConcurrencyGroup:   0,
+				provisionConcurrencyGroup: 10,
+				deployConcurrencyGroup:    10,
+			},
+		},
+		{
+			// Backward compatibility: before azd up had its own variable, users
+			// capped the whole graph with AZD_DEPLOY_CONCURRENCY.
+			name:   "deploy concurrency remains the legacy hard-ceiling fallback",
+			values: map[string]string{deployConcurrencyEnvVar: "2"},
+			max:    2,
+			groups: map[string]int{
+				packageConcurrencyGroup:   0,
+				provisionConcurrencyGroup: 0,
+				deployConcurrencyGroup:    2,
+			},
+		},
+		{
+			name: "up concurrency wins over deploy concurrency for the hard ceiling",
+			values: map[string]string{
+				upConcurrencyEnvVar:     "5",
+				deployConcurrencyEnvVar: "2",
+			},
+			max: 5,
+			groups: map[string]int{
+				packageConcurrencyGroup:   5,
+				provisionConcurrencyGroup: 5,
+				deployConcurrencyGroup:    2,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := resolveUpGraphConcurrency(lookupEnvironment(test.values))
+			assert.Equal(t, test.max, got.max)
+			assert.Equal(t, test.groups, got.groups)
+		})
+	}
+}
+
+func TestResolveDeployGraphConcurrency(t *testing.T) {
+	tests := []struct {
+		name   string
+		values map[string]string
+		max    int
+		groups map[string]int
+	}{
+		{
+			name: "unset",
+			groups: map[string]int{
+				packageConcurrencyGroup: 0,
+				deployConcurrencyGroup:  0,
+			},
+		},
+		{
+			name:   "deploy fallback",
+			values: map[string]string{deployConcurrencyEnvVar: "4"},
+			max:    4,
+			groups: map[string]int{
+				packageConcurrencyGroup: 4,
+				deployConcurrencyGroup:  4,
+			},
+		},
+		{
+			name: "specific package and global limits",
+			values: map[string]string{
+				concurrencyMaxEnvVar:     "3",
+				packageConcurrencyEnvVar: "2",
+				deployConcurrencyEnvVar:  "6",
+			},
+			max: 3,
+			groups: map[string]int{
+				packageConcurrencyGroup: 2,
+				deployConcurrencyGroup:  6,
+			},
+		},
+		{
+			name: "invalid global maximum blocks fallback",
+			values: map[string]string{
+				concurrencyMaxEnvVar:    "invalid",
+				deployConcurrencyEnvVar: "4",
+			},
+			groups: map[string]int{
+				packageConcurrencyGroup: 4,
+				deployConcurrencyGroup:  4,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := resolveDeployGraphConcurrency(lookupEnvironment(test.values))
+			assert.Equal(t, test.max, got.max)
+			assert.Equal(t, test.groups, got.groups)
+		})
+	}
+}
+
+func TestResolveProvisionGraphConcurrency(t *testing.T) {
+	tests := []struct {
+		name   string
+		values map[string]string
+		max    int
+		group  int
+	}{
+		{"unset", nil, 0, 0},
+		{"provision fallback", map[string]string{provisionConcurrencyEnvVar: "4"}, 4, 4},
+		{
+			"global maximum",
+			map[string]string{concurrencyMaxEnvVar: "2", provisionConcurrencyEnvVar: "4"},
+			2,
+			4,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := resolveProvisionGraphConcurrency(lookupEnvironment(test.values))
+			assert.Equal(t, test.max, got.max)
+			assert.Equal(t, test.group, got.groups[provisionConcurrencyGroup])
+		})
+	}
+}
diff --git a/cli/azd/internal/cmd/deploy.go b/cli/azd/internal/cmd/deploy.go
index 933fde1302e..29a864f522d 100644
--- a/cli/azd/internal/cmd/deploy.go
+++ b/cli/azd/internal/cmd/deploy.go
@@ -352,9 +352,11 @@ func (da *DeployAction) deployServicesGraph(
 
 	// Wire progress tracker to graph step lifecycle callbacks.
 	// Step names are "package-<svc>", "publish-<svc>", "deploy-<svc>".
+	concurrency := resolveDeployGraphConcurrency(da.env.LookupEnv)
 	opts := exegraph.RunOptions{
-		MaxConcurrency: da.resolveDAGConcurrency(),
-		ErrorPolicy:    exegraph.FailFast,
+		MaxConcurrency:   concurrency.max,
+		GroupConcurrency: concurrency.groups,
+		ErrorPolicy:      exegraph.FailFast,
 		OnStepStart: func(stepName string) {
 			if svc, ok := strings.CutPrefix(stepName, "package-"); ok {
 				da.updateProgress(svc, phasePackaging, "")
@@ -488,23 +490,6 @@ func (da *DeployAction) deployServicesGraph(
 	}, nil
 }
 
-// resolveDAGConcurrency reads AZD_DEPLOY_CONCURRENCY from the environment.
-// Returns 0 (unlimited) if the variable is unset or invalid.
-func (da *DeployAction) resolveDAGConcurrency() int {
-	if envVal, ok := os.LookupEnv("AZD_DEPLOY_CONCURRENCY"); ok {
-		if n, err := strconv.Atoi(envVal); err != nil {
-			log.Printf("warning: ignoring invalid AZD_DEPLOY_CONCURRENCY=%q: %v", envVal, err)
-		} else if n > 0 {
-			clamped := min(n, 64)
-			if clamped < n {
-				log.Printf("clamping deploy concurrency from %d to %d", n, clamped)
-			}
-			return clamped
-		}
-	}
-	return 0
-}
-
 func (da *DeployAction) resolveDeployTimeout() (time.Duration, error) {
 	return resolveDeployTimeout(da.flags)
 }
diff --git a/cli/azd/internal/cmd/deploy_test.go b/cli/azd/internal/cmd/deploy_test.go
index 83272f1b309..152fa862d1b 100644
--- a/cli/azd/internal/cmd/deploy_test.go
+++ b/cli/azd/internal/cmd/deploy_test.go
@@ -521,30 +521,3 @@ func TestDeploymentResultJSON(t *testing.T) {
 	require.True(t, ok)
 	require.NotContains(t, web, "warnings")
 }
-
-func TestResolveDAGConcurrency(t *testing.T) {
-	tests := []struct {
-		name     string
-		envVal   string
-		setEnv   bool
-		expected int
-	}{
-		{"Unset", "", false, 0},
-		{"Valid", "4", true, 4},
-		{"ClampedTo64", "100", true, 64},
-		{"ExactlyMax", "64", true, 64},
-		{"Invalid", "abc", true, 0},
-		{"Zero", "0", true, 0},
-		{"Negative", "-1", true, 0},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if tt.setEnv {
-				t.Setenv("AZD_DEPLOY_CONCURRENCY", tt.envVal)
-			}
-			da := &DeployAction{}
-			got := da.resolveDAGConcurrency()
-			require.Equal(t, tt.expected, got)
-		})
-	}
-}
diff --git a/cli/azd/internal/cmd/provision_graph.go b/cli/azd/internal/cmd/provision_graph.go
index 71bbd1ebc04..198d9346829 100644
--- a/cli/azd/internal/cmd/provision_graph.go
+++ b/cli/azd/internal/cmd/provision_graph.go
@@ -124,7 +124,8 @@ func (p *ProvisionAction) provisionLayersGraph(
 		layerPath := layer.AbsolutePath(p.projectConfig.Path)
 
 		if err := g.AddStep(&exegraph.Step{
-			Name: provisionLayerStepName(layer),
+			Name:             provisionLayerStepName(layer),
+			ConcurrencyGroup: provisionConcurrencyGroup,
 			Action: func(ctx context.Context) error {
 				if err := p.provisionManager.Initialize(ctx, p.projectConfig.Path, layer); err != nil {
 					return fmt.Errorf("initializing provisioning manager: %w", err)
@@ -273,8 +274,9 @@ func (p *ProvisionAction) provisionLayersGraph(
 			}
 
 			if err := g.AddStep(&exegraph.Step{
-				Name:      stepNames[i],
-				DependsOn: deps,
+				Name:             stepNames[i],
+				DependsOn:        deps,
+				ConcurrencyGroup: provisionConcurrencyGroup,
 				Action: func(ctx context.Context) error {
 					outcome, err := p.provisionSingleLayerWithOutcome(ctx, layer, stepNames[i])
 					if err != nil {
@@ -504,17 +506,9 @@ func (p *ProvisionAction) graphRunOptions(ctx context.Context, quiet bool) exegr
 		}
 	}
 
-	if v, ok := os.LookupEnv("AZD_PROVISION_CONCURRENCY"); ok {
-		if n, parseErr := strconv.Atoi(v); parseErr != nil {
-			log.Printf("warning: ignoring invalid AZD_PROVISION_CONCURRENCY=%q: %v", v, parseErr)
-		} else if n > 0 {
-			clamped := min(n, 64)
-			if clamped < n {
-				log.Printf("clamping provision concurrency from %d to %d", n, clamped)
-			}
-			opts.MaxConcurrency = clamped
-		}
-	}
+	concurrency := resolveProvisionGraphConcurrency(p.env.LookupEnv)
+	opts.MaxConcurrency = concurrency.max
+	opts.GroupConcurrency = concurrency.groups
 
 	return opts
 }
diff --git a/cli/azd/internal/cmd/service_graph.go b/cli/azd/internal/cmd/service_graph.go
index 9b97eb8d0f7..145ed6faf3b 100644
--- a/cli/azd/internal/cmd/service_graph.go
+++ b/cli/azd/internal/cmd/service_graph.go
@@ -366,9 +366,10 @@ func addServiceStepsToGraph(g *exegraph.Graph, opts serviceGraphOptions) (*servi
 		// deploy → no deps → packaging overlaps with anything upstream).
 		pkgSvc := svc
 		if err := g.AddStep(&exegraph.Step{
-			Name:      pkgStepName,
-			DependsOn: opts.packageExtraDeps,
-			Tags:      []string{"package"},
+			Name:             pkgStepName,
+			DependsOn:        opts.packageExtraDeps,
+			Tags:             []string{"package"},
+			ConcurrencyGroup: packageConcurrencyGroup,
 			Action: func(ctx context.Context) error {
 				sc := project.NewServiceContext()
 
@@ -408,9 +409,10 @@ func addServiceStepsToGraph(g *exegraph.Graph, opts serviceGraphOptions) (*servi
 
 		pubSvc := svc
 		if err := g.AddStep(&exegraph.Step{
-			Name:      publishStepName,
-			DependsOn: publishDeps,
-			Tags:      []string{"publish"},
+			Name:             publishStepName,
+			DependsOn:        publishDeps,
+			Tags:             []string{"publish"},
+			ConcurrencyGroup: deployConcurrencyGroup,
 			Action: func(stepCtx context.Context) error {
 				sc := opts.state.LoadContext(pubSvc.Name)
 
@@ -516,9 +518,10 @@ func addServiceStepsToGraph(g *exegraph.Graph, opts serviceGraphOptions) (*servi
 
 		depSvc := svc
 		if err := g.AddStep(&exegraph.Step{
-			Name:      deployStepName,
-			DependsOn: deployDeps,
-			Tags:      []string{"deploy"},
+			Name:             deployStepName,
+			DependsOn:        deployDeps,
+			Tags:             []string{"deploy"},
+			ConcurrencyGroup: deployConcurrencyGroup,
 			Action: func(stepCtx context.Context) error {
 				sc := opts.state.LoadContext(depSvc.Name)
 
diff --git a/cli/azd/internal/cmd/service_graph_test.go b/cli/azd/internal/cmd/service_graph_test.go
index a8a56cd6bf3..37fdac0e288 100644
--- a/cli/azd/internal/cmd/service_graph_test.go
+++ b/cli/azd/internal/cmd/service_graph_test.go
@@ -94,6 +94,22 @@ func newGraphOpts(services []*project.ServiceConfig) (serviceGraphOptions, *exeg
 	}, g
 }
 
+func TestServiceGraphConcurrencyGroups(t *testing.T) {
+	services := []*project.ServiceConfig{{Name: "api"}}
+	opts, g := newGraphOpts(services)
+
+	_, err := addServiceStepsToGraph(g, opts)
+	require.NoError(t, err)
+
+	groups := make(map[string]string)
+	for _, step := range g.Steps() {
+		groups[step.Name] = step.ConcurrencyGroup
+	}
+	require.Equal(t, packageConcurrencyGroup, groups["package-api"])
+	require.Equal(t, deployConcurrencyGroup, groups["publish-api"])
+	require.Equal(t, deployConcurrencyGroup, groups["deploy-api"])
+}
+
 // TestSelfRefUses verifies that a service with uses: [self] does not
 // create a self-referencing deploy step edge — the graph builder
 // filters self-references out.
diff --git a/cli/azd/internal/cmd/up_graph.go b/cli/azd/internal/cmd/up_graph.go
index 0fba333c340..1ecba2f7b76 100644
--- a/cli/azd/internal/cmd/up_graph.go
+++ b/cli/azd/internal/cmd/up_graph.go
@@ -9,9 +9,7 @@ import (
 	"fmt"
 	"io"
 	"log"
-	"os"
 	"slices"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -893,9 +891,10 @@ func (u *UpGraphAction) addProvisionSteps(
 
 		layerIdx := i
 		if err := g.AddStep(&exegraph.Step{
-			Name:      stepNames[i],
-			DependsOn: deps,
-			Tags:      []string{"provision"},
+			Name:             stepNames[i],
+			DependsOn:        deps,
+			Tags:             []string{"provision"},
+			ConcurrencyGroup: provisionConcurrencyGroup,
 			Action: func(ctx context.Context) error {
 				return provisionSingleLayer(
 					ctx, provDeps, layers[layerIdx],
@@ -932,31 +931,9 @@ func (u *UpGraphAction) runOptions() exegraph.RunOptions {
 		ErrorPolicy: exegraph.FailFast,
 	}
 
-	// Optional concurrency limit from environment. AZD_UP_CONCURRENCY is the
-	// canonical name for `azd up`; AZD_DEPLOY_CONCURRENCY is honored as a
-	// fallback so that users who already tuned `azd deploy` parallelism don't
-	// get unlimited concurrency when they switch to `azd up`.
-	if v, ok := os.LookupEnv("AZD_UP_CONCURRENCY"); ok {
-		if n, parseErr := strconv.Atoi(v); parseErr != nil {
-			log.Printf("warning: ignoring invalid AZD_UP_CONCURRENCY=%q: %v", v, parseErr)
-		} else if n > 0 {
-			clamped := min(n, 64)
-			if clamped < n {
-				log.Printf("clamping up concurrency from %d to %d", n, clamped)
-			}
-			opts.MaxConcurrency = clamped
-		}
-	} else if v, ok := os.LookupEnv("AZD_DEPLOY_CONCURRENCY"); ok {
-		if n, parseErr := strconv.Atoi(v); parseErr != nil {
-			log.Printf("warning: ignoring invalid AZD_DEPLOY_CONCURRENCY=%q: %v", v, parseErr)
-		} else if n > 0 {
-			clamped := min(n, 64)
-			if clamped < n {
-				log.Printf("clamping deploy concurrency from %d to %d", n, clamped)
-			}
-			opts.MaxConcurrency = clamped
-		}
-	}
+	concurrency := resolveUpGraphConcurrency(u.env.LookupEnv)
+	opts.MaxConcurrency = concurrency.max
+	opts.GroupConcurrency = concurrency.groups
 
 	opts.OnStepStart = func(stepName string) {
 		log.Printf("up-graph: starting %s", stepName)
diff --git a/cli/azd/internal/tracing/fields/fields.go b/cli/azd/internal/tracing/fields/fields.go
index eb5b2830fea..82044c23e44 100644
--- a/cli/azd/internal/tracing/fields/fields.go
+++ b/cli/azd/internal/tracing/fields/fields.go
@@ -804,6 +804,30 @@ var (
 		IsMeasurement:  true,
 	}
 
+	// ExeGraphPackageConcurrencyKey records the resolved package phase concurrency limit.
+	ExeGraphPackageConcurrencyKey = AttributeKey{
+		Key:            attribute.Key("exegraph.package_concurrency"),
+		Classification: SystemMetadata,
+		Purpose:        PerformanceAndHealth,
+		IsMeasurement:  true,
+	}
+
+	// ExeGraphProvisionConcurrencyKey records the resolved provision phase concurrency limit.
+	ExeGraphProvisionConcurrencyKey = AttributeKey{
+		Key:            attribute.Key("exegraph.provision_concurrency"),
+		Classification: SystemMetadata,
+		Purpose:        PerformanceAndHealth,
+		IsMeasurement:  true,
+	}
+
+	// ExeGraphDeployConcurrencyKey records the resolved deploy phase concurrency limit.
+	ExeGraphDeployConcurrencyKey = AttributeKey{
+		Key:            attribute.Key("exegraph.deploy_concurrency"),
+		Classification: SystemMetadata,
+		Purpose:        PerformanceAndHealth,
+		IsMeasurement:  true,
+	}
+
 	// ExeGraphErrorPolicyKey records the error policy (fail_fast or continue_on_error).
 	ExeGraphErrorPolicyKey = AttributeKey{
 		Key:            attribute.Key("exegraph.error_policy"),
diff --git a/cli/azd/pkg/exegraph/scheduler.go b/cli/azd/pkg/exegraph/scheduler.go
index 75f30f15755..180c26269e3 100644
--- a/cli/azd/pkg/exegraph/scheduler.go
+++ b/cli/azd/pkg/exegraph/scheduler.go
@@ -52,6 +52,11 @@ type RunOptions struct {
 	// Negative values are treated as zero.
 	MaxConcurrency int
 
+	// GroupConcurrency limits the number of simultaneously running steps in each
+	// named [Step.ConcurrencyGroup]. Non-positive limits are ignored. Group limits
+	// are enforced beneath MaxConcurrency and do not reserve worker capacity.
+	GroupConcurrency map[string]int
+
 	// ErrorPolicy determines behavior on step failure.
 	ErrorPolicy ErrorPolicy
 
@@ -67,8 +72,9 @@ type RunOptions struct {
 	// It is invoked from worker goroutines and must be safe for concurrent use.
 	OnStepStart func(stepName string)
 
-	// OnStepDone is called (if non-nil) when a step finishes, with a nil error on success.
-	// It is invoked from worker goroutines and must be safe for concurrent use.
+	// OnStepDone is called (if non-nil) when a step reaches a terminal state,
+	// including when it is skipped. It may be invoked from worker goroutines or
+	// the scheduler coordinator and must be safe for concurrent use.
 	OnStepDone func(stepName string, err error)
 }
 
@@ -98,11 +104,7 @@ func RunWithResult(ctx context.Context, g *Graph, opts RunOptions) (result *RunR
 		return result
 	}
 
-	span.SetAttributes(
-		fields.ExeGraphStepCountKey.Int(g.Len()),
-		fields.ExeGraphMaxConcurrencyKey.Int(opts.MaxConcurrency),
-		fields.ExeGraphErrorPolicyKey.String(opts.ErrorPolicy.String()),
-	)
+	setRunSpanAttributes(span, g, opts)
 
 	if g.Len() == 0 {
 		return result
@@ -118,11 +120,10 @@ func RunWithResult(ctx context.Context, g *Graph, opts RunOptions) (result *RunR
 }
 
 // execute implements an event-driven scheduler with a bounded worker pool.
-// Steps are dispatched as soon as all their predecessors complete, eliminating
-// the head-of-line blocking of a phase-based approach. A fixed pool of worker
-// goroutines pulls ready steps from a queue, executes them, and reports
-// completion back to a single coordinator goroutine that updates in-degrees
-// and enqueues newly unblocked successors.
+// A single coordinator admits ready steps in work-conserving round-robin order
+// across concurrency groups, then workers execute admitted steps and report
+// completion. Capacity is never acquired by workers, so a saturated group
+// cannot occupy workers while waiting for another step in that group to finish.
 func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 	n := g.Len()
 	result := &RunResult{
@@ -217,24 +218,101 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 
 	runStart := time.Now()
 
-	// Seed ready queue with zero in-degree steps, sorted by transitive
-	// dependent count descending (critical-path heuristic). Steps with more
-	// downstream dependents start first, reducing overall wall-clock time
-	// when parallelism is bounded.
-	inflight := 0
+	// Ready steps are partitioned by concurrency group. The order within each
+	// group preserves the graph's critical-path priority. The group cursor
+	// rotates after every admission so a continuously ready group cannot starve
+	// another group, while an uncontested group can still consume all capacity.
 	priorityOrder := g.priorityOrder()
+	priorityRank := make(map[string]int, len(priorityOrder))
+	for rank, name := range priorityOrder {
+		priorityRank[name] = rank
+	}
+	compareReady := func(a, b string) int {
+		return cmp.Compare(priorityRank[a], priorityRank[b])
+	}
+	readyByGroup := make(map[string][]string)
+	var groupOrder []string
+	groupActive := make(map[string]int)
+	groupCursor := 0
+	readyCount := 0
+	inflight := 0
+
+	enqueueReady := func(names []string) {
+		names = slices.Clone(names)
+		slices.SortFunc(names, compareReady)
+
+		touchedGroups := make(map[string]struct{})
+		for _, name := range names {
+			group := g.steps[name].ConcurrencyGroup
+			if _, ok := readyByGroup[group]; !ok {
+				groupOrder = append(groupOrder, group)
+			}
+			readyByGroup[group] = append(readyByGroup[group], name)
+			touchedGroups[group] = struct{}{}
+			readyCount++
+		}
+		for group := range touchedGroups {
+			slices.SortFunc(readyByGroup[group], compareReady)
+		}
+	}
+
+	groupHasCapacity := func(group string) bool {
+		limit := opts.GroupConcurrency[group]
+		return limit <= 0 || groupActive[group] < limit
+	}
+
+	nextReady := func() (string, bool) {
+		for offset := range len(groupOrder) {
+			index := (groupCursor + offset) % len(groupOrder)
+			group := groupOrder[index]
+			queue := readyByGroup[group]
+			if len(queue) == 0 || !groupHasCapacity(group) {
+				continue
+			}
+
+			name := queue[0]
+			readyByGroup[group] = queue[1:]
+			readyCount--
+			groupCursor = (index + 1) % len(groupOrder)
+			return name, true
+		}
+		return "", false
+	}
+
+	dispatchReady := func() {
+		for inflight < numWorkers && readyCount > 0 {
+			name, ok := nextReady()
+			if !ok {
+				return
+			}
+			groupActive[g.steps[name].ConcurrencyGroup]++
+			inflight++
+			workQueue <- name
+		}
+	}
+
+	releaseCapacity := func(name string) {
+		group := g.steps[name].ConcurrencyGroup
+		groupActive[group]--
+	}
+
+	// Seed ready queues with zero in-degree steps, sorted by transitive
+	// dependent count descending (critical-path heuristic).
+	var initialReady []string
 	for _, name := range priorityOrder {
 		if inDegree[name] == 0 {
-			workQueue <- name
-			inflight++
+			initialReady = append(initialReady, name)
 		}
 	}
+	enqueueReady(initialReady)
+	dispatchReady()
 
 	// Event loop: process completions as they arrive. Each completion may
 	// unblock successors whose in-degree drops to zero.
 	for inflight > 0 {
 		comp := <-completions
 		inflight--
+		releaseCapacity(comp.name)
 
 		status, isRealFailure := classifyStepResult(comp.err, comp.schedulerCanceled)
 		timingMu.Lock()
@@ -261,6 +339,7 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 				for inflight > 0 {
 					r := <-completions
 					inflight--
+					releaseCapacity(r.name)
 					drainStatus, drainIsReal := classifyStepResult(r.err, r.schedulerCanceled)
 					if drainIsReal {
 						allErrors = append(allErrors, r.err)
@@ -294,7 +373,7 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 		// this with a proper work-stealing queue.
 		deps := dependents[comp.name]
 		needsClone := false
-		var readyBatch []string // collect newly ready steps for priority sorting
+		var readyBatch []string // collect newly ready steps for grouped enqueue
 		for i := 0; i < len(deps); i++ {
 			dep := deps[i]
 			inDegree[dep]--
@@ -331,18 +410,10 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 			}
 		}
 
-		// Sort newly ready steps by priority (critical-path first).
-		// Stable sort so ties are deterministic across runs (tests rely on
-		// this, and users benefit from reproducible scheduling order).
-		if len(readyBatch) > 1 {
-			slices.SortStableFunc(readyBatch, func(a, b string) int {
-				return cmp.Compare(g.Priority(b), g.Priority(a))
-			})
-		}
-		for _, name := range readyBatch {
-			workQueue <- name
-			inflight++
-		}
+		// enqueueReady sorts each touched group by priority, so newly ready
+		// steps merge into their group's queue in critical-path order.
+		enqueueReady(readyBatch)
+		dispatchReady()
 	}
 
 	close(workQueue)
@@ -363,13 +434,14 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 			seen[st.Name] = true
 		}
 		now := time.Now()
-		timingMu.Lock()
+		var skippedSteps []StepTiming
 		for _, name := range g.order {
 			if seen[name] {
 				continue
 			}
 			skipErr := &StepSkippedError{StepName: name}
-			result.Steps = append(result.Steps, StepTiming{
+			safeNotifyDone(opts, name, skipErr)
+			skippedSteps = append(skippedSteps, StepTiming{
 				Name:   name,
 				Status: StepSkipped,
 				Start:  now,
@@ -378,6 +450,8 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 				Err:    skipErr,
 			})
 		}
+		timingMu.Lock()
+		result.Steps = append(result.Steps, skippedSteps...)
 		timingMu.Unlock()
 	}
 
@@ -422,7 +496,25 @@ func execute(ctx context.Context, g *Graph, opts RunOptions) *RunResult {
 	return result
 }
 
-// runStep executes a single step with panic recovery, tracing, and callbacks.
+// setRunSpanAttributes records scheduler configuration on the run span.
+func setRunSpanAttributes(span tracing.Span, g *Graph, opts RunOptions) {
+	span.SetAttributes(
+		fields.ExeGraphStepCountKey.Int(g.Len()),
+		fields.ExeGraphMaxConcurrencyKey.Int(opts.MaxConcurrency),
+		fields.ExeGraphErrorPolicyKey.String(opts.ErrorPolicy.String()),
+	)
+
+	if limit, ok := opts.GroupConcurrency["package"]; ok {
+		span.SetAttributes(fields.ExeGraphPackageConcurrencyKey.Int(limit))
+	}
+	if limit, ok := opts.GroupConcurrency["provision"]; ok {
+		span.SetAttributes(fields.ExeGraphProvisionConcurrencyKey.Int(limit))
+	}
+	if limit, ok := opts.GroupConcurrency["deploy"]; ok {
+		span.SetAttributes(fields.ExeGraphDeployConcurrencyKey.Int(limit))
+	}
+}
+
 // setStepSpanAttributes records identifying attributes for step on span. The step
 // name and DependsOn entries embed user-chosen identifiers from azure.yaml (service
 // names, layer names) and are hashed before emission per
@@ -438,6 +530,7 @@ func setStepSpanAttributes(span tracing.Span, step *Step) {
 	}
 }
 
+// runStep executes a single step with panic recovery, tracing, and callbacks.
 func runStep(ctx context.Context, step *Step, opts RunOptions) (stepErr error) {
 	ctx, span := tracing.Start(ctx, events.ExeGraphStepEvent)
 
diff --git a/cli/azd/pkg/exegraph/scheduler_test.go b/cli/azd/pkg/exegraph/scheduler_test.go
index d998941698f..9415ec3cd4b 100644
--- a/cli/azd/pkg/exegraph/scheduler_test.go
+++ b/cli/azd/pkg/exegraph/scheduler_test.go
@@ -179,6 +179,46 @@ func TestRun_FailFast_CancelsRemaining(t *testing.T) {
 	// immediately), but the error should propagate.
 }
 
+func TestRun_FailFast_NotifiesUnadmittedSteps(t *testing.T) {
+	g := NewGraph()
+	for _, step := range []*Step{
+		{
+			Name:   "fail",
+			Action: func(_ context.Context) error { return errors.New("boom") },
+		},
+		{
+			Name:   "not-admitted-a",
+			Action: func(_ context.Context) error { return nil },
+		},
+		{
+			Name:   "not-admitted-b",
+			Action: func(_ context.Context) error { return nil },
+		},
+	} {
+		require.NoError(t, g.AddStep(step))
+	}
+
+	var started []string
+	done := make(map[string]error)
+	err := Run(t.Context(), g, RunOptions{
+		MaxConcurrency: 1,
+		ErrorPolicy:    FailFast,
+		OnStepStart: func(name string) {
+			started = append(started, name)
+		},
+		OnStepDone: func(name string, err error) {
+			done[name] = err
+		},
+	})
+
+	require.Error(t, err)
+	assert.Equal(t, []string{"fail"}, started)
+	require.Len(t, done, 3)
+	assert.Error(t, done["fail"])
+	assert.True(t, IsStepSkipped(done["not-admitted-a"]))
+	assert.True(t, IsStepSkipped(done["not-admitted-b"]))
+}
+
 func TestRun_ContinueOnError_CollectsAll(t *testing.T) {
 	g := NewGraph()
 
@@ -373,6 +413,334 @@ func TestRun_MaxConcurrency(t *testing.T) {
 		"should not exceed max concurrency of 2")
 }
 
+func TestRun_GroupConcurrency(t *testing.T) {
+	g := NewGraph()
+	var concurrent atomic.Int32
+	var maxConcurrent atomic.Int32
+
+	for i := range 6 {
+		require.NoError(t, g.AddStep(&Step{
+			Name:             fmt.Sprintf("package-%d", i),
+			ConcurrencyGroup: "package",
+			Action: func(_ context.Context) error {
+				cur := concurrent.Add(1)
+				for {
+					old := maxConcurrent.Load()
+					if cur <= old || maxConcurrent.CompareAndSwap(old, cur) {
+						break
+					}
+				}
+				time.Sleep(20 * time.Millisecond)
+				concurrent.Add(-1)
+				return nil
+			},
+		}))
+	}
+
+	require.NoError(t, Run(t.Context(), g, RunOptions{
+		MaxConcurrency: 6,
+		GroupConcurrency: map[string]int{
+			"package": 2,
+		},
+	}))
+	assert.Equal(t, int32(2), maxConcurrent.Load())
+}
+
+func TestRun_GroupConcurrencyFairAcrossReadyGroups(t *testing.T) {
+	g := NewGraph()
+	started := make(chan string, 4)
+	release := make(chan struct{})
+
+	addBlockingStep := func(name, group string) {
+		require.NoError(t, g.AddStep(&Step{
+			Name:             name,
+			ConcurrencyGroup: group,
+			Action: func(_ context.Context) error {
+				started <- name
+				<-release
+				return nil
+			},
+		}))
+	}
+	addBlockingStep("package-a", "package")
+	addBlockingStep("package-b", "package")
+	addBlockingStep("package-c", "package")
+	addBlockingStep("provision-a", "provision")
+
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(t.Context(), g, RunOptions{
+			MaxConcurrency: 2,
+			GroupConcurrency: map[string]int{
+				"package":   2,
+				"provision": 2,
+			},
+		})
+	}()
+
+	first := <-started
+	second := <-started
+	assert.Contains(t, []string{first, second}, "provision-a",
+		"a ready provision step should receive one of the first global slots")
+	close(release)
+	require.NoError(t, <-done)
+}
+
+func TestRun_GroupConcurrencyPreservesInsertionOrderForPriorityTies(t *testing.T) {
+	g := NewGraph()
+	var started []string
+
+	for _, step := range []*Step{
+		{
+			Name:             "root",
+			ConcurrencyGroup: "package",
+			Action: func(_ context.Context) error {
+				started = append(started, "root")
+				return nil
+			},
+		},
+		{
+			Name:             "package-a",
+			DependsOn:        []string{"root"},
+			ConcurrencyGroup: "package",
+			Action: func(_ context.Context) error {
+				started = append(started, "package-a")
+				return nil
+			},
+		},
+		{
+			Name:             "package-b",
+			DependsOn:        []string{"root"},
+			ConcurrencyGroup: "package",
+			Action: func(_ context.Context) error {
+				started = append(started, "package-b")
+				return nil
+			},
+		},
+		{
+			Name:             "provision-a",
+			DependsOn:        []string{"root"},
+			ConcurrencyGroup: "provision",
+			Action: func(_ context.Context) error {
+				started = append(started, "provision-a")
+				return nil
+			},
+		},
+	} {
+		require.NoError(t, g.AddStep(step))
+	}
+
+	expected := []string{"root", "package-a", "provision-a", "package-b"}
+	for range 100 {
+		started = nil
+		require.NoError(t, Run(t.Context(), g, RunOptions{
+			MaxConcurrency: 1,
+			GroupConcurrency: map[string]int{
+				"package":   1,
+				"provision": 1,
+			},
+		}))
+		assert.Equal(t, expected, started)
+	}
+}
+
+func TestRun_GroupConcurrencyPreservesPriorityWithinGroup(t *testing.T) {
+	g := NewGraph()
+	releaseBlocker := make(chan struct{})
+	releaseProbe := make(chan struct{})
+	probeStarted := make(chan struct{})
+	packageStarted := make(chan string, 2)
+
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "package-blocker",
+		ConcurrencyGroup: "package",
+		Action: func(_ context.Context) error {
+			<-releaseBlocker
+			return nil
+		},
+	}))
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "package-low-priority",
+		ConcurrencyGroup: "package",
+		Action: func(_ context.Context) error {
+			packageStarted <- "package-low-priority"
+			return nil
+		},
+	}))
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "unlock",
+		ConcurrencyGroup: "provision",
+		Action:           func(_ context.Context) error { return nil },
+	}))
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "package-high-priority",
+		DependsOn:        []string{"unlock"},
+		ConcurrencyGroup: "package",
+		Action: func(_ context.Context) error {
+			packageStarted <- "package-high-priority"
+			return nil
+		},
+	}))
+	require.NoError(t, g.AddStep(&Step{
+		Name:      "high-priority-dependent",
+		DependsOn: []string{"package-high-priority"},
+		Action:    func(_ context.Context) error { return nil },
+	}))
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "probe",
+		DependsOn:        []string{"unlock"},
+		ConcurrencyGroup: "provision",
+		Action: func(_ context.Context) error {
+			close(probeStarted)
+			<-releaseProbe
+			return nil
+		},
+	}))
+
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(t.Context(), g, RunOptions{
+			MaxConcurrency: 2,
+			GroupConcurrency: map[string]int{
+				"package":   1,
+				"provision": 1,
+			},
+		})
+	}()
+
+	<-probeStarted
+	close(releaseBlocker)
+	assert.Equal(t, "package-high-priority", <-packageStarted)
+	close(releaseProbe)
+	require.NoError(t, <-done)
+}
+
+func TestRun_GroupConcurrencyBorrowsIdleCapacity(t *testing.T) {
+	g := NewGraph()
+	started := make(chan struct{}, 3)
+	release := make(chan struct{})
+
+	for i := range 3 {
+		require.NoError(t, g.AddStep(&Step{
+			Name:             fmt.Sprintf("package-%d", i),
+			ConcurrencyGroup: "package",
+			Action: func(_ context.Context) error {
+				started <- struct{}{}
+				<-release
+				return nil
+			},
+		}))
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(t.Context(), g, RunOptions{
+			MaxConcurrency: 3,
+			GroupConcurrency: map[string]int{
+				"package": 3,
+			},
+		})
+	}()
+
+	for range 3 {
+		<-started
+	}
+	close(release)
+	require.NoError(t, <-done)
+}
+
+func TestRun_GroupConcurrencySharedAcrossStepKinds(t *testing.T) {
+	g := NewGraph()
+	var concurrent atomic.Int32
+	var maxConcurrent atomic.Int32
+
+	for _, name := range []string{"publish-api", "deploy-web"} {
+		require.NoError(t, g.AddStep(&Step{
+			Name:             name,
+			ConcurrencyGroup: "deploy",
+			Action: func(_ context.Context) error {
+				cur := concurrent.Add(1)
+				for {
+					old := maxConcurrent.Load()
+					if cur <= old || maxConcurrent.CompareAndSwap(old, cur) {
+						break
+					}
+				}
+				time.Sleep(20 * time.Millisecond)
+				concurrent.Add(-1)
+				return nil
+			},
+		}))
+	}
+
+	require.NoError(t, Run(t.Context(), g, RunOptions{
+		MaxConcurrency: 2,
+		GroupConcurrency: map[string]int{
+			"deploy": 1,
+		},
+	}))
+	assert.Equal(t, int32(1), maxConcurrent.Load())
+}
+
+func TestRun_GroupConcurrencyReleasesCapacityAfterFailure(t *testing.T) {
+	g := NewGraph()
+	var completed atomic.Bool
+
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "package-fail",
+		ConcurrencyGroup: "package",
+		Action: func(_ context.Context) error {
+			return errors.New("failed")
+		},
+	}))
+	require.NoError(t, g.AddStep(&Step{
+		Name:             "package-next",
+		ConcurrencyGroup: "package",
+		Action: func(_ context.Context) error {
+			completed.Store(true)
+			return nil
+		},
+	}))
+
+	err := Run(t.Context(), g, RunOptions{
+		MaxConcurrency: 2,
+		GroupConcurrency: map[string]int{
+			"package": 1,
+		},
+		ErrorPolicy: ContinueOnError,
+	})
+	require.Error(t, err)
+	assert.True(t, completed.Load())
+}
+
+func TestRun_GroupConcurrencyNoDeadlockWhenGlobalBelowGroupCount(t *testing.T) {
+	g := NewGraph()
+	var completed atomic.Int32
+
+	for _, group := range []string{"package", "provision", "deploy"} {
+		require.NoError(t, g.AddStep(&Step{
+			Name:             group,
+			ConcurrencyGroup: group,
+			Action: func(_ context.Context) error {
+				completed.Add(1)
+				return nil
+			},
+		}))
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	require.NoError(t, Run(ctx, g, RunOptions{
+		MaxConcurrency: 1,
+		GroupConcurrency: map[string]int{
+			"package":   1,
+			"provision": 1,
+			"deploy":    1,
+		},
+	}))
+	assert.Equal(t, int32(3), completed.Load())
+}
+
 func TestRun_ContextCancellation(t *testing.T) {
 	g := NewGraph()
 	ctx, cancel := context.WithCancel(t.Context())
@@ -1191,6 +1559,31 @@ func TestSetStepSpanAttributes(t *testing.T) {
 	assert.Equal(t, []string{"deploy"}, tagsVal.AsStringSlice())
 }
 
+func TestSetRunSpanAttributes_RecordsPhaseConcurrency(t *testing.T) {
+	span := &mocktracing.Span{}
+	g := NewGraph()
+	require.NoError(t, g.AddStep(&Step{
+		Name:   "package-api",
+		Action: func(context.Context) error { return nil },
+	}))
+
+	setRunSpanAttributes(span, g, RunOptions{
+		MaxConcurrency: 8,
+		ErrorPolicy:    FailFast,
+		GroupConcurrency: map[string]int{
+			"package":   2,
+			"provision": 3,
+			"deploy":    4,
+			"custom":    5,
+		},
+	})
+
+	assert.Equal(t, int64(2), stepAttr(t, span, fields.ExeGraphPackageConcurrencyKey.Key).AsInt64())
+	assert.Equal(t, int64(3), stepAttr(t, span, fields.ExeGraphProvisionConcurrencyKey.Key).AsInt64())
+	assert.Equal(t, int64(4), stepAttr(t, span, fields.ExeGraphDeployConcurrencyKey.Key).AsInt64())
+	assert.NotContains(t, span.Attributes, attribute.Int("exegraph.custom_concurrency", 5))
+}
+
 // stepAttr returns the value of the attribute with the given key set on span.
 func stepAttr(t *testing.T, span *mocktracing.Span, key attribute.Key) attribute.Value {
 	t.Helper()
diff --git a/cli/azd/pkg/exegraph/step.go b/cli/azd/pkg/exegraph/step.go
index 5331c977bfa..728a24c5661 100644
--- a/cli/azd/pkg/exegraph/step.go
+++ b/cli/azd/pkg/exegraph/step.go
@@ -75,6 +75,10 @@ type Step struct {
 	// Tags are optional labels for querying related steps (e.g., "provision", "deploy").
 	Tags []string
 
+	// ConcurrencyGroup optionally assigns the step to a named concurrency budget.
+	// Steps in the same group share the limit configured in [RunOptions.GroupConcurrency].
+	ConcurrencyGroup string
+
 	// Action is the function to execute when all dependencies are satisfied.
 	Action StepFunc
 }
diff --git a/docs/reference/telemetry-data.md b/docs/reference/telemetry-data.md
index caff93eebd0..a31c7984b4d 100644
--- a/docs/reference/telemetry-data.md
+++ b/docs/reference/telemetry-data.md
@@ -578,6 +578,9 @@ The first-run middleware is not currently registered, so these fields are not em
 |-----------|------|-------------|
 | `exegraph.step.count` | measurement | Total steps in graph |
 | `exegraph.max_concurrency` | measurement | Effective concurrency limit |
+| `exegraph.package_concurrency` | measurement | Resolved package phase concurrency limit, when package concurrency is configured for the graph |
+| `exegraph.provision_concurrency` | measurement | Resolved provision phase concurrency limit, when provision concurrency is configured for the graph |
+| `exegraph.deploy_concurrency` | measurement | Resolved deploy phase concurrency limit, when deploy concurrency is configured for the graph |
 | `exegraph.error_policy` | string | `fail_fast` or `continue_on_error` |
 | `exegraph.step.name` | string | Step name. **SHA-256 hashed** — embeds user-defined service/layer names from `azure.yaml` |
 | `exegraph.step.deps` | string[] | Step dependencies (other step names). **SHA-256 hashed** for the same reason |
diff --git a/docs/specs/exegraph/spec.md b/docs/specs/exegraph/spec.md
index e5eee80a2f3..6f641a5455d 100644
--- a/docs/specs/exegraph/spec.md
+++ b/docs/specs/exegraph/spec.md
@@ -70,8 +70,9 @@ follow-up edits. Directory layout and file responsibilities are stable.
 Defines the unit of work:
 
 - **StepFunc** — `func(ctx context.Context) error`
-- **Step** — `Name string`, `DependsOn []string`, `Tags []string`, `Action StepFunc`
-  (per-step timeout is expressed via `RunOptions.StepTimeout`, not a `Step` field)
+- **Step** — `Name string`, `DependsOn []string`, `Tags []string`,
+  `ConcurrencyGroup string`, `Action StepFunc` (per-step timeout is expressed via
+  `RunOptions.StepTimeout`, not a `Step` field)
 - **StepStatus** — `Pending → Running → Done | Failed | Skipped`
 - **StepSkippedError** — returned by a step to mark itself skipped; downstream steps skip too
 - **RunResult** — per-step timing (`StepTiming`), status, error, plus aggregate duration
@@ -84,14 +85,18 @@ Insertion-order-deterministic DAG:
 - **Validate** — DFS cycle detection + missing-dependency check
 - **Priority** — transitive-dependent count heuristic (steps with more downstream work run first)
 - **Steps** — returns steps in insertion order (deterministic scheduling)
+- **ConcurrencyGroup** — optional step group used by the scheduler for per-group admission limits
 
 ### Scheduler (`scheduler.go`)
 
 Event-driven bounded worker pool:
 
 - **Concurrency** — `MaxConcurrency=0` (default) caps workers at `min(stepCount, GOMAXPROCS×2)`.
-  Explicit positive values override this (values larger than `min(stepCount, GOMAXPROCS×2)`
-  have no effect; the worker count never exceeds the natural cap).
+  An explicit positive value replaces the CPU-based default and is capped only by the step count.
+- **Concurrency groups** — `GroupConcurrency` limits active steps in each named
+  `Step.ConcurrencyGroup` beneath the global `MaxConcurrency` ceiling. The coordinator
+  admits ready groups in work-conserving round-robin order. Group limits do not reserve
+  capacity, and workers never wait for group capacity.
 - **Error policies** — `FailFast` cancels all on first error; `ContinueOnError` runs remaining independent steps
 - **Per-step timeout** — uniform `RunOptions.StepTimeout` wraps every step's context with
   `context.WithTimeout`. Zero (the default) means no deadline. A step that exceeds the
@@ -159,7 +164,8 @@ In-memory `sync.Map` cache keyed by SHA-256 of the full Bicep file tree:
    validation-cancel / JSON state dump / OpenAI-access / Responsible-AI wrappers are
    applied exactly once
 6. `FailFast` error policy
-7. Concurrency limit configurable via `AZD_PROVISION_CONCURRENCY` env var
+7. Hard concurrency limit configurable via `AZD_CONCURRENCY_MAX`, falling back to
+   `AZD_PROVISION_CONCURRENCY`; provision steps use the `provision` group
 8. Wraps console output in `syncConsole` (mutex-wrapped message/spinner methods)
 
 ### Graph-Driven Deploy (`deploy.go`)
@@ -173,6 +179,11 @@ Activates unconditionally. Per service, creates three steps:
 3. **`deploy-<svc>`** — depends on `publish-<svc>`; runs `serviceManager.Deploy` with
    `deployTimeout` context deadline
 
+Package steps use the `package` concurrency group. Publish and deploy steps share the
+`deploy` group so their combined active count cannot exceed `AZD_DEPLOY_CONCURRENCY`.
+`AZD_CONCURRENCY_MAX` is the hard limit for all active graph steps, falling back to
+`AZD_DEPLOY_CONCURRENCY` when unset.
+
 **Build gate (soft serialization)**: `serviceGraphOptions.buildGateKey` is an
 optional callback that returns an opaque string grouping for each service.
 Services sharing a non-empty key serialize on a "first wins, rest wait" basis
@@ -223,6 +234,11 @@ Deploy chain:
   all `deploy-<svc>` nodes)
 - `cmdhook-postdeploy` (depends on event-postdeploy)
 
+Provision, package, and combined publish/deploy steps use the `provision`, `package`,
+and `deploy` concurrency groups. The groups can overlap and share idle global capacity.
+`AZD_CONCURRENCY_MAX` is the hard limit across the unified graph, falling back to
+`AZD_UP_CONCURRENCY` and then to `AZD_DEPLOY_CONCURRENCY` when unset.
+
 Deploy timeout honors `--timeout` / `AZD_DEPLOY_TIMEOUT` via the shared
 `resolveDeployTimeout` helper.
 
@@ -254,6 +270,9 @@ OTel events and attributes added:
 | `exegraph.step` | Child span per step |
 | `exegraph.step.count` | Number of steps in graph |
 | `exegraph.max_concurrency` | Effective worker count |
+| `exegraph.package_concurrency` | Resolved package phase concurrency limit |
+| `exegraph.provision_concurrency` | Resolved provision phase concurrency limit |
+| `exegraph.deploy_concurrency` | Resolved deploy phase concurrency limit |
 | `exegraph.error_policy` | `FailFast` or `ContinueOnError` |
 | `exegraph.step.name` | Step name |
 | `exegraph.step.deps` | Step dependency list |
@@ -265,7 +284,8 @@ OTel events and attributes added:
 | Test file | Tests | Coverage |
 |-----------|-------|----------|
 | `pkg/exegraph/graph_test.go` | 15 | Mutation rules, ordering, cycles, priority, tags |
-| `pkg/exegraph/scheduler_test.go` | 33 | Execution semantics, cancellation, skip propagation, concurrency bounds, panic recovery, goroutine cleanup, timing, per-step timeout |
+| `pkg/exegraph/scheduler_test.go` | 49 | Execution semantics, cancellation, skip propagation, global and group concurrency bounds, group fairness, deterministic ordering, panic recovery, goroutine cleanup, timing, per-step timeout |
+| `internal/cmd/concurrency_test.go` | 4 | Environment parsing, phase fallback precedence, global ceiling precedence |
 | `pkg/infra/provisioning/bicep/layer_deps_test.go` | 12 | Temp file fixtures, cycles, env-skip, missing refs |
 | `internal/cmd/provision_graph_test.go` | 7 | Graph build, execution ordering, `dependsOn` edge ordering, env merge (preserves subprocess writes + concurrent merges converge), reload (refreshes `deps.env` from disk for downstream-layer clones) |
 | `internal/cmd/provision_security_test.go` | 2 | Env serialization, clone isolation |
@@ -273,21 +293,24 @@ OTel events and attributes added:
 | `internal/cmd/deploy_progress_test.go` | 13 | Interactive/non-interactive rendering, truncation, final render |
 | `pkg/tools/bicep/bicep_cache_test.go` | 6 | Cache hit/miss, hash stability, module resolution |
 
-**48 exegraph engine tests** (15 graph + 33 scheduler). Additional integration tests across
+**64 exegraph engine tests** (15 graph + 49 scheduler). Additional integration tests across
 provisioning, deployment, and thread-safety modules (see individual package `*_test.go` files).
 
 ## Environment Variables
 
 | Variable | Scope | Default | Effect |
 |----------|-------|---------|--------|
-| `AZD_PROVISION_CONCURRENCY` | `azd provision` (multi-layer) | `0` (unlimited, capped at `min(layerCount, GOMAXPROCS×2)`) | Overrides the scheduler's worker count for layer provisioning. Values `> 64` are clamped to `64`. Non-positive or non-integer values fall back to default. |
-| `AZD_DEPLOY_CONCURRENCY` | `azd deploy` | `0` (unlimited, capped at `min(stepCount, GOMAXPROCS×2)`) | Overrides the scheduler's worker count for package/publish/deploy steps. Values `> 64` are clamped to `64`. Non-positive or non-integer values fall back to default. |
-| `AZD_UP_CONCURRENCY` | `azd up` (unified DAG) | `0` (unlimited, capped at `min(stepCount, GOMAXPROCS×2)`) | Overrides the scheduler's worker count for the unified up DAG. Values `> 64` are clamped to `64`. Non-positive or non-integer values fall back to default. |
+| `AZD_CONCURRENCY_MAX` | `azd up`, `azd deploy`, `azd provision` | Command-specific limit, then scheduler default | Hard maximum across all active graph steps. |
+| `AZD_PACKAGE_CONCURRENCY` | Package group in `azd up` and `azd deploy` | `AZD_UP_CONCURRENCY` or `AZD_DEPLOY_CONCURRENCY` | Limits active package steps. Standalone `azd package` remains sequential. |
+| `AZD_PROVISION_CONCURRENCY` | Provision group in `azd up` and `azd provision` | `AZD_UP_CONCURRENCY` during `azd up`; unlimited group during `azd provision` | Limits active infrastructure layer steps. It is also the `azd provision` hard-limit fallback. |
+| `AZD_DEPLOY_CONCURRENCY` | Shared publish/deploy group in `azd up` and `azd deploy` | `AZD_UP_CONCURRENCY` during `azd up`; unlimited group during `azd deploy` | Limits the combined active publish and deploy steps. It is also the package and hard-limit fallback during `azd deploy`, and the last hard-limit fallback during `azd up`. |
+| `AZD_UP_CONCURRENCY` | `azd up` | Scheduler default | Per-group fallback and hard-limit fallback for the unified graph. |
 | `AZD_DEPLOY_TIMEOUT` | `azd deploy` / `azd up` | `1200` (20 minutes) | Per-service deploy timeout in whole seconds. Precedence: `--timeout` CLI flag first, then `AZD_DEPLOY_TIMEOUT`, then the default. Invalid or non-positive values cause an immediate error. |
 
-No new environment variables are introduced at the graph engine layer — `pkg/exegraph` is
-configuration-neutral. All three concurrency knobs live in the command layer and map to
-the scheduler's `RunOptions.MaxConcurrency` field.
+Concurrency values are positive integers clamped to `64`. An explicitly set invalid or
+non-positive value disables that limit and blocks fallback; only an unset variable falls
+back. The graph engine remains configuration-neutral. Command-layer values map to
+`RunOptions.MaxConcurrency` and `RunOptions.GroupConcurrency`.
 
 ## Known Limitations
 
diff --git a/docs/specs/metrics-audit/feature-telemetry-matrix.md b/docs/specs/metrics-audit/feature-telemetry-matrix.md
index a4dad469cf2..483da131044 100644
--- a/docs/specs/metrics-audit/feature-telemetry-matrix.md
+++ b/docs/specs/metrics-audit/feature-telemetry-matrix.md
@@ -161,7 +161,7 @@ reserved field contracts.
 | **Provision validation** | `provision` (all providers, plus Bicep `arm-provision` prior to ARM deploy) | `validation.provision` | `validation.provision.outcome`, plus peer fields covering warnings/errors counts, cancel reason, and `check_type` (dispatch site) | Local-only validation; runs for every provider via the provider-agnostic `provision` dispatch and additionally as Bicep `arm-provision`. `check_type` distinguishes the two emissions so Bicep provisions are not double-counted |
 | **ARM deployment client** | `provision` (any Bicep flow) | `arm.deploy.subscription`, `arm.deploy.resourcegroup`, `arm.stack.deploy.subscription`, `arm.stack.deploy.resourcegroup`, `arm.whatif.subscription`, `arm.whatif.resourcegroup`, `arm.validate.subscription`, `arm.validate.resourcegroup` | ARM operation status + duration | Per-call instrumentation in the ARM client; covers regular + stack deployments at both scopes |
 | **Multi-layer provision** | `provision` (when `infra.layers[]` is configured in `azure.yaml`) | (none — enriches the `provision` span) | `provision.layer.count`, `provision.layer.max_parallel`, `provision.layer.safe_fallback_count`, `provision.layer.explicit_dependson_count` | All four are integer measurements emitted from `internal/cmd/provision_graph.go`; no per-layer duration or outcome attribute is emitted |
-| **Execution graph (scheduler)** | `up`, `provision`, `deploy`, `package`, `publish`, `down` | `exegraph.run`, `exegraph.step` | `exegraph.step.count`, `exegraph.max_concurrency`, `exegraph.error_policy`, `exegraph.step.name` (hashed), `exegraph.step.deps` (hashed slice), `exegraph.step.tags` (raw — hardcoded literals only), `exegraph.step.timeout_s` | Step count, max concurrency, and timeout are integer measurements; step names embed user-defined service / layer names from `azure.yaml`; both `name` and `deps` use `fields.StringHashed` / `fields.StringSliceHashed` |
+| **Execution graph (scheduler)** | `up`, `provision`, `deploy`, `package`, `publish`, `down` | `exegraph.run`, `exegraph.step` | `exegraph.step.count`, `exegraph.max_concurrency`, `exegraph.package_concurrency`, `exegraph.provision_concurrency`, `exegraph.deploy_concurrency`, `exegraph.error_policy`, `exegraph.step.name` (hashed), `exegraph.step.deps` (hashed slice), `exegraph.step.tags` (raw — hardcoded literals only), `exegraph.step.timeout_s` | Step count, max concurrency, phase concurrency, and timeout are integer measurements; phase concurrency fields are present only when the graph configures that fixed group; step names embed user-defined service / layer names from `azure.yaml`; both `name` and `deps` use `fields.StringHashed` / `fields.StringSliceHashed` |
 | **Container lifecycle** | `package`, `deploy` (container service targets) | `container.credentials`, `container.publish`, `container.remotebuild` | `container.publish` sets a `container.remotebuild` property (bool) only; the `container.credentials` and `container.remotebuild` events set no attributes (span status carries success/failure and duration) | The hashed `pack.builder.image` / `pack.builder.tag` attributes are emitted on the separate `tools.pack.build` span, not the `container.*` spans |
 | **App Service deploy** | `deploy`, `publish` (App Service targets) | `deploy.appservice.zip` | `deploy.appservice.linux` (bool), `deploy.appservice.attempt` (retry attempt number) | Zip-deploy path only; outcome / duration are carried by the span status and span timing, not by dedicated attributes |
 | **AKS service target** | `provision` (AKS preprovision/postprovision) | `aks.postprovision.skip` | `skip.reason` (bounded enum — `cluster_not_provisioned`) | Recorded when cluster is not yet available for context setup |
diff --git a/docs/specs/metrics-audit/telemetry-schema.md b/docs/specs/metrics-audit/telemetry-schema.md
index 99cd7e26894..bf35163d605 100644
--- a/docs/specs/metrics-audit/telemetry-schema.md
+++ b/docs/specs/metrics-audit/telemetry-schema.md
@@ -416,6 +416,9 @@ The execution graph powers the parallel `up` / `provision` / `deploy` engine.
 |-------|----------|----------------|---------|-------|
 | Step count | `exegraph.step.count` | SystemMetadata | PerformanceAndHealth | **Measurement** — total number of steps in the graph |
 | Max concurrency | `exegraph.max_concurrency` | SystemMetadata | PerformanceAndHealth | **Measurement** — effective concurrency limit used for the run |
+| Package concurrency | `exegraph.package_concurrency` | SystemMetadata | PerformanceAndHealth | **Measurement** — resolved package phase concurrency limit; `0` means no dedicated phase limit |
+| Provision concurrency | `exegraph.provision_concurrency` | SystemMetadata | PerformanceAndHealth | **Measurement** — resolved provision phase concurrency limit; `0` means no dedicated phase limit |
+| Deploy concurrency | `exegraph.deploy_concurrency` | SystemMetadata | PerformanceAndHealth | **Measurement** — resolved deploy phase concurrency limit; `0` means no dedicated phase limit |
 | Error policy | `exegraph.error_policy` | SystemMetadata | PerformanceAndHealth | `fail_fast` or `continue_on_error` |
 | Step name | `exegraph.step.name` | SystemMetadata | PerformanceAndHealth | **Hashed** via `fields.StringHashed` — step names embed user-chosen service / layer names from `azure.yaml` (e.g., `deploy-<svc.Name>`, `<layer.Name>`) |
 | Step deps | `exegraph.step.deps` | SystemMetadata | PerformanceAndHealth | **Hashed slice** via `fields.StringSliceHashed` — each entry is another step name that embeds user-chosen identifiers |