diff --git a/src/sentry/api/endpoints/project_rule_actions.py b/src/sentry/api/endpoints/project_rule_actions.py
index c5e74a5f8cc920..a4e6306834e0f7 100644
--- a/src/sentry/api/endpoints/project_rule_actions.py
+++ b/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:
diff --git a/src/sentry/notifications/notification_action/types.py b/src/sentry/notifications/notification_action/types.py
index 695bd8615f4d40..46f253c9ddf480 100644
--- a/src/sentry/notifications/notification_action/types.py
+++ b/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),
diff --git a/src/sentry/notifications/notification_action/utils.py b/src/sentry/notifications/notification_action/utils.py
index b763324f0d7356..bd6c19ccd999db 100644
--- a/src/sentry/notifications/notification_action/utils.py
+++ b/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",
diff --git a/src/sentry/tasks/post_process.py b/src/sentry/tasks/post_process.py
index a0b5cb61cc3d16..bafb30dc6947ee 100644
--- a/src/sentry/tasks/post_process.py
+++ b/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"),
diff --git a/src/sentry/workflow_engine/endpoints/organization_test_fire_action.py b/src/sentry/workflow_engine/endpoints/organization_test_fire_action.py
index 6b53fe95c64fff..0e73a6cc2c89b0 100644
--- a/src/sentry/workflow_engine/endpoints/organization_test_fire_action.py
+++ b/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(
diff --git a/src/sentry/workflow_engine/handlers/condition/age_comparison_handler.py b/src/sentry/workflow_engine/handlers/condition/age_comparison_handler.py
index 7bd650af16a17d..831e85846b2b9e 100644
--- a/src/sentry/workflow_engine/handlers/condition/age_comparison_handler.py
+++ b/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"]
diff --git a/src/sentry/workflow_engine/handlers/condition/assigned_to_handler.py b/src/sentry/workflow_engine/handlers/condition/assigned_to_handler.py
index e2bb332dc1cf4a..3ee41580e20816 100644
--- a/src/sentry/workflow_engine/handlers/condition/assigned_to_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/event_attribute_handler.py b/src/sentry/workflow_engine/handlers/condition/event_attribute_handler.py
index 02e6d056c5a1b3..2da300fbd794b1 100644
--- a/src/sentry/workflow_engine/handlers/condition/event_attribute_handler.py
+++ b/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)
 
diff --git a/src/sentry/workflow_engine/handlers/condition/event_created_by_detector_handler.py b/src/sentry/workflow_engine/handlers/condition/event_created_by_detector_handler.py
index a09bc50f2c13fe..61a9daf2e0fe1f 100644
--- a/src/sentry/workflow_engine/handlers/condition/event_created_by_detector_handler.py
+++ b/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
 
diff --git a/src/sentry/workflow_engine/handlers/condition/event_seen_count_handler.py b/src/sentry/workflow_engine/handlers/condition/event_seen_count_handler.py
index 34fd71b9fadbb9..4579025615b738 100644
--- a/src/sentry/workflow_engine/handlers/condition/event_seen_count_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/existing_high_priority_issue_handler.py b/src/sentry/workflow_engine/handlers/condition/existing_high_priority_issue_handler.py
index 77324c79003127..96cb5eb06850c4 100644
--- a/src/sentry/workflow_engine/handlers/condition/existing_high_priority_issue_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/issue_category_handler.py b/src/sentry/workflow_engine/handlers/condition/issue_category_handler.py
index 8fc013f7c2979a..d6b4044ea4ce4d 100644
--- a/src/sentry/workflow_engine/handlers/condition/issue_category_handler.py
+++ b/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"]))
diff --git a/src/sentry/workflow_engine/handlers/condition/issue_occurrences_handler.py b/src/sentry/workflow_engine/handlers/condition/issue_occurrences_handler.py
index a38f67658458a8..87f20a03016884 100644
--- a/src/sentry/workflow_engine/handlers/condition/issue_occurrences_handler.py
+++ b/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):
diff --git a/src/sentry/workflow_engine/handlers/condition/issue_priority_deescalating_handler.py b/src/sentry/workflow_engine/handlers/condition/issue_priority_deescalating_handler.py
index 7b9df2b06602a3..f23ca7e409a81d 100644
--- a/src/sentry/workflow_engine/handlers/condition/issue_priority_deescalating_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/issue_priority_equals.py b/src/sentry/workflow_engine/handlers/condition/issue_priority_equals.py
index 645c2cdaf33eb8..a3d1c269383111 100644
--- a/src/sentry/workflow_engine/handlers/condition/issue_priority_equals.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/issue_priority_greater_or_equal_handler.py b/src/sentry/workflow_engine/handlers/condition/issue_priority_greater_or_equal_handler.py
index cf8bcac6a0ae35..bf3684c65ab19e 100644
--- a/src/sentry/workflow_engine/handlers/condition/issue_priority_greater_or_equal_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/issue_resolution_condition_handler.py b/src/sentry/workflow_engine/handlers/condition/issue_resolution_condition_handler.py
index b8826c86b22608..195ec02376124e 100644
--- a/src/sentry/workflow_engine/handlers/condition/issue_resolution_condition_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/latest_adopted_release_handler.py b/src/sentry/workflow_engine/handlers/condition/latest_adopted_release_handler.py
index 11d65ab7d86887..15d90e07cc1675 100644
--- a/src/sentry/workflow_engine/handlers/condition/latest_adopted_release_handler.py
+++ b/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
diff --git a/src/sentry/workflow_engine/handlers/condition/latest_release_handler.py b/src/sentry/workflow_engine/handlers/condition/latest_release_handler.py
index 15ba839cff78a0..3fcc6043a571ff 100644
--- a/src/sentry/workflow_engine/handlers/condition/latest_release_handler.py
+++ b/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:
diff --git a/src/sentry/workflow_engine/handlers/condition/level_handler.py b/src/sentry/workflow_engine/handlers/condition/level_handler.py
index b9370f9c656933..fa018221068f28 100644
--- a/src/sentry/workflow_engine/handlers/condition/level_handler.py
+++ b/src/sentry/workflow_engine/handlers/condition/level_handler.py
@@ -1,6 +1,7 @@
 from typing import Any
 
 from sentry.constants import LOG_LEVELS_MAP
+from sentry.eventstore.models import GroupEvent
 from sentry.rules import MatchType
 from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.registry import condition_handler_registry
@@ -25,6 +26,11 @@ class LevelConditionHandler(DataConditionHandler[WorkflowEventData]):
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         event = event_data.event
+
+        if not isinstance(event, GroupEvent):
+            # This condition is only applicable to GroupEvent
+            return False
+
         level_name = event.get_tag("level")
         if level_name is None:
             return False
diff --git a/src/sentry/workflow_engine/handlers/condition/new_high_priority_issue_handler.py b/src/sentry/workflow_engine/handlers/condition/new_high_priority_issue_handler.py
index c637439d73ff96..e4de218cf67d5e 100644
--- a/src/sentry/workflow_engine/handlers/condition/new_high_priority_issue_handler.py
+++ b/src/sentry/workflow_engine/handlers/condition/new_high_priority_issue_handler.py
@@ -19,4 +19,4 @@ def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         if not event.project.flags.has_high_priority_alerts:
             return is_new
 
-        return is_new and event.group.priority == PriorityLevel.HIGH
+        return is_new and event_data.group.priority == PriorityLevel.HIGH
diff --git a/src/sentry/workflow_engine/handlers/condition/tagged_event_handler.py b/src/sentry/workflow_engine/handlers/condition/tagged_event_handler.py
index 76b9a051c0e94a..e9f4df45078332 100644
--- a/src/sentry/workflow_engine/handlers/condition/tagged_event_handler.py
+++ b/src/sentry/workflow_engine/handlers/condition/tagged_event_handler.py
@@ -1,6 +1,7 @@
 from typing import Any
 
 from sentry import tagstore
+from sentry.eventstore.models import GroupEvent
 from sentry.rules import MatchType, match_values
 from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.registry import condition_handler_registry
@@ -51,6 +52,11 @@ class TaggedEventConditionHandler(DataConditionHandler[WorkflowEventData]):
     @staticmethod
     def evaluate_value(event_data: WorkflowEventData, comparison: Any) -> bool:
         event = event_data.event
+
+        if not isinstance(event, GroupEvent):
+            # We can only evaluate tagged events for GroupEvent types
+            return False
+
         raw_tags = event.tags
         key = comparison["key"]
         match = comparison["match"]
diff --git a/src/sentry/workflow_engine/processors/action.py b/src/sentry/workflow_engine/processors/action.py
index b25980fdd18c71..d436815aa210da 100644
--- a/src/sentry/workflow_engine/processors/action.py
+++ b/src/sentry/workflow_engine/processors/action.py
@@ -8,6 +8,7 @@
 from sentry import features
 from sentry.constants import ObjectStatus
 from sentry.db.models.manager.base_query_set import BaseQuerySet
+from sentry.eventstore.models import GroupEvent
 from sentry.exceptions import NotRegistered
 from sentry.integrations.base import IntegrationFeatures
 from sentry.integrations.manager import default_manager as integrations_manager
@@ -51,11 +52,16 @@ def create_workflow_fire_histories(
         ).values_list("workflow_id", flat=True)
     )
 
+    event_id = (
+        event_data.event.event_id
+        if isinstance(event_data.event, GroupEvent)
+        else event_data.event.id
+    )
     fire_histories = [
         WorkflowFireHistory(
             workflow_id=workflow_id,
             group=event_data.event.group,
-            event_id=event_data.event.event_id,
+            event_id=event_id,
         )
         for workflow_id in workflow_ids
     ]
@@ -169,7 +175,7 @@ def filter_recently_fired_workflow_actions(
 
     action_to_statuses = get_workflow_action_group_statuses(
         action_to_workflows_ids=action_to_workflows_ids,
-        group=event_data.event.group,
+        group=event_data.group,
         workflow_ids=workflow_ids,
     )
     now = timezone.now()
@@ -178,7 +184,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.event.group,
+            group=event_data.group,
             now=now,
         )
     )
@@ -266,7 +272,7 @@ def _get_integration_features(action_type: Action.Type) -> frozenset[Integration
     return integration.features
 
 
-# The features that are relevent to Action behaviors;
+# The features that are relevant to Action behaviors;
 # if the organization doesn't have access to all of the features an integration
 # requires that are in this list, the action should not be permitted.
 _ACTION_RELEVANT_INTEGRATION_FEATURES = {
diff --git a/src/sentry/workflow_engine/processors/delayed_workflow.py b/src/sentry/workflow_engine/processors/delayed_workflow.py
index 47fc66f8ec990b..819886ceaf1f5a 100644
--- a/src/sentry/workflow_engine/processors/delayed_workflow.py
+++ b/src/sentry/workflow_engine/processors/delayed_workflow.py
@@ -525,7 +525,7 @@ def fire_actions_for_groups(
     ) as tracker:
         for group, group_event in group_to_groupevent.items():
             with tracker.track(str(group.id)), log_context.new_context(group_id=group.id):
-                workflow_event_data = WorkflowEventData(event=group_event)
+                workflow_event_data = WorkflowEventData(event=group_event, group=group)
                 detector = get_detector_by_event(workflow_event_data)
 
                 workflow_triggers: set[DataConditionGroup] = set()
diff --git a/src/sentry/workflow_engine/processors/detector.py b/src/sentry/workflow_engine/processors/detector.py
index 50911f7e89b0ab..34cba996247cae 100644
--- a/src/sentry/workflow_engine/processors/detector.py
+++ b/src/sentry/workflow_engine/processors/detector.py
@@ -2,6 +2,7 @@
 
 import logging
 
+from sentry.eventstore.models import GroupEvent
 from sentry.issues.issue_occurrence import IssueOccurrence
 from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 from sentry.utils import metrics
@@ -17,6 +18,12 @@
 
 def get_detector_by_event(event_data: WorkflowEventData) -> Detector:
     evt = event_data.event
+
+    if not isinstance(evt, GroupEvent):
+        raise TypeError(
+            "Can only use `get_detector_by_event` for a new event, Activity updates are not supported"
+        )
+
     issue_occurrence = evt.occurrence
 
     try:
diff --git a/src/sentry/workflow_engine/processors/workflow.py b/src/sentry/workflow_engine/processors/workflow.py
index c1683e5e100c1f..6648b7de1ddb73 100644
--- a/src/sentry/workflow_engine/processors/workflow.py
+++ b/src/sentry/workflow_engine/processors/workflow.py
@@ -36,6 +36,7 @@
 logger = logging.getLogger(__name__)
 
 WORKFLOW_ENGINE_BUFFER_LIST_KEY = "workflow_engine_delayed_processing_buffer"
+DetectorId = int | None
 
 
 class WorkflowDataConditionGroupType(StrEnum):
@@ -145,11 +146,16 @@ def evaluate_workflow_triggers(
         except Environment.DoesNotExist:
             return set()
 
+    event_id = (
+        event_data.event.event_id
+        if isinstance(event_data.event, GroupEvent)
+        else event_data.event.id
+    )
     logger.info(
         "workflow_engine.process_workflows.triggered_workflows",
         extra={
             "group_id": event_data.event.group_id,
-            "event_id": event_data.event.event_id,
+            "event_id": event_id,
             "event_data": asdict(event_data),
             "event_environment_id": environment.id,
             "triggered_workflows": [workflow.id for workflow in triggered_workflows],
@@ -201,11 +207,17 @@ def evaluate_workflows_action_filters(
 
     enqueue_workflows(queue_items_by_project_id)
 
+    event_id = (
+        event_data.event.event_id
+        if isinstance(event_data.event, GroupEvent)
+        else event_data.event.id
+    )
+
     logger.info(
         "workflow_engine.evaluate_workflows_action_filters",
         extra={
-            "group_id": event_data.event.group_id,
-            "event_id": event_data.event.event_id,
+            "group_id": event_data.group.id,
+            "event_id": event_id,
             "workflow_ids": [workflow.id for workflow in workflows],
             "action_conditions": [action_condition.id for action_condition in action_conditions],
             "filtered_action_groups": [action_group.id for action_group in filtered_action_groups],
@@ -249,12 +261,18 @@ def _get_associated_workflows(
             len(workflows),
         )
 
+        event_id = (
+            event_data.event.event_id
+            if isinstance(event_data.event, GroupEvent)
+            else event_data.event.id
+        )
         logger.info(
             "workflow_engine.process_workflows",
             extra={
                 "payload": event_data,
                 "group_id": event_data.event.group_id,
-                "event_id": event_data.event.event_id,
+                "event_id": event_id,
+                "event_data": asdict(event_data),
                 "event_environment_id": environment.id,
                 "workflows": [workflow.id for workflow in workflows],
                 "detector_type": detector.type,
@@ -265,7 +283,9 @@ def _get_associated_workflows(
 
 
 @log_context.root()
-def process_workflows(event_data: WorkflowEventData) -> set[Workflow]:
+def process_workflows(
+    event_data: WorkflowEventData, detector_id: DetectorId = None
+) -> set[Workflow]:
     """
     This method will get the detector based on the event, and then gather the associated workflows.
     Next, it will evaluate the "when" (or trigger) conditions for each workflow, if the conditions are met,
@@ -274,8 +294,14 @@ def process_workflows(event_data: WorkflowEventData) -> set[Workflow]:
     Finally, each of the triggered workflows will have their actions evaluated and executed.
     """
     try:
-        detector = get_detector_by_event(event_data)
+        detector: Detector
+        if detector_id is not None:
+            detector = Detector.objects.get(id=detector_id)
+        elif isinstance(event_data.event, GroupEvent):
+            detector = get_detector_by_event(event_data)
+
         log_context.add_extras(detector_id=detector.id)
+
         organization = detector.project.organization
 
         # set the detector / org information asap, this is used in `get_environment_by_event` as well.
diff --git a/src/sentry/workflow_engine/types.py b/src/sentry/workflow_engine/types.py
index 862d75e3e92514..d14c198b4dd314 100644
--- a/src/sentry/workflow_engine/types.py
+++ b/src/sentry/workflow_engine/types.py
@@ -12,7 +12,9 @@
     from sentry.eventstream.base import GroupState
     from sentry.issues.issue_occurrence import IssueOccurrence
     from sentry.issues.status_change_message import StatusChangeMessage
+    from sentry.models.activity import Activity
     from sentry.models.environment import Environment
+    from sentry.models.group import Group
     from sentry.models.organization import Organization
     from sentry.snuba.models import SnubaQueryEventType
     from sentry.workflow_engine.endpoints.validators.base import BaseDetectorTypeValidator
@@ -60,7 +62,8 @@ class DetectorEvaluationResult:
 
 @dataclass(frozen=True)
 class WorkflowEventData:
-    event: GroupEvent
+    event: GroupEvent | Activity
+    group: Group
     group_state: GroupState | None = None
     has_reappeared: bool | None = None
     has_escalated: bool | None = None
diff --git a/tests/sentry/notifications/notification_action/test_group_type_notification_registry_handlers.py b/tests/sentry/notifications/notification_action/test_group_type_notification_registry_handlers.py
index 007334ace06843..3e6202bcfa8c30 100644
--- a/tests/sentry/notifications/notification_action/test_group_type_notification_registry_handlers.py
+++ b/tests/sentry/notifications/notification_action/test_group_type_notification_registry_handlers.py
@@ -21,7 +21,7 @@ def setUp(self):
         self.detector = self.create_detector(project=self.project)
         self.action = Action(type=Action.Type.DISCORD)
         self.group, self.event, self.group_event = self.create_group_event()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     @mock.patch(
         "sentry.notifications.notification_action.registry.issue_alert_handler_registry.get"
@@ -43,7 +43,7 @@ def setUp(self):
         self.detector = self.create_detector(project=self.project)
         self.action = Action(type=Action.Type.DISCORD)
         self.group, self.event, self.group_event = self.create_group_event()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     @mock.patch(
         "sentry.notifications.notification_action.registry.metric_alert_handler_registry.get"
diff --git a/tests/sentry/notifications/notification_action/test_issue_alert_registry_handlers.py b/tests/sentry/notifications/notification_action/test_issue_alert_registry_handlers.py
index 0f1377e9e811a8..1b443f953ea8ae 100644
--- a/tests/sentry/notifications/notification_action/test_issue_alert_registry_handlers.py
+++ b/tests/sentry/notifications/notification_action/test_issue_alert_registry_handlers.py
@@ -85,7 +85,9 @@ def setUp(self):
             data={"tags": "environment,user,my_tag"},
         )
         self.group, self.event, self.group_event = self.create_group_event()
-        self.event_data = WorkflowEventData(event=self.group_event, workflow_env=self.environment)
+        self.event_data = WorkflowEventData(
+            event=self.group_event, workflow_env=self.environment, group=self.group
+        )
 
         self.action.workflow_id = self.workflow.id
 
@@ -111,7 +113,9 @@ def get_additional_fields(cls, action: Action, mapping: ActionFieldMapping):
             handler.create_rule_instance_from_action(self.action, self.detector, self.event_data)
 
     def test_create_rule_instance_from_action_missing_workflow_id_raises_value_error(self):
-        job = WorkflowEventData(event=self.group_event, workflow_env=self.environment)
+        job = WorkflowEventData(
+            event=self.group_event, workflow_env=self.environment, group=self.group
+        )
         action = self.create_action(
             type=Action.Type.DISCORD,
             integration_id="1234567890",
@@ -123,7 +127,9 @@ def test_create_rule_instance_from_action_missing_workflow_id_raises_value_error
             self.handler.create_rule_instance_from_action(action, self.detector, job)
 
     def test_create_rule_instance_from_action_missing_rule_raises_value_error(self):
-        job = WorkflowEventData(event=self.group_event, workflow_env=self.environment)
+        job = WorkflowEventData(
+            event=self.group_event, workflow_env=self.environment, group=self.group
+        )
         alert_rule = self.create_alert_rule(projects=[self.project], organization=self.organization)
         self.create_alert_rule_workflow(workflow=self.workflow, alert_rule_id=alert_rule.id)
         action = self.create_action(
@@ -194,7 +200,7 @@ def test_create_rule_instance_from_action_with_workflow_engine_ui_feature_flag(s
     def test_create_rule_instance_from_action_no_environment(self):
         """Test that create_rule_instance_from_action creates a Rule with correct attributes"""
         self.create_workflow()
-        job = WorkflowEventData(event=self.group_event, workflow_env=None)
+        job = WorkflowEventData(event=self.group_event, workflow_env=None, group=self.group)
         rule = self.handler.create_rule_instance_from_action(self.action, self.detector, job)
 
         assert isinstance(rule, Rule)
@@ -222,7 +228,7 @@ def test_create_rule_instance_from_action_no_environment_with_workflow_engine_ui
     ):
         """Test that create_rule_instance_from_action creates a Rule with correct attributes"""
         self.create_workflow()
-        job = WorkflowEventData(event=self.group_event, workflow_env=None)
+        job = WorkflowEventData(event=self.group_event, workflow_env=None, group=self.group)
         rule = self.handler.create_rule_instance_from_action(self.action, self.detector, job)
 
         assert isinstance(rule, Rule)
diff --git a/tests/sentry/notifications/notification_action/test_metric_alert_registry_handlers.py b/tests/sentry/notifications/notification_action/test_metric_alert_registry_handlers.py
index 3806fc8264137a..a4b78b31c54c33 100644
--- a/tests/sentry/notifications/notification_action/test_metric_alert_registry_handlers.py
+++ b/tests/sentry/notifications/notification_action/test_metric_alert_registry_handlers.py
@@ -8,6 +8,8 @@
 import pytest
 from django.utils import timezone
 
+from sentry.db.models import NodeData
+from sentry.eventstore.models import GroupEvent
 from sentry.incidents.grouptype import MetricIssue, MetricIssueEvidenceData
 from sentry.incidents.models.alert_rule import (
     AlertRuleDetectionType,
@@ -138,7 +140,9 @@ def create_models(self):
             date_started=self.group_event.group.first_seen,
         )
         self.event_data = WorkflowEventData(
-            event=self.group_event, workflow_env=self.workflow.environment
+            event=self.group_event,
+            workflow_env=self.workflow.environment,
+            group=self.group,
         )
 
     def setUp(self):
@@ -280,7 +284,10 @@ def setUp(self):
         self.handler = TestHandler()
 
     def test_missing_occurrence_raises_value_error(self):
-        self.event_data.event._occurrence = None
+        self.event_data = WorkflowEventData(
+            event=GroupEvent(self.project.id, "test", self.group, NodeData("test-id")),
+            group=self.group,
+        )
 
         with pytest.raises(ValueError):
             self.handler.invoke_legacy_registry(self.event_data, self.action, self.detector)
diff --git a/tests/sentry/workflow_engine/handlers/action/test_action_handlers.py b/tests/sentry/workflow_engine/handlers/action/test_action_handlers.py
index 5ec67a4f44dfbb..dd2663e641d2b3 100644
--- a/tests/sentry/workflow_engine/handlers/action/test_action_handlers.py
+++ b/tests/sentry/workflow_engine/handlers/action/test_action_handlers.py
@@ -17,7 +17,7 @@ def setUp(self):
         self.detector = self.create_detector(project=self.project)
         self.action = Action(type=Action.Type.DISCORD)
         self.group, self.event, self.group_event = self.create_group_event()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def test_execute_without_group_type(self):
         """Test that execute does nothing when detector has no group_type"""
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_age_comparison_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_age_comparison_handler.py
index 0ba02f129926ad..4e8ebd113bb0aa 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_age_comparison_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_age_comparison_handler.py
@@ -17,11 +17,11 @@ class TestAgeComparisonCondition(ConditionTestCase):
 
     def setup_group_event_and_job(self):
         self.group_event = self.event.for_group(self.group)
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
         self.payload = {
             "id": AgeComparisonFilter.id,
             "comparison_type": AgeComparisonType.OLDER,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_assigned_to_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_assigned_to_handler.py
index 1bd1f935ef99e1..ba6dcc7b87670e 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_assigned_to_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_assigned_to_handler.py
@@ -18,7 +18,7 @@ class TestAssignedToCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison={
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_event_attribute_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_event_attribute_handler.py
index ffaf7f23d51a98..9d8429443fa5e8 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_event_attribute_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_event_attribute_handler.py
@@ -96,6 +96,7 @@ def setup_group_event_and_job(self):
         self.group_event = self.event.for_group(self.group)
         self.event_data = WorkflowEventData(
             event=self.group_event,
+            group=self.group,
             group_state=GroupState(
                 {
                     "id": 1,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_existing_high_priority_issue_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_existing_high_priority_issue_handler.py
index 4d85c324886dc8..49bc0d8ff764e2 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_existing_high_priority_issue_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_existing_high_priority_issue_handler.py
@@ -19,6 +19,7 @@ def setUp(self):
         super().setUp()
         self.event_data = WorkflowEventData(
             event=self.group_event,
+            group=self.group_event.group,
             group_state=GroupState(
                 {
                     "id": 1,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_first_seen_event_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_first_seen_event_handler.py
index 386ac2a3c09a0b..194529d2f24ca4 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_first_seen_event_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_first_seen_event_handler.py
@@ -18,6 +18,7 @@ def setUp(self):
         super().setUp()
         self.event_data = WorkflowEventData(
             event=self.group_event,
+            group=self.group_event.group,
             group_state=GroupState(
                 {
                     "id": 1,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_issue_category_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_issue_category_handler.py
index 22083a12757431..273f486a6f4f52 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_issue_category_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_issue_category_handler.py
@@ -19,7 +19,7 @@ class TestIssueCategoryCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison={
@@ -81,14 +81,16 @@ def test_group_event(self):
         group_event = self.event.for_group(self.group)
 
         self.dc.update(comparison={"value": GroupCategory.ERROR.value})
-        self.assert_passes(self.dc, WorkflowEventData(event=self.event))
-        self.assert_passes(self.dc, WorkflowEventData(event=group_event))
+        self.assert_passes(self.dc, WorkflowEventData(event=self.event, group=self.group))
+        self.assert_passes(self.dc, WorkflowEventData(event=group_event, group=self.group))
 
     @patch("sentry.issues.grouptype.GroupTypeRegistry.get_by_type_id")
     def test_invalid_issue_category(self, mock_get_by_type_id):
         mock_get_by_type_id.side_effect = ValueError("Invalid group type")
 
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=self.event))
+        self.assert_does_not_pass(
+            self.dc, WorkflowEventData(event=self.event, group=self.event.group)
+        )
 
     def test_category_v2(self):
         perf_group, perf_event, perf_group_event = self.create_group_event(
@@ -97,7 +99,7 @@ def test_category_v2(self):
 
         # N+1 DB query issue should pass for 'PERFORMANCE' (deprecated) as well as 'DB_QUERY' (category_v2)
         self.dc.update(comparison={"value": GroupCategory.PERFORMANCE.value})
-        self.assert_passes(self.dc, WorkflowEventData(event=perf_group_event))
+        self.assert_passes(self.dc, WorkflowEventData(event=perf_group_event, group=perf_group))
 
         self.dc.update(comparison={"value": GroupCategory.DB_QUERY.value})
-        self.assert_passes(self.dc, WorkflowEventData(event=perf_group_event))
+        self.assert_passes(self.dc, WorkflowEventData(event=perf_group_event, group=perf_group))
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_issue_occurrences_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_issue_occurrences_handler.py
index 264472e50b0e98..94789779da678e 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_issue_occurrences_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_issue_occurrences_handler.py
@@ -17,7 +17,7 @@ class TestIssueOccurrencesCondition(ConditionTestCase):
     def setUp(self):
         super().setUp()
         self.group.times_seen_pending = 0
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison={
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_deescalating_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_deescalating_handler.py
index e3fc8a33c94bb0..b11af2e3e10854 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_deescalating_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_deescalating_handler.py
@@ -16,7 +16,7 @@ class TestIssuePriorityGreaterOrEqualCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.metric_alert = self.create_alert_rule()
         self.alert_rule_trigger_warning = self.create_alert_rule_trigger(
             alert_rule=self.metric_alert, label="warning"
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_equals.py b/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_equals.py
index c96c4826c84b73..8ca5011669be7f 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_equals.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_equals.py
@@ -14,7 +14,7 @@ class TestIssuePriorityCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.metric_alert = self.create_alert_rule()
         self.alert_rule_trigger_warning = self.create_alert_rule_trigger(
             alert_rule=self.metric_alert, label="warning"
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_greater_or_equal.py b/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_greater_or_equal.py
index ac5581c586587b..e50df3fe609608 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_greater_or_equal.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_issue_priority_greater_or_equal.py
@@ -14,7 +14,7 @@ class TestIssuePriorityGreaterOrEqualCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.metric_alert = self.create_alert_rule()
         self.alert_rule_trigger_warning = self.create_alert_rule_trigger(
             alert_rule=self.metric_alert, label="warning"
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_issue_resolution_condition_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_issue_resolution_condition_handler.py
index 3b8d87ec68c885..47a64627d098af 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_issue_resolution_condition_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_issue_resolution_condition_handler.py
@@ -8,7 +8,7 @@ class TestIssueResolutionChangeCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison=1,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_latest_adopted_release_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_latest_adopted_release_handler.py
index dff037f3ebe2a4..5a9f515e6f83a4 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_latest_adopted_release_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_latest_adopted_release_handler.py
@@ -62,7 +62,7 @@ def setUp(self):
             adopted=self.now,
         )
 
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison={
@@ -107,25 +107,29 @@ def test_semver(self):
         self.assert_does_not_pass(self.dc, self.event_data)
 
         self.create_group_release(group=self.group, release=self.newest_release)
-        self.assert_passes(self.dc, WorkflowEventData(event=self.group_event))
+        self.assert_passes(
+            self.dc, WorkflowEventData(event=self.group_event, group=self.group_event.group)
+        )
 
         group_2, group_event_2 = self.create_new_group_event("group2")
         self.create_group_release(group=group_2, release=self.newest_release)
         self.create_group_release(group=group_2, release=self.oldest_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_2))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_2, group=group_2))
 
         group_3, group_event_3 = self.create_new_group_event("group3")
         self.create_group_release(group=group_3, release=self.middle_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3, group=group_3))
 
         # Check that the group cache invalidation works by adding an older release to the first group
         self.create_group_release(group=self.group, release=self.oldest_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=self.group_event))
+        self.assert_does_not_pass(
+            self.dc, WorkflowEventData(event=self.group_event, group=self.group_event.group)
+        )
 
         # Check that the project cache invalidation works by adding a newer release to the project
         group_4, group_event_4 = self.create_new_group_event("group4")
         self.create_group_release(group=group_4, release=self.newest_release)
-        self.assert_passes(self.dc, WorkflowEventData(event=group_event_4))
+        self.assert_passes(self.dc, WorkflowEventData(event=group_event_4, group=group_4))
 
         self.create_release(
             project=self.event.group.project,
@@ -134,20 +138,22 @@ def test_semver(self):
             environments=[self.prod_env],
             adopted=self.now - timedelta(days=2),
         )
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_4))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_4, group=group_4))
 
     def test_date(self):
         self.create_group_release(group=self.group, release=self.newest_release)
-        self.assert_passes(self.dc, WorkflowEventData(event=self.group_event))
+        self.assert_passes(
+            self.dc, WorkflowEventData(event=self.group_event, group=self.group_event.group)
+        )
 
         group_2, group_event_2 = self.create_new_group_event("group2")
         self.create_group_release(group=group_2, release=self.newest_release)
         self.create_group_release(group=group_2, release=self.oldest_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_2))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_2, group=group_2))
 
         group_3, group_event_3 = self.create_new_group_event("group3")
         self.create_group_release(group=group_3, release=self.middle_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3, group=group_3))
 
     def test_oldest_older(self):
         self.dc.update(
@@ -159,18 +165,20 @@ def test_oldest_older(self):
         )
 
         self.create_group_release(group=self.group, release=self.newest_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=self.group_event))
+        self.assert_does_not_pass(
+            self.dc, WorkflowEventData(event=self.group_event, group=self.group_event.group)
+        )
 
         group_2, group_event_2 = self.create_new_group_event("group2")
 
         self.create_group_release(group=group_2, release=self.newest_release)
         self.create_group_release(group=group_2, release=self.oldest_release)
-        self.assert_passes(self.dc, WorkflowEventData(event=group_event_2))
+        self.assert_passes(self.dc, WorkflowEventData(event=group_event_2, group=group_2))
 
         group_3, group_event_3 = self.create_new_group_event("group3")
 
         self.create_group_release(group=group_3, release=self.middle_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3, group=group_3))
 
     def test_newest_newer(self):
         self.dc.update(
@@ -182,18 +190,20 @@ def test_newest_newer(self):
         )
 
         self.create_group_release(group=self.group, release=self.newest_release)
-        self.assert_passes(self.dc, WorkflowEventData(event=self.group_event))
+        self.assert_passes(
+            self.dc, WorkflowEventData(event=self.group_event, group=self.group_event.group)
+        )
 
         group_2, group_event_2 = self.create_new_group_event("group2")
 
         self.create_group_release(group=group_2, release=self.newest_release)
         self.create_group_release(group=group_2, release=self.oldest_release)
-        self.assert_passes(self.dc, WorkflowEventData(event=group_event_2))
+        self.assert_passes(self.dc, WorkflowEventData(event=group_event_2, group=group_2))
 
         group_3, group_event_3 = self.create_new_group_event("group3")
 
         self.create_group_release(group=group_3, release=self.middle_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3, group=group_3))
 
     def test_newest_older(self):
         self.dc.update(
@@ -205,16 +215,18 @@ def test_newest_older(self):
         )
 
         self.create_group_release(group=self.event.group, release=self.newest_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=self.event))
+        self.assert_does_not_pass(
+            self.dc, WorkflowEventData(event=self.event, group=self.group_event.group)
+        )
 
         group_2, group_event_2 = self.create_new_group_event("group2")
         self.create_group_release(group=group_2, release=self.newest_release)
         self.create_group_release(group=group_2, release=self.oldest_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_2))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_2, group=group_2))
 
         group_3, group_event_3 = self.create_new_group_event("group3")
         self.create_group_release(group=group_3, release=self.middle_release)
-        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3))
+        self.assert_does_not_pass(self.dc, WorkflowEventData(event=group_event_3, group=group_3))
 
     def test_caching(self):
         cache_key = get_first_last_release_for_group_cache_key(
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_latest_release_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_latest_release_handler.py
index 88d935215ad4c4..b7661ef7a3066f 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_latest_release_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_latest_release_handler.py
@@ -23,7 +23,7 @@ class TestLatestReleaseCondition(ConditionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group_event.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison=True,
@@ -139,7 +139,9 @@ def test_latest_release_with_environment(self):
             date_added=datetime(2020, 9, 3, 3, 8, 24, 880386, tzinfo=UTC),
         )
 
-        self.event_data = WorkflowEventData(event=self.group_event, workflow_env=self.environment)
+        self.event_data = WorkflowEventData(
+            event=self.group_event, workflow_env=self.environment, group=self.group_event.group
+        )
 
         self.event.data["tags"] = (("release", new_release.version),)
         self.assert_passes(self.dc, self.event_data)
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_level_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_level_handler.py
index 3ba58a514c148e..c5f797db486327 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_level_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_level_handler.py
@@ -19,7 +19,7 @@ class TestLevelCondition(ConditionTestCase):
 
     def setup_group_event_and_job(self):
         self.group_event = self.event.for_group(self.group)
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def setUp(self):
         super().setUp()
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_new_high_priority_issue_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_new_high_priority_issue_handler.py
index 0a9aa37f300fd5..8d24b477c14aad 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_new_high_priority_issue_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_new_high_priority_issue_handler.py
@@ -17,6 +17,7 @@ def setUp(self):
         super().setUp()
         self.event_data = WorkflowEventData(
             event=self.group_event,
+            group=self.group_event.group,
             group_state=GroupState(
                 {
                     "id": 1,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_reappeared_event_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_reappeared_event_handler.py
index a826546a6acb9c..d6418cfbb40d31 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_reappeared_event_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_reappeared_event_handler.py
@@ -41,7 +41,9 @@ def test_json_schema(self):
             dc.save()
 
     def test(self):
-        job = WorkflowEventData(event=self.group_event, has_reappeared=True)
+        job = WorkflowEventData(
+            event=self.group_event, group=self.group_event.group, has_reappeared=True
+        )
         dc = self.create_data_condition(
             type=self.condition,
             comparison=True,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_regression_event_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_regression_event_handler.py
index 895f4a915f1039..412135981bf999 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_regression_event_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_regression_event_handler.py
@@ -42,6 +42,7 @@ def test_json_schema(self):
     def test(self):
         job = WorkflowEventData(
             event=self.group_event,
+            group=self.group_event.group,
             group_state=GroupState(
                 {
                     "id": 1,
diff --git a/tests/sentry/workflow_engine/handlers/condition/test_tagged_event_handler.py b/tests/sentry/workflow_engine/handlers/condition/test_tagged_event_handler.py
index a55cc549704c55..4fe0d45ef9c394 100644
--- a/tests/sentry/workflow_engine/handlers/condition/test_tagged_event_handler.py
+++ b/tests/sentry/workflow_engine/handlers/condition/test_tagged_event_handler.py
@@ -33,7 +33,7 @@ def setUp(self):
         self.event = self.get_event()
         self.group = self.create_group(project=self.project)
         self.group_event = self.event.for_group(self.group)
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
         self.dc = self.create_data_condition(
             type=self.condition,
             comparison={"match": MatchType.EQUAL, "key": "LOGGER", "value": "sentry.example"},
diff --git a/tests/sentry/workflow_engine/models/test_action.py b/tests/sentry/workflow_engine/models/test_action.py
index c590e50ac9abd6..dd33038c36e6cc 100644
--- a/tests/sentry/workflow_engine/models/test_action.py
+++ b/tests/sentry/workflow_engine/models/test_action.py
@@ -12,7 +12,8 @@
 
 class TestAction(TestCase):
     def setUp(self):
-        self.mock_event = WorkflowEventData(event=Mock(spec=GroupEvent))
+        mock_group = Mock(spec=GroupEvent)
+        self.mock_event = WorkflowEventData(event=mock_group, group=mock_group.group)
         self.mock_detector = Mock(name="detector")
         self.action = Action(type=Action.Type.SLACK)
         self.config_schema = {
diff --git a/tests/sentry/workflow_engine/models/test_workflow.py b/tests/sentry/workflow_engine/models/test_workflow.py
index ed3588af120d79..e2a5ec20e27cb3 100644
--- a/tests/sentry/workflow_engine/models/test_workflow.py
+++ b/tests/sentry/workflow_engine/models/test_workflow.py
@@ -15,7 +15,7 @@ def setUp(self):
         )
         self.data_condition = self.data_condition_group.conditions.first()
         self.group, self.event, self.group_event = self.create_group_event()
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def test_queryset(self):
         """
diff --git a/tests/sentry/workflow_engine/processors/test_action.py b/tests/sentry/workflow_engine/processors/test_action.py
index 6c8ba2c4941ec0..da0624b484b023 100644
--- a/tests/sentry/workflow_engine/processors/test_action.py
+++ b/tests/sentry/workflow_engine/processors/test_action.py
@@ -39,7 +39,7 @@ def setUp(self):
         self.group, self.event, self.group_event = self.create_group_event(
             occurrence=self.build_occurrence(evidence_data={"detector_id": self.detector.id})
         )
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def test(self):
         status_1 = WorkflowActionGroupStatus.objects.create(
@@ -240,7 +240,7 @@ def setUp(self):
         self.group, self.event, self.group_event = self.create_group_event(
             occurrence=self.build_occurrence(evidence_data={"detector_id": self.detector.id})
         )
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def test_create_workflow_fire_histories(self):
         create_workflow_fire_histories(Action.objects.filter(id=self.action.id), self.event_data)
diff --git a/tests/sentry/workflow_engine/processors/test_delayed_workflow.py b/tests/sentry/workflow_engine/processors/test_delayed_workflow.py
index 413f056d68e8ac..aef5d625ef81cb 100644
--- a/tests/sentry/workflow_engine/processors/test_delayed_workflow.py
+++ b/tests/sentry/workflow_engine/processors/test_delayed_workflow.py
@@ -771,11 +771,11 @@ def test_fire_actions_for_groups__fire_actions(self, mock_trigger):
 
         assert mock_trigger.call_count == 2
         assert mock_trigger.call_args_list[0][0] == (
-            WorkflowEventData(event=self.event1.for_group(self.group1)),
+            WorkflowEventData(event=self.event1.for_group(self.group1), group=self.group1),
             self.detector,
         )
         assert mock_trigger.call_args_list[1][0] == (
-            WorkflowEventData(event=self.event2.for_group(self.group2)),
+            WorkflowEventData(event=self.event2.for_group(self.group2), group=self.group2),
             self.detector,
         )
 
diff --git a/tests/sentry/workflow_engine/processors/test_workflow.py b/tests/sentry/workflow_engine/processors/test_workflow.py
index ed37635841f2c0..262249a8ea285e 100644
--- a/tests/sentry/workflow_engine/processors/test_workflow.py
+++ b/tests/sentry/workflow_engine/processors/test_workflow.py
@@ -59,6 +59,7 @@ def setUp(self):
         self.group, self.event, self.group_event = self.create_group_event()
         self.event_data = WorkflowEventData(
             event=self.group_event,
+            group=self.group,
             group_state=GroupState(
                 id=1, is_new=False, is_regression=True, is_new_group_environment=False
             ),
@@ -165,6 +166,7 @@ def test_same_environment_only(self):
         self.group, self.event, self.group_event = self.create_group_event(environment=env.name)
         self.event_data = WorkflowEventData(
             event=self.group_event,
+            group=self.group,
             group_state=GroupState(
                 id=1, is_new=False, is_regression=True, is_new_group_environment=False
             ),
@@ -341,7 +343,7 @@ def setUp(self):
         self.group, self.event, self.group_event = self.create_group_event(
             occurrence=occurrence,
         )
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     def test_workflow_trigger(self):
         triggered_workflows = evaluate_workflow_triggers({self.workflow}, self.event_data)
@@ -408,7 +410,7 @@ def test_delays_slow_conditions(self):
         )
 
         triggered_workflows = evaluate_workflow_triggers({self.workflow}, self.event_data)
-        # no workflows are triggered because the slow conditions need to be evaluted
+        # no workflows are triggered because the slow conditions need to be evaluated
         assert triggered_workflows == set()
 
 
@@ -428,7 +430,7 @@ def setUp(self):
         self.group, self.event, self.group_event = self.create_group_event(
             occurrence=occurrence,
         )
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
         self.create_workflow_action(self.workflow)
         self.mock_redis_buffer = mock_redis_buffer()
         self.mock_redis_buffer.__enter__()
@@ -556,7 +558,7 @@ def setUp(self):
         self.group, self.event, self.group_event = self.create_group_event(
             occurrence=self.build_occurrence(evidence_data={"detector_id": self.detector.id})
         )
-        self.event_data = WorkflowEventData(event=self.group_event)
+        self.event_data = WorkflowEventData(event=self.group_event, group=self.group)
 
     @with_feature("organizations:workflow-engine-process-workflows")
     @with_feature("organizations:workflow-engine-metric-alert-dual-processing-logs")
@@ -636,6 +638,10 @@ def setUp(self):
         self.data_condition_group = self.create_data_condition_group()
         self.condition = self.create_data_condition(condition_group=self.data_condition_group)
         _, self.event, self.group_event = self.create_group_event()
+        self.workflow_event_data = WorkflowEventData(
+            event=self.group_event,
+            group=self.group_event.group,
+        )
 
     @patch("sentry.buffer.backend.push_to_sorted_set")
     @patch("sentry.buffer.backend.push_to_hash_bulk")