diff --git a/products/workflows/backend/facade/api.py b/products/workflows/backend/facade/api.py
index 387a78e7ee3a..c088dbd58675 100644
--- a/products/workflows/backend/facade/api.py
+++ b/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.
diff --git a/products/workflows/backend/facade/contracts.py b/products/workflows/backend/facade/contracts.py
index 1f67b17da7c1..025372f7f94a 100644
--- a/products/workflows/backend/facade/contracts.py
+++ b/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
diff --git a/products/workflows/backend/facade/enums.py b/products/workflows/backend/facade/enums.py
index 48d92fb24a93..b4d354344a60 100644
--- a/products/workflows/backend/facade/enums.py
+++ b/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"
diff --git a/products/workflows/backend/facade/service_jwt.py b/products/workflows/backend/facade/service_jwt.py
new file mode 100644
index 000000000000..b544eafa4bf5
--- /dev/null
+++ b/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"]
diff --git a/products/workflows/backend/models/hog_flow_batch_job/hog_flow_batch_job.py b/products/workflows/backend/models/hog_flow_batch_job/hog_flow_batch_job.py
index 3c31f069e1d5..1c64c63607bf 100644
--- a/products/workflows/backend/models/hog_flow_batch_job/hog_flow_batch_job.py
+++ b/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)
diff --git a/products/workflows/backend/presentation/views/hog_flow.py b/products/workflows/backend/presentation/views/hog_flow.py
index 761eb003f2df..9e113464484a 100644
--- a/products/workflows/backend/presentation/views/hog_flow.py
+++ b/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,
@@ -6814,15 +6815,20 @@ def batch_jobs(self, request: Request, *args, **kwargs):
             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:
diff --git a/products/workflows/backend/presentation/views/hog_flow_batch_job.py b/products/workflows/backend/presentation/views/hog_flow_batch_job.py
index 6b365b1053f3..f079f2f745f3 100644
--- a/products/workflows/backend/presentation/views/hog_flow_batch_job.py
+++ b/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.",
     )
diff --git a/products/workflows/backend/presentation/views/workflow_scout_runs.py b/products/workflows/backend/presentation/views/workflow_scout_runs.py
index e4e152f2e881..b5097832fd4f 100644
--- a/products/workflows/backend/presentation/views/workflow_scout_runs.py
+++ b/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:
diff --git a/products/workflows/backend/presentation/views/workflow_tasks.py b/products/workflows/backend/presentation/views/workflow_tasks.py
index cd596fadeb21..117b0b78ce8e 100644
--- a/products/workflows/backend/presentation/views/workflow_tasks.py
+++ b/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
diff --git a/products/workflows/backend/services/batch_jobs.py b/products/workflows/backend/services/batch_jobs.py
new file mode 100644
index 000000000000..a71e68638aec
--- /dev/null
+++ b/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)
diff --git a/products/workflows/backend/tests/api/test_hog_flow.py b/products/workflows/backend/tests/api/test_hog_flow.py
index 8f1a619edf3b..6513f8097b0e 100644
--- a/products/workflows/backend/tests/api/test_hog_flow.py
+++ b/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
diff --git a/products/workflows/frontend/generated/api.schemas.ts b/products/workflows/frontend/generated/api.schemas.ts
index 5cf2a02da114..de3524f8a928 100644
--- a/products/workflows/frontend/generated/api.schemas.ts
+++ b/products/workflows/frontend/generated/api.schemas.ts
@@ -984,6 +984,7 @@ export const HogFlowBatchJobStateEnumApi = {
 } as const
 
 export interface HogFlowBatchJobApi {
+    /** ID of the batch run. */
     readonly id: string
     /** Not currently tracked — stays at its initial value. Use the workflow logs/metrics endpoints for run outcome.
      *
@@ -1000,8 +1001,11 @@ export interface HogFlowBatchJobApi {
     readonly filters: unknown
     /** Variable value overrides applied to this run. */
     variables?: unknown
+    /** When the batch run was created. */
     readonly created_at: string
+    /** User who started the batch run. */
     readonly created_by: UserBasicApi
+    /** When the batch run was last updated. */
     readonly updated_at: string
 }
 
diff --git a/pyproject.toml b/pyproject.toml
index 9812d61000d9..d5a19a95b81f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -759,14 +759,9 @@ ignore_imports = [
     "products.workflows.backend.presentation.views.hog_flow -> products.workflows.backend.utils.durations",
     "products.workflows.backend.presentation.views.hog_flow -> products.workflows.backend.utils.email_sending_tiers",
     "products.workflows.backend.presentation.views.hog_flow -> products.workflows.backend.utils.rrule_utils",
-    "products.workflows.backend.presentation.views.hog_flow_batch_job -> products.workflows.backend.models.hog_flow_batch_job",
     "products.workflows.backend.presentation.views.hog_flow_template -> products.cdp.backend.models.hog_function_template",
     "products.workflows.backend.presentation.views.hog_flow_template -> products.workflows.backend.models.hog_flow.hog_flow_template",
     "products.workflows.backend.presentation.views.hog_flow_template -> products.workflows.backend.templates",
-    "products.workflows.backend.presentation.views.workflow_scout_runs -> products.workflows.backend.models",
-    "products.workflows.backend.presentation.views.workflow_scout_runs -> products.workflows.backend.service_jwt",
-    "products.workflows.backend.presentation.views.workflow_tasks -> products.workflows.backend.models",
-    "products.workflows.backend.presentation.views.workflow_tasks -> products.workflows.backend.service_jwt",
     # TODO: access-control presentation wave — the consolidated RBAC viewsets carry their ORM
     # access with them. Thin them to parse -> facade -> serialize and delete each entry.
     "products.access_control.backend.presentation.access_control -> products.access_control.backend.models.access_control",
diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts
index 3411617aed48..de3f2e40ec82 100644
--- a/services/mcp/src/api/generated.ts
+++ b/services/mcp/src/api/generated.ts
@@ -50500,6 +50500,7 @@ export namespace Schemas {
     } as const;
 
     export interface HogFlowBatchJob {
+      /** ID of the batch run. */
       readonly id: string;
       /** Not currently tracked — stays at its initial value. Use the workflow logs/metrics endpoints for run outcome.
        *
@@ -50516,8 +50517,11 @@ export namespace Schemas {
       readonly filters: unknown;
       /** Variable value overrides applied to this run. */
       variables?: unknown;
+      /** When the batch run was created. */
       readonly created_at: string;
+      /** User who started the batch run. */
       readonly created_by: UserBasic;
+      /** When the batch run was last updated. */
       readonly updated_at: string;
     }