The 409 guard and batch creation read as methods on ModellingRunTasks 🟪

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/app/modelling/router.py

@@ -9,7 +9,6 @@ fan-out — progress and terminal state roll up from the workers.
 
 import json
 from collections.abc import Callable, Iterator
-from datetime import datetime, timezone
 from typing import Any, cast
 from uuid import uuid4
 
@@ -26,14 +25,9 @@ from backend.app.modelling.property_filters import (
     FilteredProperty,
     resolve_filtered_property_ids,
 )
+from backend.app.modelling.run_tasks import ModellingRunTasks
 from backend.app.modelling.schemas import TriggerRunRequest
 
-# DB access is parameterised raw SQL, not the SQLModel mirrors: the FastAPI
-# app registers the legacy backend.app.db.models mirrors of these tables, and
-# importing the infrastructure.postgres mirrors of the same tables into one
-# process double-registers them and crashes the app at import (see
-# property_filters — same containment, until the DDD cut-over).
-
 # Sends pre-serialised message bodies to the modelling_e2e queue. A seam so
 # tests record bodies instead of calling AWS.
 MessageSender = Callable[[list[str]], None]
@@ -77,11 +71,8 @@ async def trigger_run(
     session: Session = Depends(get_session),
     send_messages: MessageSender = Depends(get_message_sender),
 ) -> dict[str, str]:
-    already_distributed = session.connection().execute(
-        text("SELECT 1 FROM sub_task WHERE task_id = :task_id LIMIT 1"),
-        {"task_id": body.task_id},
-    ).first()
-    if already_distributed is not None:
+    run_tasks = ModellingRunTasks(session)
+    if run_tasks.already_distributed(body.task_id):
         raise HTTPException(
             status_code=409,
             detail=(
@@ -146,25 +137,7 @@ async def trigger_run(
                 }
             )
 
-    now = datetime.now(timezone.utc)
-    session.connection().execute(
-        text(
-            "INSERT INTO sub_task (id, task_id, status, inputs, updated_at)"
-            " VALUES (:id, :task_id, 'waiting', :inputs, :updated_at)"
-        ),
-        [
-            {
-                "id": message["subtask_id"],
-                "task_id": body.task_id,
-                "inputs": json.dumps(
-                    {k: v for k, v in message.items() if k != "subtask_id"}
-                ),
-                "updated_at": now,
-            }
-            for message in messages
-        ],
-    )
-    session.commit()
+    run_tasks.create_batch_subtasks(body.task_id, messages)
 
     send_messages([json.dumps(message) for message in messages])
     return {"message": "Modelling Run distributed"}

backend/app/modelling/run_tasks.py

@@ -0,0 +1,58 @@
+"""The distributor's view of the app-owned task's sub_tasks (ADR-0055).
+
+Intention-revealing methods over parameterised SQL. SQL rather than the
+SQLModel mirrors because the FastAPI app registers the legacy
+``backend.app.db.models`` mirrors of ``sub_task``, and importing the
+``infrastructure.postgres`` mirror of the same table into one process
+double-registers it and crashes the app at import (contained here until the
+DDD cut-over).
+"""
+
+import json
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy import text
+from sqlmodel import Session
+
+
+class ModellingRunTasks:
+    def __init__(self, session: Session) -> None:
+        self._session = session
+
+    def already_distributed(self, task_id: UUID) -> bool:
+        """Whether the task already has sub_tasks — i.e. a distributor has
+        already fanned it out (the 409 guard against double-submits)."""
+        row = self._session.connection().execute(
+            text("SELECT 1 FROM sub_task WHERE task_id = :task_id LIMIT 1"),
+            {"task_id": task_id},
+        ).first()
+        return row is not None
+
+    def create_batch_subtasks(
+        self, task_id: UUID, messages: list[dict[str, Any]]
+    ) -> None:
+        """Pre-create one ``waiting`` sub_task per batch message under the
+        app-owned task. Each sub_task's inputs are its exact message payload
+        (minus the self-referential subtask_id) — the fixed progress
+        denominator and the batch's re-run recipe."""
+        now = datetime.now(timezone.utc)
+        self._session.connection().execute(
+            text(
+                "INSERT INTO sub_task (id, task_id, status, inputs, updated_at)"
+                " VALUES (:id, :task_id, 'waiting', :inputs, :updated_at)"
+            ),
+            [
+                {
+                    "id": message["subtask_id"],
+                    "task_id": task_id,
+                    "inputs": json.dumps(
+                        {k: v for k, v in message.items() if k != "subtask_id"}
+                    ),
+                    "updated_at": now,
+                }
+                for message in messages
+            ],
+        )
+        self._session.commit()