Trigger-run fans out one sub_task and message per scenario batch 🟥

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

backend/app/config.py

@@ -44,6 +44,7 @@ class Settings(BaseSettings):
     COMBINER_SQS_URL: str = "changeme"
     LANDLORD_OVERRIDES_SQS_URL: str = "changeme"
     FINALISER_SQS_URL: str = "changeme"
+    MODELLING_E2E_SQS_URL: str = "changeme"
 
     # Third parties
     EPC_AUTH_TOKEN: str = "changeme"

backend/app/modelling/router.py

@@ -0,0 +1,73 @@
+"""The Modelling Run Distributor: POST /v1/modelling/trigger-run (ADR-0055).
+
+Accepts a portfolio-scoped modelling request expressed as filters, resolves
+them to a concrete property set (ADR-0056), pre-creates one batch sub_task per
+SQS message under the app-owned task, and fans the batches out to the
+modelling_e2e workers. Never models synchronously; owns nothing after the
+fan-out — progress and terminal state roll up from the workers.
+"""
+
+import json
+from collections.abc import Callable, Iterator
+from typing import Any, cast
+
+import boto3
+from fastapi import APIRouter, Depends
+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.schemas import TriggerRunRequest
+
+# 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]
+
+
+def get_session() -> Iterator[Session]:
+    with Session(db_engine) as session:
+        yield session
+
+
+def get_message_sender() -> MessageSender:
+    settings = get_settings()
+    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:
+        # send_message_batch caps at 10 entries per call — chunk accordingly.
+        for start in range(0, len(bodies), 10):
+            chunk = bodies[start : start + 10]
+            client.send_message_batch(
+                QueueUrl=queue_url,
+                Entries=[
+                    {"Id": str(index), "MessageBody": body}
+                    for index, body in enumerate(chunk)
+                ],
+            )
+
+    return send
+
+
+router = APIRouter(
+    prefix="/modelling",
+    tags=["modelling"],
+    dependencies=[Depends(validate_token)],
+)
+
+
+@router.post("/trigger-run", status_code=202)
+async def trigger_run(
+    body: TriggerRunRequest,
+    session: Session = Depends(get_session),
+    send_messages: MessageSender = Depends(get_message_sender),
+) -> dict[str, str]:
+    raise NotImplementedError
+
+
+# json is used by the implementation; referenced here so the stub imports are
+# stable for the first RED.
+_ = json

backend/app/modelling/schemas.py

@@ -0,0 +1,18 @@
+"""Request contract for the Modelling Run Distributor (ADR-0055/0056)."""
+
+from uuid import UUID
+
+from pydantic import BaseModel
+
+from backend.app.modelling.property_filters import PropertyGroupFilters
+
+
+class TriggerRunRequest(BaseModel):
+    """Exactly what the app sends: the app-created task, the portfolio scope,
+    the scenarios to model, and the property-group filters ({} = everything).
+    Anything else the distributor might want is readable from the task row."""
+
+    task_id: UUID
+    portfolio_id: int
+    scenario_ids: list[int]
+    filters: PropertyGroupFilters = PropertyGroupFilters()

tests/backend/app/modelling/test_trigger_run_router.py

@@ -0,0 +1,133 @@
+"""Tests for the Modelling Run Distributor endpoint (ADR-0055).
+
+The router's session and SQS seams are dependency-injected; tests run against
+the ephemeral Postgres and record message bodies instead of calling AWS.
+"""
+
+import json
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+from typing import Any
+from uuid import uuid4
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from backend.app.dependencies import validate_token
+from backend.app.modelling.router import get_message_sender, get_session, router
+from domain.modelling.portfolio_goal import PortfolioGoal
+from domain.tasks.tasks import Task
+from infrastructure.postgres.modelling import ScenarioModel
+from infrastructure.postgres.property_table import PropertyRow
+from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
+from repositories.tasks.task_postgres_repository import TaskPostgresRepository
+
+PORTFOLIO_ID = 814
+
+
+@dataclass
+class Api:
+    client: TestClient
+    engine: Engine
+    sent_bodies: list[str] = field(default_factory=list)
+
+    def seed_task(self) -> Task:
+        with Session(self.engine) as session:
+            return TaskPostgresRepository(session).create(
+                Task.create(task_source="app:modelling_run", service="modelling_run")
+            )
+
+    def seed_scenario(self, *, portfolio_id: int = PORTFOLIO_ID) -> int:
+        with Session(self.engine) as session:
+            row = ScenarioModel(
+                portfolio_id=portfolio_id,
+                goal=PortfolioGoal.INCREASING_EPC,
+                goal_value="C",
+            )
+            session.add(row)
+            session.commit()
+            session.refresh(row)
+            assert row.id is not None
+            return row.id
+
+    def seed_property(self, *, postcode: str = "B93 8SU") -> int:
+        with Session(self.engine) as session:
+            row = PropertyRow(portfolio_id=PORTFOLIO_ID, postcode=postcode)
+            session.add(row)
+            session.commit()
+            session.refresh(row)
+            assert row.id is not None
+            return row.id
+
+    def subtasks_for(self, task: Task) -> list[Any]:
+        with Session(self.engine) as session:
+            return list(SubTaskPostgresRepository(session).list_by_task(task.id))
+
+
+@pytest.fixture
+def api(db_engine: Engine) -> Api:
+    app = FastAPI()
+    app.include_router(router, prefix="/v1")
+    harness = Api(client=TestClient(app), engine=db_engine)
+
+    def _session() -> Iterator[Session]:
+        with Session(db_engine) as session:
+            yield session
+
+    app.dependency_overrides[get_session] = _session
+    app.dependency_overrides[get_message_sender] = lambda: harness.sent_bodies.extend
+    app.dependency_overrides[validate_token] = lambda: "test-token"
+    return harness
+
+
+def test_trigger_run_fans_out_one_subtask_and_message_per_scenario_batch(
+    api: Api,
+) -> None:
+    # arrange — 3 properties (one batch) × 2 scenarios = 2 messages
+    task = api.seed_task()
+    scenario_a = api.seed_scenario()
+    scenario_b = api.seed_scenario()
+    property_ids = [api.seed_property() for _ in range(3)]
+
+    # act
+    response = api.client.post(
+        "/v1/modelling/trigger-run",
+        json={
+            "task_id": str(task.id),
+            "portfolio_id": PORTFOLIO_ID,
+            "scenario_ids": [scenario_a, scenario_b],
+            "filters": {},
+        },
+    )
+
+    # assert
+    assert response.status_code == 202
+    subtasks = api.subtasks_for(task)
+    assert len(subtasks) == 2
+    assert {s.status.value for s in subtasks} == {"waiting"}
+
+    messages = [json.loads(body) for body in api.sent_bodies]
+    assert len(messages) == 2
+    assert {m["scenario_id"] for m in messages} == {scenario_a, scenario_b}
+    for message in messages:
+        assert message["task_id"] == str(task.id)
+        assert message["portfolio_id"] == PORTFOLIO_ID
+        assert message["property_ids"] == property_ids
+        # ADR-0055 pinned flags: live fetch, live predict, solar only-if-missing
+        assert message["refetch_epc"] is True
+        assert message["repredict_epc"] is True
+        assert message["refetch_solar"] is True
+        assert message["dry_run"] is False
+
+    # each message's subtask_id is one of the pre-created sub_tasks, and the
+    # sub_task's inputs are the message payload (its re-run recipe)
+    subtask_ids = {str(s.id) for s in subtasks}
+    assert {m["subtask_id"] for m in messages} == subtask_ids
+    by_id = {str(s.id): s for s in subtasks}
+    for message in messages:
+        inputs = by_id[message["subtask_id"]].inputs
+        assert inputs["property_ids"] == message["property_ids"]
+        assert inputs["scenario_id"] == message["scenario_id"]