From 9031c16ba76b7a4b23afa4d2806451865978088e Mon Sep 17 00:00:00 2001
From: Khalim Conn-Kowlessar <kconnkowlessar@gmail.com>
Date: Tue, 7 Jul 2026 12:03:33 +0000
Subject: [PATCH] =?UTF-8?q?Trigger-run=20fans=20out=20one=20sub=5Ftask=20a?=
 =?UTF-8?q?nd=20message=20per=20scenario=20batch=20=F0=9F=9F=A9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 backend/app/modelling/router.py | 49 ++++++++++++++++++++++++++++-----
 1 file changed, 42 insertions(+), 7 deletions(-)

diff --git a/backend/app/modelling/router.py b/backend/app/modelling/router.py
index 7eff93340..71e74ae4f 100644
--- a/backend/app/modelling/router.py
+++ b/backend/app/modelling/router.py
@@ -18,7 +18,14 @@ from sqlmodel import Session
 from backend.app.config import get_settings
 from backend.app.db.connection import db_engine
 from backend.app.dependencies import validate_token
+from backend.app.modelling.batching import pack_postcode_batches
+from backend.app.modelling.property_filters import (
+    FilteredProperty,
+    resolve_filtered_property_ids,
+)
 from backend.app.modelling.schemas import TriggerRunRequest
+from domain.tasks.subtasks import SubTask
+from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
 
 # Sends pre-serialised message bodies to the modelling_e2e queue. A seam so
 # tests record bodies instead of calling AWS.
@@ -32,9 +39,7 @@ def get_session() -> Iterator[Session]:
 
 def get_message_sender() -> MessageSender:
     settings = get_settings()
-    client: Any = cast(
-        Any, boto3.client("sqs", settings.AWS_DEFAULT_REGION)
-    )  # pyright: ignore[reportUnknownMemberType]
+    client: Any = cast(Any, boto3.client("sqs", settings.AWS_DEFAULT_REGION))  # pyright: ignore[reportUnknownMemberType]
     queue_url = settings.MODELLING_E2E_SQS_URL
 
     def send(bodies: list[str]) -> None:
@@ -65,9 +70,39 @@ async def trigger_run(
     session: Session = Depends(get_session),
     send_messages: MessageSender = Depends(get_message_sender),
 ) -> dict[str, str]:
-    raise NotImplementedError
+    properties: list[FilteredProperty] = resolve_filtered_property_ids(
+        session, body.portfolio_id, body.filters
+    )
+    batches = pack_postcode_batches(properties)
 
+    # Pre-create one sub_task per (scenario, batch) message under the
+    # app-owned task, each holding its exact message payload — the fixed
+    # progress denominator and the batch's re-run recipe (ADR-0055).
+    subtask_repo = SubTaskPostgresRepository(session)
+    subtasks: list[SubTask] = []
+    payloads: list[dict[str, Any]] = []
+    for scenario_id in body.scenario_ids:
+        for batch in batches:
+            payload: dict[str, Any] = {
+                "task_id": str(body.task_id),
+                "property_ids": [p.property_id for p in batch],
+                "portfolio_id": body.portfolio_id,
+                "scenario_id": scenario_id,
+                # ADR-0055 pinned flags: live EPC fetch, live prediction,
+                # solar fetched only where no stored row exists.
+                "refetch_epc": True,
+                "repredict_epc": True,
+                "refetch_solar": True,
+                "dry_run": False,
+            }
+            subtasks.append(SubTask.create(task_id=body.task_id, inputs=payload))
+            payloads.append(payload)
+    subtask_repo.create_many(subtasks)
 
-# json is used by the implementation; referenced here so the stub imports are
-# stable for the first RED.
-_ = json
+    send_messages(
+        [
+            json.dumps({**payload, "subtask_id": str(subtask.id)})
+            for payload, subtask in zip(payloads, subtasks)
+        ]
+    )
+    return {"message": "Modelling Run distributed"}