diff --git a/admin/jobs/river/deployments_health_check.go b/admin/jobs/river/deployments_health_check.go
index 98219f1f534..56d0a28730c 100644
--- a/admin/jobs/river/deployments_health_check.go
+++ b/admin/jobs/river/deployments_health_check.go
@@ -27,8 +27,13 @@ func (DeploymentsHealthCheckArgs) Kind() string { return "deployments_health_che
 
 type DeploymentsHealthCheckWorker struct {
 	river.WorkerDefaults[DeploymentsHealthCheckArgs]
-	admin  *admin.Service
-	logger *zap.Logger
+	admin                      *admin.Service
+	logger                     *zap.Logger
+	findDeployments            func(context.Context, string, int) ([]*database.Deployment, error)
+	findDeploymentByInstanceID func(context.Context, string) (*database.Deployment, error)
+	healthCheck                func(context.Context, *database.Deployment) ([]string, bool)
+	deploymentAnnotations      func(context.Context, *database.Deployment) (*admin.DeploymentAnnotations, error)
+	now                        func() time.Time
 }
 
 func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[DeploymentsHealthCheckArgs]) error {
@@ -39,7 +44,7 @@ func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[
 	actualInstances := map[string][]string{}
 	var mu sync.Mutex
 	for {
-		deployments, err := w.admin.DB.FindDeployments(ctx, afterID, limit)
+		deployments, err := w.loadDeployments(ctx, afterID, limit)
 		if err != nil {
 			return fmt.Errorf("deployment health check: failed to get deployments: %w", err)
 		}
@@ -52,7 +57,7 @@ func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[
 		for _, d := range deployments {
 			d := d
 			if d.Status != database.DeploymentStatusRunning {
-				if time.Since(d.UpdatedOn) > time.Hour {
+				if w.currentTime().Sub(d.UpdatedOn) > time.Hour {
 					w.logger.Error("deployment health check: deployment not ok", zap.String("project_id", d.ProjectID), zap.String("deployment_id", d.ID), zap.String("status", d.Status.String()), zap.Time("since", d.UpdatedOn), observability.ZapCtx(ctx))
 				}
 				continue
@@ -63,7 +68,7 @@ func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[
 			}
 			seenHosts[d.RuntimeHost] = true
 			group.Go(func() error {
-				instances, ok := deploymentHealthCheck(cctx, w, d)
+				instances, ok := w.checkDeploymentHealth(cctx, d)
 				if ok {
 					mu.Lock()
 					actualInstances[d.RuntimeHost] = instances
@@ -95,7 +100,7 @@ func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[
 			}
 			// an expected instance is missing
 			// re verify that the deployment is not deleted
-			d, err := w.admin.DB.FindDeploymentByInstanceID(ctx, instance)
+			d, err := w.loadDeploymentByInstanceID(ctx, instance)
 			if err != nil {
 				if errors.Is(err, database.ErrNotFound) {
 					// Deployment was deleted
@@ -104,7 +109,7 @@ func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[
 				w.logger.Error("deployment health check: failed to find deployment", zap.String("instance_id", instance), zap.Error(err), observability.ZapCtx(ctx))
 				continue
 			}
-			annotations, err := w.annotationsForDeployment(ctx, d)
+			annotations, err := w.loadDeploymentAnnotations(ctx, d)
 			if err != nil {
 				w.logger.Error("deployment health check: failed to find deployment_annotations", zap.String("project_id", d.ProjectID), zap.String("deployment_id", d.ID), zap.Error(err), observability.ZapCtx(ctx))
 				continue
@@ -119,6 +124,41 @@ func (w *DeploymentsHealthCheckWorker) Work(ctx context.Context, job *river.Job[
 	return nil
 }
 
+func (w *DeploymentsHealthCheckWorker) loadDeployments(ctx context.Context, afterID string, limit int) ([]*database.Deployment, error) {
+	if w.findDeployments != nil {
+		return w.findDeployments(ctx, afterID, limit)
+	}
+	return w.admin.DB.FindDeployments(ctx, afterID, limit)
+}
+
+func (w *DeploymentsHealthCheckWorker) loadDeploymentByInstanceID(ctx context.Context, instanceID string) (*database.Deployment, error) {
+	if w.findDeploymentByInstanceID != nil {
+		return w.findDeploymentByInstanceID(ctx, instanceID)
+	}
+	return w.admin.DB.FindDeploymentByInstanceID(ctx, instanceID)
+}
+
+func (w *DeploymentsHealthCheckWorker) checkDeploymentHealth(ctx context.Context, d *database.Deployment) ([]string, bool) {
+	if w.healthCheck != nil {
+		return w.healthCheck(ctx, d)
+	}
+	return deploymentHealthCheck(ctx, w, d)
+}
+
+func (w *DeploymentsHealthCheckWorker) loadDeploymentAnnotations(ctx context.Context, d *database.Deployment) (*admin.DeploymentAnnotations, error) {
+	if w.deploymentAnnotations != nil {
+		return w.deploymentAnnotations(ctx, d)
+	}
+	return w.annotationsForDeployment(ctx, d)
+}
+
+func (w *DeploymentsHealthCheckWorker) currentTime() time.Time {
+	if w.now != nil {
+		return w.now()
+	}
+	return time.Now()
+}
+
 func deploymentHealthCheck(ctx context.Context, w *DeploymentsHealthCheckWorker, d *database.Deployment) (instances []string, runtimeOK bool) {
 	ctx, span := tracer.Start(ctx, "deploymentHealthCheck", trace.WithAttributes(attribute.String("project_id", d.ProjectID), attribute.String("deployment_id", d.ID)))
 	defer span.End()
diff --git a/admin/jobs/river/deployments_health_check_test.go b/admin/jobs/river/deployments_health_check_test.go
new file mode 100644
index 00000000000..8e0db63da8d
--- /dev/null
+++ b/admin/jobs/river/deployments_health_check_test.go
@@ -0,0 +1,291 @@
+package river
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rilldata/rill/admin"
+	"github.com/rilldata/rill/admin/database"
+	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
+)
+
+func TestDeploymentsHealthCheckPaginationBoundaries(t *testing.T) {
+	// Exact page boundaries are easy to get wrong: a full page must trigger one
+	// more fetch, while the cursor must still let the 10,001st row be checked.
+	tests := []struct {
+		count       int
+		wantCursors []string
+	}{
+		{count: 99, wantCursors: []string{""}},
+		{count: 100, wantCursors: []string{"", "deployment-099"}},
+		{count: 101, wantCursors: []string{"", "deployment-099"}},
+	}
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("%d deployments", tt.count), func(t *testing.T) {
+			deployments := make([]*database.Deployment, tt.count)
+			for i := range deployments {
+				deployments[i] = runningDeployment(i, fmt.Sprintf("host-%03d", i))
+			}
+
+			var cursors []string
+			var healthCalls atomic.Int32
+			worker := &DeploymentsHealthCheckWorker{
+				logger: zap.NewNop(),
+				findDeployments: func(_ context.Context, afterID string, limit int) ([]*database.Deployment, error) {
+					cursors = append(cursors, afterID)
+					start := 0
+					if afterID != "" {
+						for i, deployment := range deployments {
+							if deployment.ID == afterID {
+								start = i + 1
+								break
+							}
+						}
+					}
+					end := min(start+limit, len(deployments))
+					return deployments[start:end], nil
+				},
+				healthCheck: func(_ context.Context, deployment *database.Deployment) ([]string, bool) {
+					healthCalls.Add(1)
+					return []string{deployment.RuntimeInstanceID}, true
+				},
+				findDeploymentByInstanceID: func(context.Context, string) (*database.Deployment, error) {
+					t.Fatal("a healthy instance must not be reloaded from the database")
+					return nil, nil
+				},
+			}
+
+			err := worker.Work(t.Context(), nil)
+
+			require.NoError(t, err)
+			require.Equal(t, tt.wantCursors, cursors)
+			require.Equal(t, int32(tt.count), healthCalls.Load())
+		})
+	}
+}
+
+func TestDeploymentsHealthCheckDeduplicatesHostsAndHandlesDeletionRace(t *testing.T) {
+	// One runtime health response describes every instance on its host. A
+	// deployment deleted between listing and comparison is not a missing-instance alert.
+	deployments := []*database.Deployment{
+		runningDeployment(0, "shared-host"),
+		runningDeployment(1, "shared-host"),
+		runningDeployment(2, "shared-host"),
+	}
+	core, logs := observer.New(zap.ErrorLevel)
+	var healthCalls atomic.Int32
+	worker := &DeploymentsHealthCheckWorker{
+		logger: zap.New(core),
+		findDeployments: func(_ context.Context, afterID string, _ int) ([]*database.Deployment, error) {
+			if afterID != "" {
+				return nil, nil
+			}
+			return deployments, nil
+		},
+		healthCheck: func(context.Context, *database.Deployment) ([]string, bool) {
+			healthCalls.Add(1)
+			return []string{deployments[0].RuntimeInstanceID, deployments[1].RuntimeInstanceID}, true
+		},
+		findDeploymentByInstanceID: func(_ context.Context, instanceID string) (*database.Deployment, error) {
+			require.Equal(t, deployments[2].RuntimeInstanceID, instanceID)
+			return nil, database.ErrNotFound
+		},
+	}
+
+	err := worker.Work(t.Context(), nil)
+
+	require.NoError(t, err)
+	require.Equal(t, int32(1), healthCalls.Load())
+	require.Equal(t, 0, logs.FilterMessage("deployment health check: missing instance on runtime").Len())
+}
+
+func TestDeploymentsHealthCheckReportsOnlyConfirmedMissingInstances(t *testing.T) {
+	// A successful runtime response without an expected instance is actionable
+	// only after the database confirms that the deployment still exists.
+	deployment := runningDeployment(0, "host")
+	core, logs := observer.New(zap.ErrorLevel)
+	worker := &DeploymentsHealthCheckWorker{
+		logger: zap.New(core),
+		findDeployments: func(context.Context, string, int) ([]*database.Deployment, error) {
+			return []*database.Deployment{deployment}, nil
+		},
+		healthCheck: func(context.Context, *database.Deployment) ([]string, bool) {
+			return []string{}, true
+		},
+		findDeploymentByInstanceID: func(context.Context, string) (*database.Deployment, error) {
+			return deployment, nil
+		},
+		deploymentAnnotations: func(context.Context, *database.Deployment) (*admin.DeploymentAnnotations, error) {
+			return &admin.DeploymentAnnotations{}, nil
+		},
+	}
+
+	err := worker.Work(t.Context(), nil)
+
+	require.NoError(t, err)
+	require.Equal(t, 1, logs.FilterMessage("deployment health check: missing instance on runtime").Len())
+}
+
+func TestDeploymentsHealthCheckSkipsMissingComparisonWhenRuntimeUnavailable(t *testing.T) {
+	// An unavailable runtime yields no trustworthy instance inventory, so it
+	// must not fan out misleading database lookups and missing-instance alerts.
+	deployment := runningDeployment(0, "host")
+	var lookups atomic.Int32
+	worker := &DeploymentsHealthCheckWorker{
+		logger: zap.NewNop(),
+		findDeployments: func(context.Context, string, int) ([]*database.Deployment, error) {
+			return []*database.Deployment{deployment}, nil
+		},
+		healthCheck: func(context.Context, *database.Deployment) ([]string, bool) {
+			return nil, false
+		},
+		findDeploymentByInstanceID: func(context.Context, string) (*database.Deployment, error) {
+			lookups.Add(1)
+			return nil, database.ErrNotFound
+		},
+	}
+
+	err := worker.Work(t.Context(), nil)
+
+	require.NoError(t, err)
+	require.Zero(t, lookups.Load())
+}
+
+func TestDeploymentHealthClassifiers(t *testing.T) {
+	// Health severity depends on every response field, so this table protects
+	// less common limiter, repository, metric, parsing, and reconcile failures.
+	require.False(t, runtimeUnhealthy(&runtimev1.HealthResponse{}))
+	for name, response := range map[string]*runtimev1.HealthResponse{
+		"limiter":    {LimiterError: "failed"},
+		"connection": {ConnCacheError: "failed"},
+		"metastore":  {MetastoreError: "failed"},
+		"network":    {NetworkError: "failed"},
+	} {
+		t.Run(name, func(t *testing.T) {
+			require.True(t, runtimeUnhealthy(response))
+		})
+	}
+
+	require.False(t, instanceUnhealthy(&runtimev1.InstanceHealth{}))
+	for name, health := range map[string]*runtimev1.InstanceHealth{
+		"olap":       {OlapError: "failed"},
+		"controller": {ControllerError: "failed"},
+		"repository": {RepoError: "failed"},
+		"metrics":    {MetricsViewErrors: map[string]string{"orders": "failed"}},
+		"parse":      {ParseErrorCount: 1},
+		"reconcile":  {ReconcileErrorCount: 1},
+	} {
+		t.Run(name, func(t *testing.T) {
+			require.True(t, instanceUnhealthy(health))
+		})
+	}
+}
+
+func TestValidateDeploymentsOrphanBoundary(t *testing.T) {
+	// The three-hour grace period protects a deployment that may still become
+	// primary; teardown is allowed only strictly after the boundary.
+	now := time.Date(2026, time.July, 23, 12, 0, 0, 0, time.UTC)
+	tests := []struct {
+		name      string
+		depl      *database.Deployment
+		primaryID string
+		want      bool
+	}{
+		{name: "exactly three hours old", depl: orphanDeployment("old", now.Add(-3*time.Hour)), primaryID: "new"},
+		{name: "one nanosecond older", depl: orphanDeployment("old", now.Add(-3*time.Hour-time.Nanosecond)), primaryID: "new", want: true},
+		{name: "current primary", depl: orphanDeployment("primary", now.Add(-4*time.Hour)), primaryID: "primary"},
+		{name: "already stopped", depl: withDeploymentStatus(orphanDeployment("old", now.Add(-4*time.Hour)), database.DeploymentStatusStopped), primaryID: "new"},
+		{name: "development environment", depl: withDeploymentEnvironment(orphanDeployment("old", now.Add(-4*time.Hour)), "dev"), primaryID: "new"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, shouldTeardownOrphanDeployment(tt.depl, tt.primaryID, now))
+		})
+	}
+}
+
+func TestValidateDeploymentsPrimarySwitchAndTeardownFailure(t *testing.T) {
+	// After a primary switch, the old instance is removed while the new primary
+	// is reconciled. A failed orphan teardown must not block later deployments.
+	now := time.Date(2026, time.July, 23, 12, 0, 0, 0, time.UTC)
+	primaryID := "new-primary"
+	project := &database.Project{ID: "project", OrganizationID: "org", PrimaryDeploymentID: &primaryID}
+	failedOrphan := orphanDeployment("failed-orphan", now.Add(-4*time.Hour))
+	oldPrimary := orphanDeployment("old-primary", now.Add(-4*time.Hour))
+	newPrimary := orphanDeployment(primaryID, now.Add(-time.Minute))
+
+	var mu sync.Mutex
+	var removed []string
+	var reconciled []string
+	worker := &ValidateDeploymentsWorker{
+		admin: &admin.Service{Logger: zap.NewNop()},
+		now:   func() time.Time { return now },
+		findDeploymentsForProject: func(context.Context, string, string, string) ([]*database.Deployment, error) {
+			return []*database.Deployment{failedOrphan, oldPrimary, newPrimary}, nil
+		},
+		findOrganization: func(context.Context, string) (*database.Organization, error) {
+			return &database.Organization{ID: "org"}, nil
+		},
+		teardownDeployment: func(_ context.Context, deployment *database.Deployment) error {
+			mu.Lock()
+			defer mu.Unlock()
+			removed = append(removed, deployment.ID)
+			if deployment.ID == failedOrphan.ID {
+				return fmt.Errorf("provisioner unavailable")
+			}
+			return nil
+		},
+		reconcileDeployment: func(_ context.Context, deploymentID string) error {
+			mu.Lock()
+			defer mu.Unlock()
+			reconciled = append(reconciled, deploymentID)
+			return nil
+		},
+	}
+
+	err := worker.validateDeploymentsForProject(t.Context(), project)
+
+	require.NoError(t, err)
+	require.Equal(t, []string{"failed-orphan", "old-primary"}, removed)
+	require.Equal(t, []string{"new-primary"}, reconciled)
+}
+
+func runningDeployment(index int, host string) *database.Deployment {
+	return &database.Deployment{
+		ID:                fmt.Sprintf("deployment-%03d", index),
+		ProjectID:         fmt.Sprintf("project-%03d", index),
+		Environment:       "prod",
+		RuntimeHost:       host,
+		RuntimeInstanceID: fmt.Sprintf("instance-%03d", index),
+		Status:            database.DeploymentStatusRunning,
+	}
+}
+
+func orphanDeployment(id string, updatedOn time.Time) *database.Deployment {
+	return &database.Deployment{
+		ID:          id,
+		ProjectID:   "project",
+		Environment: "prod",
+		Status:      database.DeploymentStatusRunning,
+		UpdatedOn:   updatedOn,
+	}
+}
+
+func withDeploymentStatus(deployment *database.Deployment, status database.DeploymentStatus) *database.Deployment {
+	clone := *deployment
+	clone.Status = status
+	return &clone
+}
+
+func withDeploymentEnvironment(deployment *database.Deployment, environment string) *database.Deployment {
+	clone := *deployment
+	clone.Environment = environment
+	return &clone
+}
diff --git a/admin/jobs/river/reconcile_deployment.go b/admin/jobs/river/reconcile_deployment.go
index 0428bfcdeb8..0da63c74807 100644
--- a/admin/jobs/river/reconcile_deployment.go
+++ b/admin/jobs/river/reconcile_deployment.go
@@ -21,7 +21,14 @@ func (ReconcileDeploymentArgs) Kind() string { return "reconcile_deployment" }
 
 type ReconcileDeploymentWorker struct {
 	river.WorkerDefaults[ReconcileDeploymentArgs]
-	admin *admin.Service
+	admin                  *admin.Service
+	findDeployment         func(context.Context, string) (*database.Deployment, error)
+	updateDeploymentStatus func(context.Context, string, database.DeploymentStatus, string) (*database.Deployment, error)
+	startDeployment        func(context.Context, *database.Deployment) error
+	updateDeployment       func(context.Context, *database.Deployment) error
+	stopDeployment         func(context.Context, *database.Deployment) error
+	deleteDeployment       func(context.Context, *database.Deployment) error
+	enqueueReconcile       func(context.Context, string) (int64, error)
 }
 
 // NewReconcileDeploymentWorker creates a new ReconcileDeploymentWorker. Only to be used in tests to trigger the worker directly.
@@ -37,11 +44,11 @@ func NewReconcileDeploymentWorker(admin *admin.Service) *ReconcileDeploymentWork
 // We handle all deployment state transitions in this job to ensure consistency and to avoid concurrent conflicting operations on the same deployment.
 func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[ReconcileDeploymentArgs]) error {
 	observability.AddRequestAttributes(ctx, attribute.String("args.deployment_id", job.Args.DeploymentID))
-	depl, err := w.admin.DB.FindDeployment(ctx, job.Args.DeploymentID)
+	depl, err := w.loadDeployment(ctx, job.Args.DeploymentID)
 	if err != nil {
 		if errors.Is(err, database.ErrNotFound) {
 			// If the deployment doesn't exist, we can just finish the job and do nothing more.
-			w.admin.Logger.Info("reconcile deployment: deployment not found, job succeeded", observability.ZapCtx(ctx))
+			w.logger().Info("reconcile deployment: deployment not found, job succeeded", observability.ZapCtx(ctx))
 			return nil
 		}
 		return err
@@ -60,19 +67,19 @@ func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[Rec
 			// changed, and otherwise performs a lightweight drift-aware resource check. We therefore keep the
 			// deployment in the Running status here rather than flipping it to Updating, which would otherwise flap
 			// on every periodic reconciliation.
-			err := w.admin.UpdateDeploymentInner(ctx, depl)
+			err := w.reconcileRunningDeployment(ctx, depl)
 			if err != nil {
 				return err
 			}
 		} else {
 			// Update the deployment status to pending
-			depl, err = w.admin.DB.UpdateDeploymentStatus(ctx, depl.ID, database.DeploymentStatusPending, "Provisioning...")
+			depl, err = w.persistDeploymentStatus(ctx, depl.ID, database.DeploymentStatusPending, "Provisioning...")
 			if err != nil {
 				return err
 			}
 
 			// Initialize the deployment (by provisioning a runtime and creating an instance on it)
-			err := w.admin.StartDeploymentInner(ctx, depl)
+			err := w.initializeDeployment(ctx, depl)
 			if err != nil {
 				return err
 			}
@@ -86,13 +93,13 @@ func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[Rec
 			return nil
 		}
 		// Update the deployment status to stopping
-		depl, err = w.admin.DB.UpdateDeploymentStatus(ctx, depl.ID, database.DeploymentStatusStopping, "Stopping...")
+		depl, err = w.persistDeploymentStatus(ctx, depl.ID, database.DeploymentStatusStopping, "Stopping...")
 		if err != nil {
 			return err
 		}
 
 		// Stop the deployment by tearing down its runtime instance and resources.
-		err = w.admin.StopDeploymentInner(ctx, depl)
+		err = w.hibernateDeployment(ctx, depl)
 		if err != nil {
 			return err
 		}
@@ -105,13 +112,13 @@ func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[Rec
 			return nil
 		}
 		// Update the deployment status to deleting
-		depl, err = w.admin.DB.UpdateDeploymentStatus(ctx, depl.ID, database.DeploymentStatusDeleting, "Deleting...")
+		depl, err = w.persistDeploymentStatus(ctx, depl.ID, database.DeploymentStatusDeleting, "Deleting...")
 		if err != nil {
 			return err
 		}
 
 		// Delete the deployment and all its resources.
-		err := w.admin.DeleteDeploymentInner(ctx, depl)
+		err := w.removeDeployment(ctx, depl)
 		if err != nil {
 			return err
 		}
@@ -125,7 +132,7 @@ func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[Rec
 	}
 
 	// Update the deployment status
-	depl, err = w.admin.DB.UpdateDeploymentStatus(ctx, depl.ID, newStatus, "")
+	depl, err = w.persistDeploymentStatus(ctx, depl.ID, newStatus, "")
 	if err != nil {
 		return err
 	}
@@ -133,15 +140,73 @@ func (w *ReconcileDeploymentWorker) Work(ctx context.Context, job *river.Job[Rec
 	// If current depl.DesiredStatusUpdatedOn != desiredStatusUpdatedOn when job started, then the deployment changed while we were working and we should reschedule another job.
 	if !depl.DesiredStatusUpdatedOn.Equal(desiredStatusUpdatedOn) {
 		// Deployment changed while we were working, reschedule another job to reconcile again.
-		c := river.ClientFromContext[pgx.Tx](ctx)
-		res, err := c.Insert(ctx, ReconcileDeploymentArgs{
-			DeploymentID: job.Args.DeploymentID,
-		}, nil)
+		newJobID, err := w.scheduleReconcile(ctx, job.Args.DeploymentID)
 		if err != nil {
 			return err
 		}
-		w.admin.Logger.Info("reconcile deployment: changes to deployment detected since job started, rescheduling job", observability.ZapCtx(ctx), zap.Int64("new_job_id", res.Job.ID))
+		w.logger().Info("reconcile deployment: changes to deployment detected since job started, rescheduling job", observability.ZapCtx(ctx), zap.Int64("new_job_id", newJobID))
 	}
 
 	return nil
 }
+
+func (w *ReconcileDeploymentWorker) loadDeployment(ctx context.Context, deploymentID string) (*database.Deployment, error) {
+	if w.findDeployment != nil {
+		return w.findDeployment(ctx, deploymentID)
+	}
+	return w.admin.DB.FindDeployment(ctx, deploymentID)
+}
+
+func (w *ReconcileDeploymentWorker) persistDeploymentStatus(ctx context.Context, deploymentID string, status database.DeploymentStatus, message string) (*database.Deployment, error) {
+	if w.updateDeploymentStatus != nil {
+		return w.updateDeploymentStatus(ctx, deploymentID, status, message)
+	}
+	return w.admin.DB.UpdateDeploymentStatus(ctx, deploymentID, status, message)
+}
+
+func (w *ReconcileDeploymentWorker) initializeDeployment(ctx context.Context, deployment *database.Deployment) error {
+	if w.startDeployment != nil {
+		return w.startDeployment(ctx, deployment)
+	}
+	return w.admin.StartDeploymentInner(ctx, deployment)
+}
+
+func (w *ReconcileDeploymentWorker) reconcileRunningDeployment(ctx context.Context, deployment *database.Deployment) error {
+	if w.updateDeployment != nil {
+		return w.updateDeployment(ctx, deployment)
+	}
+	return w.admin.UpdateDeploymentInner(ctx, deployment)
+}
+
+func (w *ReconcileDeploymentWorker) hibernateDeployment(ctx context.Context, deployment *database.Deployment) error {
+	if w.stopDeployment != nil {
+		return w.stopDeployment(ctx, deployment)
+	}
+	return w.admin.StopDeploymentInner(ctx, deployment)
+}
+
+func (w *ReconcileDeploymentWorker) removeDeployment(ctx context.Context, deployment *database.Deployment) error {
+	if w.deleteDeployment != nil {
+		return w.deleteDeployment(ctx, deployment)
+	}
+	return w.admin.DeleteDeploymentInner(ctx, deployment)
+}
+
+func (w *ReconcileDeploymentWorker) scheduleReconcile(ctx context.Context, deploymentID string) (int64, error) {
+	if w.enqueueReconcile != nil {
+		return w.enqueueReconcile(ctx, deploymentID)
+	}
+	c := river.ClientFromContext[pgx.Tx](ctx)
+	res, err := c.Insert(ctx, ReconcileDeploymentArgs{DeploymentID: deploymentID}, nil)
+	if err != nil {
+		return 0, err
+	}
+	return res.Job.ID, nil
+}
+
+func (w *ReconcileDeploymentWorker) logger() *zap.Logger {
+	if w.admin != nil && w.admin.Logger != nil {
+		return w.admin.Logger
+	}
+	return zap.NewNop()
+}
diff --git a/admin/jobs/river/reconcile_deployment_river_test.go b/admin/jobs/river/reconcile_deployment_river_test.go
new file mode 100644
index 00000000000..72ed5f85863
--- /dev/null
+++ b/admin/jobs/river/reconcile_deployment_river_test.go
@@ -0,0 +1,90 @@
+package river
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/rilldata/rill/admin/pkg/pgtestcontainer"
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/riverdriver/riverpgxv5"
+	"github.com/riverqueue/river/rivermigrate"
+	"github.com/riverqueue/river/rivertype"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReconcileDeploymentUniquenessIncludesRetryableJobs(t *testing.T) {
+	// River enforces uniqueness in Postgres, so this test uses the real schema
+	// and concurrent inserts instead of approximating the unique-key behavior.
+	pg := pgtestcontainer.New(t)
+	t.Cleanup(func() { pg.Terminate(t) })
+	pool, err := pgxpool.New(t.Context(), pg.DatabaseURL)
+	require.NoError(t, err)
+	t.Cleanup(pool.Close)
+
+	driver := riverpgxv5.New(pool)
+	migrator, err := rivermigrate.New(driver, nil)
+	require.NoError(t, err)
+	_, err = migrator.Migrate(t.Context(), rivermigrate.DirectionUp, nil)
+	require.NoError(t, err)
+
+	riverClient, err := river.NewClient(driver, &river.Config{TestOnly: true})
+	require.NoError(t, err)
+	client := &Client{riverClient: riverClient, dbPool: pool}
+
+	states := []rivertype.JobState{
+		rivertype.JobStateAvailable,
+		rivertype.JobStateRunning,
+		rivertype.JobStateScheduled,
+		rivertype.JobStateRetryable,
+	}
+	for _, state := range states {
+		t.Run(string(state), func(t *testing.T) {
+			// Each state represents an operative job. Sixteen simultaneous callers
+			// must all resolve to the original job, including during retry backoff.
+			deploymentID := fmt.Sprintf("deployment-%s", state)
+			original, err := client.ReconcileDeployment(t.Context(), deploymentID)
+			require.NoError(t, err)
+			require.False(t, original.Duplicate)
+
+			if state != rivertype.JobStateAvailable {
+				_, err = pool.Exec(t.Context(), "UPDATE river_job SET state = $1 WHERE id = $2", state, original.ID)
+				require.NoError(t, err)
+			}
+
+			const callers = 16
+			start := make(chan struct{})
+			results := make([]int64, callers)
+			duplicates := make([]bool, callers)
+			errs := make([]error, callers)
+			var wg sync.WaitGroup
+			for i := 0; i < callers; i++ {
+				wg.Add(1)
+				go func(index int) {
+					defer wg.Done()
+					<-start
+					result, insertErr := client.ReconcileDeployment(context.Background(), deploymentID)
+					errs[index] = insertErr
+					if result != nil {
+						results[index] = result.ID
+						duplicates[index] = result.Duplicate
+					}
+				}(i)
+			}
+			close(start)
+			wg.Wait()
+
+			for i := range callers {
+				require.NoError(t, errs[i])
+				require.True(t, duplicates[i], "caller %d inserted a conflicting job", i)
+				require.Equal(t, original.ID, results[i])
+			}
+			var count int
+			err = pool.QueryRow(t.Context(), `SELECT COUNT(*) FROM river_job WHERE kind = $1 AND args->>'DeploymentID' = $2`, ReconcileDeploymentArgs{}.Kind(), deploymentID).Scan(&count)
+			require.NoError(t, err)
+			require.Equal(t, 1, count)
+		})
+	}
+}
diff --git a/admin/jobs/river/reconcile_deployment_test.go b/admin/jobs/river/reconcile_deployment_test.go
new file mode 100644
index 00000000000..26bc998dbfe
--- /dev/null
+++ b/admin/jobs/river/reconcile_deployment_test.go
@@ -0,0 +1,272 @@
+package river
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/rilldata/rill/admin/database"
+	"github.com/riverqueue/river"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReconcileDeploymentStateMatrix(t *testing.T) {
+	// Exercise the complete desired/current status matrix so adding a status
+	// cannot silently turn a destructive transition into the default no-op path.
+	statuses := []database.DeploymentStatus{
+		database.DeploymentStatusUnspecified,
+		database.DeploymentStatusPending,
+		database.DeploymentStatusRunning,
+		database.DeploymentStatusErrored,
+		database.DeploymentStatusStopped,
+		database.DeploymentStatusUpdating,
+		database.DeploymentStatusStopping,
+		database.DeploymentStatusDeleting,
+		database.DeploymentStatusDeleted,
+	}
+	for _, desired := range statuses {
+		for _, current := range statuses {
+			name := fmt.Sprintf("desired_%s/current_%s", desired, current)
+			t.Run(name, func(t *testing.T) {
+				deployment := &database.Deployment{
+					ID:                     "deployment",
+					Status:                 current,
+					DesiredStatus:          desired,
+					DesiredStatusUpdatedOn: time.Unix(100, 0),
+				}
+				var actions []string
+				var statusWrites []database.DeploymentStatus
+				worker := newStateMatrixWorker(deployment, &actions, &statusWrites)
+
+				err := worker.Work(t.Context(), reconcileDeploymentJob())
+
+				require.NoError(t, err)
+				wantActions, wantStatuses := expectedReconcileTransition(desired, current)
+				require.Equal(t, wantActions, actions)
+				require.Equal(t, wantStatuses, statusWrites)
+			})
+		}
+	}
+}
+
+func TestReconcileDeploymentFailureBoundaries(t *testing.T) {
+	// Every database boundary and external lifecycle call must stop subsequent
+	// work, preserving an intermediate status that a retry can safely resume.
+	injected := errors.New("injected failure")
+	tests := []struct {
+		name            string
+		desired         database.DeploymentStatus
+		current         database.DeploymentStatus
+		failStatusCall  int
+		failAction      string
+		wantStatusCalls int
+		wantActionCalls int
+	}{
+		{name: "start pre-status", desired: database.DeploymentStatusRunning, current: database.DeploymentStatusStopped, failStatusCall: 1, wantStatusCalls: 1},
+		{name: "start operation", desired: database.DeploymentStatusRunning, current: database.DeploymentStatusStopped, failAction: "start", wantStatusCalls: 1, wantActionCalls: 1},
+		{name: "start final status", desired: database.DeploymentStatusRunning, current: database.DeploymentStatusStopped, failStatusCall: 2, wantStatusCalls: 2, wantActionCalls: 1},
+		{name: "running update", desired: database.DeploymentStatusRunning, current: database.DeploymentStatusRunning, failAction: "update", wantActionCalls: 1},
+		{name: "running final status", desired: database.DeploymentStatusRunning, current: database.DeploymentStatusRunning, failStatusCall: 1, wantStatusCalls: 1, wantActionCalls: 1},
+		{name: "stop pre-status", desired: database.DeploymentStatusStopped, current: database.DeploymentStatusRunning, failStatusCall: 1, wantStatusCalls: 1},
+		{name: "stop operation", desired: database.DeploymentStatusStopped, current: database.DeploymentStatusRunning, failAction: "stop", wantStatusCalls: 1, wantActionCalls: 1},
+		{name: "stop final status", desired: database.DeploymentStatusStopped, current: database.DeploymentStatusRunning, failStatusCall: 2, wantStatusCalls: 2, wantActionCalls: 1},
+		{name: "delete pre-status", desired: database.DeploymentStatusDeleted, current: database.DeploymentStatusRunning, failStatusCall: 1, wantStatusCalls: 1},
+		{name: "delete operation", desired: database.DeploymentStatusDeleted, current: database.DeploymentStatusRunning, failAction: "delete", wantStatusCalls: 1, wantActionCalls: 1},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			deployment := &database.Deployment{ID: "deployment", Status: tt.current, DesiredStatus: tt.desired}
+			statusCalls := 0
+			actionCalls := 0
+			worker := &ReconcileDeploymentWorker{
+				findDeployment: func(context.Context, string) (*database.Deployment, error) {
+					clone := *deployment
+					return &clone, nil
+				},
+				updateDeploymentStatus: func(_ context.Context, _ string, status database.DeploymentStatus, _ string) (*database.Deployment, error) {
+					statusCalls++
+					if statusCalls == tt.failStatusCall {
+						return nil, injected
+					}
+					clone := *deployment
+					clone.Status = status
+					deployment = &clone
+					return &clone, nil
+				},
+			}
+			setLifecycleAction := func(name string) func(context.Context, *database.Deployment) error {
+				return func(context.Context, *database.Deployment) error {
+					actionCalls++
+					if tt.failAction == name {
+						return injected
+					}
+					return nil
+				}
+			}
+			worker.startDeployment = setLifecycleAction("start")
+			worker.updateDeployment = setLifecycleAction("update")
+			worker.stopDeployment = setLifecycleAction("stop")
+			worker.deleteDeployment = setLifecycleAction("delete")
+
+			err := worker.Work(t.Context(), reconcileDeploymentJob())
+
+			require.ErrorIs(t, err, injected)
+			require.Equal(t, tt.wantStatusCalls, statusCalls)
+			require.Equal(t, tt.wantActionCalls, actionCalls)
+		})
+	}
+}
+
+func TestReconcileDeploymentLookupFailures(t *testing.T) {
+	// A deleted deployment is a successful idempotent no-op, while an actual
+	// database outage must remain retryable and must not invoke lifecycle work.
+	t.Run("not found", func(t *testing.T) {
+		worker := &ReconcileDeploymentWorker{findDeployment: func(context.Context, string) (*database.Deployment, error) {
+			return nil, database.ErrNotFound
+		}}
+		require.NoError(t, worker.Work(t.Context(), reconcileDeploymentJob()))
+	})
+
+	t.Run("database failure", func(t *testing.T) {
+		injected := errors.New("database unavailable")
+		worker := &ReconcileDeploymentWorker{findDeployment: func(context.Context, string) (*database.Deployment, error) {
+			return nil, injected
+		}}
+		require.ErrorIs(t, worker.Work(t.Context(), reconcileDeploymentJob()), injected)
+	})
+}
+
+func TestReconcileDeploymentSchedulesOneFollowUpForConcurrentDesiredChange(t *testing.T) {
+	// A desired-state write during an external start is detected from its
+	// timestamp after the final status write and results in exactly one follow-up.
+	originalUpdate := time.Unix(100, 0)
+	changedUpdate := originalUpdate.Add(time.Second)
+	deployment := &database.Deployment{
+		ID:                     "deployment",
+		Status:                 database.DeploymentStatusStopped,
+		DesiredStatus:          database.DeploymentStatusRunning,
+		DesiredStatusUpdatedOn: originalUpdate,
+	}
+	statusCalls := 0
+	enqueueCalls := 0
+	worker := &ReconcileDeploymentWorker{
+		findDeployment: func(context.Context, string) (*database.Deployment, error) {
+			clone := *deployment
+			return &clone, nil
+		},
+		updateDeploymentStatus: func(_ context.Context, _ string, status database.DeploymentStatus, _ string) (*database.Deployment, error) {
+			statusCalls++
+			clone := *deployment
+			clone.Status = status
+			if statusCalls == 2 {
+				clone.DesiredStatus = database.DeploymentStatusStopped
+				clone.DesiredStatusUpdatedOn = changedUpdate
+			}
+			deployment = &clone
+			return &clone, nil
+		},
+		startDeployment: func(context.Context, *database.Deployment) error { return nil },
+		enqueueReconcile: func(_ context.Context, deploymentID string) (int64, error) {
+			enqueueCalls++
+			require.Equal(t, "deployment", deploymentID)
+			return 42, nil
+		},
+	}
+
+	err := worker.Work(t.Context(), reconcileDeploymentJob())
+
+	require.NoError(t, err)
+	require.Equal(t, 1, enqueueCalls)
+}
+
+func TestReconcileDeploymentRetryRepeatsOnlyIdempotentLifecycleOperation(t *testing.T) {
+	// If the final status write fails, the next job observes the intermediate
+	// state and repeats StartDeploymentInner, whose production contract is idempotent.
+	deployment := &database.Deployment{ID: "deployment", Status: database.DeploymentStatusStopped, DesiredStatus: database.DeploymentStatusRunning}
+	startCalls := 0
+	statusCalls := 0
+	failFinalOnce := true
+	worker := &ReconcileDeploymentWorker{
+		findDeployment: func(context.Context, string) (*database.Deployment, error) {
+			clone := *deployment
+			return &clone, nil
+		},
+		updateDeploymentStatus: func(_ context.Context, _ string, status database.DeploymentStatus, _ string) (*database.Deployment, error) {
+			statusCalls++
+			if status == database.DeploymentStatusRunning && failFinalOnce {
+				failFinalOnce = false
+				return nil, errors.New("final status unavailable")
+			}
+			clone := *deployment
+			clone.Status = status
+			deployment = &clone
+			return &clone, nil
+		},
+		startDeployment: func(context.Context, *database.Deployment) error {
+			startCalls++
+			return nil
+		},
+	}
+
+	require.Error(t, worker.Work(t.Context(), reconcileDeploymentJob()))
+	require.NoError(t, worker.Work(t.Context(), reconcileDeploymentJob()))
+	require.Equal(t, 2, startCalls)
+	require.Equal(t, database.DeploymentStatusRunning, deployment.Status)
+	require.Equal(t, 4, statusCalls)
+}
+
+func newStateMatrixWorker(deployment *database.Deployment, actions *[]string, statusWrites *[]database.DeploymentStatus) *ReconcileDeploymentWorker {
+	current := *deployment
+	worker := &ReconcileDeploymentWorker{
+		findDeployment: func(context.Context, string) (*database.Deployment, error) {
+			clone := current
+			return &clone, nil
+		},
+		updateDeploymentStatus: func(_ context.Context, _ string, status database.DeploymentStatus, _ string) (*database.Deployment, error) {
+			*statusWrites = append(*statusWrites, status)
+			current.Status = status
+			clone := current
+			return &clone, nil
+		},
+	}
+	worker.startDeployment = recordLifecycleAction("start", actions)
+	worker.updateDeployment = recordLifecycleAction("update", actions)
+	worker.stopDeployment = recordLifecycleAction("stop", actions)
+	worker.deleteDeployment = recordLifecycleAction("delete", actions)
+	return worker
+}
+
+func recordLifecycleAction(name string, actions *[]string) func(context.Context, *database.Deployment) error {
+	return func(context.Context, *database.Deployment) error {
+		*actions = append(*actions, name)
+		return nil
+	}
+}
+
+func expectedReconcileTransition(desired, current database.DeploymentStatus) ([]string, []database.DeploymentStatus) {
+	switch desired {
+	case database.DeploymentStatusRunning:
+		if current == database.DeploymentStatusRunning {
+			return []string{"update"}, []database.DeploymentStatus{database.DeploymentStatusRunning}
+		}
+		return []string{"start"}, []database.DeploymentStatus{database.DeploymentStatusPending, database.DeploymentStatusRunning}
+	case database.DeploymentStatusStopped:
+		if current == database.DeploymentStatusStopped {
+			return nil, nil
+		}
+		return []string{"stop"}, []database.DeploymentStatus{database.DeploymentStatusStopping, database.DeploymentStatusStopped}
+	case database.DeploymentStatusDeleted:
+		if current == database.DeploymentStatusDeleted {
+			return nil, nil
+		}
+		return []string{"delete"}, []database.DeploymentStatus{database.DeploymentStatusDeleting}
+	default:
+		return nil, nil
+	}
+}
+
+func reconcileDeploymentJob() *river.Job[ReconcileDeploymentArgs] {
+	return &river.Job[ReconcileDeploymentArgs]{Args: ReconcileDeploymentArgs{DeploymentID: "deployment"}}
+}
diff --git a/admin/jobs/river/river.go b/admin/jobs/river/river.go
index b3d635ea5e0..1a7f47c3c99 100644
--- a/admin/jobs/river/river.go
+++ b/admin/jobs/river/river.go
@@ -345,6 +345,7 @@ func (c *Client) ReconcileDeployment(ctx context.Context, deploymentID string) (
 				rivertype.JobStateAvailable,
 				rivertype.JobStatePending,
 				rivertype.JobStateRunning,
+				rivertype.JobStateRetryable,
 				rivertype.JobStateScheduled,
 			},
 		},
diff --git a/admin/jobs/river/validate_deployments.go b/admin/jobs/river/validate_deployments.go
index ee67a43da27..e57fb01bb90 100644
--- a/admin/jobs/river/validate_deployments.go
+++ b/admin/jobs/river/validate_deployments.go
@@ -17,7 +17,13 @@ func (ValidateDeploymentsArgs) Kind() string { return "validate_deployments" }
 
 type ValidateDeploymentsWorker struct {
 	river.WorkerDefaults[ValidateDeploymentsArgs]
-	admin *admin.Service
+	admin                     *admin.Service
+	findProjects              func(context.Context, string, int) ([]*database.Project, error)
+	findDeploymentsForProject func(context.Context, string, string, string) ([]*database.Deployment, error)
+	findOrganization          func(context.Context, string) (*database.Organization, error)
+	teardownDeployment        func(context.Context, *database.Deployment) error
+	reconcileDeployment       func(context.Context, string) error
+	now                       func() time.Time
 }
 
 const validateDeploymentsForProjectTimeout = 5 * time.Minute
@@ -28,7 +34,7 @@ func (w *ValidateDeploymentsWorker) Work(ctx context.Context, job *river.Job[Val
 	limit := 100
 	afterID := ""
 	for {
-		projs, err := w.admin.DB.FindProjects(ctx, afterID, limit)
+		projs, err := w.loadProjects(ctx, afterID, limit)
 		if err != nil {
 			return err
 		}
@@ -56,7 +62,7 @@ func (w *ValidateDeploymentsWorker) validateDeploymentsForProject(ctx context.Co
 	defer cancel()
 
 	// Get all project deployments for prod environment
-	depls, err := w.admin.DB.FindDeploymentsForProject(ctx, proj.ID, "prod", "")
+	depls, err := w.loadDeploymentsForProject(ctx, proj.ID, "prod", "")
 	if err != nil {
 		return err
 	}
@@ -65,7 +71,7 @@ func (w *ValidateDeploymentsWorker) validateDeploymentsForProject(ctx context.Co
 	}
 
 	// Get project organization, we need this to create the deployment annotations
-	org, err := w.admin.DB.FindOrganization(ctx, proj.OrganizationID)
+	org, err := w.loadOrganization(ctx, proj.OrganizationID)
 	if err != nil {
 		return err
 	}
@@ -81,9 +87,9 @@ func (w *ValidateDeploymentsWorker) validateDeploymentsForProject(ctx context.Co
 		// This might for example happen if a redeploy failed after switching to the new deployment.
 		// We consider a deployment orphaned if it is not the prod deployment, is not stopped and has not been updated in 3 hours.
 		// The 3 hour delay is to ensure we don't tear down a deployment that is in the process of being created and is to become the new prod deployment.
-		if depl.Environment == "prod" && depl.ID != prodDeplID && depl.Status != database.DeploymentStatusStopped && depl.UpdatedOn.Add(3*time.Hour).Before(time.Now()) {
+		if shouldTeardownOrphanDeployment(depl, prodDeplID, w.currentTime()) {
 			w.admin.Logger.Info("validate deployments: removing deployment", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), observability.ZapCtx(ctx))
-			err = w.admin.TeardownDeployment(ctx, depl)
+			err = w.removeDeployment(ctx, depl)
 			if err != nil {
 				w.admin.Logger.Error("validate deployments: failed to remove deployment", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), observability.ZapCtx(ctx), zap.Error(err))
 				continue
@@ -98,7 +104,7 @@ func (w *ValidateDeploymentsWorker) validateDeploymentsForProject(ctx context.Co
 		// Reconcile only ever drives a deployment toward its own DesiredStatus, is a no-op for already-consistent
 		// deployments (e.g. stopped/stopped, deleted/deleted), and is de-duplicated per deployment, so scheduling
 		// for every deployment is safe and self-healing.
-		_, err := w.admin.Jobs.ReconcileDeployment(ctx, depl.ID)
+		err := w.scheduleReconcileDeployment(ctx, depl.ID)
 		if err != nil {
 			w.admin.Logger.Error("validate deployments: failed to schedule reconcile", zap.String("organization_id", org.ID), zap.String("project_id", proj.ID), zap.String("deployment_id", depl.ID), zap.String("instance_id", depl.RuntimeInstanceID), zap.Error(err), observability.ZapCtx(ctx))
 			continue
@@ -108,3 +114,53 @@ func (w *ValidateDeploymentsWorker) validateDeploymentsForProject(ctx context.Co
 
 	return nil
 }
+
+func shouldTeardownOrphanDeployment(depl *database.Deployment, primaryDeploymentID string, now time.Time) bool {
+	return depl.Environment == "prod" &&
+		depl.ID != primaryDeploymentID &&
+		depl.Status != database.DeploymentStatusStopped &&
+		depl.UpdatedOn.Add(3*time.Hour).Before(now)
+}
+
+func (w *ValidateDeploymentsWorker) loadProjects(ctx context.Context, afterID string, limit int) ([]*database.Project, error) {
+	if w.findProjects != nil {
+		return w.findProjects(ctx, afterID, limit)
+	}
+	return w.admin.DB.FindProjects(ctx, afterID, limit)
+}
+
+func (w *ValidateDeploymentsWorker) loadDeploymentsForProject(ctx context.Context, projectID, environment, branch string) ([]*database.Deployment, error) {
+	if w.findDeploymentsForProject != nil {
+		return w.findDeploymentsForProject(ctx, projectID, environment, branch)
+	}
+	return w.admin.DB.FindDeploymentsForProject(ctx, projectID, environment, branch)
+}
+
+func (w *ValidateDeploymentsWorker) loadOrganization(ctx context.Context, organizationID string) (*database.Organization, error) {
+	if w.findOrganization != nil {
+		return w.findOrganization(ctx, organizationID)
+	}
+	return w.admin.DB.FindOrganization(ctx, organizationID)
+}
+
+func (w *ValidateDeploymentsWorker) removeDeployment(ctx context.Context, depl *database.Deployment) error {
+	if w.teardownDeployment != nil {
+		return w.teardownDeployment(ctx, depl)
+	}
+	return w.admin.TeardownDeployment(ctx, depl)
+}
+
+func (w *ValidateDeploymentsWorker) scheduleReconcileDeployment(ctx context.Context, deploymentID string) error {
+	if w.reconcileDeployment != nil {
+		return w.reconcileDeployment(ctx, deploymentID)
+	}
+	_, err := w.admin.Jobs.ReconcileDeployment(ctx, deploymentID)
+	return err
+}
+
+func (w *ValidateDeploymentsWorker) currentTime() time.Time {
+	if w.now != nil {
+		return w.now()
+	}
+	return time.Now()
+}