Start with .to_logger and .to_artifact

src/sentry/workflow_engine/processors/evaluations/base.py

@@ -1,9 +1,17 @@
+from abc import abstractmethod
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass, field, replace
-from typing import Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from sentry.workflow_engine.types import ConditionError
 
+if TYPE_CHECKING:
+    from logging import Logger
+
+# The shared, static prefix for every evaluation log. Combined with each evaluation's
+# `log_name` to build a stable, searchable event name (e.g. `workflow_engine.evaluation.condition`).
+_LOG_PREFIX = "workflow_engine.evaluation"
+
 
 def _find_error(
     items: list["BaseWorkflowEngineEvaluation[Any, Any]"],
@@ -38,13 +46,40 @@ class BaseWorkflowEngineEvaluation[R, D]:
 
     triggered: bool = field(compare=False)
 
+    # The static suffix for this evaluation type, combined with `_LOG_PREFIX` to build the
+    # log event name in `to_log`. Set by each concrete subclass.
+    log_name: ClassVar[str]
+
+    @abstractmethod
+    def to_artifact(self) -> dict[str, Any]:
+        """
+        Serialize the result, data, error, and triggered state into a flat, log-safe dict
+        that can be emitted in a log or stored as an artifact.
+
+        Implementations MUST NOT include the raw evaluated value (it may be large or contain
+        PII) - use metadata (ids, types) and the derived result instead.
+        """
+        ...
+
+    def to_log(self, logger: "Logger") -> None:
+        """
+        Emit this evaluation as a structured log line under a stable, searchable event name
+        (`{_LOG_PREFIX}.{log_name}`), passing `to_artifact()` as the log `extra`.
+        """
+        event = f"{_LOG_PREFIX}.{self.log_name}"
+        logger.debug(event, extra=self.to_artifact())
+
     def is_tainted(self) -> bool:
         """
         Returns True if this result is less trustworthy due to an error during
         evaluation.
         """
         return self.error is not None
 
+    def error_message(self) -> str | None:
+        """The error's message for serialization in `to_artifact`, or None when untainted."""
+        return self.error.msg if self.error else None
+
     def with_error(self, error: ConditionError) -> "BaseWorkflowEngineEvaluation[R, D]":
         """
         Returns a copy of this evaluation with the given error. If the evaluation is

src/sentry/workflow_engine/processors/evaluations/condition.py

@@ -40,5 +40,17 @@ class DataConditionEvaluation(
     - triggered: bool - If the evaluation should consider this condition "triggered" or not.
     """
 
+    log_name = "condition"
+
     result: DataConditionResult = None
     condition: DataCondition
+
+    def to_artifact(self) -> dict[str, Any]:
+        return {
+            "condition_id": self.condition.id,
+            "type": self.condition.type,
+            # DataConditionResult may be a DetectorPriorityLevel enum; unwrap to its value.
+            "result": getattr(self.result, "value", self.result),
+            "triggered": self.triggered,
+            "error": self.error_message(),
+        }

src/sentry/workflow_engine/processors/evaluations/condition_group.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, TypedDict
+from typing import TYPE_CHECKING, Any, TypedDict
 
 from .base import BaseWorkflowEngineEvaluation
 from .condition import DataConditionEvaluation
@@ -31,4 +31,17 @@ class DataConditionGroupEvaluation(BaseWorkflowEngineEvaluation[bool, GroupEvalu
     - triggered: bool - whether the group's conditions passed
     """
 
-    pass
+    log_name = "condition_group"
+
+    def to_artifact(self) -> dict[str, Any]:
+        logic_type = self.data["logic_type"]
+        return {
+            # logic_type may be a DataConditionGroup.Type enum or a raw string; unwrap to its value.
+            "logic_type": getattr(logic_type, "value", logic_type),
+            "result": self.result,
+            "triggered": self.triggered,
+            "error": self.error_message(),
+            "condition_evaluations": [
+                condition.to_artifact() for condition in self.data["condition_evaluations"]
+            ],
+        }

src/sentry/workflow_engine/processors/evaluations/detector.py

@@ -37,5 +37,19 @@ class DetectorEvaluation(
     - triggered: bool - If there is an event that should trigger the next phase in the system.
     """
 
+    log_name = "detector"
+
     result: DetectorResult = None
     priority: DetectorPriorityLevel
+
+    def to_artifact(self) -> dict[str, Any]:
+        return {
+            "group_key": self.data["group_key"],
+            # priority is a DetectorPriorityLevel enum; unwrap to its value.
+            "priority": getattr(self.priority, "value", self.priority),
+            # result is an IssueOccurrence / StatusChangeMessage / None; log its type, not the payload.
+            "result": type(self.result).__name__ if self.result is not None else None,
+            "triggered": self.triggered,
+            "error": self.error_message(),
+            "trigger_group_evaluation": self.data["trigger_group_evaluation"].to_artifact(),
+        }

src/sentry/workflow_engine/processors/evaluations/workflow.py

@@ -3,7 +3,7 @@
 import random
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, TypedDict
+from typing import TYPE_CHECKING, Any, TypedDict
 
 from sentry_sdk import logger as sentry_logger
 
@@ -63,7 +63,27 @@ class WorkflowEvaluation(
     - `triggered`: bool - Whether the workflow's trigger (WHEN) conditions passed.
     """
 
-    pass
+    log_name = "workflow"
+
+    def to_artifact(self) -> dict[str, Any]:
+        result = self.result
+        # result is either the "deferred" sentinel (a str) or the sequence of actions that fired.
+        if isinstance(result, str):
+            result_kind = "deferred"
+            triggered_action_ids = None
+        else:
+            result_kind = "actions"
+            triggered_action_ids = [action.id for action in result]
+        return {
+            "triggered": self.triggered,
+            "error": self.error_message(),
+            "result": result_kind,
+            "triggered_action_ids": triggered_action_ids,
+            "trigger_group_eval": self.data["trigger_group_eval"].to_artifact(),
+            "filter_group_evals": [
+                filter_eval.to_artifact() for filter_eval in self.data["filter_group_evals"]
+            ],
+        }
 
 
 class WorkflowEvaluationSnapshot(TypedDict):
@@ -101,7 +121,7 @@ class GroupedWorkflowEvaluationResult:
     result: dict[WorkflowId, WorkflowEvaluation]
     tainted: bool
 
-    # Batch-level context used by log_to / get_snapshot / consumers.
+    # Batch-level context used by to_log / to_artifact / get_snapshot / consumers.
     organization: Organization
     event: GroupEvent | Activity
     group: Group | None = None
@@ -160,7 +180,42 @@ def get_snapshot(self) -> WorkflowEvaluationSnapshot:
             "triggered_actions": triggered_actions,
         }
 
-    def log_to(self, logger: Logger) -> bool:
+    def to_artifact(self) -> dict[str, Any]:
+        """
+        Flatten the batch-level context into a structured, log-safe dict (ids, counts, and
+        debug info) and embed each workflow's own `to_artifact()` under `workflow_evaluations`
+        so a log search can answer *why* a given workflow did or didn't trigger.
+
+        This is the `extra` payload emitted by `to_log`. Empty per-workflow evaluations for the
+        sentinel early-return paths, where `result` is `{}`.
+        """
+        data_snapshot = self.get_snapshot()
+        detection_type = (
+            data_snapshot["associated_detector"]["type"]
+            if data_snapshot["associated_detector"]
+            else None
+        )
+        group_id = data_snapshot["group"].id if data_snapshot["group"] else None
+        triggered_workflows = data_snapshot["triggered_workflows"] or []
+        action_filter_conditions = data_snapshot["action_filter_conditions"] or []
+        triggered_actions = data_snapshot["triggered_actions"] or []
+        return {
+            "event_id": data_snapshot["event_id"],
+            "group_id": group_id,
+            "detection_type": detection_type,
+            "workflow_ids": data_snapshot["workflow_ids"],
+            "triggered_workflow_ids": [w["id"] for w in triggered_workflows],
+            "delayed_conditions": data_snapshot["delayed_conditions"],
+            "action_filter_group_ids": [afg["id"] for afg in action_filter_conditions],
+            "triggered_action_ids": [a["id"] for a in triggered_actions],
+            "debug_msg": self.msg,
+            "workflow_evaluations": {
+                str(workflow_id): evaluation.to_artifact()
+                for workflow_id, evaluation in self.result.items()
+            },
+        }
+
+    def to_log(self, logger: Logger) -> bool:
         """
         Logs workflow evaluation data.
         Logging may be skipped if the organization isn't opted in and logs are being
@@ -189,27 +244,7 @@ def log_to(self, logger: Logger) -> bool:
         else:
             log_str = f"{log_str}.actions.triggered"
 
-        data_snapshot = self.get_snapshot()
-        detection_type = (
-            data_snapshot["associated_detector"]["type"]
-            if data_snapshot["associated_detector"]
-            else None
-        )
-        group_id = data_snapshot["group"].id if data_snapshot["group"] else None
-        triggered_workflows = data_snapshot["triggered_workflows"] or []
-        action_filter_conditions = data_snapshot["action_filter_conditions"] or []
-        triggered_actions = data_snapshot["triggered_actions"] or []
-        extra = {
-            "event_id": data_snapshot["event_id"],
-            "group_id": group_id,
-            "detection_type": detection_type,
-            "workflow_ids": data_snapshot["workflow_ids"],
-            "triggered_workflow_ids": [w["id"] for w in triggered_workflows],
-            "delayed_conditions": data_snapshot["delayed_conditions"],
-            "action_filter_group_ids": [afg["id"] for afg in action_filter_conditions],
-            "triggered_action_ids": [a["id"] for a in triggered_actions],
-            "debug_msg": self.msg,
-        }
+        extra = self.to_artifact()
 
         if direct_to_sentry:
             sentry_logger.info(log_str, attributes=extra)

src/sentry/workflow_engine/tasks/workflows.py

@@ -75,7 +75,7 @@ def process_workflow_activity(activity_id: int, group_id: int, detector_id: Dete
             batch_client, event_data, event_start_time=activity.datetime, detector=detector
         )
 
-    evaluation.log_to(logger)
+    evaluation.to_log(logger)
 
     metrics.incr(
         "workflow_engine.tasks.process_workflows.activity_update.executed",
@@ -172,7 +172,7 @@ def _process_workflows_event(
                     and len(evaluation.triggered_actions) > 0,
                 )
 
-    evaluation.log_to(logger)
+    evaluation.to_log(logger)
     duration = time.time() - start_time
     is_slow = duration > 1.0
     # We want full coverage for particularly slow cases, plus a random sampling.