feat(workflow_engine): Add support to handle Activity events in the WorkflowEventData

src/sentry/api/endpoints/project_rule_actions.py

@@ -159,6 +159,7 @@ def execute_future_on_test_event_workflow_engine(
 
         event_data = WorkflowEventData(
             event=test_event,
+            group=test_event.group,
         )
 
         for action_blob in actions:

src/sentry/notifications/notification_action/types.py

@@ -49,7 +49,9 @@ class LegacyRegistryHandler(ABC):
 
     @staticmethod
     @abstractmethod
-    def handle_workflow_action(job: WorkflowEventData, action: Action, detector: Detector) -> None:
+    def handle_workflow_action(
+        event_data: WorkflowEventData, action: Action, detector: Detector
+    ) -> None:
         """
         Implement this method to handle the specific notification logic for your handler.
         """
@@ -120,16 +122,16 @@ def create_rule_instance_from_action(
         cls,
         action: Action,
         detector: Detector,
-        job: WorkflowEventData,
+        event_data: WorkflowEventData,
     ) -> Rule:
         """
         Creates a Rule instance from the Action model.
         :param action: Action
         :param detector: Detector
-        :param job: WorkflowEventData
+        :param event_data: WorkflowEventData
         :return: Rule instance
         """
-        environment_id = job.workflow_env.id if job.workflow_env else None
+        environment_id = event_data.workflow_env.id if event_data.workflow_env else None
 
         data: RuleData = {
             "actions": [cls.build_rule_action_blob(action, detector.project.organization.id)],
@@ -194,7 +196,7 @@ def create_rule_instance_from_action(
 
     @staticmethod
     def get_rule_futures(
-        job: WorkflowEventData,
+        event_data: WorkflowEventData,
         rule: Rule,
         notification_uuid: str,
     ) -> Collection[tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], list[RuleFuture]]]:
@@ -205,12 +207,17 @@ def get_rule_futures(
         with sentry_sdk.start_span(
             op="workflow_engine.handlers.action.notification.issue_alert.invoke_legacy_registry.activate_downstream_actions"
         ):
-            grouped_futures = activate_downstream_actions(rule, job.event, notification_uuid)
+            if not isinstance(event_data.event, GroupEvent):
+                raise ValueError(
+                    "WorkflowEventData.event is not a GroupEvent when invoking legacy registry"
+                )
+
+            grouped_futures = activate_downstream_actions(rule, event_data.event, notification_uuid)
             return grouped_futures.values()
 
     @staticmethod
     def execute_futures(
-        job: WorkflowEventData,
+        event_data: WorkflowEventData,
         futures: Collection[
             tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], list[RuleFuture]]
         ],
@@ -222,12 +229,17 @@ def execute_futures(
         with sentry_sdk.start_span(
             op="workflow_engine.handlers.action.notification.issue_alert.execute_futures"
         ):
+            if not isinstance(event_data.event, GroupEvent):
+                raise ValueError(
+                    "WorkflowEventData.event is not a GroupEvent when evaluating issue alerts"
+                )
+
             for callback, future in futures:
-                safe_execute(callback, job.event, future)
+                safe_execute(callback, event_data.event, future)
 
     @staticmethod
     def send_test_notification(
-        job: WorkflowEventData,
+        event_data: WorkflowEventData,
         futures: Collection[
             tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], list[RuleFuture]]
         ],
@@ -236,22 +248,27 @@ def send_test_notification(
         This method will execute the futures.
         Based off of process_rules in post_process.py
         """
+        if not isinstance(event_data.event, GroupEvent):
+            raise ValueError(
+                "WorkflowEventData.event is not a GroupEvent when sending test notification"
+            )
+
         with sentry_sdk.start_span(
             op="workflow_engine.handlers.action.notification.issue_alert.execute_futures"
         ):
             for callback, future in futures:
-                callback(job.event, future)
+                callback(event_data.event, future)
 
     @classmethod
     def invoke_legacy_registry(
         cls,
-        job: WorkflowEventData,
+        event_data: WorkflowEventData,
         action: Action,
         detector: Detector,
     ) -> None:
         """
         This method will create a rule instance from the Action model, and then invoke the legacy registry.
-        This method encompases the following logic in our legacy system:
+        This method encompasses the following logic in our legacy system:
         1. post_process process_rules calls rule_processor.apply
         2. activate_downstream_actions
         3. execute_futures (also in post_process process_rules)
@@ -264,14 +281,14 @@ def invoke_legacy_registry(
             notification_uuid = str(uuid.uuid4())
 
             # Create a rule
-            rule = cls.create_rule_instance_from_action(action, detector, job)
+            rule = cls.create_rule_instance_from_action(action, detector, event_data)
 
             logger.info(
                 "notification_action.execute_via_issue_alert_handler",
                 extra={
                     "action_id": action.id,
                     "detector_id": detector.id,
-                    "job": asdict(job),
+                    "event_data": asdict(event_data),
                     "rule_id": rule.id,
                     "rule_project_id": rule.project.id,
                     "rule_environment_id": rule.environment_id,
@@ -280,14 +297,14 @@ def invoke_legacy_registry(
                 },
             )
             # Get the futures
-            futures = cls.get_rule_futures(job, rule, notification_uuid)
+            futures = cls.get_rule_futures(event_data, rule, notification_uuid)
 
             # Execute the futures
             # If the rule id is -1, we are sending a test notification
             if rule.id == -1:
-                cls.send_test_notification(job, futures)
+                cls.send_test_notification(event_data, futures)
             else:
-                cls.execute_futures(job, futures)
+                cls.execute_futures(event_data, futures)
 
 
 class TicketingIssueAlertHandler(BaseIssueAlertHandler):
@@ -367,27 +384,31 @@ def send_alert(
     @classmethod
     def invoke_legacy_registry(
         cls,
-        job: WorkflowEventData,
+        event_data: WorkflowEventData,
         action: Action,
         detector: Detector,
     ) -> None:
+        if not isinstance(event_data.event, GroupEvent):
+            raise ValueError(
+                "WorkflowEventData.event must be a GroupEvent to invoke metric alert legacy registry"
+            )
 
         with sentry_sdk.start_span(
             op="workflow_engine.handlers.action.notification.metric_alert.invoke_legacy_registry"
         ):
-            event = job.event
-            if not event.occurrence:
+            event = event_data.event
+            if not isinstance(event, GroupEvent) or event.occurrence is None:
                 raise ValueError("Event occurrence is required for alert context")
 
             evidence_data = MetricIssueEvidenceData(**event.occurrence.evidence_data)
 
             notification_context = cls.build_notification_context(action)
             alert_context = cls.build_alert_context(
-                detector, evidence_data, event.group.status, event.occurrence.priority
+                detector, evidence_data, event_data.group.status, event.occurrence.priority
             )
 
             metric_issue_context = cls.build_metric_issue_context(
-                event.group, evidence_data, event.occurrence.priority
+                event_data.group, evidence_data, event.occurrence.priority
             )
             open_period_context = cls.build_open_period_context(event)
 
@@ -400,7 +421,7 @@ def invoke_legacy_registry(
                 extra={
                     "action_id": action.id,
                     "detector_id": detector.id,
-                    "job": asdict(job),
+                    "event_data": asdict(event_data),
                     "notification_context": asdict(notification_context),
                     "alert_context": asdict(alert_context),
                     "metric_issue_context": asdict(metric_issue_context),

src/sentry/notifications/notification_action/utils.py

@@ -1,5 +1,6 @@
 import logging
 
+from sentry.models.activity import Activity
 from sentry.notifications.notification_action.registry import (
     group_type_notification_registry,
     issue_alert_handler_registry,
@@ -12,11 +13,27 @@
 
 
 def execute_via_group_type_registry(
-    job: WorkflowEventData, action: Action, detector: Detector
+    event_data: WorkflowEventData, action: Action, detector: Detector
 ) -> None:
+    """
+    Generic "notification action handler" this method will lookup which registry
+    to send the notification to, based on the type of detector that created it.
+
+    This currently only supported detector types: 'error', 'metric_issue'
+
+    If an `Activity` model for a `Group` is provided in the event data
+    it will send an activity notification instead.
+    """
+    if isinstance(event_data.event, Activity):
+        # TODO - this is a workaround to ensure a notification is sent about the issue.
+        # We'll need to update this in the future to read the notification configuration
+        # from the Action, then get the template for the activity, and send it to that
+        # integration.
+        return event_data.event.send_notification()
+
     try:
         handler = group_type_notification_registry.get(detector.type)
-        handler.handle_workflow_action(job, action, detector)
+        handler.handle_workflow_action(event_data, action, detector)
     except NoRegistrationExistsError:
         logger.exception(
             "No notification handler found for detector type: %s",

src/sentry/tasks/post_process.py

@@ -965,6 +965,7 @@ def process_workflow_engine(job: PostProcessJob) -> None:
     try:
         workflow_event_data = WorkflowEventData(
             event=job["event"],
+            group=job["event"].group,
             group_state=job.get("group_state"),
             has_reappeared=job.get("has_reappeared"),
             has_escalated=job.get("has_escalated"),

src/sentry/workflow_engine/endpoints/organization_test_fire_action.py

@@ -107,6 +107,7 @@ def test_fire_actions(actions: list[dict[str, Any]], project: Project):
     workflow_id = -1
     workflow_event_data = WorkflowEventData(
         event=test_event,
+        group=test_event.group,
     )
 
     detector = Detector(

src/sentry/workflow_engine/handlers/condition/age_comparison_handler.py

@@ -30,8 +30,8 @@ class AgeComparisonConditionHandler(DataConditionHandler[WorkflowEventData]):
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        event = event_data.event
-        first_seen = event.group.first_seen
+        group = event_data.group
+        first_seen = group.first_seen
         current_time = timezone.now()
         comparison_type = comparison["comparison_type"]
         time = comparison["time"]

src/sentry/workflow_engine/handlers/condition/assigned_to_handler.py

@@ -43,9 +43,9 @@ def get_assignees(group: Group) -> Sequence[GroupAssignee]:
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        event = event_data.event
+        group = event_data.group
         target_type = AssigneeTargetType(comparison.get("target_type"))
-        assignees = AssignedToConditionHandler.get_assignees(event.group)
+        assignees = AssignedToConditionHandler.get_assignees(group)
 
         if target_type == AssigneeTargetType.UNASSIGNED:
             return len(assignees) == 0

src/sentry/workflow_engine/handlers/condition/event_attribute_handler.py

@@ -77,6 +77,10 @@ def get_attribute_values(event: GroupEvent, attribute: str) -> list[str]:
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         event = event_data.event
+        if not isinstance(event, GroupEvent):
+            # cannot evaluate event attributes on non-GroupEvent types
+            return False
+
         attribute = comparison.get("attribute", "")
         attribute_values = EventAttributeConditionHandler.get_attribute_values(event, attribute)
 

src/sentry/workflow_engine/handlers/condition/event_created_by_detector_handler.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from sentry.eventstore.models import GroupEvent
 from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.registry import condition_handler_registry
 from sentry.workflow_engine.types import DataConditionHandler, WorkflowEventData
@@ -12,6 +13,9 @@ class EventCreatedByDetectorConditionHandler(DataConditionHandler[WorkflowEventD
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         event = event_data.event
+        if not isinstance(event, GroupEvent):
+            return False
+
         if event.occurrence is None or event.occurrence.evidence_data is None:
             return False
 

src/sentry/workflow_engine/handlers/condition/event_seen_count_handler.py

@@ -11,5 +11,5 @@ class EventSeenCountConditionHandler(DataConditionHandler[WorkflowEventData]):
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        event = event_data.event
-        return event.group.times_seen == comparison
+        group = event_data.group
+        return group.times_seen == comparison

src/sentry/workflow_engine/handlers/condition/existing_high_priority_issue_handler.py

@@ -18,4 +18,4 @@ def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
             return False
 
         is_escalating = bool(event_data.has_reappeared or event_data.has_escalated)
-        return is_escalating and event_data.event.group.priority == PriorityLevel.HIGH
+        return is_escalating and event_data.group.priority == PriorityLevel.HIGH

src/sentry/workflow_engine/handlers/condition/issue_category_handler.py

@@ -21,7 +21,7 @@ class IssueCategoryConditionHandler(DataConditionHandler[WorkflowEventData]):
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        group = event_data.event.group
+        group = event_data.group
 
         try:
             value: GroupCategory = GroupCategory(int(comparison["value"]))

src/sentry/workflow_engine/handlers/condition/issue_occurrences_handler.py

@@ -22,7 +22,7 @@ class IssueOccurrencesConditionHandler(DataConditionHandler[WorkflowEventData]):
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        group: Group = event_data.event.group
+        group: Group = event_data.group
         try:
             value = int(comparison["value"])
         except (TypeError, ValueError, KeyError):

src/sentry/workflow_engine/handlers/condition/issue_priority_deescalating_handler.py

@@ -14,7 +14,7 @@ class IssuePriorityDeescalatingConditionHandler(DataConditionHandler[WorkflowEve
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        group = event_data.event.group
+        group = event_data.group
 
         # we will fire actions on de-escalation if the priority seen is >= the threshold
         # priority specified in the comparison

src/sentry/workflow_engine/handlers/condition/issue_priority_equals.py

@@ -12,5 +12,5 @@ class IssuePriorityCondition(DataConditionHandler[WorkflowEventData]):
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        group = event_data.event.group
+        group = event_data.group
         return group.priority == comparison

src/sentry/workflow_engine/handlers/condition/issue_priority_greater_or_equal_handler.py

@@ -12,5 +12,5 @@ class IssuePriorityGreaterOrEqualConditionHandler(DataConditionHandler[WorkflowE
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        group = event_data.event.group
+        group = event_data.group
         return group.priority >= comparison

src/sentry/workflow_engine/handlers/condition/issue_resolution_condition_handler.py

@@ -11,5 +11,5 @@ class IssueResolutionConditionHandler(DataConditionHandler[WorkflowEventData]):
 
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
-        group = event_data.event.group
+        group = event_data.group
         return group.status == comparison

src/sentry/workflow_engine/handlers/condition/latest_adopted_release_handler.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from sentry.models.activity import Activity
 from sentry.models.environment import Environment
 from sentry.models.release import follows_semver_versioning_scheme
 from sentry.rules.age import AgeComparisonType, ModelAgeType
@@ -39,6 +40,9 @@ def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         environment_name = comparison["environment"]
 
         event = event_data.event
+        if isinstance(event, Activity):
+            # If the event is an Activity, we cannot determine the latest adopted release
+            return False
 
         if follows_semver_versioning_scheme(event.organization.id, event.project.id):
             order_type = LatestReleaseOrders.SEMVER

src/sentry/workflow_engine/handlers/condition/latest_release_handler.py

@@ -48,6 +48,8 @@ class LatestReleaseConditionHandler(DataConditionHandler[WorkflowEventData]):
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         event = event_data.event
+        if not isinstance(event, GroupEvent):
+            return False
 
         latest_release = get_latest_release_for_env(event_data.workflow_env, event)
         if not latest_release: