diff --git a/x-pack/plugins/alerting/server/alerts_client/types.ts b/x-pack/plugins/alerting/server/alerts_client/types.ts
index d043f41e1e9550..8466736dfb3c12 100644
--- a/x-pack/plugins/alerting/server/alerts_client/types.ts
+++ b/x-pack/plugins/alerting/server/alerts_client/types.ts
@@ -77,8 +77,11 @@ export interface IAlertsClient<
   processAlerts(opts: ProcessAlertsOpts): void;
   logAlerts(opts: LogAlertsOpts): void;
   getProcessedAlerts(
-    type: 'new' | 'active' | 'activeCurrent' | 'recovered' | 'recoveredCurrent'
-  ): Record<string, LegacyAlert<State, Context, ActionGroupIds | RecoveryActionGroupId>>;
+    type: 'new' | 'active' | 'activeCurrent'
+  ): Record<string, LegacyAlert<State, Context, ActionGroupIds>>;
+  getProcessedAlerts(
+    type: 'recovered' | 'recoveredCurrent'
+  ): Record<string, LegacyAlert<State, Context, RecoveryActionGroupId>>;
   persistAlerts(): Promise<{ alertIds: string[]; maintenanceWindowIds: string[] } | null>;
   isTrackedAlert(id: string): boolean;
   getSummarizedAlerts?(params: GetSummarizedAlertsParams): Promise<SummarizedAlerts>;
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.ts
index 44822657ba86fb..831aaeeac666ed 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.ts
@@ -74,9 +74,13 @@ export class ActionScheduler<
     this.schedulers.sort((a, b) => a.priority - b.priority);
   }
 
-  public async run(
-    alerts: Record<string, Alert<State, Context, ActionGroupIds | RecoveryActionGroupId>>
-  ): Promise<RunResult> {
+  public async run({
+    activeCurrentAlerts,
+    recoveredCurrentAlerts,
+  }: {
+    activeCurrentAlerts: Record<string, Alert<State, Context, ActionGroupIds>>;
+    recoveredCurrentAlerts: Record<string, Alert<State, Context, RecoveryActionGroupId>>;
+  }): Promise<RunResult> {
     const throttledSummaryActions: ThrottledActions = getSummaryActionsFromTaskState({
       actions: this.context.rule.actions,
       summaryActions: this.context.taskInstance.state?.summaryActions,
@@ -85,7 +89,11 @@ export class ActionScheduler<
     const allActionsToScheduleResult: ActionsToSchedule[] = [];
     for (const scheduler of this.schedulers) {
       allActionsToScheduleResult.push(
-        ...(await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions }))
+        ...(await scheduler.getActionsToSchedule({
+          activeCurrentAlerts,
+          recoveredCurrentAlerts,
+          throttledSummaryActions,
+        }))
       );
     }
 
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.test.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.test.ts
index 99a693133a2a6d..aaa9698619eacc 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.test.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.test.ts
@@ -213,7 +213,10 @@ describe('Per-Alert Action Scheduler', () => {
     test('should create action to schedule for each alert and each action', async () => {
       // 2 per-alert actions * 2 alerts = 4 actions to schedule
       const scheduler = new PerAlertActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).not.toHaveBeenCalled();
@@ -243,7 +246,10 @@ describe('Per-Alert Action Scheduler', () => {
         maintenanceWindowIds: ['mw-1'],
       });
       const alertsWithMaintenanceWindow = { ...newAlertWithMaintenanceWindow, ...newAlert2 };
-      const results = await scheduler.getActionsToSchedule({ alerts: alertsWithMaintenanceWindow });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alertsWithMaintenanceWindow,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(2);
@@ -281,7 +287,8 @@ describe('Per-Alert Action Scheduler', () => {
       });
       const alertsWithInvalidActionGroup = { ...newAlertInvalidActionGroup, ...newAlert2 };
       const results = await scheduler.getActionsToSchedule({
-        alerts: alertsWithInvalidActionGroup,
+        activeCurrentAlerts: alertsWithInvalidActionGroup,
+        recoveredCurrentAlerts: {},
       });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -309,6 +316,36 @@ describe('Per-Alert Action Scheduler', () => {
       ]);
     });
 
+    test('should skip creating actions to schedule when alert has no scheduled actions', async () => {
+      // 2 per-alert actions * 2 alerts = 4 actions to schedule
+      // but alert 1 has has no scheduled actions, so only actions for alert 2 should be scheduled
+      const scheduler = new PerAlertActionScheduler(getSchedulerContext());
+      const newAlertInvalidActionGroup = generateAlert({
+        id: 1,
+        scheduleActions: false,
+      });
+      const alertsWithInvalidActionGroup = { ...newAlertInvalidActionGroup, ...newAlert2 };
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alertsWithInvalidActionGroup,
+        recoveredCurrentAlerts: {},
+      });
+
+      expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
+
+      expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toEqual(2);
+      expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toEqual(2);
+      expect(ruleRunMetricsStore.getStatusByConnectorType('test')).toEqual({
+        numberOfGeneratedActions: 2,
+        numberOfTriggeredActions: 2,
+      });
+
+      expect(results).toHaveLength(2);
+      expect(results).toEqual([
+        getResult('action-1', '2', '111-111'),
+        getResult('action-2', '2', '222-222'),
+      ]);
+    });
+
     test('should skip creating actions to schedule when alert has pending recovered count greater than 0 and notifyWhen is onActiveAlert', async () => {
       // 2 per-alert actions * 2 alerts = 4 actions to schedule
       // but alert 1 has a pending recovered count > 0 & notifyWhen is onActiveAlert, so only actions for alert 2 should be scheduled
@@ -322,7 +359,8 @@ describe('Per-Alert Action Scheduler', () => {
         ...newAlert2,
       };
       const results = await scheduler.getActionsToSchedule({
-        alerts: alertsWithPendingRecoveredCount,
+        activeCurrentAlerts: alertsWithPendingRecoveredCount,
+        recoveredCurrentAlerts: {},
       });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -368,7 +406,8 @@ describe('Per-Alert Action Scheduler', () => {
         ...newAlert2,
       };
       const results = await scheduler.getActionsToSchedule({
-        alerts: alertsWithPendingRecoveredCount,
+        activeCurrentAlerts: alertsWithPendingRecoveredCount,
+        recoveredCurrentAlerts: {},
       });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -394,7 +433,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, mutedInstanceIds: ['2'] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(1);
@@ -453,7 +495,10 @@ describe('Per-Alert Action Scheduler', () => {
         rule: { ...rule, actions: [rule.actions[0], onActionGroupChangeAction] },
       });
 
-      const results = await scheduler.getActionsToSchedule({ alerts: alertsWithOngoingAlert });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alertsWithOngoingAlert,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(1);
@@ -508,7 +553,10 @@ describe('Per-Alert Action Scheduler', () => {
         rule: { ...rule, actions: [rule.actions[0], onThrottleIntervalAction] },
       });
 
-      const results = await scheduler.getActionsToSchedule({ alerts: alertsWithOngoingAlert });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alertsWithOngoingAlert,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(1);
@@ -563,7 +611,10 @@ describe('Per-Alert Action Scheduler', () => {
         rule: { ...rule, actions: [rule.actions[0], onThrottleIntervalAction] },
       });
 
-      const results = await scheduler.getActionsToSchedule({ alerts: alertsWithOngoingAlert });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alertsWithOngoingAlert,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).not.toHaveBeenCalled();
@@ -620,7 +671,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, actions: [rule.actions[0], actionWithUseAlertDataForTemplate] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -679,7 +733,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, actions: [rule.actions[0], actionWithUseAlertDataForTemplate] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -739,7 +796,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, actions: [rule.actions[0], actionWithAlertsFilter] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -799,7 +859,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, actions: [rule.actions[0], actionWithAlertsFilter] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -860,7 +923,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, actions: [rule.actions[0], actionWithAlertsFilter] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -919,7 +985,10 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, actions: [rule.actions[0], actionWithAlertsFilter] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -960,7 +1029,10 @@ describe('Per-Alert Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
 
@@ -996,7 +1068,10 @@ describe('Per-Alert Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
 
@@ -1029,7 +1104,10 @@ describe('Per-Alert Action Scheduler', () => {
 
       expect(alert.getLastScheduledActions()).toBeUndefined();
       expect(alert.hasScheduledActions()).toBe(true);
-      await scheduler.getActionsToSchedule({ alerts: { '1': alert } });
+      await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: { '1': alert },
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alert.getLastScheduledActions()).toEqual({
         date: '1970-01-01T00:00:00.000Z',
@@ -1066,7 +1144,10 @@ describe('Per-Alert Action Scheduler', () => {
         rule: { ...rule, actions: [onThrottleIntervalAction] },
       });
 
-      await scheduler.getActionsToSchedule({ alerts: { '1': alert } });
+      await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: { '1': alert },
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alert.getLastScheduledActions()).toEqual({
         date: '1970-01-01T00:00:00.000Z',
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.ts
index b35d86dff01054..1fd5dc73f8eecc 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/per_alert_action_scheduler.ts
@@ -25,8 +25,12 @@ import {
 import {
   ActionSchedulerOptions,
   ActionsToSchedule,
+  AddSummarizedAlertsOpts,
   GetActionsToScheduleOpts,
+  HelperOpts,
   IActionScheduler,
+  IsExecutableActiveAlertOpts,
+  IsExecutableAlertOpts,
 } from '../types';
 import { TransformActionParamsOptions, transformActionParams } from '../../transform_action_params';
 import { injectActionParams } from '../../inject_action_params';
@@ -96,7 +100,8 @@ export class PerAlertActionScheduler<
   }
 
   public async getActionsToSchedule({
-    alerts,
+    activeCurrentAlerts,
+    recoveredCurrentAlerts,
   }: GetActionsToScheduleOpts<State, Context, ActionGroupIds, RecoveryActionGroupId>): Promise<
     ActionsToSchedule[]
   > {
@@ -106,7 +111,9 @@ export class PerAlertActionScheduler<
     }> = [];
     const results: ActionsToSchedule[] = [];
 
-    const alertsArray = Object.entries(alerts);
+    const activeCurrentAlertsArray = Object.values(activeCurrentAlerts);
+    const recoveredCurrentAlertsArray = Object.values(recoveredCurrentAlerts);
+
     for (const action of this.actions) {
       let summarizedAlerts = null;
 
@@ -133,61 +140,26 @@ export class PerAlertActionScheduler<
 
         logNumberOfFilteredAlerts({
           logger: this.context.logger,
-          numberOfAlerts: Object.entries(alerts).length,
+          numberOfAlerts: activeCurrentAlertsArray.length + recoveredCurrentAlertsArray.length,
           numberOfSummarizedAlerts: summarizedAlerts.all.count,
           action,
         });
       }
 
-      for (const [alertId, alert] of alertsArray) {
-        const alertMaintenanceWindowIds = alert.getMaintenanceWindowIds();
-        if (alertMaintenanceWindowIds.length !== 0) {
-          this.context.logger.debug(
-            `no scheduling of summary actions "${action.id}" for rule "${
-              this.context.rule.id
-            }": has active maintenance windows ${alertMaintenanceWindowIds.join(', ')}.`
-          );
-          continue;
-        }
-
-        if (alert.isFilteredOut(summarizedAlerts)) {
-          continue;
-        }
-
-        const actionGroup =
-          alert.getScheduledActionOptions()?.actionGroup ||
-          this.context.ruleType.recoveryActionGroup.id;
-
-        if (!this.ruleTypeActionGroups!.has(actionGroup)) {
-          this.context.logger.error(
-            `Invalid action group "${actionGroup}" for rule "${this.context.ruleType.id}".`
-          );
-          continue;
-        }
-
-        // only actions with notifyWhen set to "on status change" should return
-        // notifications for flapping pending recovered alerts
+      for (const alert of activeCurrentAlertsArray) {
         if (
-          alert.getPendingRecoveredCount() > 0 &&
-          action?.frequency?.notifyWhen !== RuleNotifyWhen.CHANGE
+          this.isExecutableAlert({ alert, action, summarizedAlerts }) &&
+          this.isExecutableActiveAlert({ alert, action })
         ) {
-          continue;
-        }
-
-        if (summarizedAlerts) {
-          const alertAsData = summarizedAlerts.all.data.find(
-            (alertHit: AlertHit) => alertHit._id === alert.getUuid()
-          );
-          if (alertAsData) {
-            alert.setAlertAsData(alertAsData);
-          }
+          this.addSummarizedAlerts({ alert, summarizedAlerts });
+          executables.push({ action, alert });
         }
+      }
 
-        if (action.group === actionGroup && !this.isAlertMuted(alertId)) {
-          if (
-            this.isRecoveredAlert(action.group) ||
-            this.isExecutableActiveAlert({ alert, action })
-          ) {
+      if (this.isRecoveredAction(action.group)) {
+        for (const alert of recoveredCurrentAlertsArray) {
+          if (this.isExecutableAlert({ alert, action, summarizedAlerts })) {
+            this.addSummarizedAlerts({ alert, summarizedAlerts });
             executables.push({ action, alert });
           }
         }
@@ -285,7 +257,7 @@ export class PerAlertActionScheduler<
         },
       });
 
-      if (!this.isRecoveredAlert(actionGroup)) {
+      if (!this.isRecoveredAction(actionGroup)) {
         if (isActionOnInterval(action)) {
           alert.updateLastScheduledActions(
             action.group as ActionGroupIds,
@@ -302,30 +274,34 @@ export class PerAlertActionScheduler<
     return results;
   }
 
-  private isAlertMuted(alertId: string) {
-    const muted = this.mutedAlertIdsSet.has(alertId);
-    if (muted) {
-      if (
-        !this.skippedAlerts[alertId] ||
-        (this.skippedAlerts[alertId] && this.skippedAlerts[alertId].reason !== Reasons.MUTED)
-      ) {
-        this.context.logger.debug(
-          `skipping scheduling of actions for '${alertId}' in rule ${this.context.ruleLabel}: rule is muted`
-        );
-      }
-      this.skippedAlerts[alertId] = { reason: Reasons.MUTED };
-      return true;
-    }
-    return false;
-  }
-
-  private isExecutableActiveAlert({
+  private isExecutableAlert({
     alert,
     action,
-  }: {
-    alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds | RecoveryActionGroupId>;
-    action: RuleAction;
-  }) {
+    summarizedAlerts,
+  }: IsExecutableAlertOpts<ActionGroupIds, RecoveryActionGroupId>) {
+    return (
+      !this.hasActiveMaintenanceWindow({ alert, action }) &&
+      !this.isAlertMuted(alert) &&
+      !this.hasPendingCountButNotNotifyOnChange({ alert, action }) &&
+      !alert.isFilteredOut(summarizedAlerts)
+    );
+  }
+
+  private isExecutableActiveAlert({ alert, action }: IsExecutableActiveAlertOpts<ActionGroupIds>) {
+    if (!alert.hasScheduledActions()) {
+      return false;
+    }
+
+    const alertsActionGroup = alert.getScheduledActionOptions()?.actionGroup;
+
+    if (!this.isValidActionGroup(alertsActionGroup as ActionGroupIds)) {
+      return false;
+    }
+
+    if (action.group !== alertsActionGroup) {
+      return false;
+    }
+
     const alertId = alert.getId();
     const {
       context: { rule, logger, ruleLabel },
@@ -369,10 +345,86 @@ export class PerAlertActionScheduler<
       }
     }
 
-    return alert.hasScheduledActions();
+    return true;
   }
 
-  private isRecoveredAlert(actionGroup: string) {
+  private isRecoveredAction(actionGroup: string) {
     return actionGroup === this.context.ruleType.recoveryActionGroup.id;
   }
+
+  private isAlertMuted(
+    alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds | RecoveryActionGroupId>
+  ) {
+    const alertId = alert.getId();
+    const muted = this.mutedAlertIdsSet.has(alertId);
+    if (muted) {
+      if (
+        !this.skippedAlerts[alertId] ||
+        (this.skippedAlerts[alertId] && this.skippedAlerts[alertId].reason !== Reasons.MUTED)
+      ) {
+        this.context.logger.debug(
+          `skipping scheduling of actions for '${alertId}' in rule ${this.context.ruleLabel}: rule is muted`
+        );
+      }
+      this.skippedAlerts[alertId] = { reason: Reasons.MUTED };
+      return true;
+    }
+    return false;
+  }
+
+  private isValidActionGroup(actionGroup: ActionGroupIds | RecoveryActionGroupId) {
+    if (!this.ruleTypeActionGroups!.has(actionGroup)) {
+      this.context.logger.error(
+        `Invalid action group "${actionGroup}" for rule "${this.context.ruleType.id}".`
+      );
+      return false;
+    }
+    return true;
+  }
+
+  private hasActiveMaintenanceWindow({
+    alert,
+    action,
+  }: HelperOpts<ActionGroupIds, RecoveryActionGroupId>) {
+    const alertMaintenanceWindowIds = alert.getMaintenanceWindowIds();
+    if (alertMaintenanceWindowIds.length !== 0) {
+      this.context.logger.debug(
+        `no scheduling of summary actions "${action.id}" for rule "${
+          this.context.rule.id
+        }": has active maintenance windows ${alertMaintenanceWindowIds.join(', ')}.`
+      );
+      return true;
+    }
+
+    return false;
+  }
+
+  private addSummarizedAlerts({
+    alert,
+    summarizedAlerts,
+  }: AddSummarizedAlertsOpts<ActionGroupIds, RecoveryActionGroupId>) {
+    if (summarizedAlerts) {
+      const alertAsData = summarizedAlerts.all.data.find(
+        (alertHit: AlertHit) => alertHit._id === alert.getUuid()
+      );
+      if (alertAsData) {
+        alert.setAlertAsData(alertAsData);
+      }
+    }
+  }
+
+  private hasPendingCountButNotNotifyOnChange({
+    alert,
+    action,
+  }: HelperOpts<ActionGroupIds, RecoveryActionGroupId>) {
+    // only actions with notifyWhen set to "on status change" should return
+    // notifications for flapping pending recovered alerts
+    if (
+      alert.getPendingRecoveredCount() > 0 &&
+      action?.frequency?.notifyWhen !== RuleNotifyWhen.CHANGE
+    ) {
+      return true;
+    }
+    return false;
+  }
 }
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.test.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.test.ts
index fc810fc4ef34c1..e250ba8106ad98 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.test.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.test.ts
@@ -217,7 +217,11 @@ describe('Summary Action Scheduler', () => {
 
       const throttledSummaryActions = {};
       const scheduler = new SummaryActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
@@ -266,7 +270,11 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -307,7 +315,11 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({ '444-444': { date: '1970-01-01T00:00:00.000Z' } });
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -340,7 +352,11 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = { '444-444': { date: '1969-12-31T13:00:00.000Z' } };
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({ '444-444': { date: '1969-12-31T13:00:00.000Z' } });
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -374,7 +390,11 @@ describe('Summary Action Scheduler', () => {
       const scheduler = new SummaryActionScheduler(getSchedulerContext());
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
@@ -436,7 +456,11 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -480,7 +504,11 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -507,7 +535,11 @@ describe('Summary Action Scheduler', () => {
       const scheduler = new SummaryActionScheduler(getSchedulerContext());
 
       try {
-        await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions: {} });
+        await scheduler.getActionsToSchedule({
+          activeCurrentAlerts: alerts,
+          recoveredCurrentAlerts: {},
+          throttledSummaryActions: {},
+        });
       } catch (err) {
         expect(err.message).toEqual(`no alerts for you`);
         expect(getErrorSource(err)).toBe(TaskErrorSource.FRAMEWORK);
@@ -533,7 +565,11 @@ describe('Summary Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions: {} });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
@@ -587,7 +623,11 @@ describe('Summary Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions: {} });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+        throttledSummaryActions: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.ts
index 050eea352f0d59..db53f15be21804 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/summary_action_scheduler.ts
@@ -81,11 +81,13 @@ export class SummaryActionScheduler<
   }
 
   public async getActionsToSchedule({
-    alerts,
+    activeCurrentAlerts,
+    recoveredCurrentAlerts,
     throttledSummaryActions,
   }: GetActionsToScheduleOpts<State, Context, ActionGroupIds, RecoveryActionGroupId>): Promise<
     ActionsToSchedule[]
   > {
+    const alerts = { ...activeCurrentAlerts, ...recoveredCurrentAlerts };
     const executables: Array<{
       action: RuleAction;
       summarizedAlerts: CombinedSummarizedAlerts;
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/system_action_scheduler.test.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/system_action_scheduler.test.ts
index 28bf58a30c6893..87d0b497b60e0d 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/system_action_scheduler.test.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/system_action_scheduler.test.ts
@@ -160,7 +160,10 @@ describe('System Action Scheduler', () => {
       alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
 
       const scheduler = new SystemActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -202,7 +205,10 @@ describe('System Action Scheduler', () => {
       alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
 
       const scheduler = new SystemActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -240,7 +246,10 @@ describe('System Action Scheduler', () => {
       alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
 
       const scheduler = new SystemActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -265,7 +274,10 @@ describe('System Action Scheduler', () => {
       const scheduler = new SystemActionScheduler(getSchedulerContext());
 
       try {
-        await scheduler.getActionsToSchedule({ alerts });
+        await scheduler.getActionsToSchedule({
+          activeCurrentAlerts: alerts,
+          recoveredCurrentAlerts: {},
+        });
       } catch (err) {
         expect(err.message).toEqual(`no alerts for you`);
         expect(getErrorSource(err)).toBe(TaskErrorSource.FRAMEWORK);
@@ -299,7 +311,10 @@ describe('System Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
@@ -361,7 +376,10 @@ describe('System Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
@@ -416,7 +434,10 @@ describe('System Action Scheduler', () => {
         ...defaultContext,
         rule: { ...rule, systemActions: [differentSystemAction] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
diff --git a/x-pack/plugins/alerting/server/task_runner/action_scheduler/types.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/types.ts
index b90ffb88d541bb..cfc8a3aa8b4212 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/types.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/types.ts
@@ -21,6 +21,7 @@ import {
   RuleAlertData,
   RuleSystemAction,
   ThrottledActions,
+  RuleAction,
 } from '../../../common';
 import { NormalizedRuleType } from '../../rule_type_registry';
 import { CombinedSummarizedAlerts, RawRule } from '../../types';
@@ -90,7 +91,8 @@ export interface GetActionsToScheduleOpts<
   ActionGroupIds extends string,
   RecoveryActionGroupId extends string
 > {
-  alerts: Record<string, Alert<State, Context, ActionGroupIds | RecoveryActionGroupId>>;
+  activeCurrentAlerts: Record<string, Alert<State, Context, ActionGroupIds>>;
+  recoveredCurrentAlerts: Record<string, Alert<State, Context, RecoveryActionGroupId>>;
   throttledSummaryActions?: ThrottledActions;
 }
 
@@ -118,3 +120,30 @@ export interface RuleUrl {
   spaceIdSegment?: string;
   relativePath?: string;
 }
+
+export interface IsExecutableAlertOpts<
+  ActionGroupIds extends string,
+  RecoveryActionGroupId extends string
+> {
+  alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds | RecoveryActionGroupId>;
+  action: RuleAction;
+  summarizedAlerts: CombinedSummarizedAlerts | null;
+}
+
+export interface IsExecutableActiveAlertOpts<ActionGroupIds extends string> {
+  alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds>;
+  action: RuleAction;
+}
+
+export interface HelperOpts<ActionGroupIds extends string, RecoveryActionGroupId extends string> {
+  alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds | RecoveryActionGroupId>;
+  action: RuleAction;
+}
+
+export interface AddSummarizedAlertsOpts<
+  ActionGroupIds extends string,
+  RecoveryActionGroupId extends string
+> {
+  alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds | RecoveryActionGroupId>;
+  summarizedAlerts: CombinedSummarizedAlerts | null;
+}
diff --git a/x-pack/plugins/alerting/server/task_runner/task_runner.ts b/x-pack/plugins/alerting/server/task_runner/task_runner.ts
index 897937ce55a0ab..89432e18220292 100644
--- a/x-pack/plugins/alerting/server/task_runner/task_runner.ts
+++ b/x-pack/plugins/alerting/server/task_runner/task_runner.ts
@@ -414,8 +414,8 @@ export class TaskRunner<
           this.countUsageOfActionExecutionAfterRuleCancellation();
         } else {
           actionSchedulerResult = await actionScheduler.run({
-            ...alertsClient.getProcessedAlerts('activeCurrent'),
-            ...alertsClient.getProcessedAlerts('recoveredCurrent'),
+            activeCurrentAlerts: alertsClient.getProcessedAlerts('activeCurrent'),
+            recoveredCurrentAlerts: alertsClient.getProcessedAlerts('recoveredCurrent'),
           });
         }
       })