diff --git a/runtime/reconcilers/alert.go b/runtime/reconcilers/alert.go
index cd44f55ac47..f113f819fb8 100644
--- a/runtime/reconcilers/alert.go
+++ b/runtime/reconcilers/alert.go
@@ -188,7 +188,7 @@ func (r *AlertReconciler) Reconcile(ctx context.Context, n *runtimev1.ResourceNa
 
 	// If we were cancelled, exit without updating any other trigger-related state.
 	// NOTE: We don't set Retrigger here because we'll leave re-scheduling to whatever cancelled the reconciler.
-	if errors.Is(executeErr, context.Canceled) {
+	if isAlertContextInterruption(executeErr) {
 		return runtime.ReconcileResult{Err: executeErr}
 	}
 
@@ -521,38 +521,43 @@ func (r *AlertReconciler) executeAll(ctx context.Context, self *runtimev1.Resour
 
 	// Get admin metadata for the alert (if an admin service does not support sending alerts, alerts will still work, the notifications just won't have open/edit links).
 	var adminMeta *drivers.AlertMetadata
-	admin, release, err := r.C.Runtime.Admin(ctx, r.C.InstanceID)
-	if err != nil {
-		return fmt.Errorf("failed to get admin client: %w", err)
-	}
-	defer release()
-	anonRecipients := false
-	var emailRecipients []string
-	for _, notifier := range a.Spec.Notifiers {
-		if notifier.Connector == "email" {
-			emailRecipients = pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"])
-		} else {
-			anonRecipients = true
+	admin, release, executeErr := r.C.Runtime.Admin(ctx, r.C.InstanceID)
+	if executeErr != nil {
+		executeErr = fmt.Errorf("failed to get admin client: %w", executeErr)
+	} else {
+		defer release()
+		anonRecipients := false
+		var emailRecipients []string
+		for _, notifier := range a.Spec.Notifiers {
+			if notifier.Connector == "email" {
+				emailRecipients = pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"])
+			} else {
+				anonRecipients = true
+			}
+		}
+		ownerID := ""
+		if a.Spec.Annotations != nil {
+			ownerID = a.Spec.Annotations["admin_owner_user_id"]
+		}
+		adminMeta, executeErr = admin.GetAlertMetadata(ctx, self.Meta.Name.Name, ownerID, emailRecipients, anonRecipients, a.Spec.Annotations, a.Spec.GetQueryForUserId(), a.Spec.GetQueryForUserEmail())
+		if errors.Is(executeErr, drivers.ErrNotImplemented) {
+			executeErr = nil
+		} else if executeErr != nil {
+			executeErr = fmt.Errorf("failed to get alert metadata: %w", executeErr)
 		}
-	}
-	ownerID := ""
-	if a.Spec.Annotations != nil {
-		ownerID = a.Spec.Annotations["admin_owner_user_id"]
-	}
-	adminMeta, err = admin.GetAlertMetadata(ctx, self.Meta.Name.Name, ownerID, emailRecipients, anonRecipients, a.Spec.Annotations, a.Spec.GetQueryForUserId(), a.Spec.GetQueryForUserEmail())
-	if err != nil && !errors.Is(err, drivers.ErrNotImplemented) {
-		return fmt.Errorf("failed to get alert metadata: %w", err)
 	}
 
 	// Run alert queries and send notifications
-	executeErr := r.executeAllWrapped(ctx, self, a, adminMeta, triggerTime, adhocTrigger)
+	if executeErr == nil {
+		executeErr = r.executeAllWrapped(ctx, self, a, adminMeta, triggerTime, adhocTrigger)
+	}
 	if executeErr == nil {
 		return nil
 	}
 
 	// If it's a cancellation, don't add the error to the execution history.
 	// The controller may for example cancel if the runtime is restarting or the underlying source is scheduled to refresh.
-	if errors.Is(executeErr, context.Canceled) {
+	if isAlertContextInterruption(executeErr) {
 		// If there's a CurrentExecution, pretend it never happened
 		if a.State.CurrentExecution != nil {
 			a.State.CurrentExecution = nil
@@ -578,7 +583,7 @@ func (r *AlertReconciler) executeAll(ctx context.Context, self *runtimev1.Resour
 		ErrorMessage: executeErr.Error(),
 	}
 	a.State.CurrentExecution.FinishedOn = timestamppb.Now()
-	err = r.popCurrentExecution(ctx, self, a, adminMeta)
+	err := r.popCurrentExecution(ctx, self, a, adminMeta)
 	if err != nil {
 		return err
 	}
@@ -586,6 +591,10 @@ func (r *AlertReconciler) executeAll(ctx context.Context, self *runtimev1.Resour
 	return executeErr
 }
 
+func isAlertContextInterruption(err error) bool {
+	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
+}
+
 // executeAllWrapped is called by executeAll, which wraps it with timeout and writing of errors to the execution history.
 func (r *AlertReconciler) executeAllWrapped(ctx context.Context, self *runtimev1.Resource, a *runtimev1.Alert, adminMeta *drivers.AlertMetadata, triggerTime time.Time, adhocTrigger bool) error {
 	// Check refs
@@ -751,227 +760,153 @@ func (r *AlertReconciler) executeSingleWrapped(ctx context.Context, self *runtim
 	}, nil
 }
 
-// popCurrentExecution moves the current execution into the execution history and sends notifications if the execution matched the notification criteria.
-// At a certain limit, it trims old executions from the history to prevent it from growing unboundedly.
-func (r *AlertReconciler) popCurrentExecution(ctx context.Context, self *runtimev1.Resource, a *runtimev1.Alert, adminMeta *drivers.AlertMetadata) error {
-	if a.State.CurrentExecution == nil {
-		panic(fmt.Errorf("attempting to pop current execution when there is none"))
+// prepareAlertNotification decides whether the current execution should emit a
+// notification. Keeping this state transition separate makes the renotify
+// boundary deterministic and lets malformed persisted history fail closed.
+func prepareAlertNotification(a *runtimev1.Alert, current *runtimev1.AlertExecution) (*drivers.AlertStatus, error) {
+	if a == nil || a.Spec == nil || a.State == nil {
+		return nil, errors.New("alert notification state is incomplete")
+	}
+	if current == nil || current.Result == nil {
+		return nil, errors.New("alert execution result is missing")
 	}
 
-	current := a.State.CurrentExecution
-
-	// td represents the amount of time since we last sent a notification for the current status AND where all intervening executions have returned the same status.
+	// td is the duration since the last notification for an uninterrupted run
+	// of the same status.
 	var td *time.Duration
 	var lastNotifyTime time.Time
 	if current.ExecutionTime != nil {
-		var currT time.Time
-		if current.ExecutionTime != nil {
-			currT = current.ExecutionTime.AsTime()
-		} else {
-			currT = current.FinishedOn.AsTime()
-		}
-
+		currT := current.ExecutionTime.AsTime()
 		for _, prev := range a.State.ExecutionHistory {
-			if prev.Result.Status != current.Result.Status {
+			if prev == nil || prev.Result == nil || prev.Result.Status != current.Result.Status {
 				break
 			}
 			if !prev.SentNotifications {
-				// If notifications were not sent we store since when we are suppressing
 				if prev.SuppressedSince != nil {
 					lastNotifyTime = prev.SuppressedSince.AsTime()
 					v := currT.Sub(lastNotifyTime)
 					td = &v
 					break
 				}
-				// backward compatibility since we did not store the suppressed time earlier
+				// Older versions did not retain a suppression timestamp.
 				continue
 			}
 
-			var prevT time.Time
 			if prev.ExecutionTime != nil {
-				prevT = prev.ExecutionTime.AsTime()
+				lastNotifyTime = prev.ExecutionTime.AsTime()
+			} else if prev.FinishedOn != nil {
+				lastNotifyTime = prev.FinishedOn.AsTime()
 			} else {
-				prevT = prev.FinishedOn.AsTime()
+				break
 			}
-
-			v := currT.Sub(prevT)
+			v := currT.Sub(lastNotifyTime)
 			td = &v
-			lastNotifyTime = prevT
 			break
 		}
 	}
 
-	// Determine if we should notify/renotify using td
-	var notify bool
-	if td == nil {
-		// The status has changed since the last execution, so we should notify.
-		// NOTE: This case may also match in an edge case of execution history limits, but that's fine.
-		notify = true
-	} else if a.Spec.Renotify {
-		if a.Spec.RenotifyAfterSeconds == 0 {
-			// The status has not changed since the last execution and there's no renotify suppression period, so we should notify.
-			notify = true
-		} else if int(td.Seconds()) >= int(a.Spec.RenotifyAfterSeconds) {
-			// The status has not changed since the last notification and the last notification was sent more than the renotify suppression period ago, so we should notify.
+	notify := td == nil
+	if td != nil && a.Spec.Renotify {
+		if a.Spec.RenotifyAfterSeconds == 0 || int(td.Seconds()) >= int(a.Spec.RenotifyAfterSeconds) {
 			notify = true
 		} else {
 			current.SuppressedSince = timestamppb.New(lastNotifyTime)
 		}
 	}
+	if !notify {
+		return nil, nil
+	}
 
-	// Get execution time
 	var executionTime time.Time
 	if current.ExecutionTime != nil {
 		executionTime = current.ExecutionTime.AsTime()
 	}
-
-	// Generate the notification message to send (if any)
-	var msg *drivers.AlertStatus
-	if notify {
-		switch current.Result.Status {
-		case runtimev1.AssertionStatus_ASSERTION_STATUS_PASS:
-			if !a.Spec.NotifyOnRecover {
-				break
+	switch current.Result.Status {
+	case runtimev1.AssertionStatus_ASSERTION_STATUS_PASS:
+		if !a.Spec.NotifyOnRecover || len(a.State.ExecutionHistory) == 0 {
+			return nil, nil
+		}
+		for _, prev := range a.State.ExecutionHistory {
+			if prev == nil || prev.Result == nil {
+				return nil, nil
 			}
-
-			// Check this is a recovery, i.e. that the previous status was something other than a PASS
-			if len(a.State.ExecutionHistory) == 0 {
-				break
+			if prev.Result.Status != runtimev1.AssertionStatus_ASSERTION_STATUS_PASS {
+				return &drivers.AlertStatus{
+					DisplayName:   a.Spec.DisplayName,
+					ExecutionTime: executionTime,
+					Status:        current.Result.Status,
+					IsRecover:     true,
+				}, nil
 			}
-			prev := a.State.ExecutionHistory[0]
-			if prev.Result.Status == runtimev1.AssertionStatus_ASSERTION_STATUS_PASS {
-				break
+			// Retry a recovery notification only when its previous attempt failed
+			// before any recipient succeeded.
+			if prev.SentNotifications || prev.Result.ErrorMessage == "" {
+				return nil, nil
 			}
+		}
+		return nil, nil
+	case runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL:
+		if !a.Spec.NotifyOnFail {
+			return nil, nil
+		}
+		if current.Result.FailRow == nil {
+			return nil, errors.New("failed alert execution is missing its fail row")
+		}
+		return &drivers.AlertStatus{
+			DisplayName:   a.Spec.DisplayName,
+			ExecutionTime: executionTime,
+			Status:        current.Result.Status,
+			FailRow:       current.Result.FailRow.AsMap(),
+		}, nil
+	case runtimev1.AssertionStatus_ASSERTION_STATUS_ERROR:
+		if !a.Spec.NotifyOnError {
+			return nil, nil
+		}
+		return &drivers.AlertStatus{
+			DisplayName:    a.Spec.DisplayName,
+			ExecutionTime:  executionTime,
+			Status:         current.Result.Status,
+			ExecutionError: current.Result.ErrorMessage,
+		}, nil
+	default:
+		return nil, fmt.Errorf("unexpected assertion result status: %v", current.Result.Status)
+	}
+}
 
-			msg = &drivers.AlertStatus{
-				DisplayName:   a.Spec.DisplayName,
-				ExecutionTime: executionTime,
-				Status:        current.Result.Status,
-				IsRecover:     true,
-			}
-		case runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL:
-			if !a.Spec.NotifyOnFail {
-				break
-			}
+// popCurrentExecution moves the current execution into the execution history and sends notifications if the execution matched the notification criteria.
+// At a certain limit, it trims old executions from the history to prevent it from growing unboundedly.
+func (r *AlertReconciler) popCurrentExecution(ctx context.Context, self *runtimev1.Resource, a *runtimev1.Alert, adminMeta *drivers.AlertMetadata) error {
+	if a.State.CurrentExecution == nil {
+		panic(fmt.Errorf("attempting to pop current execution when there is none"))
+	}
 
-			msg = &drivers.AlertStatus{
-				DisplayName:   a.Spec.DisplayName,
-				ExecutionTime: executionTime,
-				Status:        current.Result.Status,
-				FailRow:       current.Result.FailRow.AsMap(),
-			}
-		case runtimev1.AssertionStatus_ASSERTION_STATUS_ERROR:
-			if !a.Spec.NotifyOnError {
-				break
-			}
+	current := a.State.CurrentExecution
+	msg, err := prepareAlertNotification(a, current)
+	if err != nil {
+		return err
+	}
 
-			msg = &drivers.AlertStatus{
-				DisplayName:    a.Spec.DisplayName,
-				ExecutionTime:  executionTime,
-				Status:         current.Result.Status,
-				ExecutionError: current.Result.ErrorMessage,
-			}
-		default:
-			return fmt.Errorf("unexpected assertion result status: %v", current.Result.Status)
-		}
+	var executionTime time.Time
+	if current.ExecutionTime != nil {
+		executionTime = current.ExecutionTime.AsTime()
 	}
 
-	// Send a notification (if applicable)
-	var notificationErr error
+	// Send a notification (if applicable). A context interruption is retryable
+	// only when no external delivery has succeeded yet.
 	var sentNotifications bool
+	var notificationErr error
 	if msg != nil {
-		for _, notifier := range a.Spec.Notifiers {
-			switch notifier.Connector {
-			// TODO: transform email client to notifier
-			case "email":
-				recipients := pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"])
-				for _, recipient := range recipients {
-					msg.ToEmail = recipient
-
-					// Set recipient-specific URLs if available from admin metadata
-					if adminMeta != nil && adminMeta.RecipientURLs != nil {
-						if recipientURLs, ok := adminMeta.RecipientURLs[recipient]; ok {
-							// Use recipient-specific URLs (with magic token)
-							openLink, err := addExecutionTime(recipientURLs.OpenURL, executionTime)
-							if err != nil {
-								return fmt.Errorf("failed to build recipient open url: %w", err)
-							}
-							msg.OpenLink = openLink
-							msg.EditLink = recipientURLs.EditURL
-							msg.UnsubscribeLink = recipientURLs.UnsubscribeURL
-						} else {
-							// Note: adminMeta may not always be available (if outside of cloud) or no links sent for this recipient. In those cases, we leave the links blank (no clickthrough available).
-							msg.OpenLink = ""
-							msg.EditLink = ""
-							msg.UnsubscribeLink = ""
-						}
-					}
-
-					err := r.C.Runtime.Email.SendAlertStatus(msg)
-					if err != nil {
-						notificationErr = fmt.Errorf("failed to send email to %q: %w", recipient, err)
-						break
-					}
-				}
-			default:
-				err := func() (outErr error) {
-					conn, release, err := r.C.Runtime.AcquireHandle(ctx, r.C.InstanceID, notifier.Connector)
-					if err != nil {
-						return err
-					}
-					defer release()
-					n, err := conn.AsNotifier(notifier.Properties.AsMap())
-					if err != nil {
-						return err
-					}
-					// Note: adminMeta may not always be available (if outside of cloud). In that case, we leave the links blank (no clickthrough available).
-					msg.OpenLink = ""
-					msg.EditLink = ""
-					msg.UnsubscribeLink = ""
-					if adminMeta != nil && adminMeta.RecipientURLs != nil {
-						urls, ok := adminMeta.RecipientURLs[""]
-						if !ok {
-							return fmt.Errorf("failed to get recipient URLs for anon user")
-						}
-						openLink, err := addExecutionTime(urls.OpenURL, executionTime)
-						if err != nil {
-							return fmt.Errorf("failed to build recipient open url: %w", err)
-						}
-						msg.OpenLink = openLink
-						msg.EditLink = urls.EditURL
-					}
-					start := time.Now()
-					defer func() {
-						totalLatency := time.Since(start).Milliseconds()
-
-						if r.C.Activity != nil {
-							r.C.Activity.RecordMetric(ctx, "notifier_total_latency_ms", float64(totalLatency),
-								attribute.Bool("failed", outErr != nil),
-								attribute.String("connector", notifier.Connector),
-								attribute.String("notification_type", "alert_status"),
-							)
-						}
-					}()
-					err = n.SendAlertStatus(msg)
-					if err != nil {
-						notificationErr = fmt.Errorf("failed to send %s notification: %w", notifier.Connector, err)
-					}
-					return nil
-				}()
-				if err != nil {
-					return err
-				}
-			}
-		}
-		sentNotifications = true
+		sentNotifications, notificationErr = r.sendAlertNotifications(ctx, a, adminMeta, msg, executionTime)
+	}
+	if notificationErr != nil && !sentNotifications && isAlertContextInterruption(notificationErr) {
+		return notificationErr
 	}
 
-	// If sending notifications failed, add the error as an execution error.
+	// Preserve the assertion status on delivery failure. Changing it to ERROR
+	// would manufacture a status transition and resend already-successful
+	// recipients on the next identical alert execution.
 	if notificationErr != nil {
-		a.State.CurrentExecution.Result = &runtimev1.AssertionResult{
-			Status:       runtimev1.AssertionStatus_ASSERTION_STATUS_ERROR,
-			ErrorMessage: notificationErr.Error(),
-		}
+		a.State.CurrentExecution.Result.ErrorMessage = notificationErr.Error()
 	}
 
 	a.State.CurrentExecution.SentNotifications = sentNotifications
@@ -987,6 +922,85 @@ func (r *AlertReconciler) popCurrentExecution(ctx context.Context, self *runtime
 	return r.C.UpdateState(ctx, self.Meta.Name, self)
 }
 
+func (r *AlertReconciler) sendAlertNotifications(ctx context.Context, a *runtimev1.Alert, adminMeta *drivers.AlertMetadata, base *drivers.AlertStatus, executionTime time.Time) (bool, error) {
+	sentAny := false
+	for _, notifier := range a.Spec.Notifiers {
+		switch notifier.Connector {
+		case "email": // TODO: transform email client to notifier
+			recipients := pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"])
+			for _, recipient := range recipients {
+				if err := ctx.Err(); err != nil {
+					return sentAny, err
+				}
+
+				msg := *base
+				msg.ToEmail = recipient
+				if adminMeta != nil && adminMeta.RecipientURLs != nil {
+					if recipientURLs, ok := adminMeta.RecipientURLs[recipient]; ok {
+						openLink, err := addExecutionTime(recipientURLs.OpenURL, executionTime)
+						if err != nil {
+							return sentAny, fmt.Errorf("failed to build recipient open url: %w", err)
+						}
+						msg.OpenLink = openLink
+						msg.EditLink = recipientURLs.EditURL
+						msg.UnsubscribeLink = recipientURLs.UnsubscribeURL
+					}
+				}
+
+				if err := r.C.Runtime.Email.SendAlertStatus(&msg); err != nil {
+					return sentAny, fmt.Errorf("failed to send email to %q: %w", recipient, err)
+				}
+				sentAny = true
+			}
+		default:
+			if err := ctx.Err(); err != nil {
+				return sentAny, err
+			}
+			conn, release, err := r.C.Runtime.AcquireHandle(ctx, r.C.InstanceID, notifier.Connector)
+			if err != nil {
+				return sentAny, err
+			}
+			n, err := conn.AsNotifier(notifier.Properties.AsMap())
+			if err != nil {
+				release()
+				return sentAny, err
+			}
+
+			msg := *base
+			if adminMeta != nil && adminMeta.RecipientURLs != nil {
+				urls, ok := adminMeta.RecipientURLs[""]
+				if !ok {
+					release()
+					return sentAny, fmt.Errorf("failed to get recipient URLs for anon user")
+				}
+				openLink, err := addExecutionTime(urls.OpenURL, executionTime)
+				if err != nil {
+					release()
+					return sentAny, fmt.Errorf("failed to build recipient open url: %w", err)
+				}
+				msg.OpenLink = openLink
+				msg.EditLink = urls.EditURL
+			}
+
+			start := time.Now()
+			err = n.SendAlertStatus(&msg)
+			release()
+			if r.C.Activity != nil {
+				r.C.Activity.RecordMetric(ctx, "notifier_total_latency_ms", float64(time.Since(start).Milliseconds()),
+					attribute.Bool("failed", err != nil),
+					attribute.String("connector", notifier.Connector),
+					attribute.String("notification_type", "alert_status"),
+				)
+			}
+			if err != nil {
+				return sentAny, fmt.Errorf("failed to send %s notification: %w", notifier.Connector, err)
+			}
+			sentAny = true
+		}
+	}
+	return sentAny, nil
+}
+
 // computeInheritedWatermark computes the inherited watermark for the alert.
 // It returns false if the watermark could not be computed.
 func (r *AlertReconciler) computeInheritedWatermark(ctx context.Context, refs []*runtimev1.ResourceName) (time.Time, bool, error) {
@@ -1079,7 +1093,7 @@ func calculateAlertExecutionTimes(a *runtimev1.Alert, watermark, previousWaterma
 
 	// Calculate the execution times
 	ts := []time.Time{end}
-	for i := 0; i < limit; i++ {
+	for len(ts) < limit {
 		t := ts[len(ts)-1]
 		t = d.Sub(t)
 		if !t.After(previousWatermark) {
@@ -1119,8 +1133,9 @@ func (s skipError) Error() string {
 }
 
 func latestAlertWarnings(a *runtimev1.Alert) []string {
-	if a.State != nil && len(a.State.ExecutionHistory) > 0 {
-		return a.State.ExecutionHistory[len(a.State.ExecutionHistory)-1].Result.Warnings
+	if a.State != nil && len(a.State.ExecutionHistory) > 0 && a.State.ExecutionHistory[0].Result != nil {
+		// Execution history is newest-first; popCurrentExecution inserts at index 0.
+		return a.State.ExecutionHistory[0].Result.Warnings
 	}
 	return nil
 }
diff --git a/runtime/reconcilers/alert_failure_test.go b/runtime/reconcilers/alert_failure_test.go
new file mode 100644
index 00000000000..21391bd199c
--- /dev/null
+++ b/runtime/reconcilers/alert_failure_test.go
@@ -0,0 +1,361 @@
+package reconcilers
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
+	"github.com/rilldata/rill/runtime"
+	"github.com/rilldata/rill/runtime/drivers"
+	"github.com/rilldata/rill/runtime/pkg/email"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func TestCalculateAlertExecutionTimesHonorsConfiguredLimit(t *testing.T) {
+	// The interval limit counts actual checks. An off-by-one here can fan out one
+	// extra query and notification every time a delayed alert catches up.
+	alert := &runtimev1.Alert{Spec: &runtimev1.AlertSpec{
+		IntervalsIsoDuration: "P1D",
+		IntervalsLimit:       2,
+	}}
+	watermark := time.Date(2026, time.January, 10, 12, 0, 0, 0, time.UTC)
+	previous := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+
+	got, err := calculateAlertExecutionTimes(alert, watermark, previous)
+	require.NoError(t, err)
+	require.Equal(t, []time.Time{
+		time.Date(2026, time.January, 9, 0, 0, 0, 0, time.UTC),
+		time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC),
+	}, got)
+}
+
+func TestCalculateAlertExecutionTimesFailureBoundaries(t *testing.T) {
+	watermark := time.Date(2026, time.January, 10, 12, 0, 0, 0, time.UTC)
+
+	// Reconciler inputs can bypass parser validation during recovery or upgrades,
+	// so malformed durations and time zones must return errors rather than panic.
+	tests := []struct {
+		name     string
+		interval string
+		timezone string
+		want     string
+	}{
+		{name: "malformed duration", interval: "not-a-duration", want: "failed to parse interval duration"},
+		{name: "nonstandard duration", interval: "inf", want: "is not a standard ISO 8601 duration"},
+		{name: "invalid timezone", interval: "P1D", timezone: "Mars/Olympus", want: "failed to load time zone"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Each malformed persisted scheduling value must produce a stable,
+			// actionable error instead of panicking the reconciler.
+			alert := &runtimev1.Alert{Spec: &runtimev1.AlertSpec{
+				IntervalsIsoDuration: tt.interval,
+				RefreshSchedule:      &runtimev1.Schedule{TimeZone: tt.timezone},
+			}}
+			_, err := calculateAlertExecutionTimes(alert, watermark, time.Time{})
+			require.ErrorContains(t, err, tt.want)
+		})
+	}
+}
+
+func TestLatestAlertWarningsUsesNewestValidHistoryEntry(t *testing.T) {
+	// Histories are newest-first. Reading the tail exposes stale warnings, while
+	// a partially persisted nil result must not crash reconciliation.
+	alert := &runtimev1.Alert{State: &runtimev1.AlertState{ExecutionHistory: []*runtimev1.AlertExecution{
+		{Result: &runtimev1.AssertionResult{Warnings: []string{"newest warning"}}},
+		{Result: &runtimev1.AssertionResult{Warnings: []string{"oldest warning"}}},
+	}}}
+	require.Equal(t, []string{"newest warning"}, latestAlertWarnings(alert))
+	require.Nil(t, latestAlertWarnings(&runtimev1.Alert{State: &runtimev1.AlertState{ExecutionHistory: []*runtimev1.AlertExecution{{}}}}))
+}
+
+func TestPrepareAlertNotificationRenotifyBoundaries(t *testing.T) {
+	// Renotify uses the last actual send time, and the configured delay is
+	// inclusive: 59 seconds is suppressed while exactly 60 seconds is sent.
+	lastSent := time.Date(2026, time.July, 23, 10, 0, 0, 0, time.UTC)
+	alert := &runtimev1.Alert{
+		Spec: &runtimev1.AlertSpec{
+			DisplayName:          "Orders failed",
+			NotifyOnFail:         true,
+			Renotify:             true,
+			RenotifyAfterSeconds: 60,
+		},
+		State: &runtimev1.AlertState{ExecutionHistory: []*runtimev1.AlertExecution{{
+			Result:            alertFailureResult(t),
+			ExecutionTime:     timestamppb.New(lastSent),
+			SentNotifications: true,
+		}}},
+	}
+
+	suppressed := &runtimev1.AlertExecution{
+		Result:        alertFailureResult(t),
+		ExecutionTime: timestamppb.New(lastSent.Add(59 * time.Second)),
+	}
+	msg, err := prepareAlertNotification(alert, suppressed)
+	require.NoError(t, err)
+	require.Nil(t, msg)
+	require.Equal(t, lastSent, suppressed.SuppressedSince.AsTime())
+
+	atBoundary := &runtimev1.AlertExecution{
+		Result:        alertFailureResult(t),
+		ExecutionTime: timestamppb.New(lastSent.Add(60 * time.Second)),
+	}
+	msg, err = prepareAlertNotification(alert, atBoundary)
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+	require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, msg.Status)
+}
+
+func TestPrepareAlertNotificationRenotifyModes(t *testing.T) {
+	// Zero seconds has different meaning when renotify is enabled, so exercise
+	// both modes without relying on elapsed-time rounding.
+	lastSent := time.Date(2026, time.July, 23, 10, 0, 0, 0, time.UTC)
+	tests := []struct {
+		name       string
+		renotify   bool
+		after      uint32
+		wantNotify bool
+	}{
+		{name: "disabled suppresses unchanged status", wantNotify: false},
+		{name: "zero delay resends unchanged status", renotify: true, wantNotify: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Renotify policy must distinguish a disabled policy from an enabled
+			// zero-second policy even though both store a zero duration.
+			alert := alertForNotification(&runtimev1.AlertSpec{
+				NotifyOnFail:         true,
+				Renotify:             tt.renotify,
+				RenotifyAfterSeconds: tt.after,
+			}, []*runtimev1.AlertExecution{{
+				Result:            alertFailureResult(t),
+				ExecutionTime:     timestamppb.New(lastSent),
+				SentNotifications: true,
+			}})
+			msg, err := prepareAlertNotification(alert, alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, lastSent.Add(time.Second)))
+			require.NoError(t, err)
+			require.Equal(t, tt.wantNotify, msg != nil)
+		})
+	}
+}
+
+func TestSendAlertNotificationsStopsBeforeCanceledDelivery(t *testing.T) {
+	// A pre-existing context interruption must be detected before the first
+	// transport call for both cancellation and deadline expiry.
+	tests := []struct {
+		name string
+		ctx  func() context.Context
+		err  error
+	}{
+		{
+			name: "cancellation",
+			ctx: func() context.Context {
+				ctx, cancel := context.WithCancel(context.Background())
+				cancel()
+				return ctx
+			},
+			err: context.Canceled,
+		},
+		{
+			name: "deadline",
+			ctx: func() context.Context {
+				ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
+				cancel()
+				return ctx
+			},
+			err: context.DeadlineExceeded,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// An interruption already visible before the first side effect must
+			// leave the delivery clean and must not call the email transport.
+			sender := &alertCallbackSender{}
+			reconciler, alert, msg := newAlertDeliverySeam(t, sender, "first@example.com")
+			sent, err := reconciler.sendAlertNotifications(tt.ctx(), alert, nil, msg, msg.ExecutionTime)
+			require.ErrorIs(t, err, tt.err)
+			require.False(t, sent)
+			require.Empty(t, sender.recipients)
+		})
+	}
+}
+
+func TestSendAlertNotificationsMarksInterruptionAfterDeliveryDirty(t *testing.T) {
+	// Once one recipient succeeds, later interruption must be reported as dirty
+	// so callers can avoid replaying that successful side effect.
+	t.Run("cancellation", func(t *testing.T) {
+		// Cancellation observed after the first successful recipient is dirty;
+		// the second recipient must not be attempted by this invocation.
+		ctx, cancel := context.WithCancel(context.Background())
+		sender := &alertCallbackSender{afterSend: func(call int) {
+			if call == 1 {
+				cancel()
+			}
+		}}
+		reconciler, alert, msg := newAlertDeliverySeam(t, sender, "first@example.com", "second@example.com")
+		sent, err := reconciler.sendAlertNotifications(ctx, alert, nil, msg, msg.ExecutionTime)
+		require.ErrorIs(t, err, context.Canceled)
+		require.True(t, sent)
+		require.Equal(t, []string{"first@example.com"}, sender.recipients)
+	})
+
+	t.Run("deadline", func(t *testing.T) {
+		// A deadline that expires immediately after the first transport success
+		// has the same dirty semantics as cancellation.
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond)
+		defer cancel()
+		sender := &alertCallbackSender{afterSend: func(call int) {
+			if call == 1 {
+				<-ctx.Done()
+			}
+		}}
+		reconciler, alert, msg := newAlertDeliverySeam(t, sender, "first@example.com", "second@example.com")
+		sent, err := reconciler.sendAlertNotifications(ctx, alert, nil, msg, msg.ExecutionTime)
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+		require.True(t, sent)
+		require.Equal(t, []string{"first@example.com"}, sender.recipients)
+	})
+}
+
+func TestPrepareAlertNotificationStatusTransitions(t *testing.T) {
+	// Notification policy distinguishes an initial pass from a real recovery,
+	// while fail and error transitions retain their semantic payloads.
+	executionTime := time.Date(2026, time.July, 23, 11, 0, 0, 0, time.UTC)
+	t.Run("initial pass is silent", func(t *testing.T) {
+		// A pass without a preceding non-pass status is not a recovery event.
+		alert := alertForNotification(&runtimev1.AlertSpec{NotifyOnRecover: true}, nil)
+		msg, err := prepareAlertNotification(alert, alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_PASS, executionTime))
+		require.NoError(t, err)
+		require.Nil(t, msg)
+	})
+
+	t.Run("recovery follows a failure", func(t *testing.T) {
+		// A pass immediately after failure carries the recovery marker.
+		previous := alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, executionTime.Add(-time.Minute))
+		alert := alertForNotification(&runtimev1.AlertSpec{DisplayName: "Orders", NotifyOnRecover: true}, []*runtimev1.AlertExecution{previous})
+		msg, err := prepareAlertNotification(alert, alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_PASS, executionTime))
+		require.NoError(t, err)
+		require.True(t, msg.IsRecover)
+		require.Equal(t, "Orders", msg.DisplayName)
+	})
+
+	t.Run("cleanly failed recovery delivery retries", func(t *testing.T) {
+		// A recovery send that reached no recipient remains recoverable, while
+		// retaining PASS as the actual assertion status in persisted history.
+		failedRecovery := alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_PASS, executionTime.Add(-time.Minute))
+		failedRecovery.Result.ErrorMessage = "failed to send recovery email"
+		previousFailure := alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, executionTime.Add(-2*time.Minute))
+		alert := alertForNotification(&runtimev1.AlertSpec{NotifyOnRecover: true}, []*runtimev1.AlertExecution{failedRecovery, previousFailure})
+		msg, err := prepareAlertNotification(alert, alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_PASS, executionTime))
+		require.NoError(t, err)
+		require.True(t, msg.IsRecover)
+	})
+
+	t.Run("error carries execution failure", func(t *testing.T) {
+		// Evaluation errors retain their diagnostic text in the outgoing payload.
+		alert := alertForNotification(&runtimev1.AlertSpec{NotifyOnError: true}, nil)
+		current := alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_ERROR, executionTime)
+		current.Result.ErrorMessage = "query unavailable"
+		msg, err := prepareAlertNotification(alert, current)
+		require.NoError(t, err)
+		require.Equal(t, "query unavailable", msg.ExecutionError)
+	})
+
+	t.Run("disabled failure notification is silent", func(t *testing.T) {
+		// A failing assertion must remain silent when fail notifications are off.
+		alert := alertForNotification(&runtimev1.AlertSpec{NotifyOnFail: false}, nil)
+		msg, err := prepareAlertNotification(alert, alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, executionTime))
+		require.NoError(t, err)
+		require.Nil(t, msg)
+	})
+}
+
+func TestPrepareAlertNotificationMalformedPersistedState(t *testing.T) {
+	// Reconciliation can encounter partially persisted state after a crash, so
+	// missing results and fail rows must return errors instead of panicking.
+	alert := alertForNotification(&runtimev1.AlertSpec{NotifyOnFail: true}, nil)
+	_, err := prepareAlertNotification(alert, &runtimev1.AlertExecution{})
+	require.ErrorContains(t, err, "result is missing")
+
+	current := alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, time.Now())
+	current.Result.FailRow = nil
+	_, err = prepareAlertNotification(alert, current)
+	require.ErrorContains(t, err, "fail row")
+
+	current = alertExecution(runtimev1.AssertionStatus(99), time.Now())
+	_, err = prepareAlertNotification(alert, current)
+	require.ErrorContains(t, err, "unexpected assertion result status")
+
+	// A malformed history entry is treated as a status boundary rather than
+	// dereferenced, allowing the current valid failure to notify safely.
+	alert.State.ExecutionHistory = []*runtimev1.AlertExecution{nil}
+	msg, err := prepareAlertNotification(alert, alertExecution(runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, time.Now()))
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+}
+
+func TestAddExecutionTimePreservesAndValidatesURL(t *testing.T) {
+	// Recipient links retain existing query parameters and fragments, and a
+	// malformed encoded query is rejected before any notification is sent.
+	executionTime := time.Date(2026, time.July, 23, 12, 30, 0, 0, time.FixedZone("offset", 5*60*60+30*60))
+	got, err := addExecutionTime("https://example.com/explore?foo=bar#chart", executionTime)
+	require.NoError(t, err)
+	require.Equal(t, "https://example.com/explore?execution_time=2026-07-23T07%3A00%3A00Z&foo=bar#chart", got)
+
+	_, err = addExecutionTime("https://example.com/explore?bad=%zz", executionTime)
+	require.Error(t, err)
+}
+
+func alertFailureResult(t *testing.T) *runtimev1.AssertionResult {
+	t.Helper()
+	row, err := structpb.NewStruct(map[string]any{"orders": 1})
+	require.NoError(t, err)
+	return &runtimev1.AssertionResult{Status: runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, FailRow: row}
+}
+
+func alertExecution(status runtimev1.AssertionStatus, executionTime time.Time) *runtimev1.AlertExecution {
+	result := &runtimev1.AssertionResult{Status: status}
+	if status == runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL {
+		result.FailRow, _ = structpb.NewStruct(map[string]any{"orders": 1})
+	}
+	return &runtimev1.AlertExecution{Result: result, ExecutionTime: timestamppb.New(executionTime)}
+}
+
+func alertForNotification(spec *runtimev1.AlertSpec, history []*runtimev1.AlertExecution) *runtimev1.Alert {
+	return &runtimev1.Alert{Spec: spec, State: &runtimev1.AlertState{ExecutionHistory: history}}
+}
+
+type alertCallbackSender struct {
+	recipients []string
+	afterSend  func(call int)
+}
+
+func (s *alertCallbackSender) Send(toEmail, _, _, _ string) error {
+	s.recipients = append(s.recipients, toEmail)
+	if s.afterSend != nil {
+		s.afterSend(len(s.recipients))
+	}
+	return nil
+}
+
+func newAlertDeliverySeam(t *testing.T, sender email.Sender, recipients ...string) (*AlertReconciler, *runtimev1.Alert, *drivers.AlertStatus) {
+	t.Helper()
+	values := make([]any, len(recipients))
+	for i, recipient := range recipients {
+		values[i] = recipient
+	}
+	properties, err := structpb.NewStruct(map[string]any{"recipients": values})
+	require.NoError(t, err)
+	reconciler := &AlertReconciler{C: &runtime.Controller{Runtime: &runtime.Runtime{Email: email.New(sender)}}}
+	alert := &runtimev1.Alert{Spec: &runtimev1.AlertSpec{Notifiers: []*runtimev1.Notifier{{Connector: "email", Properties: properties}}}}
+	msg := &drivers.AlertStatus{
+		DisplayName:   "Delivery seam",
+		ExecutionTime: time.Date(2026, time.July, 23, 12, 0, 0, 0, time.UTC),
+		Status:        runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL,
+		FailRow:       map[string]any{"orders": 1},
+	}
+	return reconciler, alert, msg
+}
diff --git a/runtime/reconcilers/alert_transition_test.go b/runtime/reconcilers/alert_transition_test.go
new file mode 100644
index 00000000000..de020867dc7
--- /dev/null
+++ b/runtime/reconcilers/alert_transition_test.go
@@ -0,0 +1,615 @@
+package reconcilers_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
+	"github.com/rilldata/rill/runtime"
+	"github.com/rilldata/rill/runtime/drivers"
+	"github.com/rilldata/rill/runtime/pkg/activity"
+	"github.com/rilldata/rill/runtime/pkg/email"
+	"github.com/rilldata/rill/runtime/storage"
+	"github.com/rilldata/rill/runtime/testruntime"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+const (
+	alertTransitionResolverName = "alert_transition_test"
+	alertTransitionDriverName   = "alert_transition_test"
+	alertTransitionAdminName    = "alert_transition_admin"
+	alertTransitionNotifierName = "alert_transition_notifier"
+)
+
+var (
+	alertTransitionScenarios sync.Map
+	alertTransitionSequence  atomic.Uint64
+)
+
+func init() {
+	runtime.RegisterResolverInitializer(alertTransitionResolverName, newAlertTransitionResolver)
+	drivers.Register(alertTransitionDriverName, &alertTransitionDriver{})
+}
+
+func TestAlertResolverFailuresCommitExecutionState(t *testing.T) {
+	// Resolver setup and iteration errors are alert outcomes: each must finish
+	// one execution, advance scheduling, and clear the one-shot trigger.
+	tests := []struct {
+		name         string
+		configure    func(*alertTransitionScenario)
+		wantError    string
+		wantValidate int
+		wantResolve  int
+		wantNext     int
+	}{
+		{
+			name: "initializer failure",
+			configure: func(s *alertTransitionScenario) {
+				s.initializerErr = errors.New("initializer unavailable")
+			},
+			wantError: "initializer unavailable",
+		},
+		{
+			name: "validation failure",
+			configure: func(s *alertTransitionScenario) {
+				s.validateErr = errors.New("resolver properties invalid")
+			},
+			wantError:    "resolver properties invalid",
+			wantValidate: 1,
+		},
+		{
+			name: "next failure",
+			configure: func(s *alertTransitionScenario) {
+				s.nextErr = errors.New("row stream interrupted")
+			},
+			wantError:    "failed to get row from alert resolver: row stream interrupted",
+			wantValidate: 1,
+			wantResolve:  1,
+			wantNext:     1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Phase-specific call counts prove the failure is handled at the
+			// intended resolver boundary instead of by an earlier shortcut.
+			scenario, scenarioID := newAlertTransitionScenario(t)
+			tt.configure(scenario)
+			fixture := newAlertTransitionFixture(t, scenarioID, nil, alertTransitionSpec(t, scenarioID), false, false)
+
+			res := requireAlertCompleted(t, fixture, 1)
+			execution := res.GetAlert().State.ExecutionHistory[0]
+			require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_ERROR, execution.Result.Status)
+			require.ErrorContains(t, errors.New(execution.Result.ErrorMessage), tt.wantError)
+			require.NotNil(t, execution.ExecutionTime)
+			require.Empty(t, res.Meta.ReconcileError)
+
+			snapshot := scenario.snapshot()
+			require.Equal(t, 1, snapshot.initializerCalls)
+			require.Equal(t, tt.wantValidate, snapshot.validateCalls)
+			require.Equal(t, tt.wantResolve, snapshot.resolveCalls)
+			require.Equal(t, tt.wantNext, snapshot.nextCalls)
+		})
+	}
+}
+
+func TestAlertAdminMetadataFailureCommitsErrorHistory(t *testing.T) {
+	// Admin failure happens before query initialization, but it is still a
+	// completed alert attempt and must be visible in history and reconcile state.
+	scenario, scenarioID := newAlertTransitionScenario(t)
+	scenario.adminErr = errors.New("admin metadata unavailable")
+	fixture := newAlertTransitionFixture(t, scenarioID, nil, alertTransitionSpec(t, scenarioID), true, false)
+
+	res := requireAlertCompleted(t, fixture, 1)
+	execution := res.GetAlert().State.ExecutionHistory[0]
+	require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_ERROR, execution.Result.Status)
+	require.ErrorContains(t, errors.New(execution.Result.ErrorMessage), "failed to get alert metadata: admin metadata unavailable")
+	require.Nil(t, execution.ExecutionTime)
+	require.ErrorContains(t, errors.New(res.Meta.ReconcileError), "failed to get alert metadata")
+
+	snapshot := scenario.snapshot()
+	require.Equal(t, 1, snapshot.adminCalls)
+	require.Zero(t, snapshot.initializerCalls)
+}
+
+func TestAlertEmailFailureRetriesOnlyWhenNoDeliverySucceeded(t *testing.T) {
+	// A failed first recipient leaves SentNotifications false, allowing the next
+	// explicit execution to retry without manufacturing an assertion transition.
+	scenario, scenarioID := newAlertTransitionScenario(t)
+	scenario.rows = []map[string]any{{"orders": 1}}
+	sender := &alertTransitionEmailSender{errs: []error{errors.New("smtp unavailable"), nil}}
+	spec := alertTransitionSpec(t, scenarioID)
+	spec.NotifyOnFail = true
+	spec.Notifiers = []*runtimev1.Notifier{alertTransitionEmailNotifier(t, "ops@example.com")}
+	fixture := newAlertTransitionFixture(t, scenarioID, sender, spec, false, false)
+
+	res := requireAlertCompleted(t, fixture, 1)
+	first := res.GetAlert().State.ExecutionHistory[0]
+	require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, first.Result.Status)
+	require.ErrorContains(t, errors.New(first.Result.ErrorMessage), "smtp unavailable")
+	require.False(t, first.SentNotifications)
+	require.Equal(t, []string{"ops@example.com"}, sender.snapshot())
+
+	triggerAlertAgain(t, fixture)
+	res = requireAlertCompleted(t, fixture, 2)
+	second := res.GetAlert().State.ExecutionHistory[0]
+	require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, second.Result.Status)
+	require.Empty(t, second.Result.ErrorMessage)
+	require.True(t, second.SentNotifications)
+	require.Equal(t, []string{"ops@example.com", "ops@example.com"}, sender.snapshot())
+}
+
+func TestAlertPartialMultiNotifierFailureDoesNotDuplicateSuccess(t *testing.T) {
+	// Email succeeds before the non-email notifier fails. The dirty marker must
+	// suppress both transports on the next unchanged failure.
+	scenario, scenarioID := newAlertTransitionScenario(t)
+	scenario.rows = []map[string]any{{"orders": 1}}
+	scenario.notifierErrs = []error{errors.New("webhook unavailable")}
+	sender := &alertTransitionEmailSender{}
+	spec := alertTransitionSpec(t, scenarioID)
+	spec.NotifyOnFail = true
+	spec.Notifiers = []*runtimev1.Notifier{
+		alertTransitionEmailNotifier(t, "ops@example.com"),
+		{Connector: alertTransitionNotifierName, Properties: alertTransitionProperties(t, map[string]any{})},
+	}
+	fixture := newAlertTransitionFixture(t, scenarioID, sender, spec, false, true)
+
+	res := requireAlertCompleted(t, fixture, 1)
+	first := res.GetAlert().State.ExecutionHistory[0]
+	require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, first.Result.Status)
+	require.ErrorContains(t, errors.New(first.Result.ErrorMessage), "webhook unavailable")
+	require.True(t, first.SentNotifications)
+	require.Equal(t, []string{"ops@example.com"}, sender.snapshot())
+	require.Equal(t, 1, scenario.snapshot().notifierCalls)
+
+	triggerAlertAgain(t, fixture)
+	res = requireAlertCompleted(t, fixture, 2)
+	second := res.GetAlert().State.ExecutionHistory[0]
+	require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, second.Result.Status)
+	require.False(t, second.SentNotifications)
+	require.Empty(t, second.Result.ErrorMessage)
+	require.Equal(t, []string{"ops@example.com"}, sender.snapshot())
+	require.Equal(t, 1, scenario.snapshot().notifierCalls)
+}
+
+func TestAlertContextInterruptionBeforeDeliveryRemainsRetryable(t *testing.T) {
+	// Clean interruption should preserve the ad-hoc trigger and empty history so
+	// a later controller invocation can complete exactly one delivery.
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{name: "cancellation", err: context.Canceled},
+		{name: "deadline", err: context.DeadlineExceeded},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// With no external side effect, cancellation and deadline both leave
+			// the execution clean, keep Trigger set, and retry exactly once later.
+			scenario, scenarioID := newAlertTransitionScenario(t)
+			scenario.resolveErr = tt.err
+			sender := &alertTransitionEmailSender{}
+			spec := alertTransitionSpec(t, scenarioID)
+			spec.NotifyOnFail = true
+			spec.Notifiers = []*runtimev1.Notifier{alertTransitionEmailNotifier(t, "ops@example.com")}
+			fixture := newAlertTransitionFixture(t, scenarioID, sender, spec, false, false)
+
+			res := requireAlertRetryableInterruption(t, fixture)
+			require.Contains(t, res.Meta.ReconcileError, tt.err.Error())
+			require.Empty(t, sender.snapshot())
+
+			scenario.setResolveOutcome(nil, []map[string]any{{"orders": 1}})
+			reconcileAlert(t, fixture)
+			res = requireAlertCompleted(t, fixture, 1)
+			require.True(t, res.GetAlert().State.ExecutionHistory[0].SentNotifications)
+			require.Equal(t, []string{"ops@example.com"}, sender.snapshot())
+		})
+	}
+}
+
+func TestAlertContextInterruptionAfterDeliveryDoesNotRetry(t *testing.T) {
+	// Dirty interruption should commit history and scheduling so reconciliation
+	// cannot automatically replay an already-successful recipient.
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{name: "cancellation", err: context.Canceled},
+		{name: "deadline", err: context.DeadlineExceeded},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// The first recipient succeeds and the second reports interruption;
+			// the committed dirty execution must suppress a duplicate first send.
+			scenario, scenarioID := newAlertTransitionScenario(t)
+			scenario.rows = []map[string]any{{"orders": 1}}
+			sender := &alertTransitionEmailSender{errs: []error{nil, tt.err}}
+			spec := alertTransitionSpec(t, scenarioID)
+			spec.NotifyOnFail = true
+			spec.Notifiers = []*runtimev1.Notifier{alertTransitionEmailNotifier(t, "first@example.com", "second@example.com")}
+			fixture := newAlertTransitionFixture(t, scenarioID, sender, spec, false, false)
+
+			res := requireAlertCompleted(t, fixture, 1)
+			first := res.GetAlert().State.ExecutionHistory[0]
+			require.Equal(t, runtimev1.AssertionStatus_ASSERTION_STATUS_FAIL, first.Result.Status)
+			require.Contains(t, first.Result.ErrorMessage, tt.err.Error())
+			require.True(t, first.SentNotifications)
+			require.Equal(t, []string{"first@example.com", "second@example.com"}, sender.snapshot())
+
+			triggerAlertAgain(t, fixture)
+			res = requireAlertCompleted(t, fixture, 2)
+			require.False(t, res.GetAlert().State.ExecutionHistory[0].SentNotifications)
+			require.Equal(t, []string{"first@example.com", "second@example.com"}, sender.snapshot())
+		})
+	}
+}
+
+type alertTransitionFixture struct {
+	rt   *runtime.Runtime
+	id   string
+	ctrl *runtime.Controller
+	name *runtimev1.ResourceName
+}
+
+func newAlertTransitionFixture(t *testing.T, scenarioID string, sender email.Sender, spec *runtimev1.AlertSpec, useAdmin, useNotifier bool) *alertTransitionFixture {
+	t.Helper()
+	rt, id := testruntime.NewInstance(t)
+	if sender != nil {
+		rt.Email = email.New(sender)
+	}
+
+	inst, err := rt.Instance(t.Context(), id)
+	require.NoError(t, err)
+	edited := *inst
+	edited.Connectors = append([]*runtimev1.Connector(nil), inst.Connectors...)
+	properties := alertTransitionProperties(t, map[string]any{"scenario": scenarioID})
+	if useAdmin {
+		edited.AdminConnector = alertTransitionAdminName
+		edited.Connectors = append(edited.Connectors, &runtimev1.Connector{Name: alertTransitionAdminName, Type: alertTransitionDriverName, Config: properties})
+	}
+	if useNotifier {
+		edited.Connectors = append(edited.Connectors, &runtimev1.Connector{Name: alertTransitionNotifierName, Type: alertTransitionDriverName, Config: properties})
+	}
+	require.NoError(t, rt.EditInstance(t.Context(), &edited, false))
+
+	ctrl, err := rt.Controller(t.Context(), id)
+	require.NoError(t, err)
+	name := &runtimev1.ResourceName{Kind: runtime.ResourceKindAlert, Name: "transition"}
+	require.NoError(t, ctrl.Create(t.Context(), name, nil, nil, nil, nil, false, &runtimev1.Resource{
+		Resource: &runtimev1.Resource_Alert{Alert: &runtimev1.Alert{Spec: spec, State: &runtimev1.AlertState{}}},
+	}))
+	require.NoError(t, ctrl.WaitUntilIdle(t.Context(), false))
+	return &alertTransitionFixture{rt: rt, id: id, ctrl: ctrl, name: name}
+}
+
+func alertTransitionSpec(t *testing.T, scenarioID string) *runtimev1.AlertSpec {
+	t.Helper()
+	return &runtimev1.AlertSpec{
+		DisplayName:        "Transition test",
+		Trigger:            true,
+		RefreshSchedule:    &runtimev1.Schedule{TickerSeconds: 3600},
+		Resolver:           alertTransitionResolverName,
+		ResolverProperties: alertTransitionProperties(t, map[string]any{"scenario": scenarioID}),
+	}
+}
+
+func alertTransitionEmailNotifier(t *testing.T, recipients ...string) *runtimev1.Notifier {
+	t.Helper()
+	values := make([]any, len(recipients))
+	for i, recipient := range recipients {
+		values[i] = recipient
+	}
+	return &runtimev1.Notifier{
+		Connector:  "email",
+		Properties: alertTransitionProperties(t, map[string]any{"recipients": values}),
+	}
+}
+
+func alertTransitionProperties(t *testing.T, values map[string]any) *structpb.Struct {
+	t.Helper()
+	properties, err := structpb.NewStruct(values)
+	require.NoError(t, err)
+	return properties
+}
+
+func requireAlertCompleted(t *testing.T, fixture *alertTransitionFixture, count uint32) *runtimev1.Resource {
+	t.Helper()
+	res, err := fixture.ctrl.Get(t.Context(), fixture.name, true)
+	require.NoError(t, err)
+	alert := res.GetAlert()
+	require.Nil(t, alert.State.CurrentExecution)
+	require.Equal(t, count, alert.State.ExecutionCount)
+	require.Len(t, alert.State.ExecutionHistory, int(count))
+	require.NotNil(t, alert.State.NextRunOn)
+	require.True(t, alert.State.NextRunOn.AsTime().After(time.Now()))
+	require.False(t, alert.Spec.Trigger)
+	require.NotNil(t, res.Meta.ReconcileOn)
+	require.Equal(t, alert.State.NextRunOn.AsTime(), res.Meta.ReconcileOn.AsTime())
+	return res
+}
+
+func requireAlertRetryableInterruption(t *testing.T, fixture *alertTransitionFixture) *runtimev1.Resource {
+	t.Helper()
+	res, err := fixture.ctrl.Get(t.Context(), fixture.name, true)
+	require.NoError(t, err)
+	alert := res.GetAlert()
+	require.Nil(t, alert.State.CurrentExecution)
+	require.Zero(t, alert.State.ExecutionCount)
+	require.Empty(t, alert.State.ExecutionHistory)
+	require.Nil(t, alert.State.NextRunOn)
+	require.True(t, alert.Spec.Trigger)
+	return res
+}
+
+func triggerAlertAgain(t *testing.T, fixture *alertTransitionFixture) {
+	t.Helper()
+	res, err := fixture.ctrl.Get(t.Context(), fixture.name, true)
+	require.NoError(t, err)
+	res.GetAlert().Spec.Trigger = true
+	require.NoError(t, fixture.ctrl.UpdateSpec(t.Context(), fixture.name, res))
+	require.NoError(t, fixture.ctrl.WaitUntilIdle(t.Context(), false))
+}
+
+func reconcileAlert(t *testing.T, fixture *alertTransitionFixture) {
+	t.Helper()
+	require.NoError(t, fixture.ctrl.Reconcile(t.Context(), fixture.name))
+	require.NoError(t, fixture.ctrl.WaitUntilIdle(t.Context(), false))
+}
+
+type alertTransitionScenario struct {
+	mu sync.Mutex
+
+	initializerErr error
+	validateErr    error
+	resolveErr     error
+	nextErr        error
+	rows           []map[string]any
+	adminErr       error
+	adminMeta      *drivers.AlertMetadata
+	notifierErrs   []error
+
+	initializerCalls int
+	validateCalls    int
+	resolveCalls     int
+	nextCalls        int
+	adminCalls       int
+	notifierCalls    int
+}
+
+type alertTransitionSnapshot struct {
+	initializerCalls int
+	validateCalls    int
+	resolveCalls     int
+	nextCalls        int
+	adminCalls       int
+	notifierCalls    int
+}
+
+func newAlertTransitionScenario(t *testing.T) (*alertTransitionScenario, string) {
+	t.Helper()
+	id := fmt.Sprintf("scenario-%d", alertTransitionSequence.Add(1))
+	scenario := &alertTransitionScenario{}
+	alertTransitionScenarios.Store(id, scenario)
+	t.Cleanup(func() { alertTransitionScenarios.Delete(id) })
+	return scenario, id
+}
+
+func (s *alertTransitionScenario) snapshot() alertTransitionSnapshot {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return alertTransitionSnapshot{
+		initializerCalls: s.initializerCalls,
+		validateCalls:    s.validateCalls,
+		resolveCalls:     s.resolveCalls,
+		nextCalls:        s.nextCalls,
+		adminCalls:       s.adminCalls,
+		notifierCalls:    s.notifierCalls,
+	}
+}
+
+func (s *alertTransitionScenario) setResolveOutcome(err error, rows []map[string]any) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.resolveErr = err
+	s.rows = rows
+}
+
+func loadAlertTransitionScenario(id string) (*alertTransitionScenario, error) {
+	value, ok := alertTransitionScenarios.Load(id)
+	if !ok {
+		return nil, fmt.Errorf("unknown alert transition scenario %q", id)
+	}
+	return value.(*alertTransitionScenario), nil
+}
+
+type alertTransitionResolver struct {
+	scenario *alertTransitionScenario
+}
+
+func newAlertTransitionResolver(_ context.Context, opts *runtime.ResolverOptions) (runtime.Resolver, error) {
+	id, _ := opts.Properties["scenario"].(string)
+	scenario, err := loadAlertTransitionScenario(id)
+	if err != nil {
+		return nil, err
+	}
+	scenario.mu.Lock()
+	defer scenario.mu.Unlock()
+	scenario.initializerCalls++
+	if scenario.initializerErr != nil {
+		return nil, scenario.initializerErr
+	}
+	return &alertTransitionResolver{scenario: scenario}, nil
+}
+
+func (r *alertTransitionResolver) Close() error { return nil }
+
+func (r *alertTransitionResolver) CacheKey(context.Context) ([]byte, bool, error) {
+	return nil, false, nil
+}
+
+func (r *alertTransitionResolver) Refs() []*runtimev1.ResourceName { return nil }
+
+func (r *alertTransitionResolver) Validate(context.Context) error {
+	r.scenario.mu.Lock()
+	defer r.scenario.mu.Unlock()
+	r.scenario.validateCalls++
+	return r.scenario.validateErr
+}
+
+func (r *alertTransitionResolver) ResolveInteractive(context.Context) (runtime.ResolverResult, error) {
+	r.scenario.mu.Lock()
+	defer r.scenario.mu.Unlock()
+	r.scenario.resolveCalls++
+	if r.scenario.resolveErr != nil {
+		return nil, r.scenario.resolveErr
+	}
+	rows := append([]map[string]any(nil), r.scenario.rows...)
+	return &alertTransitionResult{scenario: r.scenario, rows: rows, nextErr: r.scenario.nextErr}, nil
+}
+
+func (r *alertTransitionResolver) ResolveExport(context.Context, io.Writer, *runtime.ResolverExportOptions) error {
+	return drivers.ErrNotImplemented
+}
+
+func (r *alertTransitionResolver) InferRequiredSecurityRules() ([]*runtimev1.SecurityRule, error) {
+	return nil, nil
+}
+
+type alertTransitionResult struct {
+	scenario *alertTransitionScenario
+	rows     []map[string]any
+	nextErr  error
+	index    int
+}
+
+func (r *alertTransitionResult) Close() error                  { return nil }
+func (r *alertTransitionResult) Meta() map[string]any          { return nil }
+func (r *alertTransitionResult) Schema() *runtimev1.StructType { return nil }
+
+func (r *alertTransitionResult) Next() (map[string]any, error) {
+	r.scenario.mu.Lock()
+	r.scenario.nextCalls++
+	r.scenario.mu.Unlock()
+	if r.nextErr != nil {
+		return nil, r.nextErr
+	}
+	if r.index >= len(r.rows) {
+		return nil, io.EOF
+	}
+	row := r.rows[r.index]
+	r.index++
+	return row, nil
+}
+
+func (r *alertTransitionResult) MarshalJSON() ([]byte, error) { return nil, drivers.ErrNotImplemented }
+
+type alertTransitionDriver struct{}
+
+func (d *alertTransitionDriver) Spec() drivers.Spec {
+	return drivers.Spec{ImplementsAdmin: true, ImplementsNotifier: true}
+}
+
+func (d *alertTransitionDriver) Open(_ string, _ string, config map[string]any, _ *storage.Client, _ *activity.Client, _ *zap.Logger) (drivers.Handle, error) {
+	id, _ := config["scenario"].(string)
+	scenario, err := loadAlertTransitionScenario(id)
+	if err != nil {
+		return nil, err
+	}
+	return &alertTransitionHandle{scenario: scenario, config: config}, nil
+}
+
+func (d *alertTransitionDriver) HasAnonymousSourceAccess(context.Context, map[string]any, *zap.Logger) (bool, error) {
+	return true, nil
+}
+
+func (d *alertTransitionDriver) TertiarySourceConnectors(context.Context, map[string]any, *zap.Logger) ([]string, error) {
+	return nil, nil
+}
+
+type alertTransitionHandle struct {
+	drivers.Handle
+	scenario *alertTransitionScenario
+	config   map[string]any
+}
+
+func (h *alertTransitionHandle) Ping(context.Context) error    { return nil }
+func (h *alertTransitionHandle) Driver() string                { return alertTransitionDriverName }
+func (h *alertTransitionHandle) Config() map[string]any        { return h.config }
+func (h *alertTransitionHandle) Migrate(context.Context) error { return nil }
+func (h *alertTransitionHandle) MigrationStatus(context.Context) (int, int, error) {
+	return 0, 0, nil
+}
+func (h *alertTransitionHandle) Close() error { return nil }
+
+func (h *alertTransitionHandle) AsAdmin(string) (drivers.AdminService, bool) {
+	return &alertTransitionAdmin{scenario: h.scenario}, true
+}
+
+func (h *alertTransitionHandle) AsNotifier(map[string]any) (drivers.Notifier, error) {
+	return &alertTransitionNotifier{scenario: h.scenario}, nil
+}
+
+type alertTransitionAdmin struct {
+	drivers.AdminService
+	scenario *alertTransitionScenario
+}
+
+func (a *alertTransitionAdmin) GetAlertMetadata(context.Context, string, string, []string, bool, map[string]string, string, string) (*drivers.AlertMetadata, error) {
+	a.scenario.mu.Lock()
+	defer a.scenario.mu.Unlock()
+	a.scenario.adminCalls++
+	return a.scenario.adminMeta, a.scenario.adminErr
+}
+
+func (a *alertTransitionAdmin) GetConfig(context.Context) (*drivers.Config, error) {
+	return &drivers.Config{}, nil
+}
+
+type alertTransitionNotifier struct {
+	scenario *alertTransitionScenario
+}
+
+func (n *alertTransitionNotifier) SendAlertStatus(*drivers.AlertStatus) error {
+	n.scenario.mu.Lock()
+	defer n.scenario.mu.Unlock()
+	index := n.scenario.notifierCalls
+	n.scenario.notifierCalls++
+	if index < len(n.scenario.notifierErrs) {
+		return n.scenario.notifierErrs[index]
+	}
+	return nil
+}
+
+func (n *alertTransitionNotifier) SendScheduledReport(*drivers.ScheduledReport) error {
+	return drivers.ErrNotImplemented
+}
+
+type alertTransitionEmailSender struct {
+	mu         sync.Mutex
+	errs       []error
+	recipients []string
+}
+
+func (s *alertTransitionEmailSender) Send(toEmail, _, _, _ string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	index := len(s.recipients)
+	s.recipients = append(s.recipients, toEmail)
+	if index < len(s.errs) {
+		return s.errs[index]
+	}
+	return nil
+}
+
+func (s *alertTransitionEmailSender) snapshot() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]string(nil), s.recipients...)
+}