diff --git a/docs/concepts.md b/docs/concepts.md
index 05e9e197..673d21b5 100644
--- a/docs/concepts.md
+++ b/docs/concepts.md
@@ -125,6 +125,12 @@ Defines how Drained versions are cleaned up:
 - **scaledownDelay**: How long to wait after a version has been Drained before scaling pods to zero
 - **deleteDelay**: How long to wait after a version has been Drained before deleting the Kubernetes `Deployment`
 
+> **NOTE**: Versions superseded before ever becoming Current or Ramping remain Inactive in Temporal;
+> they never acquire a drainage timestamp. The controller retires these versions after
+> their Deployment has fully scaled to zero, visibility reports no running pinned workflows,
+> and Temporal accepts the normal version deletion request. The drainage-based sunset delays
+> do not apply to these unused versions.
+
 ### Template
 The pod template used for the target version of this worker deployment. Similar to the pod template used in a standar Kubernetes `Deployment`, but managed by the controller.
 
diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go
index 0dac93d9..3de22605 100644
--- a/internal/controller/execplan.go
+++ b/internal/controller/execplan.go
@@ -21,6 +21,7 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/api/workflowservice/v1"
 	sdkclient "go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/converter"
 	"go.temporal.io/sdk/worker"
@@ -664,13 +665,20 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations(
 	return errors.Join(append(applyErrs, statusErrs...)...)
 }
 
-// deleteDrainedVersions prunes the Temporal server-side Worker Deployment Version
+// deleteDeprecatedVersions prunes the Temporal server-side Worker Deployment Version
 // record for each k8s Deployment in DeleteDeployments, before executeK8sOperations
 // deletes them. It is mutated to remove the k8s Deployments that should not be
 // deleted because their Temporal server-side WDV record could not be removed or
 // because the k8s Deployment has no build ID label. The removed k8s deployments
 // stay in the cluster so that a later reconcile retries the pruning.
 //
+// Because DeleteVersion does not check to see if there are open workflows using
+// pinned execution for Inactive versions, we query the Temporal visibility service
+// for open pinned workflows before deleting Inactive versions.
+// Like Temporal drainage, visibility is eventually consistent: callers must stop
+// sending new pinned overrides to a version being retired. This is not an atomic
+// exclusion against concurrent workflow starts or override updates.
+//
 // The planner only adds a drained version to DeleteDeployments once it is
 // EligibleForDeletion (see planner.getDeleteDeployments): drained past the sunset
 // delays with no active worker pods. Deleting the Kubernetes Deployment alone would
@@ -689,11 +697,12 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations(
 // phase without reaching back into k8sState. NotRegistered Deployments are also carried
 // in DeleteDeployments; they have no server-side version, so they skip DeleteVersion and
 // are retained for deletion.
-func (r *WorkerDeploymentReconciler) deleteDrainedVersions(
+func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions(
 	ctx context.Context,
 	l logr.Logger,
 	workerDeploy *temporaliov1alpha1.WorkerDeployment,
 	depHandle sdkclient.WorkerDeploymentHandle,
+	temporalClient sdkclient.Client,
 	p *plan,
 ) {
 	identity := getControllerIdentity()
@@ -712,6 +721,19 @@ func (r *WorkerDeploymentReconciler) deleteDrainedVersions(
 			markedForDeletion = append(markedForDeletion, d)
 			continue
 		}
+		if slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool {
+			return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive
+		}) {
+			count, err := getOpenPinnedWorkflowExecutions(ctx, temporalClient, p.WorkerDeploymentName, buildID)
+			if err != nil || count == nil {
+				l.Info("could not confirm inactive version has no running pinned workflows, keeping its Deployment", "buildID", buildID, "error", err)
+				continue
+			}
+			if count.Count != 0 {
+				l.Info("inactive version has running pinned workflows, keeping its Deployment", "buildID", buildID, "count", count.Count)
+				continue
+			}
+		}
 		backoffKey := k8s.ComputeWorkerDeploymentName(workerDeploy) + "/" + buildID
 		if r.skipVersionDelete(backoffKey) {
 			continue
@@ -733,7 +755,7 @@ func (r *WorkerDeploymentReconciler) deleteDrainedVersions(
 			}
 			l.Info("worker deployment version already deleted", "buildID", buildID)
 		} else {
-			l.Info("deleted drained worker deployment version", "buildID", buildID)
+			l.Info("deleted deprecated worker deployment version", "buildID", buildID)
 		}
 		r.noteVersionDeleteSuccess(backoffKey)
 		markedForDeletion = append(markedForDeletion, d)
@@ -741,6 +763,25 @@ func (r *WorkerDeploymentReconciler) deleteDrainedVersions(
 	p.DeleteDeployments = markedForDeletion
 }
 
+func getOpenPinnedWorkflowExecutions(
+	ctx context.Context,
+	temporalClient sdkclient.Client,
+	deploymentName string,
+	buildID string,
+) (*workflowservice.CountWorkflowExecutionsResponse, error) {
+	// Visibility can contain either the legacy dot separator or the newer colon form.
+	legacyVersion := strings.ReplaceAll(deploymentName+"."+buildID, "'", "''")
+	version := strings.ReplaceAll(deploymentName+":"+buildID, "'", "''")
+	qs := fmt.Sprintf(
+		"TemporalWorkerDeploymentVersion IN ('%s', '%s') "+
+			"AND TemporalWorkflowVersioningBehavior = 'Pinned' "+
+			"AND ExecutionStatus = 'Running'",
+		legacyVersion, version,
+	)
+	req := &workflowservice.CountWorkflowExecutionsRequest{Query: qs}
+	return temporalClient.CountWorkflow(ctx, req)
+}
+
 // versionDeleteBackoff returns the per-version DeleteVersion backoff.
 func (r *WorkerDeploymentReconciler) versionDeleteBackoff() *flowcontrol.Backoff {
 	r.deleteBackoffOnce.Do(func() {
@@ -795,8 +836,8 @@ func (r *WorkerDeploymentReconciler) executePlan(
 	// Prune the Temporal server-side version records before their k8s Deployments are
 	// deleted, and narrow the plan to the versions the server confirmed gone. A Deployment
 	// held back here keeps its version nominated for deletion, so a failed deletion is retried
-	// on the next reconcile instead of orphaning the record; see deleteDrainedVersions.
-	r.deleteDrainedVersions(ctx, l, workerDeploy, deploymentHandler, p)
+	// on the next reconcile instead of orphaning the record; see deleteDeprecatedVersions.
+	r.deleteDeprecatedVersions(ctx, l, workerDeploy, deploymentHandler, temporalClient, p)
 	deletedWorkerResources, err := r.executeK8sOperations(ctx, l, workerDeploy, p)
 	if err != nil {
 		return err
diff --git a/internal/controller/execplan_test.go b/internal/controller/execplan_test.go
index ec443f11..9df964db 100644
--- a/internal/controller/execplan_test.go
+++ b/internal/controller/execplan_test.go
@@ -18,6 +18,7 @@ import (
 	"github.com/temporalio/temporal-worker-controller/internal/k8s"
 	"github.com/temporalio/temporal-worker-controller/internal/temporal"
 	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/api/workflowservice/v1"
 	sdkclient "go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/converter"
 	appsv1 "k8s.io/api/apps/v1"
@@ -663,3 +664,84 @@ func TestGeneratePlan_CarriesEncodingAndMessageType(t *testing.T) {
 	require.Equal(t, "my.package.DeployRequest", wf.messageType)
 	require.Equal(t, []byte(`{"service":"checkout"}`), wf.input)
 }
+
+// A visibility failure must retain the Deployment so the next reconciliation can retry.
+type inactivePruneClient struct {
+	*stubTemporalClient
+	response *workflowservice.CountWorkflowExecutionsResponse
+	err      error
+	query    string
+}
+
+func (c *inactivePruneClient) CountWorkflow(_ context.Context, request *workflowservice.CountWorkflowExecutionsRequest) (*workflowservice.CountWorkflowExecutionsResponse, error) {
+	c.query = request.Query
+	return c.response, c.err
+}
+
+func TestExecutePlan_InactiveVersionDeletion(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		response    *workflowservice.CountWorkflowExecutionsResponse
+		countErr    error
+		deleteErr   error
+		wantAttempt bool
+		wantDelete  bool
+	}{
+		{name: "unused version", response: &workflowservice.CountWorkflowExecutionsResponse{}, wantAttempt: true, wantDelete: true},
+		{name: "running pinned workflow", response: &workflowservice.CountWorkflowExecutionsResponse{Count: 1}},
+		{name: "visibility failure", countErr: errors.New("visibility unavailable")},
+		{name: "missing response"},
+		{name: "server rejects deletion", response: &workflowservice.CountWorkflowExecutionsResponse{}, deleteErr: serviceerror.NewFailedPrecondition("active pollers"), wantAttempt: true},
+		{name: "Temporal version already deleted", response: &workflowservice.CountWorkflowExecutionsResponse{}, deleteErr: serviceerror.NewNotFound("version"), wantAttempt: true, wantDelete: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			connection := temporaliov1alpha1.ConnectionSpec{HostPort: "test:7233"}
+			twd := makeExecplanTWD("my-worker", "default")
+			old := makeVersionedDeployment(twd, "old", 0, connection)
+			current := makeVersionedDeployment(twd, "current", 1, connection)
+			wrt := makeExecplanWRT("config", twd)
+			wrt.Spec.Template.Raw = []byte(`{"apiVersion":"v1","kind":"ConfigMap","data":{"test":"present"}}`)
+			oldConfig, oldHash := renderWRT(t, wrt, old, "old", testTemporalNamespace)
+			currentConfig, currentHash := renderWRT(t, wrt, current, "current", testTemporalNamespace)
+			// This resource outlived its Deployment and must still be cleaned up,
+			// even when the old version's Deployment and resources are retained.
+			orphan := makeVersionedDeployment(twd, "orphan", 0, connection)
+			orphanConfig, orphanHash := renderWRT(t, wrt, orphan, "orphan", testTemporalNamespace)
+			wrt.Status.Versions = []temporaliov1alpha1.WorkerResourceTemplateVersionStatus{
+				k8s.WorkerResourceTemplateVersionStatusForBuildID("old", oldConfig.GetName(), 1, oldHash, ""),
+				k8s.WorkerResourceTemplateVersionStatusForBuildID("current", currentConfig.GetName(), 1, currentHash, ""),
+				k8s.WorkerResourceTemplateVersionStatusForBuildID("orphan", orphanConfig.GetName(), 1, orphanHash, ""),
+			}
+			r, _ := newTestReconciler([]client.Object{twd, old, current, wrt, oldConfig, currentConfig, orphanConfig})
+			handle := newPruneStubHandle(tc.deleteErr)
+			c := &inactivePruneClient{stubTemporalClient: newStubTemporalClientWithHandle(handle), response: tc.response, err: tc.countErr}
+			status := statusWithDeprecated("current", current, &temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{
+				BaseWorkerDeploymentVersion: baseVersion("old", old, temporaliov1alpha1.VersionStatusInactive),
+			})
+			twd.Status = status
+			state := &temporal.TemporalWorkerState{Versions: map[string]*temporal.VersionInfo{
+				"old":     {Status: temporaliov1alpha1.VersionStatusInactive},
+				"current": {Status: temporaliov1alpha1.VersionStatusCurrent},
+			}}
+			p, err := r.generatePlan(context.Background(), logr.Discard(), twd, connection, state)
+			require.NoError(t, err)
+			require.NoError(t, r.executePlan(context.Background(), logr.Discard(), twd, c, p))
+			require.Equal(t, tc.wantAttempt, len(handle.deletedVersions) == 1)
+			require.Equal(t, tc.wantDelete, len(p.DeleteDeployments) == 1)
+			require.Equal(t, !tc.wantDelete, deploymentExists(t, r, "default", old.Name))
+			require.True(t, deploymentExists(t, r, "default", current.Name))
+			configErr := r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: oldConfig.GetName()}, &corev1.ConfigMap{})
+			if tc.wantDelete {
+				require.True(t, apierrors.IsNotFound(configErr))
+			} else {
+				require.NoError(t, configErr, "retained version must keep its rendered resources")
+			}
+			require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: currentConfig.GetName()}, &corev1.ConfigMap{}))
+			orphanErr := r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: orphanConfig.GetName()}, &corev1.ConfigMap{})
+			require.True(t, apierrors.IsNotFound(orphanErr), "orphan cleanup must continue when another version is retained")
+			require.Contains(t, c.query, "TemporalWorkerDeploymentVersion IN ('default/my-worker.old', 'default/my-worker:old')")
+			require.Contains(t, c.query, "TemporalWorkflowVersioningBehavior = 'Pinned'")
+			require.Contains(t, c.query, "ExecutionStatus = 'Running'")
+		})
+	}
+}
diff --git a/internal/planner/planner.go b/internal/planner/planner.go
index 53ea4b1d..a42ed1a3 100644
--- a/internal/planner/planner.go
+++ b/internal/planner/planner.go
@@ -744,6 +744,16 @@ func getDeleteDeployments(
 		}
 
 		switch version.Status {
+		case temporaliov1alpha1.VersionStatusInactive:
+			// Superseded versions that never received routed traffic never become Drained.
+			// Wait for scale-down to finish; execution checks pinned workflows before pruning.
+			if foundDeploymentInTemporal && status.TargetVersion.BuildID != version.BuildID &&
+				(status.CurrentVersion == nil || status.CurrentVersion.BuildID != version.BuildID) &&
+				d.Spec.Replicas != nil && *d.Spec.Replicas == 0 &&
+				d.Status.ObservedGeneration >= d.Generation && d.Status.Replicas == 0 &&
+				(d.Status.TerminatingReplicas == nil || *d.Status.TerminatingReplicas == 0) {
+				deleteDeployments = append(deleteDeployments, d)
+			}
 		case temporaliov1alpha1.VersionStatusDrained:
 			// Deleting a deployment is only possible when:
 			// 1. The deployment has been drained for deleteDelay + scaledownDelay.
@@ -754,7 +764,7 @@ func getDeleteDeployments(
 			//    reconcile as the Deployment delete: EligibleForDeletion is only
 			//    computable while the Deployment (and thus this DeprecatedVersions
 			//    entry) still exists, so this is the only point that can reliably
-			//    prune it. See execplan.deleteDrainedVersions.
+			//    prune it. See execplan.deleteDeprecatedVersions.
 			if version.DrainedSince != nil &&
 				(time.Since(version.DrainedSince.Time) > spec.SunsetStrategy.DeleteDelay.Duration+spec.SunsetStrategy.ScaledownDelay.Duration) &&
 				d.Spec.Replicas != nil && *d.Spec.Replicas == 0 &&
diff --git a/internal/planner/planner_test.go b/internal/planner/planner_test.go
index d9105b7b..9352abe9 100644
--- a/internal/planner/planner_test.go
+++ b/internal/planner/planner_test.go
@@ -734,7 +734,7 @@ func TestGetDeleteDeployments(t *testing.T) {
 			// EligibleForDeletion: worker pods have not fully terminated (Status.Replicas > 0),
 			// so versioned pollers may still be registered and the Temporal-side DeleteVersion
 			// would fail. Deleting the Deployment now would strand that server-side version
-			// record with no way to retry (see execplan.deleteDrainedVersions), so hold off.
+			// record with no way to retry (see execplan.deleteDeprecatedVersions), so hold off.
 			name: "drained long enough and scaled to zero in spec, but not eligible for deletion - not deleted",
 			k8sState: &k8s.DeploymentState{
 				Deployments: map[string]*appsv1.Deployment{
@@ -4793,3 +4793,46 @@ func TestGetSunsetScaleDownBuildIDs(t *testing.T) {
 		})
 	}
 }
+
+func TestGetDeleteDeployments_Inactive(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		mutate     func(*appsv1.Deployment, *temporaliov1alpha1.WorkerDeploymentStatus)
+		wantDelete bool
+	}{
+		{name: "superseded and fully scaled down", wantDelete: true},
+		{name: "scale down requested but pods remain", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { d.Status.Replicas = 1 }},
+		{name: "terminating pods remain", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) {
+			n := int32(1)
+			d.Status.TerminatingReplicas = &n
+		}},
+		{name: "scale down not observed", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { d.Generation = 1 }},
+		{name: "replicas unspecified", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { d.Spec.Replicas = nil }},
+		{name: "replicas positive", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { *d.Spec.Replicas = 1 }},
+		{name: "target retained", mutate: func(_ *appsv1.Deployment, s *temporaliov1alpha1.WorkerDeploymentStatus) {
+			s.TargetVersion.BuildID = "old"
+		}},
+		{name: "current retained", mutate: func(_ *appsv1.Deployment, s *temporaliov1alpha1.WorkerDeploymentStatus) {
+			s.CurrentVersion = &temporaliov1alpha1.CurrentWorkerDeploymentVersion{BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{BuildID: "old"}}
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			d := createDeploymentWithDefaultConnectionSpecHash(0)
+			s := &temporaliov1alpha1.WorkerDeploymentStatus{
+				DeprecatedVersions: []*temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{{
+					BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{
+						BuildID: "old", Status: temporaliov1alpha1.VersionStatusInactive,
+						Deployment: &corev1.ObjectReference{Name: "old"},
+					},
+				}},
+			}
+			if tc.mutate != nil {
+				tc.mutate(d, s)
+			}
+			state := &k8s.DeploymentState{Deployments: map[string]*appsv1.Deployment{"old": d}}
+			deleted := getDeleteDeployments(state, s, &temporaliov1alpha1.WorkerDeploymentSpec{}, true)
+			assert.Equal(t, tc.wantDelete, len(deleted) == 1)
+			assert.Empty(t, getDeleteDeployments(state, s, &temporaliov1alpha1.WorkerDeploymentSpec{}, false))
+		})
+	}
+}
diff --git a/internal/tests/internal/deletion_integration_test.go b/internal/tests/internal/deletion_integration_test.go
index 31eb0d5a..3491ea5d 100644
--- a/internal/tests/internal/deletion_integration_test.go
+++ b/internal/tests/internal/deletion_integration_test.go
@@ -49,6 +49,12 @@ func runDeletionTests(
 	t.Run("drained-version-pruned-from-temporal-on-sunset", func(t *testing.T) {
 		testDrainedVersionPrunedOnSunset(t, k8sClient, ts, testNamespace)
 	})
+
+	for _, pinned := range []bool{false, true} {
+		t.Run(fmt.Sprintf("inactive-version-retirement/pinned-%t", pinned), func(t *testing.T) {
+			testInactiveVersionRetirement(t, k8sClient, ts, testNamespace, pinned)
+		})
+	}
 }
 
 // testDeletionSetsCurrentToUnversioned verifies the core fix: when a WD is deleted,
diff --git a/internal/tests/internal/deployment_controller.go b/internal/tests/internal/deployment_controller.go
index 160c9fa2..7cc4cb27 100644
--- a/internal/tests/internal/deployment_controller.go
+++ b/internal/tests/internal/deployment_controller.go
@@ -177,6 +177,7 @@ func scaleDeploymentToZero(t *testing.T, ctx context.Context, k8sClient client.C
 		dep.Status.ReadyReplicas = 0
 		dep.Status.AvailableReplicas = 0
 		dep.Status.UpdatedReplicas = 0
+		dep.Status.ObservedGeneration = dep.Generation
 		return k8sClient.Status().Update(ctx, &dep)
 	}); err != nil {
 		t.Fatalf("failed to zero status of deployment %s: %v", name, err)
diff --git a/internal/tests/internal/inactive_retirement_integration_test.go b/internal/tests/internal/inactive_retirement_integration_test.go
new file mode 100644
index 00000000..d0b6c380
--- /dev/null
+++ b/internal/tests/internal/inactive_retirement_integration_test.go
@@ -0,0 +1,161 @@
+package internal
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
+	"github.com/temporalio/temporal-worker-controller/internal/k8s"
+	"github.com/temporalio/temporal-worker-controller/internal/testhelpers"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/api/workflowservice/v1"
+	sdkclient "go.temporal.io/sdk/client"
+	sdkworker "go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+	"go.temporal.io/server/temporaltest"
+	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// A superseded rollout that was never Current or Ramping stays Inactive forever.
+// Exercise real version registration, visibility, poller expiry and DeleteVersion,
+// including a pinned override on a version which has never received routed traffic.
+func testInactiveVersionRetirement(t *testing.T, k8sClient client.Client, ts *temporaltest.TestServer, namespace string, pinned bool) {
+	ctx := context.Background()
+	name := fmt.Sprintf("inactive-retire-%t", pinned)
+	tc := testhelpers.NewTestCase().WithInput(testhelpers.NewWorkerDeploymentBuilder().
+		WithManualStrategy().WithTargetTemplate("v1.0")).
+		BuildWithValues(name, namespace, ts.GetDefaultNamespace())
+	twd := tc.GetTWD()
+	connection := &temporaliov1alpha1.Connection{
+		ObjectMeta: metav1.ObjectMeta{Name: twd.Spec.WorkerOptions.ConnectionRef.Name, Namespace: namespace},
+		Spec:       temporaliov1alpha1.ConnectionSpec{HostPort: ts.GetFrontendHostPort()},
+	}
+	if err := k8sClient.Create(ctx, connection); err != nil {
+		t.Fatal(err)
+	}
+	if err := k8sClient.Create(ctx, twd); err != nil {
+		t.Fatal(err)
+	}
+	deploymentName := k8s.ComputeWorkerDeploymentName(twd)
+	buildID := k8s.ComputeBuildID(twd)
+	oldKey := types.NamespacedName{Namespace: namespace, Name: k8s.ComputeVersionedDeploymentName(twd.Name, buildID)}
+	var old appsv1.Deployment
+	eventually(t, 30*time.Second, time.Second, func() error { return k8sClient.Get(ctx, oldKey, &old) })
+	w, stop, err := testhelpers.NewWorker(ctx, deploymentName, buildID, name, ts.GetFrontendHostPort(), ts.GetDefaultNamespace(), true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	stopOnce := sync.OnceFunc(stop)
+	defer stopOnce()
+	w.RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
+		return workflow.Await(ctx, func() bool { return false })
+	}, workflow.RegisterOptions{Name: "inactivePinnedWorkflow"})
+	if err := w.Start(); err != nil {
+		t.Fatal(err)
+	}
+	setHealthyDeploymentStatus(t, ctx, k8sClient, old)
+	version := sdkworker.WorkerDeploymentVersion{DeploymentName: deploymentName, BuildID: buildID}
+	waitForVersionRegistrationInDeployment(t, ctx, ts, &version)
+	handle := ts.GetDefaultClient().WorkerDeploymentClient().GetHandle(deploymentName)
+
+	var run sdkclient.WorkflowRun
+	if pinned {
+		run, err = ts.GetDefaultClient().ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
+			ID: name, TaskQueue: name, VersioningOverride: &sdkclient.PinnedVersioningOverride{Version: version},
+		}, "inactivePinnedWorkflow")
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() { _ = ts.GetDefaultClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "test cleanup") }()
+		// Wait for the actual visibility index, not just the start RPC, before retiring.
+		eventually(t, 30*time.Second, time.Second, func() error {
+			count, err := ts.GetDefaultClient().CountWorkflow(ctx, &workflowservice.CountWorkflowExecutionsRequest{
+				Query: fmt.Sprintf("WorkflowId = '%s' AND TemporalWorkflowVersioningBehavior = 'Pinned' AND ExecutionStatus = 'Running'", name),
+			})
+			if err != nil {
+				return err
+			}
+			if count.Count != 1 {
+				return fmt.Errorf("pinned workflow not yet visible")
+			}
+			return nil
+		})
+	}
+
+	// Replace the target before ever making v1 current or ramping.
+	var next temporaliov1alpha1.WorkerDeployment
+	key := types.NamespacedName{Name: twd.Name, Namespace: namespace}
+	if err := k8sClient.Get(ctx, key, &next); err != nil {
+		t.Fatal(err)
+	}
+	next.Spec.Template.Spec.Containers[0].Image = "v2.0"
+	newBuildID := k8s.ComputeBuildID(&next)
+	if err := k8sClient.Update(ctx, &next); err != nil {
+		t.Fatal(err)
+	}
+	newKey := types.NamespacedName{Name: k8s.ComputeVersionedDeploymentName(twd.Name, newBuildID), Namespace: namespace}
+	eventually(t, 30*time.Second, time.Second, func() error {
+		var dep appsv1.Deployment
+		return k8sClient.Get(ctx, newKey, &dep)
+	})
+	stops := applyDeployment(t, ctx, k8sClient, newKey.Name, namespace)
+	defer handleStopFuncs(stops)
+	setCurrentVersion(t, ctx, ts, deploymentName, newBuildID)
+	eventually(t, 30*time.Second, time.Second, func() error {
+		if err := k8sClient.Get(ctx, key, &next); err != nil {
+			return err
+		}
+		for _, v := range next.Status.DeprecatedVersions {
+			if v.BuildID == buildID {
+				if v.Status == temporaliov1alpha1.VersionStatusInactive && v.DrainedSince == nil {
+					return nil
+				}
+				return fmt.Errorf("superseded version has status %s and drainedSince %v", v.Status, v.DrainedSince)
+			}
+		}
+		return fmt.Errorf("superseded version %s is not yet in deprecated versions", buildID)
+	})
+	stopOnce()
+	scaleDeploymentToZero(t, ctx, k8sClient, oldKey.Name, namespace)
+	if pinned {
+		// Longer than both the poller TTL and several controller reconciliations.
+		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Second) {
+			if err := k8sClient.Get(ctx, oldKey, &old); err != nil {
+				t.Fatalf("deleted a version with a pinned workflow: %v", err)
+			}
+			if _, err := handle.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{BuildID: buildID}); err != nil {
+				t.Fatal(err)
+			}
+		}
+		if err := ts.GetDefaultClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "release pinned version"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	eventually(t, 90*time.Second, time.Second, func() error {
+		if err := k8sClient.Get(ctx, oldKey, &old); err == nil {
+			return fmt.Errorf("superseded Inactive Deployment still exists")
+		} else if client.IgnoreNotFound(err) != nil {
+			return err
+		}
+		_, err := handle.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{BuildID: buildID})
+		var notFound *serviceerror.NotFound
+		if !errors.As(err, &notFound) {
+			return fmt.Errorf("expected retired Temporal version, got %v", err)
+		}
+		return nil
+	})
+	var current appsv1.Deployment
+	if err := k8sClient.Get(ctx, newKey, &current); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := handle.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{BuildID: newBuildID}); err != nil {
+		t.Fatal(err)
+	}
+}