diff --git a/.github/workflows/functional-test-cloud.yaml b/.github/workflows/functional-test-cloud.yaml
index e403bf6507..1f1d4cc928 100644
--- a/.github/workflows/functional-test-cloud.yaml
+++ b/.github/workflows/functional-test-cloud.yaml
@@ -64,8 +64,16 @@ env:
   CONTAINER_REGISTRY: ${{ vars.FUNCTIONAL_TEST_CONTAINER_REGISTRY }}
   # Container registry for storing Bicep recipe artifacts
   BICEP_RECIPE_REGISTRY: ${{ vars.FUNCTIONAL_TEST_BICEP_RECIPE_REGISTRY }}
-  # The radius functional test timeout
-  FUNCTIONALTEST_TIMEOUT: 60m
+  # The radius functional test timeout. This bounds `go test` and must stay
+  # comfortably below the job's timeout-minutes: the job also pays for ~10
+  # minutes of setup (KinD cluster, Radius install, Helm charts, cloud logins)
+  # before the tests start, and must still have time afterwards to collect
+  # cluster diagnostics and upload artifacts. If `go test` is allowed to run to
+  # the job timeout, GitHub cancels the job mid-test and the post-test steps -
+  # including diagnostics collection - never run, which is exactly the evidence
+  # needed to debug a hang. The cloud legs normally spend about 5 minutes in
+  # `go test`, so this leaves a wide margin.
+  FUNCTIONALTEST_TIMEOUT: 30m
   # The Azure Location to store test resources
   AZURE_LOCATION: ${{ vars.AZURE_LOCATION }}
   # The base directory for storing test logs
@@ -983,6 +991,12 @@ jobs:
           # Restore AWS Bicep types
           bicep restore ./test/functional-portable/corerp/cloud/resources/testdata/aws-logs-loggroup.bicep --force
       - name: Run functional tests
+        # A step-level timeout, unlike the job-level one, still lets the
+        # `if: always()` steps below run - so a hung suite is followed by
+        # diagnostics collection rather than a cancelled job. Kept above
+        # FUNCTIONALTEST_TIMEOUT so `go test` times out first and prints its
+        # goroutine dump. Mirrors the pattern in long-running-azure.yaml.
+        timeout-minutes: 35
         run: |
           set -euo pipefail
 
diff --git a/.github/workflows/long-running-azure.yaml b/.github/workflows/long-running-azure.yaml
index 7949a8d5a0..aa05149786 100644
--- a/.github/workflows/long-running-azure.yaml
+++ b/.github/workflows/long-running-azure.yaml
@@ -452,21 +452,16 @@ jobs:
           make test-functional-all-cloud
           make test-functional-all-noncloud
 
-      - name: Collect Pod details
+      - name: Collect cluster diagnostics
         if: always()
-        # Diagnostic-only snapshot of cluster-wide pod state for post-mortem. A
-        # `kubectl describe pods -A` can race pod teardown and return a transient
-        # NotFound, so never let it fail the job.
+        # Diagnostic-only post-mortem snapshot. The script records each command's
+        # failure inline and keeps going, because the failures worth debugging are
+        # usually the ones where the control plane is unreachable.
         continue-on-error: true
-        run: |
-          POD_STATE_LOG_FILENAME="${RADIUS_CONTAINER_LOG_BASE}/all-tests-pod-states.log"
-          mkdir -p $(dirname $POD_STATE_LOG_FILENAME)
-          {
-            echo "kubectl get pods -A"
-            kubectl get pods -A
-            echo "kubectl describe pods -A"
-            kubectl describe pods -A
-          } >> "${POD_STATE_LOG_FILENAME}"
+        env:
+          RADIUS_CONTAINER_LOG_PATH: ${{ github.workspace }}/${{ env.RADIUS_CONTAINER_LOG_BASE }}
+          DIAGNOSTICS_NAME: all
+        run: ./.github/scripts/collect-cluster-diagnostics.sh
 
       - name: Upload container logs
         if: always()
diff --git a/test/functional-portable/corerp/cloud/resources/aci_test.go b/test/functional-portable/corerp/cloud/resources/aci_test.go
index 949b7383c8..c7e8cf8b72 100644
--- a/test/functional-portable/corerp/cloud/resources/aci_test.go
+++ b/test/functional-portable/corerp/cloud/resources/aci_test.go
@@ -40,12 +40,13 @@ import (
 func Test_ACI(t *testing.T) {
 	// Disabled: this test dominates the corerp-cloud functional leg's wall-clock
 	// time. It provisions real Azure Container Instances (plus a VNet/NSG/ILB) and
-	// is wrapped in a 2x retry with 60s backoff to tolerate the subscription-shared
-	// 'StandardCores' ACI quota (ContainerGroupQuotaReached). When the quota is
-	// exhausted by concurrent CI runs the deploy is retried end-to-end, so a single
-	// run can take ~11-12 minutes (vs <1 minute for every other test in the leg) and
-	// roughly doubles the cloud workflow's total time. Re-enable once the ACI tests
-	// are isolated onto their own quota-aware lane. See #12044 and #12163.
+	// is wrapped in a 5 minute retry budget with 60s backoff to tolerate the
+	// subscription-shared 'StandardCores' ACI quota (ContainerGroupQuotaReached).
+	// When the quota is exhausted by concurrent CI runs the deploy is retried
+	// end-to-end, so a single run can take ~11-12 minutes (vs <1 minute for every
+	// other test in the leg) and roughly doubles the cloud workflow's total time.
+	// Re-enable once the ACI tests are isolated onto their own quota-aware lane.
+	// See #12044 and #12163.
 	t.Skip("Test_ACI is temporarily disabled: real ACI provisioning + quota retries dominate corerp-cloud CI time. See #12044, #12163.")
 
 	name := "aci-app"
@@ -56,7 +57,7 @@ func Test_ACI(t *testing.T) {
 
 	test := rp.NewRPTest(t, name, []rp.TestStep{
 		{
-			Executor:             step.NewDeployExecutor(template).WithRetry(2, 60*time.Second, isTransientAzureError),
+			Executor:             step.NewDeployExecutor(template).WithRetry(5*time.Minute, 60*time.Second, isTransientCloudDeployError),
 			SkipObjectValidation: true,
 			RPResources: &validation.RPResourceSet{
 				Resources: []validation.RPResource{
@@ -188,15 +189,15 @@ var transientAzureErrorMarkers = []string{
 	"ContainerGroupQuotaReached",
 }
 
-// isTransientAzureError returns true if the error is a known transient Azure
-// error that may succeed on retry. It delegates to step.ErrorContainsAny, which
-// flattens the nested ARM error details rad surfaces inside a CLIError so the
-// match covers causes such as the ACI container group quota error.
-func isTransientAzureError(err error) bool {
-	return step.ErrorContainsAny(err, transientAzureErrorMarkers...)
+// isTransientCloudDeployError extends the default deployment predicate with
+// Azure-specific failures. Custom retry configuration must retain the default
+// image-pull and control-plane recovery behavior.
+func isTransientCloudDeployError(err error) bool {
+	return step.IsTransientDeployError(err) ||
+		step.ErrorContainsAny(err, transientAzureErrorMarkers...)
 }
 
-func Test_isTransientAzureError(t *testing.T) {
+func Test_isTransientCloudDeployError(t *testing.T) {
 	// aciQuotaError mirrors how rad surfaces an ACI quota failure: the quota
 	// error code only appears inside a deeply nested details[].message field,
 	// while the top-level code/message returned by CLIError.Error() is the
@@ -258,13 +259,15 @@ func Test_isTransientAzureError(t *testing.T) {
 		{name: "nested ACI quota error", err: aciQuotaError, expected: true},
 		{name: "nested managed identity error", err: msiError, expected: true},
 		{name: "plain transient error string", err: errors.New("deployment failed: ManagedServiceIdentityNotFound"), expected: true},
+		{name: "apiserver connection reset", err: errors.New("read: connection reset by peer"), expected: true},
+		{name: "apiserver restart response", err: errors.New("before all known HTTP paths have been installed"), expected: true},
 		{name: "non-transient CLIError", err: nonTransientError, expected: false},
 		{name: "unrelated error", err: errors.New("connection refused"), expected: false},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			require.Equal(t, tc.expected, isTransientAzureError(tc.err))
+			require.Equal(t, tc.expected, isTransientCloudDeployError(tc.err))
 		})
 	}
 }
diff --git a/test/functional-portable/corerp/cloud/resources/radiuscore_azure_mysql_portallink_test.go b/test/functional-portable/corerp/cloud/resources/radiuscore_azure_mysql_portallink_test.go
index 4763fc45dc..eb6a2a98ce 100644
--- a/test/functional-portable/corerp/cloud/resources/radiuscore_azure_mysql_portallink_test.go
+++ b/test/functional-portable/corerp/cloud/resources/radiuscore_azure_mysql_portallink_test.go
@@ -127,7 +127,7 @@ func Test_RadiusCore_AzureMySql_PortalLink(t *testing.T) {
 				"azureSubscriptionId="+azureSubscriptionID,
 				"azureResourceGroupName="+azureResourceGroupName,
 				"password=not-prod-password",
-			).WithRetry(2, 60*time.Second, isTransientAzureError),
+			).WithRetry(5*time.Minute, 60*time.Second, isTransientCloudDeployError),
 			RPResources: &validation.RPResourceSet{
 				Resources: []validation.RPResource{
 					{Name: recipePackName, Type: validation.CoreRecipePacksResource},
diff --git a/test/step/controlplane.go b/test/step/controlplane.go
new file mode 100644
index 0000000000..98ed8c32fa
--- /dev/null
+++ b/test/step/controlplane.go
@@ -0,0 +1,101 @@
+/*
+Copyright 2023 The Radius Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package step
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	k8s "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+const (
+	// controlPlaneReadyPollInterval is how long to wait between readiness
+	// probes while the control plane is recovering.
+	controlPlaneReadyPollInterval = 5 * time.Second
+
+	// controlPlaneProbeTimeout bounds a single readiness probe so a hung
+	// kube-apiserver cannot consume the caller's entire retry budget in one
+	// request.
+	controlPlaneProbeTimeout = 10 * time.Second
+
+	// radiusAggregatedAPIPath is the aggregated API discovery path that rad
+	// itself requests when it opens a workspace connection (see
+	// pkg/cli/workspaces.Connection). Probing it exercises the whole chain a
+	// deployment depends on: the kube-apiserver, its aggregation layer, and the
+	// UCP APIService behind it. A healthy response here implies the
+	// kube-apiserver is serving, so no separate /readyz probe is needed.
+	radiusAggregatedAPIPath = "/apis/api.ucp.dev/v1alpha3"
+)
+
+// newControlPlaneReadyWaiter returns a function that blocks until the Radius
+// aggregated API is serving again, or until ctx is done.
+//
+// It exists because a fixed sleep is not a useful gate between deployment
+// retries. When the kind control plane goes down, every queued retry fails
+// instantly against a dead socket and burns the retry budget in seconds while
+// the outage lasts minutes. Waiting for readiness instead makes each retry
+// attempt meaningful.
+//
+// It returns nil when no client is available, which disables the gate and
+// leaves the retry loop with its delay-only behavior.
+func newControlPlaneReadyWaiter(t *testing.T, client k8s.Interface) func(context.Context) error {
+	if client == nil {
+		return nil
+	}
+
+	restClient := client.Discovery().RESTClient()
+	if restClient == nil {
+		return nil
+	}
+
+	return func(ctx context.Context) error {
+		return waitForControlPlaneReady(ctx, t, restClient, controlPlaneReadyPollInterval)
+	}
+}
+
+// waitForControlPlaneReady polls until the control plane is ready or ctx is
+// done. The error returned on timeout includes the last probe failure so the
+// test log records what was still unavailable.
+func waitForControlPlaneReady(ctx context.Context, t *testing.T, client rest.Interface, pollInterval time.Duration) error {
+	for {
+		lastErr := probe(ctx, client, radiusAggregatedAPIPath)
+		if lastErr == nil {
+			return nil
+		}
+
+		t.Logf("waiting for the Radius control plane to become ready: %v", lastErr)
+
+		timer := time.NewTimer(pollInterval)
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			timer.Stop()
+			return fmt.Errorf("control plane did not become ready: %w (last probe failure: %v)", ctx.Err(), lastErr)
+		}
+	}
+}
+
+func probe(ctx context.Context, client rest.Interface, path string) error {
+	probeCtx, cancel := context.WithTimeout(ctx, controlPlaneProbeTimeout)
+	defer cancel()
+
+	return client.Get().AbsPath(path).Do(probeCtx).Error()
+}
diff --git a/test/step/controlplane_test.go b/test/step/controlplane_test.go
new file mode 100644
index 0000000000..2c9bf4507c
--- /dev/null
+++ b/test/step/controlplane_test.go
@@ -0,0 +1,107 @@
+/*
+Copyright 2023 The Radius Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package step
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+// newTestRESTClient returns a REST client pointed at a test HTTP server that
+// serves the given handler.
+func newTestRESTClient(t *testing.T, handler http.Handler) rest.Interface {
+	t.Helper()
+
+	server := httptest.NewServer(handler)
+	t.Cleanup(server.Close)
+
+	client, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
+	require.NoError(t, err)
+
+	return client.Discovery().RESTClient()
+}
+
+func Test_WaitForControlPlaneReady_ReturnsWhenAggregatedAPIServes(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	var paths []string
+	client := newTestRESTClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		paths = append(paths, r.URL.Path)
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+
+	require.NoError(t, waitForControlPlaneReady(t.Context(), t, client, time.Millisecond))
+
+	// A single probe of the aggregated path is enough: it cannot succeed unless
+	// the kube-apiserver and its aggregation layer are both serving.
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, []string{radiusAggregatedAPIPath}, paths)
+}
+
+func Test_WaitForControlPlaneReady_PollsUntilHealthy(t *testing.T) {
+	t.Parallel()
+	var requests atomic.Int32
+	// The kube-apiserver can be up while the UCP APIService behind the
+	// aggregation layer is still unavailable, which surfaces as a 503. The gate
+	// must keep polling rather than let a retry proceed in that window.
+	client := newTestRESTClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if requests.Add(1) < 3 {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+
+	require.NoError(t, waitForControlPlaneReady(ctx, t, client, time.Millisecond))
+	assert.Equal(t, int32(3), requests.Load())
+}
+
+func Test_WaitForControlPlaneReady_ReportsLastProbeFailureOnTimeout(t *testing.T) {
+	t.Parallel()
+	client := newTestRESTClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+
+	ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+	defer cancel()
+
+	err := waitForControlPlaneReady(ctx, t, client, time.Millisecond)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Contains(t, err.Error(), "last probe failure")
+}
+
+func Test_NewControlPlaneReadyWaiter_NilClientDisablesGate(t *testing.T) {
+	t.Parallel()
+	assert.Nil(t, newControlPlaneReadyWaiter(t, nil))
+}
diff --git a/test/step/deployexecutor.go b/test/step/deployexecutor.go
index d614febae7..a0e697fd63 100644
--- a/test/step/deployexecutor.go
+++ b/test/step/deployexecutor.go
@@ -46,36 +46,43 @@ type DeployExecutor struct {
 	// the environment is not defined in bicep.
 	Environment string
 
-	// MaxRetries is the maximum number of retry attempts after the initial deployment fails.
-	// Zero means no retries (default behavior).
-	MaxRetries int
-
-	// RetryDelay is the base duration to wait between retry attempts. The wait
-	// doubles after each attempt, up to maxTransientRetryDelay.
+	// RetryDelay is the minimum time to wait between retry attempts.
 	RetryDelay time.Duration
 
+	// RetryBudget bounds how long the executor keeps starting new attempts,
+	// measured from the moment the first attempt fails. Once it elapses no
+	// further attempt begins, but an attempt already in flight runs to
+	// completion under the caller's context. Zero disables retries.
+	RetryBudget time.Duration
+
 	// ShouldRetry is a predicate that determines whether a failed deployment should be retried.
-	// If nil, no retries are attempted regardless of MaxRetries.
+	// If nil, no retries are attempted regardless of RetryBudget.
 	ShouldRetry func(error) bool
+
+	// waitForReady, when set, blocks before each retry until the Radius control
+	// plane is serving again. Execute populates it from the test's Kubernetes
+	// client; a nil value falls back to RetryDelay alone.
+	waitForReady func(context.Context) error
 }
 
 // Default retry behavior applied by NewDeployExecutor for transient deployment
-// failures. Functional tests hit two classes of environmental flake in CI:
-// container image pulls from shared registries (for example the
-// ghcr.io/radius-project/* images) that occasionally fail due to registry or
-// network blips, and UCP connection resets/EOFs when the kind control-plane
-// restarts under runner resource pressure and drops the port-forward tunnel.
-// Callers can override these defaults with WithRetry.
+// failures: image pull blips from shared registries, and the UCP connection
+// resets, EOFs and restart-phase HTTP errors produced when the kind control
+// plane restarts under runner pressure.
 //
-// The delay doubles per attempt, so the defaults give a recovery window of
-// 30s + 60s + 120s = 3m30s. A fixed 30s delay left the budget exhausted while a
-// restarted kube-apiserver was still initializing (see
-// transientAPIServerRestartErrorMarkers), and only transient-classified failures
-// wait at all - a genuine deployment failure still fails on the first attempt.
+// The bound is wall-clock time rather than an attempt count because a
+// control-plane outage makes every attempt fail instantly against a dead
+// socket, so "2 retries, 30s apart" is spent in about a minute while the outage
+// lasts several. Waiting for readiness between attempts (see waitForReady)
+// spends that budget on recovery instead of on attempts that cannot succeed.
 const (
-	defaultTransientMaxRetries = 3
 	defaultTransientRetryDelay = 30 * time.Second
-	maxTransientRetryDelay     = 2 * time.Minute
+
+	// defaultTransientRetryBudget covers the longest control-plane outage
+	// observed in CI (about 3.5 minutes between the first etcd request timeout
+	// and the kube-apiserver serving again) with a little margin. It is an upper
+	// bound: effectiveRetryBudget shortens it to fit the test binary's deadline.
+	defaultTransientRetryBudget = 5 * time.Minute
 )
 
 // transientImagePullErrorMarkers are substrings that indicate a container image
@@ -110,31 +117,31 @@ func IsTransientImagePullError(err error) bool {
 // failed because the connection between rad and the UCP API server was reset or
 // closed mid-request, rather than because the deployment itself was invalid.
 //
-// In CI the Radius workspace reaches UCP through a local `kubectl port-forward`
-// tunnel that proxies through the kind cluster's kube-apiserver. When the
-// GitHub-hosted runner is under resource pressure the kind static control-plane
-// pods (kube-apiserver/controller-manager/scheduler) restart, which drops every
-// in-flight port-forward tunnel at once and resets all parallel `rad deploy`
-// connections simultaneously. UCP and the Radius pods do not crash, so
-// re-running the deployment once the control-plane recovers typically succeeds.
+// rad reaches UCP over the aggregated API path (/apis/api.ucp.dev/v1alpha3/...)
+// using the connection built from the test kubeconfig, which for kind points at
+// the apiserver port published on 127.0.0.1. When the control plane goes down -
+// observed in CI as etcd failing to serve requests, followed a few minutes
+// later by the kube-apiserver restarting - every in-flight connection is reset
+// at once, so all parallel `rad deploy` invocations fail together. UCP and the
+// Radius pods do not crash, so re-running the deployment once the control plane
+// recovers typically succeeds.
 var transientConnectionErrorMarkers = []string{
-	// The socket to the port-forward tunnel was reset when the kube-apiserver
-	// (which the tunnel proxies through) bounced, e.g.
+	// The socket to the apiserver was reset when it bounced, e.g.
 	// `read tcp 127.0.0.1:38764->127.0.0.1:37481: read: connection reset by peer`.
 	"connection reset by peer",
-	// rad's HTTP client observed a clean close of the tunnel mid-response, e.g.
+	// rad's HTTP client observed a clean close mid-response, e.g.
 	// `Get "https://127.0.0.1:37481/.../operationStatuses/...": EOF`.
 	": EOF",
 	// The pod log-stream tailers and larger response bodies surface this variant
-	// when the tunnel closes partway through a read.
+	// when the connection closes partway through a read.
 	"unexpected EOF",
-	// The write side of the tunnel was torn down while rad was still sending.
+	// The write side was torn down while rad was still sending.
 	"broken pipe",
 }
 
 // IsTransientConnectionError reports whether err was caused by a transient
 // network disruption between rad and the UCP API server (a reset or closed
-// port-forward tunnel) that is likely to succeed on retry. See
+// connection) that is likely to succeed on retry. See
 // transientConnectionErrorMarkers for the environmental root cause.
 //
 // A connection reset/EOF is a transport-level failure that rad surfaces as a
@@ -209,16 +216,16 @@ func IsTransientDeployError(err error) bool {
 // NewDeployExecutor creates a new DeployExecutor instance with the given template and parameters.
 //
 // By default the executor retries a deployment that fails with a transient
-// error - either a container image pull blip or a UCP connection reset/EOF (see
-// IsTransientDeployError). Use WithRetry to override the retry count, delay, and
-// predicate.
+// error (see IsTransientDeployError) for up to defaultTransientRetryBudget,
+// waiting for the control plane to become ready again before each attempt. Use
+// WithRetry to override the budget, delay, and predicate.
 func NewDeployExecutor(template string, parameters ...string) *DeployExecutor {
 	return &DeployExecutor{
 		Description: fmt.Sprintf("deploy %s", template),
 		Template:    template,
 		Parameters:  parameters,
-		MaxRetries:  defaultTransientMaxRetries,
 		RetryDelay:  defaultTransientRetryDelay,
+		RetryBudget: defaultTransientRetryBudget,
 		ShouldRetry: IsTransientDeployError,
 	}
 }
@@ -236,13 +243,11 @@ func (d *DeployExecutor) WithEnvironment(environment string) *DeployExecutor {
 }
 
 // WithRetry configures retry behavior for transient deployment failures,
-// replacing the default transient image pull retry set by NewDeployExecutor.
-// maxRetries is the number of additional attempts after the first failure.
-// delay is the base wait time between attempts, doubling per attempt up to
-// maxTransientRetryDelay. shouldRetry determines whether a given error is
-// eligible for retry.
-func (d *DeployExecutor) WithRetry(maxRetries int, delay time.Duration, shouldRetry func(error) bool) *DeployExecutor {
-	d.MaxRetries = maxRetries
+// replacing the default set by NewDeployExecutor. budget bounds how long new
+// attempts keep being started, delay is the minimum wait between them, and
+// shouldRetry decides which errors are eligible.
+func (d *DeployExecutor) WithRetry(budget time.Duration, delay time.Duration, shouldRetry func(error) bool) *DeployExecutor {
+	d.RetryBudget = budget
 	d.RetryDelay = delay
 	d.ShouldRetry = shouldRetry
 	return d
@@ -262,6 +267,10 @@ func (d *DeployExecutor) Execute(ctx context.Context, t *testing.T, options test
 	t.Logf("deploying %s from file %s", d.Description, d.Template)
 	cli := radcli.NewCLI(t, options.ConfigFilePath)
 
+	if d.waitForReady == nil && options.K8sClient != nil {
+		d.waitForReady = newControlPlaneReadyWaiter(t, options.K8sClient)
+	}
+
 	deployFunc := func() error {
 		return cli.Deploy(ctx, templateFilePath, d.Environment, d.Application, d.Parameters...)
 	}
@@ -271,57 +280,100 @@ func (d *DeployExecutor) Execute(ctx context.Context, t *testing.T, options test
 	t.Logf("finished deploying %s from file %s", d.Description, d.Template)
 }
 
-// executeWithRetry runs the deploy function with optional retry logic.
+// executeWithRetry runs the deploy function, retrying transient failures until
+// the retry budget is spent or the context is cancelled.
+//
+// The bound is wall-clock time because a control-plane outage makes a
+// deployment fail immediately rather than slowly, so a fixed attempt count is
+// consumed long before the control plane returns. Between attempts the loop
+// waits for the delay and then for readiness, spending the budget on recovery
+// rather than on attempts that cannot succeed.
 func (d *DeployExecutor) executeWithRetry(ctx context.Context, t *testing.T, deployFunc func() error) error {
-	maxAttempts := 1
-	if d.MaxRetries > 0 && d.ShouldRetry != nil {
-		maxAttempts = d.MaxRetries + 1
+	err := deployFunc()
+	if err == nil || d.ShouldRetry == nil {
+		return err
 	}
 
-	var lastErr error
-	for attempt := 1; attempt <= maxAttempts; attempt++ {
-		if attempt > 1 {
-			delay := d.retryDelayForAttempt(attempt)
-			t.Logf("waiting %s before retry attempt %d/%d", delay, attempt, maxAttempts)
-			timer := time.NewTimer(delay)
-			select {
-			case <-timer.C:
-			case <-ctx.Done():
-				timer.Stop()
-				lastErr = ctx.Err()
-			}
+	budget := d.effectiveRetryBudget(t)
+	if budget <= 0 {
+		return err
+	}
+
+	retryCtx, cancel := context.WithTimeout(ctx, budget)
+	defer cancel()
+
+	for attempt := 1; d.ShouldRetry(err); attempt++ {
+		t.Logf("deployment attempt %d failed with a retryable error: %v", attempt, err)
+
+		if waitErr := d.waitBeforeRetry(retryCtx, t); waitErr != nil {
+			// A cancelled parent context is a caller-driven abort and is
+			// reported as such. Exhausting the budget is not: the deployment
+			// error is the meaningful failure to surface.
 			if ctx.Err() != nil {
-				break
+				return ctx.Err()
 			}
-		}
 
-		lastErr = deployFunc()
-		if lastErr == nil {
-			break
+			t.Logf("no longer retrying the deployment after %d attempts: %v", attempt, waitErr)
+			return err
 		}
 
-		if attempt == maxAttempts || !d.ShouldRetry(lastErr) {
-			break
+		if err = deployFunc(); err == nil {
+			return nil
 		}
+	}
+
+	return err
+}
 
-		t.Logf("deployment attempt %d/%d failed with retryable error: %v", attempt, maxAttempts, lastErr)
+// effectiveRetryBudget returns RetryBudget shortened so that retrying can never
+// run the test binary out of time.
+//
+// The budget is per deploy step, so a control plane that never recovers makes
+// every test pay it. The suites run in parallel batches - corerp-noncloud is 52
+// tests at -parallel 10 - and their `go test -timeout` is as low as 15 minutes
+// (FUNCTIONALTEST_TIMEOUT in functional-test-noncloud.yaml, versus 30 minutes
+// for the cloud suites). A fixed budget large enough to cover a multi-minute
+// outage in one suite would blow the deadline in another, replacing clear
+// per-test failures with an opaque panic and risking the loss of the
+// diagnostics collected on failure.
+//
+// Spending at most half the remaining time leaves each subsequent batch half of
+// what is left, so the deadline is approached but never reached, without
+// needing a per-suite constant.
+func (d *DeployExecutor) effectiveRetryBudget(t *testing.T) time.Duration {
+	deadline, ok := t.Deadline()
+	if !ok {
+		return d.RetryBudget
 	}
 
-	return lastErr
+	if half := time.Until(deadline) / 2; half < d.RetryBudget {
+		return half
+	}
+
+	return d.RetryBudget
 }
 
-// retryDelayForAttempt returns how long to wait before the given attempt, which
-// is always 2 or greater. The wait starts at RetryDelay and doubles per attempt
-// so the retry budget spans a control-plane restart, clamped to
-// maxTransientRetryDelay. A caller-supplied delay is never shortened.
-func (d *DeployExecutor) retryDelayForAttempt(attempt int) time.Duration {
-	// Taking the max keeps a caller-supplied delay that already exceeds the cap.
-	limit := max(d.RetryDelay, maxTransientRetryDelay)
-
-	delay := d.RetryDelay
-	for i := 2; i < attempt && delay < limit; i++ {
-		delay *= 2
+// waitBeforeRetry waits out the retry delay and then blocks until the control
+// plane is ready, bounded by ctx.
+func (d *DeployExecutor) waitBeforeRetry(ctx context.Context, t *testing.T) error {
+	if d.RetryDelay > 0 {
+		timer := time.NewTimer(d.RetryDelay)
+		defer timer.Stop()
+
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+
+	if d.waitForReady == nil {
+		// ctx.Err() is nil while budget remains. Returning it rather than a
+		// bare nil matters when RetryDelay is also zero: with neither a delay
+		// nor a readiness gate to block on, nothing else would consult the
+		// deadline and the loop would keep starting attempts past the budget.
+		return ctx.Err()
 	}
 
-	return min(delay, limit)
+	return d.waitForReady(ctx)
 }
diff --git a/test/step/deployexecutor_test.go b/test/step/deployexecutor_test.go
index c8d1e28e10..e28369a818 100644
--- a/test/step/deployexecutor_test.go
+++ b/test/step/deployexecutor_test.go
@@ -33,7 +33,7 @@ import (
 func Test_ExecuteWithRetry_SucceedsOnFirstAttempt(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
-	d := NewDeployExecutor("test.bicep").WithRetry(2, 10*time.Millisecond, func(error) bool { return true })
+	d := NewDeployExecutor("test.bicep").WithRetry(time.Second, 10*time.Millisecond, func(error) bool { return true })
 
 	err := d.executeWithRetry(t.Context(), t, func() error {
 		calls.Add(1)
@@ -48,7 +48,7 @@ func Test_ExecuteWithRetry_RetriesOnTransientThenSucceeds(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
 	transientErr := errors.New("ManagedServiceIdentityNotFound")
-	d := NewDeployExecutor("test.bicep").WithRetry(2, 10*time.Millisecond, func(err error) bool {
+	d := NewDeployExecutor("test.bicep").WithRetry(time.Second, 10*time.Millisecond, func(err error) bool {
 		return err.Error() == "ManagedServiceIdentityNotFound"
 	})
 
@@ -67,7 +67,7 @@ func Test_ExecuteWithRetry_RetriesOnTransientThenSucceeds(t *testing.T) {
 func Test_ExecuteWithRetry_DoesNotRetryNonTransientError(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
-	d := NewDeployExecutor("test.bicep").WithRetry(2, 10*time.Millisecond, func(err error) bool {
+	d := NewDeployExecutor("test.bicep").WithRetry(time.Second, 10*time.Millisecond, func(err error) bool {
 		return err.Error() == "transient"
 	})
 
@@ -81,10 +81,10 @@ func Test_ExecuteWithRetry_DoesNotRetryNonTransientError(t *testing.T) {
 	assert.Equal(t, int32(1), calls.Load())
 }
 
-func Test_ExecuteWithRetry_ExhaustsAllRetries(t *testing.T) {
+func Test_ExecuteWithRetry_StopsWhenBudgetIsExhausted(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
-	d := NewDeployExecutor("test.bicep").WithRetry(2, 10*time.Millisecond, func(error) bool { return true })
+	d := NewDeployExecutor("test.bicep").WithRetry(50*time.Millisecond, 10*time.Millisecond, func(error) bool { return true })
 
 	err := d.executeWithRetry(t.Context(), t, func() error {
 		calls.Add(1)
@@ -93,7 +93,9 @@ func Test_ExecuteWithRetry_ExhaustsAllRetries(t *testing.T) {
 
 	require.Error(t, err)
 	assert.Equal(t, "always fails", err.Error())
-	assert.Equal(t, int32(3), calls.Load()) // 1 initial + 2 retries
+	// The initial attempt plus at least one retry, and the loop must have
+	// stopped once the budget elapsed rather than run forever.
+	assert.Greater(t, calls.Load(), int32(1))
 }
 
 func Test_ExecuteWithRetry_DefaultDoesNotRetryNonTransientError(t *testing.T) {
@@ -135,8 +137,9 @@ func Test_ExecuteWithRetry_DefaultRetriesTransientImagePullError(t *testing.T) {
 func Test_ExecuteWithRetry_DefaultRetriesTransientConnectionError(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
-	// NewDeployExecutor retries transient UCP connection resets by default, which
-	// occur when the kind control-plane restarts and drops the port-forward tunnel.
+	// NewDeployExecutor retries transient UCP connection resets by default,
+	// which occur when the kind control plane restarts and drops every
+	// in-flight connection to the API server.
 	d := NewDeployExecutor("test.bicep")
 	d.RetryDelay = 10 * time.Millisecond // shorten the delay for the test
 
@@ -183,7 +186,7 @@ func Test_ExecuteWithRetry_ContextCancelledDuringDelay(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
 	ctx, cancel := context.WithCancel(t.Context())
-	d := NewDeployExecutor("test.bicep").WithRetry(2, 5*time.Second, func(error) bool { return true })
+	d := NewDeployExecutor("test.bicep").WithRetry(time.Minute, 5*time.Second, func(error) bool { return true })
 
 	// Cancel context immediately after first deploy attempt
 	err := d.executeWithRetry(ctx, t, func() error {
@@ -204,7 +207,6 @@ func Test_ExecuteWithRetry_NilShouldRetryDisablesRetries(t *testing.T) {
 	t.Parallel()
 	var calls atomic.Int32
 	d := NewDeployExecutor("test.bicep")
-	d.MaxRetries = 3
 	d.ShouldRetry = nil // nil predicate
 
 	err := d.executeWithRetry(t.Context(), t, func() error {
@@ -216,37 +218,163 @@ func Test_ExecuteWithRetry_NilShouldRetryDisablesRetries(t *testing.T) {
 	assert.Equal(t, int32(1), calls.Load())
 }
 
-func Test_RetryDelayForAttempt(t *testing.T) {
+func Test_ExecuteWithRetry_ZeroBudgetDisablesRetries(t *testing.T) {
 	t.Parallel()
-	tests := []struct {
-		name     string
-		base     time.Duration
-		attempt  int
-		expected time.Duration
-	}{
-		{name: "first retry uses the base delay", base: 30 * time.Second, attempt: 2, expected: 30 * time.Second},
-		{name: "delay doubles per attempt", base: 30 * time.Second, attempt: 3, expected: 60 * time.Second},
-		{name: "delay reaches the cap", base: 30 * time.Second, attempt: 4, expected: maxTransientRetryDelay},
-		{name: "delay stops growing at the cap", base: 30 * time.Second, attempt: 9, expected: maxTransientRetryDelay},
-		// A base above half the cap makes the next doubling overshoot it, so the
-		// result has to be clamped rather than just left ungrown.
-		{name: "doubling overshoots the cap", base: 90 * time.Second, attempt: 3, expected: maxTransientRetryDelay},
-		{name: "overshooting base stays at the cap", base: 90 * time.Second, attempt: 9, expected: maxTransientRetryDelay},
-		// A caller that asks for a longer delay than the cap keeps it: the cap only
-		// bounds the doubling, it never shortens a caller-supplied delay.
-		{name: "caller delay above the cap is preserved", base: 5 * time.Minute, attempt: 3, expected: 5 * time.Minute},
-		{name: "zero delay stays zero", base: 0, attempt: 5, expected: 0},
+	var calls atomic.Int32
+	d := NewDeployExecutor("test.bicep")
+	d.RetryDelay = 0
+	d.RetryBudget = 0
+	d.ShouldRetry = func(error) bool { return true }
+
+	err := d.executeWithRetry(t.Context(), t, func() error {
+		calls.Add(1)
+		return errors.New("transient")
+	})
+
+	require.Error(t, err)
+	assert.Equal(t, int32(1), calls.Load())
+}
+
+func Test_ExecuteWithRetry_UnreadyControlPlaneStopsRetrying(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+	var waits atomic.Int32
+
+	d := NewDeployExecutor("test.bicep")
+	d.RetryDelay = time.Millisecond
+	d.ShouldRetry = func(error) bool { return true }
+	// The gate stands in for a control plane that never came back within the
+	// budget. Retrying against it cannot succeed, so the loop must stop.
+	d.waitForReady = func(ctx context.Context) error {
+		waits.Add(1)
+		return errors.New("kube-apiserver is not ready")
 	}
 
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-			d := NewDeployExecutor("test.bicep")
-			d.RetryDelay = tc.base
+	err := d.executeWithRetry(t.Context(), t, func() error {
+		n := calls.Add(1)
+		if n == 1 {
+			return errors.New("connection reset by peer")
+		}
+		return nil
+	})
 
-			require.Equal(t, tc.expected, d.retryDelayForAttempt(tc.attempt))
-		})
+	// The deployment error is surfaced rather than the gate's error, and no
+	// retry is attempted against an unavailable control plane.
+	require.Error(t, err)
+	assert.Equal(t, "connection reset by peer", err.Error())
+	assert.Equal(t, int32(1), calls.Load())
+	assert.Equal(t, int32(1), waits.Load())
+}
+
+func Test_ExecuteWithRetry_RetriesOnceControlPlaneIsReady(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+	var waits atomic.Int32
+
+	d := NewDeployExecutor("test.bicep")
+	d.RetryDelay = time.Millisecond
+	d.ShouldRetry = func(error) bool { return true }
+	d.waitForReady = func(ctx context.Context) error {
+		waits.Add(1)
+		return nil
 	}
+
+	err := d.executeWithRetry(t.Context(), t, func() error {
+		n := calls.Add(1)
+		if n == 1 {
+			return errors.New("connection reset by peer")
+		}
+		return nil
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, int32(2), calls.Load())
+	assert.Equal(t, int32(1), waits.Load())
+}
+
+func Test_ExecuteWithRetry_BudgetBoundsTheReadinessWait(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+
+	d := NewDeployExecutor("test.bicep")
+	d.RetryDelay = 0
+	d.RetryBudget = 50 * time.Millisecond
+	d.ShouldRetry = func(error) bool { return true }
+	// A control plane that never recovers must not block past the budget.
+	d.waitForReady = func(ctx context.Context) error {
+		<-ctx.Done()
+		return ctx.Err()
+	}
+
+	start := time.Now()
+	err := d.executeWithRetry(t.Context(), t, func() error {
+		calls.Add(1)
+		return errors.New("connection reset by peer")
+	})
+
+	require.Error(t, err)
+	assert.Equal(t, "connection reset by peer", err.Error())
+	assert.Equal(t, int32(1), calls.Load())
+	assert.Less(t, time.Since(start), 5*time.Second)
+}
+
+func Test_EffectiveRetryBudget_CappedByTestDeadline(t *testing.T) {
+	t.Parallel()
+	deadline, ok := t.Deadline()
+	if !ok {
+		t.Skip("go test was run without -timeout, so there is no deadline to cap against")
+	}
+
+	// A budget far larger than the time the test binary has left must be cut
+	// down, so a broken cluster produces per-test failures rather than an
+	// opaque `go test -timeout` panic.
+	d := NewDeployExecutor("test.bicep")
+	d.RetryBudget = time.Hour
+
+	halfRemainingBefore := time.Until(deadline) / 2
+	budget := d.effectiveRetryBudget(t)
+	halfRemainingAfter := time.Until(deadline) / 2
+
+	assert.LessOrEqual(t, budget, halfRemainingBefore)
+	assert.GreaterOrEqual(t, budget, halfRemainingAfter)
+	assert.Positive(t, budget)
+}
+
+func Test_EffectiveRetryBudget_UnchangedWhenItFitsTheDeadline(t *testing.T) {
+	t.Parallel()
+	d := NewDeployExecutor("test.bicep")
+	d.RetryBudget = time.Millisecond
+
+	assert.Equal(t, time.Millisecond, d.effectiveRetryBudget(t))
+}
+
+func Test_ExecuteWithRetry_BudgetBoundsRetriesWithoutDelayOrReadinessGate(t *testing.T) {
+	t.Parallel()
+	var calls atomic.Int32
+
+	// With neither a retry delay nor a readiness gate, the budget is the only
+	// thing that can stop the loop, so waitBeforeRetry has to consult it on
+	// that path too. Without this the loop runs until the parent context or the
+	// test binary's timeout, not the budget.
+	d := NewDeployExecutor("test.bicep")
+	d.RetryDelay = 0
+	d.RetryBudget = 20 * time.Millisecond
+	d.ShouldRetry = func(error) bool { return true }
+	d.waitForReady = nil
+
+	start := time.Now()
+	err := d.executeWithRetry(t.Context(), t, func() error {
+		calls.Add(1)
+		// Keeps the attempt count (and the log volume) bounded while the
+		// budget elapses.
+		time.Sleep(5 * time.Millisecond)
+		return errors.New("connection reset by peer")
+	})
+
+	require.Error(t, err)
+	assert.Equal(t, "connection reset by peer", err.Error())
+	assert.Less(t, time.Since(start), 5*time.Second)
+	assert.Greater(t, calls.Load(), int32(1))
 }
 
 func Test_IsTransientImagePullError(t *testing.T) {
@@ -341,7 +469,7 @@ func Test_IsTransientConnectionError(t *testing.T) {
 	}{
 		{name: "nil error", err: nil, expected: false},
 		{
-			name:     "port-forward connection reset",
+			name:     "apiserver connection reset",
 			err:      errors.New(`Get "https://127.0.0.1:37481/.../operationStatuses/...": read tcp 127.0.0.1:38764->127.0.0.1:37481: read: connection reset by peer`),
 			expected: true,
 		},