feat(workflows): show waiting for domain and count it as enabled

frontend/src/scenes/cohorts/CohortBlockingWorkflowsModal.tsx

@@ -23,7 +23,7 @@ export function CohortBlockingWorkflowsModal(): JSX.Element {
         <LemonModal
             isOpen={!!blocking}
             onClose={closeBlockingWorkflowsModal}
-            title="This cohort is used by active workflows"
+            title="This cohort is used by enabled workflows"
             width={480}
             footer={
                 <LemonButton type="secondary" onClick={closeBlockingWorkflowsModal}>

frontend/src/scenes/cohorts/cohortBlockingWorkflows.ts

@@ -9,7 +9,7 @@ import type { CohortUsedInWorkflowsBlockApi } from 'products/cohorts/frontend/ge
 export const COHORT_USED_BY_WORKFLOWS_ERROR_CODE = 'cohort_used_by_workflows'
 
 /**
- * The active workflows that block deleting this cohort or editing it so it stops updating in real
+ * The enabled workflows that block deleting this cohort or editing it so it stops updating in real
  * time, or null when nothing blocks it. A failed check returns null, because the API still refuses
  * the change and the caller then shows the API error.
  */

posthog/api/cohort.py

@@ -141,15 +141,15 @@
 from products.surveys.backend.models import Survey
 from products.workflows.backend.facade.api import (
     filter_hog_flow_references_by_access_level,
-    get_active_workflows_using_realtime_cohort,
-    team_has_active_workflows_using_cohorts,
+    get_enabled_workflows_using_realtime_cohort,
+    team_has_enabled_workflows_using_cohorts,
 )
 from products.workflows.backend.facade.contracts import CohortDependentWorkflow, HogFlowReference
 
 
 def _workflows_relying_on_cohort(cohort: Cohort) -> list[CohortDependentWorkflow]:
     # Most teams have no active cohort workflows, so skip loading every cohort in the project for them.
-    if not team_has_active_workflows_using_cohorts(team_id=cohort.team_id):
+    if not team_has_enabled_workflows_using_cohorts(team_id=cohort.team_id):
         return []
     # A cohort that includes this one as a criterion reads its membership, so a workflow on that cohort
     # also relies on this one. Load the project's cohort references in one query and walk them in memory.
@@ -165,7 +165,7 @@ def _workflows_relying_on_cohort(cohort: Cohort) -> list[CohortDependentWorkflow
         for dependent_id in dependents[pending.pop()] - cohort_ids:
             cohort_ids.add(dependent_id)
             pending.append(dependent_id)
-    return get_active_workflows_using_realtime_cohort(team_id=cohort.team_id, cohort_ids=cohort_ids)
+    return get_enabled_workflows_using_realtime_cohort(team_id=cohort.team_id, cohort_ids=cohort_ids)
 
 
 def _token_can_read(request: Request, scope_object: str) -> bool:
@@ -688,7 +688,7 @@ class CohortUsedInWorkflowSerializer(serializers.Serializer):
     id = serializers.CharField(help_text="Workflow ID (UUID)")
     name = serializers.CharField(allow_blank=True, help_text="Workflow display name")
     status = serializers.CharField(
-        help_text="Workflow status. Always 'active', because only active workflows rely on the cohort"
+        help_text="Workflow status: 'active', or 'waiting_for_domain' for an enabled workflow that goes live once its email domain is verified"
     )
 
 
@@ -747,11 +747,11 @@ class CohortUsedInWorkflowsBlockSerializer(serializers.Serializer):
     results = CohortUsedInWorkflowSerializer(
         many=True,
         help_text=(
-            f"Active workflows relying on this cohort that the caller can view, capped at {COHORT_USED_IN_PAGE_SIZE} results"
+            f"Enabled workflows relying on this cohort that the caller can view, capped at {COHORT_USED_IN_PAGE_SIZE} results"
         ),
     )
     total = serializers.IntegerField(
-        help_text="Total number of active workflows relying on this cohort, including ones the caller cannot view"
+        help_text="Total number of enabled workflows relying on this cohort, including ones the caller cannot view"
     )
     hidden = serializers.IntegerField(
         help_text="Number of relying workflows the caller cannot view. They count toward total but are never listed"
@@ -1923,12 +1923,12 @@ def update(self, cohort: Cohort, validated_data: dict, *args: Any, **kwargs: Any
                 if stops_realtime:
                     raise ValidationError(
                         f"This change stops the cohort updating in real time, and {count} "
-                        f"active workflow(s) rely on that{names}. "
+                        f"enabled workflow(s) rely on that{names}. "
                         "Please remove the cohort from these workflows before making this change.",
                         code=COHORT_USED_BY_WORKFLOWS_ERROR_CODE,
                     )
                 raise ValidationError(
-                    f"This cohort is used in {count} active workflow(s){names}. Changing its criteria would send "
+                    f"This cohort is used in {count} enabled workflow(s){names}. Changing its criteria would send "
                     "its current members through them again. "
                     "Please remove the cohort from these workflows before changing its criteria.",
                     code=COHORT_USED_BY_WORKFLOWS_ERROR_CODE,
@@ -1977,7 +1977,7 @@ def update(self, cohort: Cohort, validated_data: dict, *args: Any, **kwargs: Any
                     self._reject_delete(
                         cohort,
                         "workflows",
-                        f"This cohort is used in {len(workflows_with_cohort)} active workflow(s){names}. "
+                        f"This cohort is used in {len(workflows_with_cohort)} enabled workflow(s){names}. "
                         "Please remove the cohort from these workflows before deleting it.",
                         code=COHORT_USED_BY_WORKFLOWS_ERROR_CODE,
                     )

posthog/api/test/test_cohort.py

@@ -7541,18 +7541,19 @@ def _create_workflow_using_cohort(self, cohort_id: int, usage: str, status: str)
             ("active_trigger", "trigger", "active", 400),
             ("active_branch", "branch", "active", 400),
             ("active_branch_precalculated", "branch_precalculated", "active", 400),
+            ("waiting_trigger", "trigger", "waiting_for_domain", 400),
             ("draft_trigger", "trigger", "draft", 200),
         ]
     )
-    def test_delete_cohort_is_blocked_by_active_workflows_using_it(self, _name, usage, workflow_status, expected):
+    def test_delete_cohort_is_blocked_by_enabled_workflows_using_it(self, _name, usage, workflow_status, expected):
         cohort_id = self._create_realtime_cohort()
         self._create_workflow_using_cohort(cohort_id, usage, workflow_status)
 
         response = self.client.patch(f"/api/projects/{self.team.id}/cohorts/{cohort_id}", {"deleted": True})
 
         assert response.status_code == expected, response.json()
         if expected == 400:
-            assert f"used in 1 active workflow(s): Workflow {usage}" in response.json()["detail"]
+            assert f"used in 1 enabled workflow(s): Workflow {usage}" in response.json()["detail"]
             assert response.json()["code"] == "cohort_used_by_workflows"
 
     @parameterized.expand(
@@ -7572,7 +7573,7 @@ def test_blocked_delete_names_workflows_only_for_workflow_readers(self, _name, s
         )
 
         assert response.status_code == 400, response.json()
-        assert "used in 1 active workflow(s)" in response.json()["detail"]
+        assert "used in 1 enabled workflow(s)" in response.json()["detail"]
         assert ("Workflow trigger" in response.json()["detail"]) == names_shown
 
     def test_blocked_delete_hides_names_of_workflows_the_member_cannot_view(self):
@@ -7594,7 +7595,7 @@ def test_blocked_delete_hides_names_of_workflows_the_member_cannot_view(self):
         response = self.client.patch(f"/api/projects/{self.team.id}/cohorts/{cohort_id}", {"deleted": True})
 
         assert response.status_code == 400, response.json()
-        assert "used in 2 active workflow(s): Workflow branch and 1 more." in response.json()["detail"]
+        assert "used in 2 enabled workflow(s): Workflow branch and 1 more." in response.json()["detail"]
         assert "Workflow trigger" not in response.json()["detail"]
 
     @parameterized.expand([("used_in", "workflows"), ("blocking_workflows", None)])
@@ -7728,7 +7729,7 @@ def test_edit_that_stops_realtime_is_blocked_by_workflows_on_a_cohort_that_inclu
         )
 
         assert response.status_code == 400, response.json()
-        assert "1 active workflow(s) rely on that: Workflow trigger." in response.json()["detail"]
+        assert "1 enabled workflow(s) rely on that: Workflow trigger." in response.json()["detail"]
 
     def test_cohort_type_not_set_when_not_provided(self):
         """cohort_type should remain None when not provided"""

posthog/api/test/test_search.py

@@ -5,6 +5,8 @@
 from django.db import connection
 from django.test.utils import CaptureQueriesContext
 
+from parameterized import parameterized
+
 from posthog.api.search import ENTITY_MAP, SEARCHABLE_ENTITIES, class_queryset, search_entities
 from posthog.helpers.full_text_search import build_search_vector, process_query
 from posthog.models import OrganizationMembership, Team, User
@@ -247,9 +249,10 @@ def test_early_access_features(self):
         self.assertEqual(results[0]["type"], "early_access_feature")
         self.assertEqual(results[0]["extra_fields"]["name"], "second feature")
 
-    def test_hog_flows(self):
+    @parameterized.expand([("draft",), ("active",), ("waiting_for_domain",)])
+    def test_hog_flows(self, status):
         create_workflow_for_test(team_id=self.team.id, name="first workflow")
-        create_workflow_for_test(team_id=self.team.id, name="second workflow")
+        create_workflow_for_test(team_id=self.team.id, name="second workflow", status=status)
         create_workflow_for_test(team_id=self.team.id, name="third workflow")
 
         response = self.client.get("/api/projects/@current/search?q=sec&entities=hog_flow")

products/workflows/backend/facade/api.py

@@ -28,6 +28,7 @@
     WorkflowTaskDailyLimits,
     WorkflowVisionScanDailyLimits,
 )
+from products.workflows.backend.facade.enums import ENABLED_HOG_FLOW_STATES
 from products.workflows.backend.models import HogFlow, TeamWorkflowsConfig
 from products.workflows.backend.models.hog_flow.hog_flow import opt_out_category_ids, static_cohort_ids
 from products.workflows.backend.services import email_utm_defaults
@@ -98,7 +99,7 @@ def search_workflows(
     ``with_access_levels`` adds ``user_access_level``, the user's resolved level for each workflow.
     ``include_count=False`` skips the count query and reports a total of 0.
     """
-    statuses = [HogFlow.State.DRAFT, HogFlow.State.ACTIVE]
+    statuses = [HogFlow.State.DRAFT, *ENABLED_HOG_FLOW_STATES]
     if include_archived:
         statuses.append(HogFlow.State.ARCHIVED)
     queryset = access_control.filter_queryset_by_access_level(
@@ -291,28 +292,28 @@ def enable_waiting_workflows(*, team_id: int, sender_ids: Iterable[int], workers
         )
 
 
-def get_active_workflows_using_realtime_cohort(
+def get_enabled_workflows_using_realtime_cohort(
     *, team_id: int, cohort_ids: Collection[int]
 ) -> list[CohortDependentWorkflow]:
     # Deferred to keep the cdp filter compiler, and the HogQL schema it loads, off the django.setup() path.
     from products.workflows.backend.services.cohort_usage import (  # noqa: PLC0415
-        get_active_hog_flows_using_realtime_cohort,
+        get_enabled_hog_flows_using_realtime_cohort,
     )
 
     return [
         CohortDependentWorkflow(id=str(flow.id), name=flow.name or "", status=flow.status)
-        for flow in get_active_hog_flows_using_realtime_cohort(team_id=team_id, cohort_ids=cohort_ids)
+        for flow in get_enabled_hog_flows_using_realtime_cohort(team_id=team_id, cohort_ids=cohort_ids)
     ]
 
 
-def team_has_active_workflows_using_cohorts(*, team_id: int) -> bool:
-    """Cheap pre-check: False means no active workflow in the team can rely on any cohort."""
+def team_has_enabled_workflows_using_cohorts(*, team_id: int) -> bool:
+    """Cheap pre-check: False means no enabled workflow in the team can rely on any cohort."""
     # Deferred for the same reason as get_active_workflows_using_realtime_cohort.
     from products.workflows.backend.services.cohort_usage import (  # noqa: PLC0415
-        team_has_active_hog_flows_mentioning_cohorts,
+        team_has_enabled_hog_flows_mentioning_cohorts,
     )
 
-    return team_has_active_hog_flows_mentioning_cohorts(team_id)
+    return team_has_enabled_hog_flows_mentioning_cohorts(team_id)
 
 
 def recompute_email_sending_tier(team_id: int) -> TierDecision | None:

products/workflows/backend/presentation/views/hog_flow.py

@@ -229,6 +229,7 @@
     verified_email_domains,
 )
 from products.workflows.backend.facade.enums import (
+    ENABLED_HOG_FLOW_STATES,
     HogFlowBatchJobState,
     HogFlowExitCondition,
     HogFlowOriginProduct,
@@ -5679,23 +5680,29 @@ def perform_update(self, serializer):
         )
         self._emit_resource_edited(serializer.instance)
 
-        self._report_went_live(result, serializer.instance)
+        self._report_turned_on(result, serializer.instance)
 
-    def _report_went_live(self, result: WorkflowWriteResult, instance: Workflow) -> None:
-        if result.previous.get("status") == HogFlowState.ACTIVE or instance.status != HogFlowState.ACTIVE:
-            return
-        # A workflow that kept suggestions on while it was off needs its scout back when it goes live again.
-        if is_optimization_enabled(instance.id):
+    def _report_turned_on(self, result: WorkflowWriteResult, instance: Workflow) -> None:
+        previous = result.previous.get("status")
+        # A workflow that kept suggestions on while it was off needs its scout back when it is enabled again.
+        # Going live from waiting checks again, because the enable that started the wait may not have granted it.
+        if (
+            previous != HogFlowState.ACTIVE
+            and previous != instance.status
+            and instance.status in ENABLED_HOG_FLOW_STATES
+            and is_optimization_enabled(instance.id)
+        ):
             self._sync_suggestions_scout(self.request)
-        self._report_workflow_action(
-            "hog_flow_activated",
-            instance,
-            {
-                "activated_by": "user",
-                "edges_count": len(instance.edges or []),
-                "actions_count": len(instance.actions or []),
-            },
-        )
+        if previous != HogFlowState.ACTIVE and instance.status == HogFlowState.ACTIVE:
+            self._report_workflow_action(
+                "hog_flow_activated",
+                instance,
+                {
+                    "activated_by": "user",
+                    "edges_count": len(instance.edges or []),
+                    "actions_count": len(instance.actions or []),
+                },
+            )
 
     def update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         partial = kwargs.pop("partial", False)
@@ -5816,7 +5823,7 @@ def edit(state: WorkflowEditState, base_actions: list[dict], base_edges: list[di
             instance,
             {"operations_count": len(operations), "routed_to_draft": result.routed_to_draft},
         )
-        self._report_went_live(result, instance)
+        self._report_turned_on(result, instance)
 
         return Response(self.get_serializer(instance).data)
 
@@ -5887,7 +5894,7 @@ def edit(state: WorkflowEditState, base_actions: list[dict], base_edges: list[di
                 "routed_to_draft": result.routed_to_draft,
             },
         )
-        self._report_went_live(result, instance)
+        self._report_turned_on(result, instance)
 
         return Response(self.get_serializer(instance).data)
 
@@ -6035,7 +6042,7 @@ def validate(state: WorkflowEditState, content: dict) -> dict:
         self._log_activity(locked.id, locked.name, "published", previous=result.previous, current=result.current)
         self._emit_resource_edited(locked)
         self._report_workflow_action("hog_flow_draft_published", locked)
-        self._report_went_live(result, locked)
+        self._report_turned_on(result, locked)
 
         return Response(
             {

products/workflows/backend/services/cohort_usage.py

@@ -5,6 +5,7 @@
 from django.db.models.functions import Cast
 
 from products.cdp.backend.presentation.hog_functions import filter_cohort_ids
+from products.workflows.backend.facade.enums import ENABLED_HOG_FLOW_STATES
 from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
 
 
@@ -23,25 +24,25 @@ def _action_uses_realtime_cohort(action: Any, cohort_ids: set[int]) -> bool:
     return False
 
 
-def _active_hog_flows_mentioning_cohorts(team_id: int) -> QuerySet[HogFlow]:
+def _enabled_hog_flows_mentioning_cohorts(team_id: int) -> QuerySet[HogFlow]:
     # Postgres skips workflows whose actions never mention a cohort, so only those graphs reach Python.
     return (
-        HogFlow.objects.filter(team_id=team_id, status=HogFlow.State.ACTIVE)
+        HogFlow.objects.filter(team_id=team_id, status__in=ENABLED_HOG_FLOW_STATES)
         .alias(actions_text=Cast("actions", output_field=TextField()))
         .filter(actions_text__contains="cohort")
     )
 
 
-def team_has_active_hog_flows_mentioning_cohorts(team_id: int) -> bool:
-    return _active_hog_flows_mentioning_cohorts(team_id).exists()
+def team_has_enabled_hog_flows_mentioning_cohorts(team_id: int) -> bool:
+    return _enabled_hog_flows_mentioning_cohorts(team_id).exists()
 
 
-def get_active_hog_flows_using_realtime_cohort(team_id: int, cohort_ids: Collection[int]) -> Iterator[HogFlow]:
-    """Active workflows in the environment that read the realtime membership of any of these cohorts, as a
+def get_enabled_hog_flows_using_realtime_cohort(team_id: int, cohort_ids: Collection[int]) -> Iterator[HogFlow]:
+    """Enabled workflows in the environment that read the realtime membership of any of these cohorts, as a
     cohort trigger or in a conditional branch."""
     # Only the fields the check reads, streamed: full rows carry encrypted inputs and the whole graph.
     ids = set(cohort_ids)
-    flows = _active_hog_flows_mentioning_cohorts(team_id).only("id", "name", "status", "actions")
+    flows = _enabled_hog_flows_mentioning_cohorts(team_id).only("id", "name", "status", "actions")
     for flow in flows.iterator():
         if any(_action_uses_realtime_cohort(action, ids) for action in (flow.actions or [])):
             yield flow