feat(workflow_engine): Bypass Inbox group frequency suppression

src/sentry/workflow_engine/docs/execution.md

@@ -362,6 +362,9 @@ before event-level deduplication and task execution. A tuple is suppressed when
 passed within `Workflow.config.frequency` minutes, even if a later equivalent action was
 deduplicated or an external handler failed.
 
+Only alert workflows use group-frequency suppression; Inbox workflows bypass it,
+including when both match the same issue Activity.
+
 ### Event-level deduplication
 
 `get_unique_active_actions` first removes actions whose status is not active, then keeps
@@ -373,11 +376,12 @@ imply Workflow execution order.
 ### History and task dispatch
 
 The processor updates frequency statuses and creates
-[`WorkflowFireHistory`](../models/workflow_fire_history.py) rows before `fire_actions`
+[`WorkflowFireHistory`](../models/workflow_fire_history.py) rows for eligible workflows before `fire_actions`
 removes inactive actions and deduplicates by key. It then schedules
 [`trigger_action`](../tasks/actions.py) for retained actions. Fire-history rows describe
 scheduling, not successful external delivery, and can exist for actions that are later
-filtered or deduplicated.
+filtered or deduplicated. Sharing an action with an eligible workflow does not create
+history for a suppressed workflow.
 
 The action task:
 

src/sentry/workflow_engine/processors/action.py

@@ -295,34 +295,14 @@ def fire_actions(
         trigger_action.apply_async(kwargs=task_params, headers={"sentry-propagate-traces": False})
 
 
-def filter_recently_fired_workflow_actions(
-    filtered_action_groups: set[DataConditionGroup], event_data: WorkflowEventData
-) -> tuple[BaseQuerySet[Action], dict[ActionId, WorkflowId]]:
-    """
-    Returns actions associated with the provided DataConditionGroups, excluding
-    those that have been recently fired, along with a mapping of action_id to
-    the workflow_id each action is firing for. Also updates associated
-    WorkflowActionGroupStatus objects.
-    """
-
-    data_condition_group_actions = DataConditionGroupAction.objects.filter(
-        condition_group__in=filtered_action_groups
-    ).values_list("action_id", "condition_group__workflowdataconditiongroup__workflow_id")
-
-    action_to_workflows_ids: dict[int, set[int]] = defaultdict(set)
-    workflow_ids: set[int] = set()
-
-    for action_id, workflow_id in data_condition_group_actions:
-        # If we are mid-deletion, the workflow_id can be none.
-        if workflow_id is not None:
-            action_to_workflows_ids[action_id].add(workflow_id)
-            workflow_ids.add(workflow_id)
-
+def _filter_recently_fired_alert_actions(
+    action_to_workflows_ids: dict[int, set[int]], group: Group
+) -> dict[int, set[int]]:
+    workflow_ids = {workflow_id for ids in action_to_workflows_ids.values() for workflow_id in ids}
     workflows = Workflow.objects.filter(id__in=workflow_ids)
-
     action_to_statuses = get_workflow_action_group_statuses(
         action_to_workflows_ids=action_to_workflows_ids,
-        group=event_data.require_group(),
+        group=group,
         workflow_ids=workflow_ids,
     )
     now = timezone.now()
@@ -331,7 +311,7 @@ def filter_recently_fired_workflow_actions(
             action_to_workflows_ids=action_to_workflows_ids,
             action_to_statuses=action_to_statuses,
             workflows=workflows,
-            group=event_data.require_group(),
+            group=group,
             now=now,
         )
     )
@@ -343,14 +323,48 @@ def filter_recently_fired_workflow_actions(
         if not action_to_workflows_ids[action_id]:
             action_to_workflows_ids.pop(action_id)
 
-    actions_queryset = Action.objects.filter(id__in=list(action_to_workflows_ids.keys()))
+    return action_to_workflows_ids
 
-    # Build a mapping of action_id -> workflow_id (pick one arbitrarily but deterministically)
-    action_to_workflow_id: dict[ActionId, WorkflowId] = {
-        action_id: min(wf_ids) for action_id, wf_ids in action_to_workflows_ids.items()
-    }
 
-    return actions_queryset, action_to_workflow_id
+def filter_recently_fired_workflow_actions(
+    filtered_action_groups: set[DataConditionGroup], event_data: WorkflowEventData
+) -> tuple[BaseQuerySet[Action], dict[ActionId, WorkflowId], set[WorkflowId]]:
+    """Select actions and their eligible workflows before equivalent-action deduplication.
+
+    Inbox workflows bypass group-frequency state. Alert workflows retain their
+    existing per-group suppression. The complete eligible workflow set is returned
+    for history, while each action is attributed to one workflow for task dispatch.
+    """
+    organization = event_data.get_organization()
+    associations = DataConditionGroupAction.objects.filter(
+        condition_group__in=filtered_action_groups,
+        condition_group__workflowdataconditiongroup__workflow__organization_id=organization.id,
+    ).values_list(
+        "action_id",
+        "condition_group__workflowdataconditiongroup__workflow_id",
+        "condition_group__workflowdataconditiongroup__workflow__type",
+    )
+    inbox_actions: dict[int, set[int]] = defaultdict(set)
+    alert_actions: dict[int, set[int]] = defaultdict(set)
+    for action_id, workflow_id, workflow_type in associations:
+        if workflow_type == Workflow.Type.INBOX:
+            inbox_actions[action_id].add(workflow_id)
+        elif event_data.group is not None:
+            alert_actions[action_id].add(workflow_id)
+
+    eligible_actions = inbox_actions
+    if alert_actions and event_data.group is not None:
+        for action_id, workflow_ids in _filter_recently_fired_alert_actions(
+            alert_actions, event_data.group
+        ).items():
+            eligible_actions[action_id].update(workflow_ids)
+
+    actions = get_supported_actions(Action.objects.filter(id__in=eligible_actions), event_data)
+    action_to_workflow_id = {action.id: min(eligible_actions[action.id]) for action in actions}
+    eligible_workflow_ids = {
+        workflow_id for action in actions for workflow_id in eligible_actions[action.id]
+    }
+    return actions, action_to_workflow_id, eligible_workflow_ids
 
 
 def get_available_action_integrations_for_org(organization: Organization) -> list[RpcIntegration]:

src/sentry/workflow_engine/processors/delayed_workflow.py

@@ -848,8 +848,8 @@ def fire_actions_for_groups(
                 )
 
                 dcgs_for_group = groups_to_fire.get(group.id, set())
-                filtered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions(
-                    dcgs_for_group, workflow_event_data
+                filtered_actions, action_to_workflow_id, eligible_workflow_ids = (
+                    filter_recently_fired_workflow_actions(dcgs_for_group, workflow_event_data)
                 )
                 # TODO: trigger service hooks from here
 
@@ -864,6 +864,7 @@ def fire_actions_for_groups(
                     workflow_event_data,
                     is_delayed=True,
                     start_timestamp=start_timestamp,
+                    eligible_workflow_ids=eligible_workflow_ids,
                 )
 
                 # Create mapping: workflow_id -> notification_uuid for propagation

src/sentry/workflow_engine/processors/workflow.py

@@ -643,7 +643,7 @@ def process_workflows(
         return result
     enqueue_workflows(batch_client, delayed_items)
 
-    actions, action_to_workflow_id = filter_recently_fired_workflow_actions(
+    actions, action_to_workflow_id, eligible_workflow_ids = filter_recently_fired_workflow_actions(
         action_groups, event_data
     )
 
@@ -657,6 +657,7 @@ def process_workflows(
             event_data,
             is_delayed=False,
             start_timestamp=event_start_time,
+            eligible_workflow_ids=eligible_workflow_ids,
         )
 
         # Create mapping: workflow_id -> notification_uuid for propagation

src/sentry/workflow_engine/processors/workflow_fire_history.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Sequence
+from collections.abc import Collection, Sequence
 from datetime import datetime, timezone
 
 from django.db.models import OuterRef, Subquery
@@ -30,6 +30,8 @@ def create_workflow_fire_histories(
     event_data: WorkflowEventData,
     is_delayed: bool = False,
     start_timestamp: datetime | None = None,
+    *,
+    eligible_workflow_ids: Collection[int] | None = None,
 ) -> list[WorkflowFireHistory]:
     """
     Record that the workflows associated with these actions were fired for this
@@ -46,6 +48,9 @@ def create_workflow_fire_histories(
         ).values_list("workflow_id", flat=True)
     )
 
+    if eligible_workflow_ids is not None:
+        workflow_ids.intersection_update(eligible_workflow_ids)
+
     event_id = event_data.event_id
 
     if start_timestamp: