diff --git a/x-pack/plugins/alerting/server/alerts_client/types.ts b/x-pack/plugins/alerting/server/alerts_client/types.ts
index d043f41e1e955..f3c4a85fa1b71 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.test.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.test.ts
index b6f250b47205e..00f1a87aefd71 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.test.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/action_scheduler.test.ts
@@ -95,6 +95,7 @@ describe('Action Scheduler', () => {
     );
     ruleRunMetricsStore = new RuleRunMetricsStore();
     actionsClient.bulkEnqueueExecution.mockResolvedValue(defaultExecutionResponse);
+    alertsClient.getProcessedAlerts.mockReturnValue({});
   });
   beforeAll(() => {
     clock = sinon.useFakeTimers();
@@ -104,7 +105,7 @@ describe('Action Scheduler', () => {
   test('schedules execution per selected action', async () => {
     const alerts = generateAlert({ id: 1 });
     const actionScheduler = new ActionScheduler(getSchedulerContext());
-    await actionScheduler.run(alerts);
+    await actionScheduler.run({ activeCurrentAlerts: alerts, recoveredCurrentAlerts: {} });
 
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(1);
@@ -204,7 +205,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    await actionScheduler.run(generateAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 1 }),
+      recoveredCurrentAlerts: {},
+    });
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(2);
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
@@ -269,7 +273,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    await actionScheduler.run(generateAlert({ id: 2 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2 }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(2);
@@ -281,7 +288,10 @@ describe('Action Scheduler', () => {
       ruleRunMetricsStore,
     });
 
-    await actionSchedulerForPreconfiguredAction.run(generateAlert({ id: 2 }));
+    await actionSchedulerForPreconfiguredAction.run({
+      activeCurrentAlerts: generateAlert({ id: 2 }),
+      recoveredCurrentAlerts: {},
+    });
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
   });
 
@@ -321,7 +331,10 @@ describe('Action Scheduler', () => {
     );
 
     try {
-      await actionScheduler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 2, state: { value: 'state-val' } }),
+        recoveredCurrentAlerts: {},
+      });
     } catch (err) {
       expect(getErrorSource(err)).toBe(TaskErrorSource.USER);
     }
@@ -329,7 +342,10 @@ describe('Action Scheduler', () => {
 
   test('limits actionsPlugin.execute per action group', async () => {
     const actionScheduler = new ActionScheduler(getSchedulerContext());
-    await actionScheduler.run(generateAlert({ id: 2, group: 'other-group' }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2, group: 'other-group' }),
+      recoveredCurrentAlerts: {},
+    });
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(0);
     expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
@@ -337,7 +353,10 @@ describe('Action Scheduler', () => {
 
   test('context attribute gets parameterized', async () => {
     const actionScheduler = new ActionScheduler(getSchedulerContext());
-    await actionScheduler.run(generateAlert({ id: 2, context: { value: 'context-val' } }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2, context: { value: 'context-val' } }),
+      recoveredCurrentAlerts: {},
+    });
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(1);
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
@@ -381,7 +400,10 @@ describe('Action Scheduler', () => {
 
   test('state attribute gets parameterized', async () => {
     const actionScheduler = new ActionScheduler(getSchedulerContext());
-    await actionScheduler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2, state: { value: 'state-val' } }),
+      recoveredCurrentAlerts: {},
+    });
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
     expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
       Array [
@@ -423,9 +445,13 @@ describe('Action Scheduler', () => {
 
   test(`logs an error when action group isn't part of actionGroups available for the ruleType`, async () => {
     const actionScheduler = new ActionScheduler(getSchedulerContext());
-    await actionScheduler.run(
-      generateAlert({ id: 2, group: 'invalid-group' as 'default' | 'other-group' })
-    );
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({
+        id: 2,
+        group: 'invalid-group' as 'default' | 'other-group',
+      }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(defaultSchedulerContext.logger.error).toHaveBeenCalledWith(
       'Invalid action group "invalid-group" for rule "test".'
@@ -503,7 +529,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2, state: { value: 'state-val' } }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(2);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(3);
@@ -604,7 +633,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2, state: { value: 'state-val' } }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(4);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(5);
@@ -688,7 +720,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2, state: { value: 'state-val' } }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(2);
     expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(3);
@@ -722,7 +757,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateRecoveredAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: generateRecoveredAlert({ id: 1 }),
+    });
 
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
     expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
@@ -787,7 +825,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateRecoveredAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: generateRecoveredAlert({ id: 1 }),
+    });
 
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
     expect(defaultSchedulerContext.logger.debug).nthCalledWith(
@@ -807,7 +848,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 1 }),
+      recoveredCurrentAlerts: {},
+    });
 
     clock.tick(30000);
 
@@ -837,12 +881,13 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(
-      generateAlert({
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({
         id: 1,
         throttledActions: { '111-111': { date: new Date(DATE_1970).toISOString() } },
-      })
-    );
+      }),
+      recoveredCurrentAlerts: {},
+    });
 
     clock.tick(30000);
 
@@ -872,7 +917,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 1, lastScheduledActionsGroup: 'recovered' }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 1, lastScheduledActionsGroup: 'recovered' }),
+      recoveredCurrentAlerts: {},
+    });
 
     clock.tick(30000);
 
@@ -890,7 +938,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 1 }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
     expect(defaultSchedulerContext.logger.debug).nthCalledWith(
@@ -945,7 +996,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    await actionScheduler.run(generateAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 1 }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
       executionUuid: '5f6aa57d-3e22-484e-bae8-cbed868f4d28',
@@ -1026,7 +1080,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    await actionScheduler.run({});
+    await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: {},
+    });
 
     expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
     expect(alertingEventLogger.logAction).not.toHaveBeenCalled();
@@ -1078,7 +1135,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    const result = await actionScheduler.run({});
+    const result = await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: {},
+    });
 
     expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
       start: new Date('1969-12-31T00:01:30.000Z'),
@@ -1174,7 +1234,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    await actionScheduler.run({});
+    await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: {},
+    });
     expect(defaultSchedulerContext.logger.debug).toHaveBeenCalledTimes(1);
     expect(defaultSchedulerContext.logger.debug).toHaveBeenCalledWith(
       "skipping scheduling the action 'testActionTypeId:1', summary action is still being throttled"
@@ -1236,7 +1299,10 @@ describe('Action Scheduler', () => {
       })
     );
 
-    const result = await actionScheduler.run({});
+    const result = await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: {},
+    });
     expect(result).toEqual({
       throttledSummaryActions: {
         '111-111': {
@@ -1271,7 +1337,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateAlert({ id: 2 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: generateAlert({ id: 2 }),
+      recoveredCurrentAlerts: {},
+    });
 
     expect(defaultSchedulerContext.logger.error).toHaveBeenCalledWith(
       'Skipping action "1" for rule "1" because the rule type "Test" does not support alert-as-data.'
@@ -1332,7 +1401,10 @@ describe('Action Scheduler', () => {
         },
       })
     );
-    await actionScheduler.run(generateRecoveredAlert({ id: 1 }));
+    await actionScheduler.run({
+      activeCurrentAlerts: {},
+      recoveredCurrentAlerts: generateRecoveredAlert({ id: 1 }),
+    });
 
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
     expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
@@ -1455,8 +1527,11 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1 }),
-      ...generateAlert({ id: 2 }),
+      activeCurrentAlerts: {
+        ...generateAlert({ id: 1 }),
+        ...generateAlert({ id: 2 }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -1529,8 +1604,11 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1 }),
-      ...generateAlert({ id: 2 }),
+      activeCurrentAlerts: {
+        ...generateAlert({ id: 1 }),
+        ...generateAlert({ id: 2 }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -1597,9 +1675,12 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1 }),
-      ...generateAlert({ id: 2 }),
-      ...generateAlert({ id: 3 }),
+      activeCurrentAlerts: {
+        ...generateAlert({ id: 1 }),
+        ...generateAlert({ id: 2 }),
+        ...generateAlert({ id: 3 }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -1706,9 +1787,12 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1, maintenanceWindowIds: ['test-id-1'] }),
-      ...generateAlert({ id: 2, maintenanceWindowIds: ['test-id-2'] }),
-      ...generateAlert({ id: 3, maintenanceWindowIds: ['test-id-3'] }),
+      activeCurrentAlerts: {
+        ...generateAlert({ id: 1, maintenanceWindowIds: ['test-id-1'] }),
+        ...generateAlert({ id: 2, maintenanceWindowIds: ['test-id-2'] }),
+        ...generateAlert({ id: 3, maintenanceWindowIds: ['test-id-3'] }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
@@ -1755,9 +1839,12 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1, maintenanceWindowIds: ['test-id-1'] }),
-      ...generateAlert({ id: 2, maintenanceWindowIds: ['test-id-2'] }),
-      ...generateAlert({ id: 3, maintenanceWindowIds: ['test-id-3'] }),
+      activeCurrentAlerts: {
+        ...generateAlert({ id: 1, maintenanceWindowIds: ['test-id-1'] }),
+        ...generateAlert({ id: 2, maintenanceWindowIds: ['test-id-2'] }),
+        ...generateAlert({ id: 3, maintenanceWindowIds: ['test-id-3'] }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
@@ -1773,9 +1860,12 @@ describe('Action Scheduler', () => {
     const actionScheduler = new ActionScheduler(getSchedulerContext());
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1, maintenanceWindowIds: ['test-id-1'] }),
-      ...generateAlert({ id: 2, maintenanceWindowIds: ['test-id-2'] }),
-      ...generateAlert({ id: 3, maintenanceWindowIds: ['test-id-3'] }),
+      activeCurrentAlerts: {
+        ...generateAlert({ id: 1, maintenanceWindowIds: ['test-id-1'] }),
+        ...generateAlert({ id: 2, maintenanceWindowIds: ['test-id-2'] }),
+        ...generateAlert({ id: 3, maintenanceWindowIds: ['test-id-3'] }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
@@ -1813,9 +1903,24 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1, pendingRecoveredCount: 1, lastScheduledActionsGroup: 'recovered' }),
-      ...generateAlert({ id: 2, pendingRecoveredCount: 1, lastScheduledActionsGroup: 'recovered' }),
-      ...generateAlert({ id: 3, pendingRecoveredCount: 1, lastScheduledActionsGroup: 'recovered' }),
+      activeCurrentAlerts: {
+        ...generateAlert({
+          id: 1,
+          pendingRecoveredCount: 1,
+          lastScheduledActionsGroup: 'recovered',
+        }),
+        ...generateAlert({
+          id: 2,
+          pendingRecoveredCount: 1,
+          lastScheduledActionsGroup: 'recovered',
+        }),
+        ...generateAlert({
+          id: 3,
+          pendingRecoveredCount: 1,
+          lastScheduledActionsGroup: 'recovered',
+        }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
@@ -1842,9 +1947,24 @@ describe('Action Scheduler', () => {
     );
 
     await actionScheduler.run({
-      ...generateAlert({ id: 1, pendingRecoveredCount: 1, lastScheduledActionsGroup: 'recovered' }),
-      ...generateAlert({ id: 2, pendingRecoveredCount: 1, lastScheduledActionsGroup: 'recovered' }),
-      ...generateAlert({ id: 3, pendingRecoveredCount: 1, lastScheduledActionsGroup: 'recovered' }),
+      activeCurrentAlerts: {
+        ...generateAlert({
+          id: 1,
+          pendingRecoveredCount: 1,
+          lastScheduledActionsGroup: 'recovered',
+        }),
+        ...generateAlert({
+          id: 2,
+          pendingRecoveredCount: 1,
+          lastScheduledActionsGroup: 'recovered',
+        }),
+        ...generateAlert({
+          id: 3,
+          pendingRecoveredCount: 1,
+          lastScheduledActionsGroup: 'recovered',
+        }),
+      },
+      recoveredCurrentAlerts: {},
     });
 
     expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
@@ -1991,7 +2111,11 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
+
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
 
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
@@ -2024,7 +2148,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
 
       expect(injectActionParamsMock.mock.calls[0][0].actionParams).toEqual({
         val: 'rule url: http://localhost:12345/kbn/s/test1/app/management/insightsAndAlerting/triggersActions/rule/1',
@@ -2064,8 +2191,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2100,8 +2229,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2133,8 +2264,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2166,8 +2299,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2196,8 +2331,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2226,8 +2363,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2259,8 +2398,10 @@ describe('Action Scheduler', () => {
       };
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(execParams));
-      await actionScheduler.run(generateAlert({ id: 1 }));
-
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       expect(injectActionParamsMock.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           Object {
@@ -2328,8 +2469,10 @@ describe('Action Scheduler', () => {
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(executorParams));
 
-      const res = await actionScheduler.run(generateAlert({ id: 1 }));
-
+      const res = await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
       /**
        * Verifies that system actions are not throttled
        */
@@ -2451,7 +2594,10 @@ describe('Action Scheduler', () => {
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(executorParams));
 
-      const res = await actionScheduler.run(generateAlert({ id: 1 }));
+      const res = await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
 
       /**
        * Verifies that system actions are not throttled
@@ -2508,7 +2654,10 @@ describe('Action Scheduler', () => {
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(executorParams));
 
-      const res = await actionScheduler.run(generateAlert({ id: 1 }));
+      const res = await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
 
       expect(res).toEqual({ throttledSummaryActions: {} });
       expect(buildActionParams).not.toHaveBeenCalled();
@@ -2547,7 +2696,10 @@ describe('Action Scheduler', () => {
 
       const actionScheduler = new ActionScheduler(getSchedulerContext(executorParams));
 
-      await actionScheduler.run(generateAlert({ id: 1 }));
+      await actionScheduler.run({
+        activeCurrentAlerts: generateAlert({ id: 1 }),
+        recoveredCurrentAlerts: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(buildActionParams).not.toHaveBeenCalled();
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 44822657ba86f..fa16cfcabb094 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/lib/get_summarized_alerts.ts b/x-pack/plugins/alerting/server/task_runner/action_scheduler/lib/get_summarized_alerts.ts
index 00e155856d946..56d9c08c8b98f 100644
--- a/x-pack/plugins/alerting/server/task_runner/action_scheduler/lib/get_summarized_alerts.ts
+++ b/x-pack/plugins/alerting/server/task_runner/action_scheduler/lib/get_summarized_alerts.ts
@@ -56,7 +56,7 @@ export const getSummarizedAlerts = async <
    * yet (the update call uses refresh: false). So we need to rely on the in
    * memory alerts to do this.
    */
-  const newAlertsInMemory = Object.values(alertsClient.getProcessedAlerts('new') || {}) || [];
+  const newAlertsInMemory = Object.values(alertsClient.getProcessedAlerts('new'));
 
   const newAlertsWithMaintenanceWindowIds = newAlertsInMemory.reduce<string[]>((result, alert) => {
     if (alert.getMaintenanceWindowIds().length > 0) {
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 99a693133a2a6..62e501f6963af 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,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).not.toHaveBeenCalled();
@@ -243,7 +245,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(2);
@@ -281,7 +285,7 @@ describe('Per-Alert Action Scheduler', () => {
       });
       const alertsWithInvalidActionGroup = { ...newAlertInvalidActionGroup, ...newAlert2 };
       const results = await scheduler.getActionsToSchedule({
-        alerts: alertsWithInvalidActionGroup,
+        activeCurrentAlerts: alertsWithInvalidActionGroup,
       });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -309,6 +313,35 @@ 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,
+      });
+
+      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 +355,7 @@ describe('Per-Alert Action Scheduler', () => {
         ...newAlert2,
       };
       const results = await scheduler.getActionsToSchedule({
-        alerts: alertsWithPendingRecoveredCount,
+        activeCurrentAlerts: alertsWithPendingRecoveredCount,
       });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -368,7 +401,7 @@ describe('Per-Alert Action Scheduler', () => {
         ...newAlert2,
       };
       const results = await scheduler.getActionsToSchedule({
-        alerts: alertsWithPendingRecoveredCount,
+        activeCurrentAlerts: alertsWithPendingRecoveredCount,
       });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -394,7 +427,9 @@ describe('Per-Alert Action Scheduler', () => {
         ...getSchedulerContext(),
         rule: { ...rule, mutedInstanceIds: ['2'] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(1);
@@ -453,7 +488,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(1);
@@ -508,7 +545,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).toHaveBeenCalledTimes(1);
@@ -563,7 +602,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
       expect(logger.debug).not.toHaveBeenCalled();
@@ -620,7 +661,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -679,7 +722,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -739,7 +784,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -799,7 +846,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -860,7 +909,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -919,7 +970,9 @@ 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,
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -960,7 +1013,9 @@ describe('Per-Alert Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
 
@@ -996,7 +1051,9 @@ describe('Per-Alert Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+      });
 
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
 
@@ -1029,7 +1086,9 @@ 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 },
+      });
 
       expect(alert.getLastScheduledActions()).toEqual({
         date: '1970-01-01T00:00:00.000Z',
@@ -1066,7 +1125,9 @@ describe('Per-Alert Action Scheduler', () => {
         rule: { ...rule, actions: [onThrottleIntervalAction] },
       });
 
-      await scheduler.getActionsToSchedule({ alerts: { '1': alert } });
+      await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: { '1': alert },
+      });
 
       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 b35d86dff0105..28b35d885b3d2 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 fc810fc4ef34c..cb19cb781ae3e 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
@@ -13,7 +13,13 @@ import { alertingEventLoggerMock } from '../../../lib/alerting_event_logger/aler
 import { RuleRunMetricsStore } from '../../../lib/rule_run_metrics_store';
 import { mockAAD } from '../../fixtures';
 import { SummaryActionScheduler } from './summary_action_scheduler';
-import { getRule, getRuleType, getDefaultSchedulerContext, generateAlert } from '../test_fixtures';
+import {
+  getRule,
+  getRuleType,
+  getDefaultSchedulerContext,
+  generateAlert,
+  generateRecoveredAlert,
+} from '../test_fixtures';
 import { RuleAction } from '@kbn/alerting-types';
 import { ALERT_UUID } from '@kbn/rule-data-utils';
 import {
@@ -165,6 +171,7 @@ describe('Summary Action Scheduler', () => {
   describe('getActionsToSchedule', () => {
     const newAlert1 = generateAlert({ id: 1 });
     const newAlert2 = generateAlert({ id: 2 });
+    const recoveredAlert = generateRecoveredAlert({ id: 3 });
     const alerts = { ...newAlert1, ...newAlert2 };
 
     const summaryActionWithAlertFilter: RuleAction = {
@@ -217,7 +224,10 @@ 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,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
@@ -266,7 +276,10 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -307,7 +320,10 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({ '444-444': { date: '1970-01-01T00:00:00.000Z' } });
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -340,7 +356,10 @@ 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,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({ '444-444': { date: '1969-12-31T13:00:00.000Z' } });
       expect(alertsClient.getSummarizedAlerts).not.toHaveBeenCalled();
@@ -374,7 +393,10 @@ 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,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
@@ -436,7 +458,11 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        recoveredCurrentAlerts: recoveredAlert,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -449,7 +475,7 @@ describe('Summary Action Scheduler', () => {
       });
       expect(logger.debug).toHaveBeenCalledTimes(1);
       expect(logger.debug).toHaveBeenCalledWith(
-        `(1) alert has been filtered out for: test:333-333`
+        `(2) alerts have been filtered out for: test:333-333`
       );
 
       expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toEqual(1);
@@ -480,7 +506,10 @@ describe('Summary Action Scheduler', () => {
       });
 
       const throttledSummaryActions = {};
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        throttledSummaryActions,
+      });
 
       expect(throttledSummaryActions).toEqual({});
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
@@ -507,7 +536,10 @@ describe('Summary Action Scheduler', () => {
       const scheduler = new SummaryActionScheduler(getSchedulerContext());
 
       try {
-        await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions: {} });
+        await scheduler.getActionsToSchedule({
+          activeCurrentAlerts: alerts,
+          throttledSummaryActions: {},
+        });
       } catch (err) {
         expect(err.message).toEqual(`no alerts for you`);
         expect(getErrorSource(err)).toBe(TaskErrorSource.FRAMEWORK);
@@ -533,7 +565,10 @@ describe('Summary Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions: {} });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        throttledSummaryActions: {},
+      });
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
@@ -587,7 +622,10 @@ describe('Summary Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts, throttledSummaryActions: {} });
+      const results = await scheduler.getActionsToSchedule({
+        activeCurrentAlerts: alerts,
+        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 050eea352f0d5..db53f15be2180 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 28bf58a30c689..71a7584c7280b 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,7 @@ describe('System Action Scheduler', () => {
       alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
 
       const scheduler = new SystemActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({});
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -202,7 +202,7 @@ describe('System Action Scheduler', () => {
       alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
 
       const scheduler = new SystemActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({});
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -240,7 +240,7 @@ describe('System Action Scheduler', () => {
       alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
 
       const scheduler = new SystemActionScheduler(getSchedulerContext());
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({});
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(1);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
@@ -265,7 +265,7 @@ describe('System Action Scheduler', () => {
       const scheduler = new SystemActionScheduler(getSchedulerContext());
 
       try {
-        await scheduler.getActionsToSchedule({ alerts });
+        await scheduler.getActionsToSchedule({});
       } catch (err) {
         expect(err.message).toEqual(`no alerts for you`);
         expect(getErrorSource(err)).toBe(TaskErrorSource.FRAMEWORK);
@@ -299,7 +299,7 @@ describe('System Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({});
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
@@ -361,7 +361,7 @@ describe('System Action Scheduler', () => {
           },
         },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({});
 
       expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledTimes(2);
       expect(alertsClient.getSummarizedAlerts).toHaveBeenNthCalledWith(1, {
@@ -416,7 +416,7 @@ describe('System Action Scheduler', () => {
         ...defaultContext,
         rule: { ...rule, systemActions: [differentSystemAction] },
       });
-      const results = await scheduler.getActionsToSchedule({ alerts });
+      const results = await scheduler.getActionsToSchedule({});
 
       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 b90ffb88d541b..02b9647f91866 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
@@ -90,7 +90,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 +119,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.test.ts b/x-pack/plugins/alerting/server/task_runner/task_runner.test.ts
index b6e59402ba4c6..a79dfe8f59c73 100644
--- a/x-pack/plugins/alerting/server/task_runner/task_runner.test.ts
+++ b/x-pack/plugins/alerting/server/task_runner/task_runner.test.ts
@@ -1677,6 +1677,7 @@ describe('Task Runner', () => {
         return { state: {} };
       });
 
+      alertsClient.getProcessedAlerts.mockReturnValue({});
       alertsClient.getSummarizedAlerts.mockResolvedValue({
         new: {
           count: 1,
@@ -1738,7 +1739,7 @@ describe('Task Runner', () => {
       ruleType.executor.mockImplementation(async () => {
         return { state: {} };
       });
-
+      alertsClient.getProcessedAlerts.mockReturnValue({});
       alertsClient.getSummarizedAlerts.mockResolvedValue({
         new: {
           count: 1,
@@ -1747,6 +1748,7 @@ describe('Task Runner', () => {
         ongoing: { count: 0, data: [] },
         recovered: { count: 0, data: [] },
       });
+
       alertsClient.getAlertsToSerialize.mockResolvedValueOnce({ state: {}, meta: {} });
       alertsService.createAlertsClient.mockImplementation(() => alertsClient);
 
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 897937ce55a0a..89432e1822029 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'),
           });
         }
       })