chore(workflows): route the small workflows views through the facade

products/workflows/backend/facade/api.py

@@ -17,8 +17,10 @@
     TwilioPhoneNumber,
     WorkflowActivitySummary,
     WorkflowSummary,
+    WorkflowTaskDailyLimits,
 )
-from products.workflows.backend.models import HogFlow
+from products.workflows.backend.models import HogFlow, TeamWorkflowsConfig
+from products.workflows.backend.services.batch_jobs import create_batch_job
 from products.workflows.backend.services.email_sending_controls import (
     ensure_workflows_config,
     get_email_sending_state,
@@ -41,6 +43,7 @@
 __all__ = [
     "MIN_EMAIL_SENDING_TIER",
     "compute_next_occurrences",
+    "create_batch_job",
     "ensure_workflows_config",
     "filter_hog_flow_references_by_access_level",
     "get_email_sending_state",
@@ -134,6 +137,24 @@ def get_workflow_owner_id(*, team_id: int, workflow_id: UUID) -> int | None:
         raise WorkflowNotFound() from None
 
 
+def workflow_exists(*, team_id: int, workflow_id: UUID) -> bool:
+    return HogFlow.objects.filter(team_id=team_id, id=workflow_id).exists()
+
+
+def get_workflow_task_daily_limits(*, team_id: int) -> WorkflowTaskDailyLimits:
+    config = (
+        TeamWorkflowsConfig.objects.filter(team_id=team_id)
+        .only("workflow_task_rate_limit_per_day", "workflow_task_team_rate_limit_per_day")
+        .first()
+    )
+    if config is None:
+        return WorkflowTaskDailyLimits(per_workflow=None, per_team=None)
+    return WorkflowTaskDailyLimits(
+        per_workflow=config.workflow_task_rate_limit_per_day,
+        per_team=config.workflow_task_team_rate_limit_per_day,
+    )
+
+
 def accept_github_event(delivery: WebhookDelivery) -> None:
     """The inbound GitHub App webhook enters workflows here, so its consumer needs no internal import."""
     # Deferred to keep the Kafka producer off the facade import path.

products/workflows/backend/facade/contracts.py

@@ -5,8 +5,11 @@
 
 from posthog.dataclasses import frozen
 
+from products.workflows.backend.facade.enums import HogFlowBatchJobState
+
 if TYPE_CHECKING:
     from posthog.models.team.team import Team
+    from posthog.models.user import User
 
 
 @frozen
@@ -31,6 +34,33 @@ class WorkflowActivitySummary:
     recent: tuple[RecentWorkflow, ...]
 
 
+@frozen
+class WorkflowTaskDailyLimits:
+    """A team's daily caps on tasks created by workflows. None means the default cap applies."""
+
+    per_workflow: int | None
+    per_team: int | None
+
+
+@frozen
+class WorkflowBatchJob:
+    """One batch run of a workflow.
+
+    ``created_by`` carries the core ``User`` row rather than a projection of it, so the
+    presentation layer keeps serializing it through core's ``UserBasicSerializer`` and the
+    generated ``UserBasic`` component stays as it was.
+    """
+
+    id: UUID
+    hog_flow_id: UUID
+    status: HogFlowBatchJobState
+    filters: dict[str, Any]
+    variables: dict[str, Any]
+    created_at: datetime
+    updated_at: datetime
+    created_by: "User | None"
+
+
 @dataclass(frozen=True)
 class HogFlowReference:
     id: str

products/workflows/backend/facade/enums.py

@@ -1,5 +1,7 @@
 from enum import StrEnum
 
+from django.db import models
+
 
 class EmailTrackingConsentMode(StrEnum):
     # No consent enforcement: tracking follows the email step's own setting only.
@@ -15,3 +17,12 @@ class EmailTrackingConsentMode(StrEnum):
 EMAIL_TRACKING_CONSENT_MODE_CHOICES = [
     (mode.value, mode.name.replace("_", " ").title()) for mode in EmailTrackingConsentMode
 ]
+
+
+class HogFlowBatchJobState(models.TextChoices):
+    WAITING = "waiting"
+    QUEUED = "queued"
+    ACTIVE = "active"
+    COMPLETED = "completed"
+    CANCELLED = "cancelled"
+    FAILED = "failed"

products/workflows/backend/facade/service_jwt.py

@@ -0,0 +1,8 @@
+"""The scoped service JWT purposes that the workflow callback endpoints verify.
+
+The plugin server mints these tokens for the "Create AI task" and "Run scout" workflow steps.
+"""
+
+from products.workflows.backend.service_jwt import TASKS_CREATE_PURPOSE, WORKFLOW_SCOUT_RUN_PURPOSE
+
+__all__ = ["TASKS_CREATE_PURPOSE", "WORKFLOW_SCOUT_RUN_PURPOSE"]

products/workflows/backend/models/hog_flow_batch_job/hog_flow_batch_job.py

@@ -7,6 +7,7 @@
 from posthog.models.utils import RootTeamMixin, UUIDTModel
 from posthog.plugins.plugin_server_api import create_batch_hog_flow_job_invocation
 
+from products.workflows.backend.facade.enums import HogFlowBatchJobState
 from products.workflows.backend.utils.batch_trigger_limit import get_hogflow_batch_trigger_limit, hog_flow_sends_email
 
 logger = structlog.get_logger(__name__)
@@ -22,13 +23,7 @@ class Meta:
             models.Index(fields=["team"]),
         ]
 
-    class State(models.TextChoices):
-        WAITING = "waiting"
-        QUEUED = "queued"
-        ACTIVE = "active"
-        COMPLETED = "completed"
-        CANCELLED = "cancelled"
-        FAILED = "failed"
+    State = HogFlowBatchJobState
 
     team = models.ForeignKey("posthog.Team", on_delete=models.DO_NOTHING, related_name="+")
     hog_flow = models.ForeignKey("workflows.HogFlow", on_delete=models.DO_NOTHING)

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

@@ -87,7 +87,7 @@
 from posthog.clickhouse.query_tagging import Feature, tag_queries
 from posthog.dataclasses import frozen
 from posthog.event_usage import AGENT_EVENT_SOURCES, EventSource, get_event_source, report_user_action
-from posthog.models import Team
+from posthog.models import Team, User
 from posthog.models.filters import Filter
 from posthog.models.integration import Integration
 from posthog.permissions import posthog_feature_flag_enabled
@@ -128,6 +128,7 @@
     resolve_connectors,
     validate_skill_names,
 )
+from products.workflows.backend.facade.api import create_batch_job
 from products.workflows.backend.metrics import (
     GUARDRAIL_LABELS,
     GUARDRAIL_METRICS,
@@ -1443,7 +1444,7 @@
                         }
                     )
 
     def _validate_create_task_action(self, inputs: dict) -> None:
         """Save-time checks for the "Create AI task" step beyond input shape: whether the
         chosen connectors, model and repository are actually usable, and the parallel-run
         limit is sane - so a misconfigured step fails here instead of only when it fires."""
@@ -1541,7 +1542,7 @@
                 {"template_id": "Run scout is only available in the project's main environment."}
             )
 
     def validate(self, data):
         is_draft = self.context.get("is_draft")
         # Drafts from the web builder stay lenient (incomplete graphs save fine); programmatic callers
         # (MCP/API) get full validation even on drafts so a broken or unsupported config fails at create
@@ -1945,7 +1946,7 @@
 
         return data
 
     def _validate_delay(self, data: dict, strict: bool) -> None:
         """A delay waits either a fixed span or until a date carried by the person or event, never both."""
         config = data.get("config") or {}
         delay_until = config.get("delay_until")
@@ -3295,7 +3296,7 @@
             "last_run",
         ]
 
     def validate(self, data):
         instance = cast(Optional[HogFlow], self.instance)
         is_draft = self.context.get("is_draft")
 
@@ -4818,7 +4819,7 @@
             self._workflow_last_runs = list_workflow_last_runs(self.team_id, self.request.user.id, loop_ids)
         return page
 
     def safely_get_queryset(self, queryset: QuerySet) -> QuerySet:
         if self.action == "list":
             # `id` breaks ties so LIMIT/OFFSET paging stays stable: rows sharing an updated_at can
             # otherwise repeat on one page and never appear on another.
@@ -6814,15 +6815,20 @@
             if get_event_source(request) in AGENT_EVENT_SOURCES:
                 self._require_audience_confirm_token(request, hog_flow)
 
-            serializer = HogFlowBatchJobSerializer(
-                data={**request.data, "hog_flow": hog_flow.id}, context={**self.get_serializer_context()}
-            )
+            serializer = HogFlowBatchJobSerializer(data={**request.data, "hog_flow": hog_flow.id})
             if not serializer.is_valid():
                 return Response(serializer.errors, status=400)
 
-            # The consumer fans out to the trigger's stored filters, so snapshot those on the job -
-            # caller-supplied filters are never what actually runs.
-            batch_job = serializer.save(filters=(hog_flow.trigger or {}).get("filters") or {})
+            batch_job = create_batch_job(
+                team_id=self.team_id,
+                hog_flow_id=hog_flow.id,
+                created_by_id=cast(User, request.user).id,
+                variables=serializer.validated_data.get("variables", {}),
+                status=serializer.validated_data.get("status"),
+                # The consumer fans out to the trigger's stored filters, so snapshot those on the job -
+                # caller-supplied filters are never what actually runs.
+                filters=(hog_flow.trigger or {}).get("filters") or {},
+            )
             self._report_workflow_action("hog_flow_batch_job_created", hog_flow, {"batch_job_id": str(batch_job.id)})
             return Response(HogFlowBatchJobSerializer(batch_job).data)
         else:

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

@@ -3,54 +3,30 @@
 
 from posthog.api.shared import UserBasicSerializer
 
-from products.workflows.backend.models.hog_flow_batch_job import HogFlowBatchJob
+from products.workflows.backend.facade.enums import HogFlowBatchJobState
 
 logger = structlog.get_logger(__name__)
 
 
-class HogFlowBatchJobSerializer(serializers.ModelSerializer):
-    created_by = UserBasicSerializer(read_only=True)
-
-    class Meta:
-        model = HogFlowBatchJob
-        fields = [
-            "id",
-            "status",
-            "hog_flow",
-            "filters",
-            "variables",
-            "created_at",
-            "created_by",
-            "updated_at",
-        ]
-        read_only_fields = [
-            "id",
-            "filters",
-            "created_at",
-            "created_by",
-            "updated_at",
-        ]
-        extra_kwargs = {
-            "status": {
-                "help_text": (
-                    "Not currently tracked — stays at its initial value. Use the workflow logs/metrics "
-                    "endpoints for run outcome."
-                )
-            },
-            "hog_flow": {"help_text": "ID of the workflow this batch run belongs to."},
-            "filters": {
-                "help_text": "Audience snapshot the run fanned out to, taken from the workflow's batch trigger filters."
-            },
-            "variables": {"help_text": "Variable value overrides applied to this run."},
-        }
-
-    def create(self, validated_data: dict, *args, **kwargs) -> HogFlowBatchJob:
-        request = self.context["request"]
-        team_id = self.context["team_id"]
-        validated_data["created_by"] = request.user
-        validated_data["team_id"] = team_id
-
-        return super().create(validated_data=validated_data)
+class HogFlowBatchJobSerializer(serializers.Serializer):
+    id = serializers.UUIDField(read_only=True, help_text="ID of the batch run.")
+    status = serializers.ChoiceField(
+        choices=HogFlowBatchJobState.choices,
+        required=False,
+        help_text=(
+            "Not currently tracked — stays at its initial value. Use the workflow logs/metrics "
+            "endpoints for run outcome."
+        ),
+    )
+    hog_flow = serializers.UUIDField(source="hog_flow_id", help_text="ID of the workflow this batch run belongs to.")
+    filters = serializers.JSONField(
+        read_only=True,
+        help_text="Audience snapshot the run fanned out to, taken from the workflow's batch trigger filters.",
+    )
+    variables = serializers.JSONField(required=False, help_text="Variable value overrides applied to this run.")
+    created_at = serializers.DateTimeField(read_only=True, help_text="When the batch run was created.")
+    created_by = UserBasicSerializer(read_only=True, help_text="User who started the batch run.")
+    updated_at = serializers.DateTimeField(read_only=True, help_text="When the batch run was last updated.")
 
 
 class HogFlowBatchJobCancelResponseSerializer(serializers.Serializer):
@@ -61,7 +37,7 @@ class HogFlowBatchJobCancelResponseSerializer(serializers.Serializer):
     """
 
     status = serializers.ChoiceField(
-        choices=HogFlowBatchJob.State.choices,
+        choices=HogFlowBatchJobState.choices,
         help_text="The batch run's status after this request. 'cancelled' once every in-flight run is flagged; "
         "a completion that raced the stop wins and is reported instead.",
     )

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

@@ -17,8 +17,8 @@
     WorkflowScoutRunRejected,
     start_workflow_scout_run,
 )
-from products.workflows.backend.models import HogFlow
-from products.workflows.backend.service_jwt import WORKFLOW_SCOUT_RUN_PURPOSE
+from products.workflows.backend.facade.api import workflow_exists
+from products.workflows.backend.facade.service_jwt import WORKFLOW_SCOUT_RUN_PURPOSE
 
 logger = structlog.get_logger(__name__)
 
@@ -144,7 +144,7 @@ def create(self, request: Request, **kwargs: Any) -> Response:
 
         # A token outlives the workflow it was minted for (its TTL covers the whole fetch retry
         # chain), so a deleted workflow must not still be able to spend scout runs.
-        if not HogFlow.objects.filter(team_id=team_id, id=hog_flow_id).exists():
+        if not workflow_exists(team_id=team_id, workflow_id=hog_flow_id):
             return _rejected("Workflow no longer exists.", status.HTTP_422_UNPROCESSABLE_ENTITY)
 
         try:

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

@@ -28,8 +28,12 @@
     build_output_schema,
     create_workflow_task,
 )
-from products.workflows.backend.models import HogFlow, TeamWorkflowsConfig
-from products.workflows.backend.service_jwt import TASKS_CREATE_PURPOSE
+from products.workflows.backend.facade.api import (
+    WorkflowNotFound,
+    get_workflow_owner_id,
+    get_workflow_task_daily_limits,
+)
+from products.workflows.backend.facade.service_jwt import TASKS_CREATE_PURPOSE
 
 logger = structlog.get_logger(__name__)
 
@@ -218,15 +222,8 @@ def create(self, request: Request, **kwargs: Any) -> Response:
         if owner_id is None:
             return _rejected("Workflow has no owner who can run tasks.", status.HTTP_422_UNPROCESSABLE_ENTITY)
 
-        config = (
-            TeamWorkflowsConfig.objects.filter(team_id=team_id)
-            .only("workflow_task_rate_limit_per_day", "workflow_task_team_rate_limit_per_day")
-            .first()
-        )
-        rate_limits = WorkflowTaskRateLimits(
-            per_workflow=config.workflow_task_rate_limit_per_day if config is not None else None,
-            per_team=config.workflow_task_team_rate_limit_per_day if config is not None else None,
-        )
+        daily_limits = get_workflow_task_daily_limits(team_id=team_id)
+        rate_limits = WorkflowTaskRateLimits(per_workflow=daily_limits.per_workflow, per_team=daily_limits.per_team)
 
         try:
             result = create_workflow_task(
@@ -326,7 +323,7 @@ def _resolve_workflow_owner(team_id: int, hog_flow_id: uuid.UUID) -> int | None:
     """The workflow's creator, who the run executes as. Read from the row rather than the
     request so a token can never assert a different user. Eligibility (active account,
     current project access) is enforced in-transaction by the tasks service."""
-    hog_flow = HogFlow.objects.filter(team_id=team_id, id=hog_flow_id).only("created_by_id").first()
-    if hog_flow is None or hog_flow.created_by_id is None:
+    try:
+        return get_workflow_owner_id(team_id=team_id, workflow_id=hog_flow_id)
+    except WorkflowNotFound:
         return None
-    return hog_flow.created_by_id

products/workflows/backend/services/batch_jobs.py

@@ -0,0 +1,40 @@
+from typing import Any
+from uuid import UUID
+
+from products.workflows.backend.facade.contracts import WorkflowBatchJob
+from products.workflows.backend.facade.enums import HogFlowBatchJobState
+from products.workflows.backend.models.hog_flow_batch_job import HogFlowBatchJob
+
+
+def to_batch_job(job: HogFlowBatchJob) -> WorkflowBatchJob:
+    return WorkflowBatchJob(
+        id=job.id,
+        hog_flow_id=job.hog_flow_id,
+        status=HogFlowBatchJobState(job.status),
+        filters=job.filters,
+        variables=job.variables,
+        created_at=job.created_at,
+        updated_at=job.updated_at,
+        created_by=job.created_by,
+    )
+
+
+def create_batch_job(
+    *,
+    team_id: int,
+    hog_flow_id: UUID,
+    created_by_id: int,
+    variables: dict[str, Any],
+    filters: dict[str, Any],
+    status: HogFlowBatchJobState | None = None,
+) -> WorkflowBatchJob:
+    """Save the job row. Its post_save receiver dispatches the run to the plugin server."""
+    job = HogFlowBatchJob.objects.create(
+        team_id=team_id,
+        hog_flow_id=hog_flow_id,
+        created_by_id=created_by_id,
+        variables=variables,
+        filters=filters,
+        **({} if status is None else {"status": status}),
+    )
+    return to_batch_job(job)

products/workflows/backend/tests/api/test_hog_flow.py

@@ -4325,6 +4325,8 @@ def test_post_hog_flow_batch_jobs_endpoint_creates_job(self, mock_create_invocat
         assert response.json()["hog_flow"] == flow_id
         assert response.json()["variables"] == batch_job_data["variables"]
         assert response.json()["status"] == "queued"
+        assert response.json()["created_by"]["id"] == self.user.id
+        assert response.json()["filters"] == HogFlow.objects.get(pk=flow_id).trigger["filters"]
         mock_create_invocation.assert_called_once()
         # The per-team audience cap must ride on the invocation so the consumer enforces the team's limit.
         assert mock_create_invocation.call_args.kwargs["max_audience_size"] == 5000