Merge branch 'main' into feature/abri-api-integration

.github/workflows/deploy_terraform.yml

@@ -538,7 +538,7 @@ jobs:
   # Deploy FastAPI Lambda
   # ============================================================
   fast_api_lambda:
-    needs: [determine_stage, ara_engine_lambda, categorisation_lambda, postcodeSplitter_lambda, bulk_address2uprn_combiner_lambda, bulkUploadFinaliser_lambda]
+    needs: [determine_stage, ara_engine_lambda, categorisation_lambda, postcodeSplitter_lambda, bulk_address2uprn_combiner_lambda, bulkUploadFinaliser_lambda, modelling_e2e_lambda]
     uses: ./.github/workflows/_deploy_lambda.yml
     with:
       lambda_name: ara_fast_api

CONTEXT.md

@@ -207,8 +207,16 @@ _Avoid_: rebaseline (that is a specific ML trigger — see Rebaselining), enrich
 The third stage. Takes the baselined Property plus a set of **Scenarios** and produces **Recommendations** → an **Optimised Package** → **Plans**, persisted to repos. A separate orchestrator from Baseline so the single-property flow can stop after Baseline and only run Modelling when the user hits "play".
 _Avoid_: scoring (overloaded), recommendation engine
 
+**Modelling Run**:
+One triggered unit of modelling over a portfolio: a target set of **Properties** resolved from user-chosen filters (no filters = the whole portfolio) crossed with one or more **Scenarios**, producing one **Plan** per (Scenario, Property). Tracked as a single task; its batch sub_tasks are the units of execution, of failure, and of re-run. Re-runs append Plans; readers take the latest per Property.
+_Avoid_: modelling job (ambiguous with one lambda invocation), batch (that is one message-worth of a run), trigger run
+
+**Distributor**:
+The role of the Modelling Run entry point: validate the request, resolve the filters to a concrete Property set, pre-create the run's batch sub_tasks, and fan the work out to the modelling workers. It never models synchronously and owns nothing after the fan-out — progress and terminal state roll up from the workers' sub_task statuses.
+_Avoid_: trigger endpoint (names the URL, not the role), orchestrator (taken — stage orchestrators, TaskOrchestrator)
+
 **First Run**:
-The use case where a Property has only a row in the property table (post address→UPRN matching) and no existing **Plan**: the pipeline runs Ingestion → Baseline → Modelling end-to-end over a batch. The first sibling lambda being built (`ara_first_run`).
+The special case of a **Modelling Run** where a Property has only a row in the property table (post address→UPRN matching) and no existing **Plan**: the pipeline runs Ingestion → Baseline → Modelling end-to-end. Executed by the same `modelling_e2e` worker as re-runs — the lambda originally planned as `ara_first_run` serves both.
 _Avoid_: initial run, cold run
 
 ### ML training
@@ -246,12 +254,8 @@ _Avoid_: emission factors (ambiguous), CO2 rates
 ### Outputs
 
 **Scenario**:
-A named portfolio-level retrofit plan, built by a user in the scenario-builder UI and persisted before any modelling fires; carries the overall goal (e.g. Increasing EPC), budget, exclusions, housing type, and the set of measure types it permits. The model is triggered against one or more Scenarios at once; each Scenario yields one Plan per Property.
-_Avoid_: project, batch, run-set
-
-**Scenario Snapshot**:
-A frozen copy of a Scenario pinned at trigger time, keyed by (task, scenario); used by the modelling pipeline so mid-run edits to the live Scenario do not affect an in-flight job. Snapshots are read-only and may be garbage-collected after the task completes.
-_Avoid_: scenario version, frozen scenario, pinned scenario
+A named portfolio-level retrofit plan, built by a user in the scenario-builder UI and persisted before any modelling fires; carries the overall goal (e.g. Increasing EPC), budget, exclusions, housing type, and the set of measure types it permits. The model is triggered against one or more Scenarios at once; each Scenario yields one Plan per Property. Scenarios are **immutable after creation**: they may be renamed, and deleted only while no Plans reference them — their modelling-relevant values never change, so an in-flight run can safely read the live row (no snapshot/pinning machinery is needed or exists).
+_Avoid_: project, batch, run-set, scenario snapshot (described pinning machinery that immutability makes unnecessary; removed 2026-07)
 
 **Plan**:
 The per-Property output of one Scenario's modelling run; carries the **Optimised Package** selected for the Property (its **Plan Measures**) and the Property's post-retrofit figures (SAP / kWh / CO₂ / bills). A Property modelled against N Scenarios in one trigger ends up with N Plans.
@@ -451,7 +455,7 @@ addresses `<michaelduong@Michaels-MacBook-Pro.local>`, `<michael@11s-MacBook.loc
 - **Bill Derivation** derives **fuel split** and **bills** from kWh values (sourced from the EPC's `renewable_heat_incentive` fields for baseline SAP10 properties, or from ML when Rebaselining fires), reading current **Fuel Rates** and **Carbon Factors** from their respective repos.
 - **EPC Prediction** uses **Comparable Properties** for both gap-filling (the no-EPC path) and producing **EPC Anomaly Flags** (the has-EPC path).
 - Triggering the model against N **Scenarios** produces N **Plans** per Property. Each **Plan** holds one **Optimised Package** — its selected **Plan Measures** — plus the Property's post-retrofit figures.
-- A **Scenario Snapshot** is pinned at trigger time per (task, scenario) so mid-run edits to the live Scenario do not affect an in-flight modelling job.
+- **Scenarios** are immutable after creation (rename-only; deletable only while no Plans reference them), so in-flight modelling jobs read the live Scenario row safely.
 - A **Recommendation** references one **Measure Type** and carries property-specific cost and impact.
 - A **Property Valuation** (current market value) is a Baseline attribute and is mostly absent; a **Valuation Uplift** is a Plan output, always a percentage from the **EPC Band** jump and an absolute £ only when a Property Valuation exists.
 - **Address Matching** uses a **User Address** and **Postcode** to find a **UPRN** by scoring **UPRN Candidates** from an EPC search. A **Lexirank** of 1 with no **Ambiguous Match** and a **Lexiscore** ≥ the **Score Threshold** produces a **Best Match**.

applications/modelling_e2e/handler.py

@@ -65,7 +65,7 @@ from domain.modelling.plan import Plan
 from domain.property.property import Property, PropertyIdentity
 from domain.property_baseline.calculator_rebaseliner import CalculatorRebaseliner
 from domain.sap10_calculator.calculator import Sap10Calculator
-from domain.tasks.subtasks import SubTask
+from domain.tasks.subtasks import SubTask, SubTaskFailure
 from domain.tasks.tasks import Source
 from harness.console import run_modelling
 from orchestration.task_orchestrator import TaskOrchestrator
@@ -536,7 +536,9 @@ def handler(
 
         def _work(subtask: SubTask) -> None:
             inputs = subtask.inputs or {}
-            pid = int(inputs["property_id"])
+            _model_property(int(inputs["property_id"]))
+
+        def _model_property(pid: int) -> None:
             uprn = uprns[pid]
             postcode = postcodes.get(pid, "")
             logger.info(f"property={pid} uprn={uprn} postcode={postcode!r}")
@@ -685,16 +687,32 @@ def handler(
             )
             logger.info(f"property={pid} queued for write")
 
-        # Fan the batch out into one child SubTask per property and run them in
-        # a single batched pass: create all children, model each (failures
-        # isolated per child), then persist all their statuses in two writes +
-        # one cascade — not ~5 writes and a full parent re-roll-up per property
-        # (see TaskOrchestrator.run_subtasks).
-        orchestrator.run_subtasks(
-            task_id,
-            [{"property_id": pid} for pid in property_ids],
-            work=_work,
-        )
+        failed_properties: list[dict[str, Any]] = []
+        if trigger.subtask_id is not None:
+            # Attach mode (ADR-0055): the whole batch runs under the
+            # distributor's pre-created sub_task — the @task_handler wrapper is
+            # already inside it — so no per-property children are created.
+            # Failures stay isolated per property (siblings continue) and are
+            # recorded on the sub_task after the surviving writes flush.
+            for pid in property_ids:
+                try:
+                    _model_property(pid)
+                except Exception as exc:  # noqa: BLE001 — recorded below
+                    logger.exception(f"property={pid} failed")
+                    failed_properties.append(
+                        {"property_id": pid, "error": str(exc)}
+                    )
+        else:
+            # Fan the batch out into one child SubTask per property and run
+            # them in a single batched pass: create all children, model each
+            # (failures isolated per child), then persist all their statuses in
+            # two writes + one cascade — not ~5 writes and a full parent
+            # re-roll-up per property (see TaskOrchestrator.run_subtasks).
+            orchestrator.run_subtasks(
+                task_id,
+                [{"property_id": pid} for pid in property_ids],
+                work=_work,
+            )
 
         # Persist the whole batch in one transaction, then re-establish every
         # written Property's Baseline (the orchestrator batches its own UoW). The
@@ -723,5 +741,18 @@ def handler(
                 f"{[s['certificate_number'] for s in skipped_certs]}"
             )
 
+        # Raised only after the surviving writes flushed: the failure is a
+        # record on the sub_task, never an SQS retry (ADR-0055). Recovery is a
+        # deliberate re-send of the sub_task's own inputs.
+        if failed_properties:
+            raise SubTaskFailure(
+                f"{len(failed_properties)} of {len(property_ids)} "
+                "properties failed",
+                details={
+                    "succeeded": len(property_ids) - len(failed_properties),
+                    "failed": failed_properties,
+                },
+            )
+
     finally:
         read_session.close()

applications/modelling_e2e/modelling_e2e_trigger_body.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from pydantic import BaseModel, ConfigDict
 
 
@@ -11,3 +13,8 @@ class ModellingE2ETriggerBody(BaseModel):
     refetch_epc: bool = True
     repredict_epc: bool = True
     dry_run: bool = False
+    # Attach mode (ADR-0055): a Modelling Run batch carries the app-owned task
+    # and the distributor's pre-created sub_task; the handler then creates no
+    # per-property child sub_tasks. Absent on classic (script) messages.
+    task_id: Optional[str] = None
+    subtask_id: Optional[str] = None

backend/app/config.py

@@ -45,6 +45,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/db/models/tasks.py

@@ -25,6 +25,11 @@ class Task(SQLModel, table=True):
     job_completed: Optional[datetime] = None
     status: str = Field(default="In Progress")
     service: Optional[str] = None
+    # FE-owned column (Drizzle migration): the app stores a task's original
+    # request here — for a Modelling Run, the full trigger-run payload
+    # (ADR-0055). The backend reads it at most; per-batch inputs live on the
+    # sub_tasks.
+    inputs: Optional[str] = None
     updated_at: datetime = Field(default_factory=datetime.utcnow)
 
     # source: Mapped[Optional[SourceEnum]] = mapped_column(Enum(SourceEnum)) <- SQLAlchemy not SQLModel

backend/app/main.py

@@ -11,6 +11,7 @@ from backend.app.whlg import router as whlg_router
 from backend.app.plan import router as plan_router
 from backend.app.tasks import router as tasks_router
 from backend.app.bulk_uploads import router as bulk_uploads_router
+from backend.app.modelling import router as modelling_router
 from backend.app.dependencies import validate_api_key
 from backend.app.config import get_settings
 
@@ -63,6 +64,7 @@ app.include_router(portfolio_router.router, prefix="/v1")
 app.include_router(plan_router.router, prefix="/v1")
 app.include_router(whlg_router.router, prefix="/v1")
 app.include_router(bulk_uploads_router.router, prefix="/v1")
+app.include_router(modelling_router.router, prefix="/v1")
 
 if get_settings().ENVIRONMENT == "local":
     from backend.app.local import router as local_router

backend/app/modelling/batching.py

@@ -0,0 +1,24 @@
+"""Pack a Modelling Run's resolved properties into SQS-message-sized batches.
+
+Batches stay postcode-grouped — properties sharing a postcode ride the same
+message so the workers' prediction cohort cache keeps paying (ADR-0055), the
+same packing the trigger script has been running.
+"""
+
+from backend.app.modelling.property_filters import FilteredProperty
+from utilities.grouped_batching import iter_grouped_batches
+
+BATCH_SIZE = 50
+
+
+def pack_postcode_batches(
+    properties: list[FilteredProperty], batch_size: int = BATCH_SIZE
+) -> list[list[FilteredProperty]]:
+    """Pack *properties* into batches of ~*batch_size*, never splitting a
+    postcode across batches. A single postcode larger than *batch_size*
+    becomes its own oversized batch."""
+    return list(
+        iter_grouped_batches(
+            properties, key=lambda p: p.postcode or "", max_batch_size=batch_size
+        )
+    )

backend/app/modelling/property_filters.py

@@ -0,0 +1,182 @@
+"""Resolve a Modelling Run's property-group filters to a concrete property set.
+
+The resolution rule is a shared contract with the app's preview (ADR-0056):
+the "N properties will be modelled" count the user approves is computed by the
+Next.js app from this same rule, so any change here must land in both
+codebases and amend the ADR.
+
+Reads go through parameterised raw SQL rather than the SQLModel table mirrors:
+the FastAPI app registers the legacy ``backend.app.db.models`` mirrors of
+``property``/``sub_task``, and importing the ``infrastructure.postgres``
+mirrors of the same tables into one process double-registers them in the
+shared SQLModel metadata and crashes the app at import. Until the DDD
+cut-over unifies the stacks, this module stays model-free.
+"""
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from pydantic import BaseModel, ConfigDict
+from sqlalchemy import text
+from sqlmodel import Session
+
+
+class PropertyGroupFilters(BaseModel):
+    """The filters key of a trigger-run request. An absent key is
+    unconstrained; present keys combine with AND."""
+
+    model_config = ConfigDict(frozen=True)
+
+    postcodes: Optional[list[str]] = None
+    property_types: Optional[list[str]] = None
+    built_forms: Optional[list[str]] = None
+
+
+@dataclass(frozen=True)
+class FilteredProperty:
+    """One property the filters resolved: the id to model and the postcode the
+    distributor batches by (postcode grouping keeps the workers' prediction
+    cohort cache effective)."""
+
+    property_id: int
+    postcode: Optional[str]
+
+
+# "Unknown" is itself a selectable filter value: exactly the bucket of
+# properties whose value resolves at no precedence level (ADR-0056).
+_UNKNOWN = "Unknown"
+
+
+@dataclass(frozen=True)
+class _ComponentResolution:
+    """How one filterable component (property_type / built_form) resolves at
+    each ADR-0056 precedence level: which override row names it, how its RdSAP
+    numeric codes map (text labels pass through as-is), and which columns hold
+    it on the EPC. The legacy property.property_type / built_form columns are
+    NOT consulted (ADR-0056 amendment — overrides own those facts)."""
+
+    override_component: str
+    codes: dict[str, str]
+    epc_columns: tuple[str, ...]
+
+
+_PROPERTY_TYPE = _ComponentResolution(
+    override_component="property_type",
+    codes={
+        "0": "House",
+        "1": "Bungalow",
+        "2": "Flat",
+        "3": "Maisonette",
+        "4": "Park home",
+    },
+    # A cert without the property_type column falls back to its dwelling_type.
+    epc_columns=("property_type", "dwelling_type"),
+)
+
+_BUILT_FORM = _ComponentResolution(
+    override_component="built_form_type",
+    codes={
+        "1": "Detached",
+        "2": "Semi-Detached",
+        "3": "End-Terrace",
+        "4": "Mid-Terrace",
+        "5": "Enclosed End-Terrace",
+        "6": "Enclosed Mid-Terrace",
+    },
+    epc_columns=("built_form",),
+)
+
+
+def resolve_filtered_property_ids(
+    session: Session, portfolio_id: int, filters: PropertyGroupFilters
+) -> list[FilteredProperty]:
+    """Resolve *filters* against *portfolio_id* per ADR-0056: base set is the
+    portfolio's rows with ``marked_for_deletion = false``; property_type /
+    built_form resolve override → lodged EPC → predicted EPC → "Unknown".
+    Filters AND-combine; an absent key is unconstrained."""
+    # IS NOT TRUE rather than = false: the FE schema defaults the flag, but a
+    # NULL must never silently exclude a property.
+    base_sql = (
+        "SELECT id, postcode FROM property"
+        " WHERE portfolio_id = :portfolio_id"
+        " AND marked_for_deletion IS NOT TRUE"
+    )
+    params: dict[str, Any] = {"portfolio_id": portfolio_id}
+    if filters.postcodes is not None:
+        base_sql += " AND postcode = ANY(:postcodes)"
+        params["postcodes"] = filters.postcodes
+    base_sql += " ORDER BY id"
+    candidates = [
+        _Candidate(property_id=int(row[0]), postcode=row[1])
+        for row in session.connection().execute(text(base_sql), params)
+    ]
+
+    for wanted_values, component in (
+        (filters.property_types, _PROPERTY_TYPE),
+        (filters.built_forms, _BUILT_FORM),
+    ):
+        if wanted_values is None:
+            continue
+        wanted = set(wanted_values)
+        resolved = _resolved_component_values(session, candidates, component)
+        candidates = [
+            c for c in candidates if resolved.get(c.property_id) in wanted
+        ]
+
+    return [
+        FilteredProperty(property_id=c.property_id, postcode=c.postcode)
+        for c in candidates
+    ]
+
+
+@dataclass(frozen=True)
+class _Candidate:
+    property_id: int
+    postcode: Optional[str]
+
+
+def _resolved_component_values(
+    session: Session, candidates: list[_Candidate], component: _ComponentResolution
+) -> dict[int, str]:
+    """Each candidate's effective value for *component* per the ADR-0056
+    precedence: override (building_part 0) → lodged EPC → predicted EPC →
+    "Unknown"."""
+    property_ids = [c.property_id for c in candidates]
+
+    override_rows = session.connection().execute(
+        text(
+            "SELECT property_id, override_value FROM property_overrides"
+            " WHERE property_id = ANY(:ids)"
+            " AND building_part = 0"
+            " AND override_component = :component"
+        ),
+        {"ids": property_ids, "component": component.override_component},
+    )
+    by_override = {int(row[0]): str(row[1]) for row in override_rows}
+
+    epc_columns = ", ".join(component.epc_columns)
+    epc_rows = session.connection().execute(
+        text(
+            f"SELECT property_id, source, {epc_columns} FROM epc_property"
+            " WHERE property_id = ANY(:ids)"
+            " AND source IN ('lodged', 'predicted')"
+        ),
+        {"ids": property_ids},
+    )
+    by_epc_source: dict[str, dict[int, str]] = {"lodged": {}, "predicted": {}}
+    for row in epc_rows:
+        value = next((v for v in row[2:] if v), None)
+        if value is not None:
+            by_epc_source[str(row[1])][int(row[0])] = component.codes.get(
+                str(value), str(value)
+            )
+
+    return {
+        c.property_id: (
+            by_override.get(c.property_id)
+            or by_epc_source["lodged"].get(c.property_id)
+            or by_epc_source["predicted"].get(c.property_id)
+            or _UNKNOWN
+        )
+        for c in candidates
+    }

backend/app/modelling/router.py

@@ -0,0 +1,143 @@
+"""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
+from uuid import uuid4
+
+import boto3
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import text
+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.run_tasks import ModellingRunTasks
+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]:
+    run_tasks = ModellingRunTasks(session)
+    if run_tasks.already_distributed(body.task_id):
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Task {body.task_id} already has sub_tasks — it has been "
+                "distributed. Check its progress instead of re-triggering."
+            ),
+        )
+
+    scenario_rows = session.connection().execute(
+        text("SELECT id, portfolio_id FROM scenario WHERE id = ANY(:ids)"),
+        {"ids": body.scenario_ids},
+    )
+    portfolio_by_scenario = {int(row[0]): row[1] for row in scenario_rows}
+    invalid = [
+        scenario_id
+        for scenario_id in body.scenario_ids
+        if portfolio_by_scenario.get(scenario_id) != body.portfolio_id
+    ]
+    if invalid:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Scenarios {invalid} do not belong to portfolio "
+                f"{body.portfolio_id}."
+            ),
+        )
+
+    properties: list[FilteredProperty] = resolve_filtered_property_ids(
+        session, body.portfolio_id, body.filters
+    )
+    if not properties:
+        # A task with zero sub_tasks could never roll up to complete; the
+        # app's preview shows the same zero from the same rule (ADR-0056).
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"The filters resolve to no properties in portfolio "
+                f"{body.portfolio_id} — nothing to distribute."
+            ),
+        )
+    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).
+    messages: list[dict[str, Any]] = []
+    for scenario_id in body.scenario_ids:
+        for batch in batches:
+            messages.append(
+                {
+                    "task_id": str(body.task_id),
+                    "subtask_id": str(uuid4()),
+                    "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,
+                }
+            )
+
+    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()

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()

deployment/terraform/lambda/fast-api/main.tf

@@ -64,6 +64,15 @@ data "terraform_remote_state" "landlord_description_overrides" {
   }
 }
 
+data "terraform_remote_state" "modelling_e2e" {
+  backend = "s3"
+  config = {
+    bucket = "modelling-e2e-terraform-state",
+    key    = "env:/${var.stage}/terraform.tfstate"
+    region = "eu-west-2"
+  }
+}
+
 ############################################
 # Load Credentials
 ############################################
@@ -126,6 +135,7 @@ module "fastapi" {
     COMBINER_SQS_URL           = data.terraform_remote_state.bulk_address2uprn_combiner.outputs.bulk_address2uprn_combiner_queue_url
     FINALISER_SQS_URL          = data.terraform_remote_state.bulk_upload_finaliser.outputs.bulk_upload_finaliser_queue_url
     LANDLORD_OVERRIDES_SQS_URL = data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_url
+    MODELLING_E2E_SQS_URL      = data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_url
   }
 }
 
@@ -149,7 +159,8 @@ module "fastapi_sqs_policy" {
     data.terraform_remote_state.postcode_splitter.outputs.postcode_splitter_queue_arn,
     data.terraform_remote_state.bulk_address2uprn_combiner.outputs.bulk_address2uprn_combiner_queue_arn,
     data.terraform_remote_state.bulk_upload_finaliser.outputs.bulk_upload_finaliser_queue_arn,
-    data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_arn
+    data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_arn,
+    data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_arn
   ]
 
   conditions = null

docs/adr/0055-modelling-run-batches-attach-to-the-app-owned-task.md

@@ -0,0 +1,103 @@
+# A Modelling Run's batch sub_tasks are pre-created by the distributor and attach to the app-owned task; their status is a record, not retry logic
+
+## Status
+
+accepted
+
+## Context
+
+The Ara app is gaining a filter-scoped modelling trigger: `POST
+/v1/modelling/trigger-run` accepts a portfolio, a set of scenario ids, and
+property-group filters, and fans one plan-generation job per (Scenario,
+Property) out to the existing `modelling_e2e` workers. The app creates the
+task before calling (service `modelling_run`, status `in progress`, the full
+request JSON in the new `tasks.inputs` column) and renders progress as
+`count(completed)/count(all)` over the task's sub_tasks.
+
+Two prior patterns conflicted:
+
+- The legacy `/v1/plan/trigger` route creates the task **and** its per-chunk
+  sub_tasks in the API, passing `task_id`/`subtask_id` in each SQS message.
+- The `modelling_e2e` worker's `@task_handler` creates **its own** Task per
+  SQS message and fans per-property child sub_tasks under it
+  (`TaskOrchestrator.run_subtasks`); nothing can attach to a caller's task.
+
+The parent-status roll-up (`Task.recalculate_from_subtasks`) had three
+properties that break shared-parent fan-out: a complete+waiting mix rolls up
+to `waiting` (a status the app does not render); `SubTask.start()` refuses to
+start from `failed` (breaking any redelivery/re-run of a failed batch); and
+`TaskOrchestrator._cascade` short-circuits on a FAILED parent, making failure
+permanently sticky.
+
+## Decisions
+
+### 1. The distributor pre-creates one sub_task per SQS message at accept time
+
+One task spans the whole run. The distributor resolves the filters, batches
+the properties (50 per message, postcode-grouped so the workers' prediction
+cohort cache keeps paying, one message per (scenario, batch)), bulk-creates
+one `waiting` sub_task per message under the app's task, and sends each
+message carrying `task_id`, `subtask_id`, and the batch payload. Each
+sub_task's `inputs` is the **exact message payload** — the run's precise
+what-ran record, and a failed batch's re-run recipe (re-send its own inputs).
+
+The progress denominator is therefore fixed and correct the moment the
+endpoint returns 202, and it counts **batches, not properties**.
+
+### 2. The worker gains an attach mode; the script path is untouched
+
+When a message carries `task_id` + `subtask_id`, `task_handler` creates no
+Task and the handler runs the whole batch under the supplied sub_task —
+no per-property child sub_tasks. Without those fields, current behaviour
+(own task, per-property children) is unchanged, so
+`scripts/trigger_modelling_e2e_sqs.py` keeps working.
+
+### 3. Sub_task status is a database record, not retry fuel
+
+Per-property failures inside a batch (unresolvable type, degenerate
+prediction, …) stay isolated as today — the surviving properties model and
+persist. But if **any** property failed, the batch sub_task is marked
+`failed` with `outputs = {succeeded: n, failed: [{property_id, error}]}`,
+and the parent task rolls up `failed`. The lambda still succeeds to SQS:
+a `failed` sub_task **never** triggers redelivery. Recovery is deliberate —
+fix the cause, re-send the sub_task's own `inputs`, the sub_task completes,
+and the parent task un-fails to `complete`. Batch-level crashes (OOM,
+timeout, unhandled infra errors) keep normal SQS redelivery semantics; that
+is infrastructure retry, not status-driven retry.
+
+### 4. The roll-up rules change to make failure recoverable and progress honest
+
+- A complete/failed + waiting mix rolls up `in progress`, not `waiting`.
+- `SubTask.start()` may start from `failed` (redelivery and re-runs).
+- `_cascade` no longer short-circuits on a FAILED parent: the parent status
+  is always recomputed from the children, so completing a re-run batch
+  flips the task from `failed` to `complete`.
+
+### 5. Contract guards
+
+- `409` if the task already has sub_tasks (double-submit / blind retry
+  protection — the app checks progress instead of re-POSTing).
+- `400` if the filters resolve to zero properties (a zero-sub_task task
+  could never complete; the app's preview shows the same count first).
+- `4xx` if any scenario does not belong to the portfolio. Scenarios are
+  immutable after creation (rename-only; deletable only while no Plans
+  reference them), so workers read the live Scenario row safely.
+- Pipeline flags are pinned by the endpoint, not the caller:
+  `refetch_epc=True, repredict_epc=True, refetch_solar=True, dry_run=False`.
+  Note `refetch_solar` is a misnomer: it only gates the Google fetch for
+  UPRNs with **no stored row**; stored Solar is never re-fetched.
+
+## Consequences
+
+- The app's progress bar denominator is batch count (≈ `ceil(N/50)` ×
+  scenarios), not the preview's property count. Property-level failure
+  detail is summed from sub_task `outputs`.
+- A 10k-property run whose single bad batch is fixed and re-run ends
+  `complete` — no permanently red runs over recoverable data issues.
+- The roll-up changes are global to the task machinery; other task_handler
+  lambdas keep their semantics (their per-message tasks have no shared
+  parent), but FAILED-parent recomputation now costs one children read.
+- FastAPI needs the modelling_e2e queue wired: `MODELLING_E2E_SQS_URL` from
+  the modelling_e2e remote state (outputs already exported), its ARN in the
+  `fastapi-sqs-send` policy, and `deploy_terraform.yml`'s `fast_api_lambda`
+  job ordered after `modelling_e2e_lambda`.

docs/adr/0056-modelling-run-filter-resolution-is-a-shared-contract-with-the-app-preview.md

@@ -0,0 +1,65 @@
+# Modelling Run filter resolution is a shared contract with the app's preview; the precedence order is authoritative here
+
+## Status
+
+accepted
+
+## Context
+
+The trigger-run request carries filters (`postcodes`, `property_types`,
+`built_forms`) that the distributor resolves to a concrete property set. The
+Next.js app shows users "N properties will be modelled" **before** POSTing,
+computed by its own implementation of the same rule. If the two
+implementations drift, the preview lies — the run models a different set
+than the user approved. There is no shared code path between the two
+codebases, so the rule itself must be the contract, written down once.
+
+A property's type and built form have no single column of truth: they may be
+overridden by the landlord, derived from a lodged or predicted EPC (as text
+labels or RdSAP numeric codes), or sit in legacy `property` columns — or be
+unknowable.
+
+## Decision
+
+Both implementations resolve the filters with exactly this rule; any change
+lands in both codebases and amends this ADR.
+
+1. **Base set**: `property` rows where `portfolio_id` matches and
+   `marked_for_deletion = false`.
+2. **postcodes**: exact match on `property.postcode`, canonical form
+   (uppercase, single space). The app pre-normalises and caps the list at 40.
+3. **property_types / built_forms** — resolve each property's value by
+   precedence, then filter:
+   1. `property_overrides` snapshot where `building_part = 0` and
+      `override_component` is `property_type` / `built_form_type` → the
+      `override_value`;
+   2. else EPC-derived: `epc_property` for the property, `source='lodged'`
+      preferred over `'predicted'`. Values are text labels **or** RdSAP
+      numeric codes; codes map as — property_type: 0=House, 1=Bungalow,
+      2=Flat, 3=Maisonette, 4=Park home, falling back to `dwelling_type`;
+      built_form: 1=Detached, 2=Semi-Detached, 3=End-Terrace, 4=Mid-Terrace,
+      5=Enclosed End-Terrace, 6=Enclosed Mid-Terrace. Text passes through
+      as-is;
+   3. else **"Unknown"** — a selectable filter value meaning exactly this
+      bucket (no resolvable value at any level).
+4. An absent filter key is unconstrained; present keys combine with **AND**.
+
+## Consequences
+
+- "N properties will be modelled" in the preview equals the number the
+  distributor fans out (the run's property count; the task's sub_task count
+  is batches, per ADR-0055).
+- The precedence means an override always beats a cert and a cert always
+  beats the legacy columns — consistent with how the modelling pipeline
+  itself treats Landlord Overrides as strongest truth.
+- Drift risk is accepted and mitigated by this document; a cross-repo
+  contract test (same fixture set, both implementations) is the natural
+  follow-on if drift ever bites.
+
+## Amendment (2026-07-08): the legacy property columns are not consulted
+
+The original rule fell back to `property.property_type` / `property.built_form`
+before "Unknown". Those columns are legacy: the override layer owns
+user-supplied type/form facts, so a property with no override and no EPC is
+**"Unknown"** whatever the legacy columns say. The app's preview must apply
+the same amendment.

domain/addresses/postcode_batching.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Iterable, Iterator
 
 from domain.addresses.unstandardised_address import AddressList, UnstandardisedAddress
-from domain.postcode import Postcode
+from utilities.grouped_batching import iter_grouped_batches
 
 
 def iter_postcode_grouped_batches(
@@ -11,41 +11,7 @@ def iter_postcode_grouped_batches(
     *,
     max_batch_size: int = 500,
 ) -> Iterator[AddressList]:
-    if max_batch_size < 1:
-        raise ValueError("max_batch_size must be >= 1")
-
-    groups = _group_by_postcode_in_order(addresses)
-
-    buffer: AddressList = AddressList([])
-    for group in groups.values():
-        group_len = len(group)
-
-        # Oversize single-Postcode group: flush buffer first, then dispatch
-        # the group as its own batch. Mirrors the legacy
-        # ``if group_len >= batch_size`` branch.
-        if group_len >= max_batch_size:
-            if buffer:
-                yield buffer
-                buffer = AddressList([])
-            yield group
-            continue
-
-        # Adding this group would overflow: flush buffer before appending.
-        if len(buffer) + group_len > max_batch_size:
-            yield buffer
-            buffer = AddressList([])
-
-        buffer.extend(group)
-
-    # Final flush.
-    if buffer:
-        yield buffer
-
-
-def _group_by_postcode_in_order(
-    addresses: Iterable[UnstandardisedAddress],
-) -> dict[Postcode, AddressList]:
-    groups: dict[Postcode, AddressList] = {}
-    for address in addresses:
-        groups.setdefault(address.postcode, AddressList([])).append(address)
-    return groups
+    for batch in iter_grouped_batches(
+        addresses, key=lambda a: a.postcode, max_batch_size=max_batch_size
+    ):
+        yield AddressList(batch)

domain/tasks/subtasks.py

@@ -5,6 +5,19 @@ from typing import Any, Optional
 from uuid import UUID, uuid4
 
 
+class SubTaskFailure(Exception):
+    """A *recorded* failure: the work ran, its outcome belongs on the SubTask
+    record, and the message must NOT be retried (ADR-0055 — status is a
+    record, not retry fuel). Infra crashes raise anything else and keep their
+    retry semantics. ``details`` lands on the failed SubTask's outputs."""
+
+    def __init__(
+        self, message: str, details: Optional[dict[str, Any]] = None
+    ) -> None:
+        super().__init__(message)
+        self.details = details
+
+
 class SubTaskStatus(str, Enum):
     WAITING = "waiting"
     IN_PROGRESS = "in progress"
@@ -35,11 +48,14 @@ class SubTask:
         )
 
     def start(self, cloud_logs_url: Optional[str] = None) -> None:
-        if self.status not in (SubTaskStatus.WAITING, SubTaskStatus.IN_PROGRESS):
+        # FAILED may restart: a failed batch is re-run by re-sending its own
+        # inputs, and its completion un-fails the parent Task (ADR-0055).
+        if self.status is SubTaskStatus.COMPLETE:
             raise ValueError(f"cannot start subtask in status {self.status}")
         if self.job_started is None:
             self.job_started = datetime.now(timezone.utc)
         self.status = SubTaskStatus.IN_PROGRESS
+        self.job_completed = None
         if cloud_logs_url is not None:
             self.cloud_logs_url = cloud_logs_url
 
@@ -53,3 +69,5 @@ class SubTask:
         self.status = SubTaskStatus.FAILED
         self.job_completed = datetime.now(timezone.utc)
         self.outputs = {"error": str(error)}
+        if isinstance(error, SubTaskFailure) and error.details is not None:
+            self.outputs.update(error.details)

domain/tasks/tasks.py

@@ -70,11 +70,12 @@ class Task:
     def recalculate_from_subtasks(self, statuses: list[SubTaskStatus]) -> None:
         """Recompute Task.status from its SubTasks' statuses.
 
-        Rule (preserved from legacy _update_task_progress):
-          - any FAILED       → FAILED
-          - all COMPLETE     → COMPLETE
-          - any IN_PROGRESS  → IN_PROGRESS
-          - otherwise        → WAITING
+        Rule:
+          - any FAILED                  → FAILED
+          - all COMPLETE                → COMPLETE
+          - any IN_PROGRESS or COMPLETE → IN_PROGRESS (finished batches plus
+            queued batches is a run in progress, not a waiting one — ADR-0055)
+          - all WAITING                 → WAITING
 
         Empty list is a no-op (newly-created task with no subtasks).
         """
@@ -87,7 +88,10 @@ class Task:
         elif all(s is SubTaskStatus.COMPLETE for s in statuses):
             self.status = TaskStatus.COMPLETE
             self.job_completed = now
-        elif SubTaskStatus.IN_PROGRESS in statuses:
+        elif (
+            SubTaskStatus.IN_PROGRESS in statuses
+            or SubTaskStatus.COMPLETE in statuses
+        ):
             self.status = TaskStatus.IN_PROGRESS
             self.job_completed = None
         else:

infrastructure/postgres/property_table.py

@@ -56,3 +56,8 @@ class PropertyRow(SQLModel, table=True):
     # `updated_at >= 2026-06-01`, the cutoff the old pipeline predates).
     has_recommendations: Optional[bool] = Field(default=None)
     updated_at: Optional[datetime] = Field(default=None)
+
+    # The Modelling Run filter resolution's base-set exclusion flag (ADR-0056).
+    # The legacy property_type / built_form columns are deliberately NOT
+    # mirrored: overrides and EPCs own those facts (ADR-0056 amendment).
+    marked_for_deletion: Optional[bool] = Field(default=None)

orchestration/task_orchestrator.py

@@ -2,7 +2,7 @@ from typing import Any, Callable, Optional
 from uuid import UUID
 
 from domain.tasks.subtasks import SubTask
-from domain.tasks.tasks import Source, Task, TaskStatus
+from domain.tasks.tasks import Source, Task
 from repositories.tasks.subtask_repository import SubTaskRepository
 from repositories.tasks.task_repository import TaskRepository
 from utilities.private import private
@@ -146,11 +146,10 @@ class TaskOrchestrator:
     @private
     def _cascade(self, task_id: UUID) -> None:
         task = self._tasks.get(task_id)
-        # FAILED is terminal: once any SubTask has failed the Task is failed and
-        # stays failed, so skip the (potentially large) sibling roll-up entirely —
-        # no need to list and re-check the SubTasks.
-        if task.status is TaskStatus.FAILED:
-            return
+        # Always recompute, even from FAILED: a failed batch SubTask may be
+        # re-run (re-sending its own inputs), and its completion must be able
+        # to un-fail the parent (ADR-0055). While any child is still failed,
+        # the recompute keeps the Task failed anyway.
         statuses = [s.status for s in self._subtasks.list_by_task(task_id)]
         task.recalculate_from_subtasks(statuses)
         self._tasks.save(task)

tests/applications/modelling_e2e/test_handler.py

@@ -378,6 +378,139 @@ def test_handler_creates_one_child_subtask_per_property_id() -> None:
     assert [i["property_id"] for i in inputs_per_subtask] == [pid1, pid2, pid3]
 
 
+def test_attach_mode_models_the_batch_without_child_subtasks() -> None:
+    """A batch carrying task_id + subtask_id (a Modelling Run, ADR-0055) runs
+    entirely under the distributor's pre-created sub_task: no per-property
+    child SubTasks are created, and the batch still persists."""
+    # Arrange
+    pid1, pid2 = 111, 222
+    mock_engine = _engine_mock([pid1, pid2], [1001, 1002], [POSTCODE, POSTCODE])
+    mock_orch = _mock_orchestrator()
+    task_id = uuid4()
+
+    with ExitStack() as stack:
+        stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine)
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.EpcClientService")
+        ).return_value.get_by_uprn.return_value = MagicMock()
+        stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository"))
+        stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient"))
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.overlays_from", return_value=[])
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader")
+        ).return_value.overrides_for_many.return_value = {}
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
+        ).return_value.get_many.return_value = [MagicMock()]
+        stack.enter_context(patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides"))
+        stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.run_modelling", return_value=_plan_mock())
+        )
+        MockUoW = stack.enter_context(patch("applications.modelling_e2e.handler.PostgresUnitOfWork"))
+        mock_uow = MagicMock()
+        MockUoW.return_value.__enter__.return_value = mock_uow
+        MockUoW.return_value.__exit__.return_value = False
+
+        # Act
+        from applications.modelling_e2e.handler import handler
+        handler.__wrapped__(  # type: ignore[attr-defined]
+            {"task_id": str(task_id), "subtask_id": str(uuid4()),
+             "property_ids": [pid1, pid2], "portfolio_id": PORTFOLIO_ID,
+             "scenario_id": SCENARIO_ID, "refetch_solar": False, "dry_run": False},
+            None, mock_orch, task_id,
+        )
+
+    # Assert — no child SubTasks; the whole batch persisted in the one UoW
+    mock_orch.run_subtasks.assert_not_called()
+    plan_requests = mock_uow.plan.save_batch.call_args.args[0]
+    assert [r.property_id for r in plan_requests] == [pid1, pid2]
+
+
+def test_attach_mode_partial_failure_persists_successes_then_records_failure() -> None:
+    """In attach mode a failing property doesn't stop its siblings: the batch
+    flushes the successes, then raises SubTaskFailure carrying
+    {succeeded, failed} so the sub_task record holds the outcome (ADR-0055)."""
+    # Arrange — property 222's EPC fetch blows up; 111 models fine
+    pid_ok, pid_bad = 111, 222
+    mock_engine = _engine_mock([pid_ok, pid_bad], [1001, 1002], [POSTCODE, POSTCODE])
+    mock_orch = _mock_orchestrator()
+    task_id = uuid4()
+
+    with ExitStack() as stack:
+        stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine)
+        )
+        epc_client = stack.enter_context(
+            patch("applications.modelling_e2e.handler.EpcClientService")
+        ).return_value
+        epc_client.get_by_uprn.side_effect = [
+            MagicMock(),
+            RuntimeError("gov API exploded"),
+        ]
+        stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository"))
+        stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient"))
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.overlays_from", return_value=[])
+        )
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader")
+        ).return_value.overrides_for_many.return_value = {}
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
+        ).return_value.get_many.return_value = [MagicMock()]
+        stack.enter_context(patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides"))
+        stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
+        stack.enter_context(
+            patch("applications.modelling_e2e.handler.run_modelling", return_value=_plan_mock())
+        )
+        MockUoW = stack.enter_context(patch("applications.modelling_e2e.handler.PostgresUnitOfWork"))
+        mock_uow = MagicMock()
+        MockUoW.return_value.__enter__.return_value = mock_uow
+        MockUoW.return_value.__exit__.return_value = False
+
+        # Act
+        from applications.modelling_e2e.handler import handler
+        from domain.tasks.subtasks import SubTaskFailure
+
+        with pytest.raises(SubTaskFailure) as raised:
+            handler.__wrapped__(  # type: ignore[attr-defined]
+                {"task_id": str(task_id), "subtask_id": str(uuid4()),
+                 "property_ids": [pid_ok, pid_bad], "portfolio_id": PORTFOLIO_ID,
+                 "scenario_id": SCENARIO_ID, "refetch_solar": False,
+                 "dry_run": False},
+                None, mock_orch, task_id,
+            )
+
+    # Assert — the success flushed before the failure was recorded
+    plan_requests = mock_uow.plan.save_batch.call_args.args[0]
+    assert [r.property_id for r in plan_requests] == [pid_ok]
+    details = raised.value.details
+    assert details is not None
+    assert details["succeeded"] == 1
+    assert details["failed"] == [
+        {"property_id": pid_bad, "error": "gov API exploded"}
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Lodged EPC path
 # ---------------------------------------------------------------------------

tests/backend/app/modelling/test_batching.py

@@ -0,0 +1,43 @@
+from backend.app.modelling.batching import pack_postcode_batches
+from backend.app.modelling.property_filters import FilteredProperty
+
+
+def _properties(postcode: str, count: int, start_id: int) -> list[FilteredProperty]:
+    return [
+        FilteredProperty(property_id=start_id + i, postcode=postcode)
+        for i in range(count)
+    ]
+
+
+def test_postcodes_are_never_split_across_batches() -> None:
+    # arrange — 30 + 30 + 10 with a cap of 50: the second postcode won't fit
+    # alongside the first, the third rides with the second
+    properties = (
+        _properties("B93 8SU", 30, start_id=0)
+        + _properties("M20 4TF", 30, start_id=100)
+        + _properties("SW1A 1AA", 10, start_id=200)
+    )
+
+    # act
+    batches = pack_postcode_batches(properties, batch_size=50)
+
+    # assert
+    assert [len(b) for b in batches] == [30, 40]
+    for batch in batches:
+        for postcode in {p.postcode for p in batch}:
+            in_batch = [p for p in batch if p.postcode == postcode]
+            everywhere = [p for p in properties if p.postcode == postcode]
+            assert len(in_batch) == len(everywhere)  # whole postcode, one batch
+
+
+def test_an_oversized_postcode_becomes_its_own_batch() -> None:
+    # arrange — one postcode alone exceeds the cap; neighbours are unaffected
+    properties = _properties("B93 8SU", 60, start_id=0) + _properties(
+        "M20 4TF", 20, start_id=100
+    )
+
+    # act
+    batches = pack_postcode_batches(properties, batch_size=50)
+
+    # assert
+    assert [len(b) for b in batches] == [60, 20]

tests/backend/app/modelling/test_property_filters.py

@@ -0,0 +1,343 @@
+from typing import Optional
+
+from sqlalchemy import Engine, text
+from sqlmodel import Session
+
+from backend.app.modelling.property_filters import (
+    FilteredProperty,
+    PropertyGroupFilters,
+    resolve_filtered_property_ids,
+)
+from infrastructure.postgres.epc_property_table import EpcPropertyModel
+from infrastructure.postgres.property_override_table import PropertyOverrideRow
+from infrastructure.postgres.property_table import PropertyRow
+
+
+def _seed_property(
+    session: Session,
+    *,
+    portfolio_id: int,
+    postcode: str = "B93 8SU",
+    marked_for_deletion: bool = False,
+) -> int:
+    row = PropertyRow(
+        portfolio_id=portfolio_id,
+        postcode=postcode,
+        marked_for_deletion=marked_for_deletion,
+    )
+    session.add(row)
+    session.commit()
+    session.refresh(row)
+    assert row.id is not None
+    return row.id
+
+
+def _seed_override(
+    session: Session,
+    *,
+    property_id: int,
+    component: str,
+    value: str,
+    building_part: int = 0,
+) -> None:
+    session.add(
+        PropertyOverrideRow(
+            property_id=property_id,
+            portfolio_id=814,
+            building_part=building_part,
+            override_component=component,
+            override_value=value,
+            original_spreadsheet_description=value,
+        )
+    )
+    session.commit()
+
+
+def _seed_epc(
+    session: Session,
+    *,
+    property_id: int,
+    source: str = "lodged",
+    property_type: Optional[str] = None,
+    built_form: Optional[str] = None,
+    dwelling_type: str = "Mid-terrace house",
+) -> None:
+    session.add(
+        EpcPropertyModel(
+            property_id=property_id,
+            portfolio_id=814,
+            source=source,
+            dwelling_type=dwelling_type,
+            property_type=property_type,
+            built_form=built_form,
+            # Minimal NOT NULL ballast — irrelevant to filter resolution.
+            tenure="rental (social)",
+            transaction_type="assessment for green deal",
+            inspection_date="2024-01-01",
+            total_floor_area_m2=80.0,
+            solar_water_heating=False,
+            has_hot_water_cylinder=False,
+            has_fixed_air_conditioning=False,
+            door_count=1,
+            wet_rooms_count=1,
+            extensions_count=0,
+            heated_rooms_count=4,
+            open_chimneys_count=0,
+            habitable_rooms_count=4,
+            insulated_door_count=0,
+            cfl_fixed_lighting_bulbs_count=0,
+            led_fixed_lighting_bulbs_count=4,
+            incandescent_fixed_lighting_bulbs_count=0,
+            energy_gas_connection_available=True,
+            energy_meter_type="single",
+            energy_pv_battery_count=0,
+            energy_wind_turbines_count=0,
+            energy_gas_smart_meter_present=False,
+            energy_is_dwelling_export_capable=False,
+            energy_wind_turbines_terrain_type="",
+            energy_electricity_smart_meter_present=False,
+            energy_pv_diverter_present=False,
+            ventilation_present=False,
+        )
+    )
+    session.commit()
+
+
+def test_no_filters_resolves_the_whole_portfolio_minus_deletions(
+    db_engine: Engine,
+) -> None:
+    # arrange
+    with Session(db_engine) as session:
+        kept = _seed_property(session, portfolio_id=814, postcode="B93 8SU")
+        deleted = _seed_property(
+            session, portfolio_id=814, postcode="M20 4TF", marked_for_deletion=True
+        )
+        _other_portfolio = _seed_property(session, portfolio_id=999)
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session, portfolio_id=814, filters=PropertyGroupFilters()
+        )
+
+    # assert
+    assert resolved == [FilteredProperty(property_id=kept, postcode="B93 8SU")]
+    assert deleted not in [p.property_id for p in resolved]
+
+
+def test_property_type_override_beats_the_lodged_epc(db_engine: Engine) -> None:
+    # arrange — both properties' lodged EPCs say House; one is overridden
+    with Session(db_engine) as session:
+        overridden = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=overridden, property_type="House")
+        _seed_override(
+            session,
+            property_id=overridden,
+            component="property_type",
+            value="Bungalow",
+        )
+        plain = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=plain, property_type="House")
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Bungalow"]),
+        )
+
+    # assert
+    assert [p.property_id for p in resolved] == [overridden]
+
+
+def test_epc_property_type_prefers_lodged_over_predicted(db_engine: Engine) -> None:
+    # arrange — one property has both sources; the other only a prediction
+    with Session(db_engine) as session:
+        both = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=both, source="lodged", property_type="House")
+        _seed_epc(
+            session, property_id=both, source="predicted", property_type="Bungalow"
+        )
+        predicted_only = _seed_property(session, portfolio_id=814)
+        _seed_epc(
+            session,
+            property_id=predicted_only,
+            source="predicted",
+            property_type="Bungalow",
+        )
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Bungalow"]),
+        )
+
+    # assert — 'both' resolves House (lodged wins); only the prediction-backed
+    # property is a Bungalow
+    assert [p.property_id for p in resolved] == [predicted_only]
+
+
+def test_epc_numeric_property_type_codes_map_to_labels(db_engine: Engine) -> None:
+    # arrange — RdSAP code 1 = Bungalow
+    with Session(db_engine) as session:
+        coded = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=coded, property_type="1")
+        _house = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=_house, property_type="0")
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Bungalow"]),
+        )
+
+    # assert
+    assert [p.property_id for p in resolved] == [coded]
+
+
+def test_epc_without_property_type_falls_back_to_dwelling_type(
+    db_engine: Engine,
+) -> None:
+    # arrange — cert lodged with no property_type column, only dwelling_type
+    with Session(db_engine) as session:
+        fallback = _seed_property(session, portfolio_id=814)
+        _seed_epc(
+            session,
+            property_id=fallback,
+            property_type=None,
+            dwelling_type="Bungalow",
+        )
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Bungalow"]),
+        )
+
+    # assert
+    assert [p.property_id for p in resolved] == [fallback]
+
+
+def test_legacy_property_columns_are_ignored(db_engine: Engine) -> None:
+    """property.property_type / built_form are legacy: with no override and no
+    EPC the property is "Unknown", whatever the legacy columns say (ADR-0056
+    amendment)."""
+    # arrange — no override, no EPC rows; the FE-owned legacy column exists in
+    # prod but is no longer mirrored, so add it here and set a value that must
+    # not count
+    with Session(db_engine) as session:
+        legacy = _seed_property(session, portfolio_id=814)
+        session.connection().execute(
+            text("ALTER TABLE property ADD COLUMN IF NOT EXISTS property_type TEXT")
+        )
+        session.connection().execute(
+            text("UPDATE property SET property_type = 'Bungalow' WHERE id = :id"),
+            {"id": legacy},
+        )
+        session.commit()
+
+        # act
+        as_bungalow = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Bungalow"]),
+        )
+        as_unknown = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Unknown"]),
+        )
+
+    # assert
+    assert as_bungalow == []
+    assert [p.property_id for p in as_unknown] == [legacy]
+
+
+def test_unknown_is_a_selectable_property_type_bucket(db_engine: Engine) -> None:
+    # arrange — one property resolvable at no level, one resolvable
+    with Session(db_engine) as session:
+        unknowable = _seed_property(session, portfolio_id=814)
+        typed = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=typed, property_type="House")
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(property_types=["Unknown"]),
+        )
+
+    # assert
+    assert [p.property_id for p in resolved] == [unknowable]
+
+
+def test_built_form_filter_resolves_with_the_same_precedence(
+    db_engine: Engine,
+) -> None:
+    # arrange — both lodged as code 4 (Mid-Terrace); one overridden to Detached
+    with Session(db_engine) as session:
+        overridden = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=overridden, built_form="4")
+        _seed_override(
+            session,
+            property_id=overridden,
+            component="built_form_type",
+            value="Detached",
+        )
+        mid_terrace = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=mid_terrace, built_form="4")
+        semi = _seed_property(session, portfolio_id=814)
+        _seed_epc(session, property_id=semi, built_form="2")
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(built_forms=["Detached", "Mid-Terrace"]),
+        )
+
+    # assert — the override wins for one; the code maps for the other
+    assert [p.property_id for p in resolved] == [overridden, mid_terrace]
+
+
+def test_filters_combine_with_and(db_engine: Engine) -> None:
+    # arrange — right postcode + right type, right postcode + wrong type,
+    # wrong postcode + right type
+    with Session(db_engine) as session:
+        match = _seed_property(session, portfolio_id=814, postcode="B93 8SU")
+        _seed_epc(session, property_id=match, property_type="House")
+        wrong_type = _seed_property(session, portfolio_id=814, postcode="B93 8SU")
+        _seed_epc(session, property_id=wrong_type, property_type="Flat")
+        wrong_postcode = _seed_property(session, portfolio_id=814, postcode="M20 4TF")
+        _seed_epc(session, property_id=wrong_postcode, property_type="House")
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(
+                postcodes=["B93 8SU"], property_types=["House"]
+            ),
+        )
+
+    # assert
+    assert [p.property_id for p in resolved] == [match]
+
+
+def test_postcode_filter_matches_exactly(db_engine: Engine) -> None:
+    # arrange
+    with Session(db_engine) as session:
+        in_area = _seed_property(session, portfolio_id=814, postcode="B93 8SU")
+        _elsewhere = _seed_property(session, portfolio_id=814, postcode="M20 4TF")
+
+        # act
+        resolved = resolve_filtered_property_ids(
+            session,
+            portfolio_id=814,
+            filters=PropertyGroupFilters(postcodes=["B93 8SU"]),
+        )
+
+    # assert
+    assert [p.property_id for p in resolved] == [in_area]

tests/backend/app/modelling/test_trigger_run_router.py

@@ -0,0 +1,197 @@
+"""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 _trigger(api: Api, task: Task, scenario_ids: list[int]) -> Any:
+    return api.client.post(
+        "/v1/modelling/trigger-run",
+        json={
+            "task_id": str(task.id),
+            "portfolio_id": PORTFOLIO_ID,
+            "scenario_ids": scenario_ids,
+            "filters": {},
+        },
+    )
+
+
+def test_trigger_run_rejects_a_task_that_already_has_subtasks(api: Api) -> None:
+    """A blind retry or double-submit must not double the fan-out (ADR-0055):
+    the app checks the task's progress instead of re-POSTing."""
+    # arrange — a task that has already been distributed
+    task = api.seed_task()
+    scenario = api.seed_scenario()
+    api.seed_property()
+    assert _trigger(api, task, [scenario]).status_code == 202
+    already_sent = len(api.sent_bodies)
+
+    # act
+    response = _trigger(api, task, [scenario])
+
+    # assert — refused, nothing new created or sent
+    assert response.status_code == 409
+    assert len(api.sent_bodies) == already_sent
+    assert len(api.subtasks_for(task)) == 1
+
+
+def test_trigger_run_rejects_filters_that_match_no_properties(api: Api) -> None:
+    """Zero resolved properties is a caller error (ADR-0055): a task with no
+    sub_tasks could never roll up to complete, and the app's preview shows the
+    same zero from the same rule before POSTing."""
+    # arrange — a portfolio with no properties at all
+    task = api.seed_task()
+    scenario = api.seed_scenario()
+
+    # act
+    response = _trigger(api, task, [scenario])
+
+    # assert
+    assert response.status_code == 400
+    assert api.sent_bodies == []
+    assert api.subtasks_for(task) == []
+
+
+def test_trigger_run_rejects_scenarios_outside_the_portfolio(api: Api) -> None:
+    # arrange — one valid scenario, one belonging to a different portfolio
+    task = api.seed_task()
+    ours = api.seed_scenario()
+    theirs = api.seed_scenario(portfolio_id=999)
+    api.seed_property()
+
+    # act
+    response = _trigger(api, task, [ours, theirs])
+
+    # assert — refused outright; nothing partially distributed
+    assert response.status_code == 400
+    assert api.sent_bodies == []
+    assert api.subtasks_for(task) == []
+
+
+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"]

tests/domain/tasks/test_subtasks.py

@@ -2,7 +2,7 @@ from uuid import uuid4
 
 import pytest
 
-from domain.tasks.subtasks import SubTask, SubTaskStatus
+from domain.tasks.subtasks import SubTask, SubTaskFailure, SubTaskStatus
 
 
 def test_create_subtask_starts_waiting() -> None:
@@ -49,7 +49,7 @@ def test_start_is_idempotent_from_in_progress() -> None:
     assert st.cloud_logs_url == "https://other"
 
 
-def test_start_rejects_from_terminal_status() -> None:
+def test_start_rejects_from_complete() -> None:
     # arrange
     st = SubTask.create(task_id=uuid4())
     st.complete()
@@ -58,6 +58,20 @@ def test_start_rejects_from_terminal_status() -> None:
         st.start()
 
 
+def test_start_restarts_a_failed_subtask_for_a_rerun() -> None:
+    # arrange
+    st = SubTask.create(task_id=uuid4())
+    st.start()
+    st.fail(RuntimeError("degenerate prediction"))
+
+    # act
+    st.start()
+
+    # assert
+    assert st.status is SubTaskStatus.IN_PROGRESS
+    assert st.job_completed is None
+
+
 def test_complete_marks_outputs_and_job_completed() -> None:
     # arrange
     st = SubTask.create(task_id=uuid4())
@@ -81,6 +95,31 @@ def test_complete_without_result_leaves_outputs_unset() -> None:
     assert st.outputs is None
 
 
+def test_fail_with_subtask_failure_records_structured_details() -> None:
+    # arrange
+    st = SubTask.create(task_id=uuid4())
+    st.start()
+
+    # act
+    st.fail(
+        SubTaskFailure(
+            "1 of 2 properties failed",
+            details={
+                "succeeded": 1,
+                "failed": [{"property_id": 9, "error": "degenerate prediction"}],
+            },
+        )
+    )
+
+    # assert
+    assert st.status is SubTaskStatus.FAILED
+    assert st.outputs == {
+        "error": "1 of 2 properties failed",
+        "succeeded": 1,
+        "failed": [{"property_id": 9, "error": "degenerate prediction"}],
+    }
+
+
 def test_fail_records_error_in_outputs() -> None:
     # arrange
     st = SubTask.create(task_id=uuid4())

tests/domain/tasks/test_tasks.py

@@ -105,6 +105,18 @@ def test_recalculate_any_in_progress_marks_in_progress() -> None:
     assert t.job_completed is None
 
 
+def test_recalculate_complete_and_waiting_mix_marks_in_progress() -> None:
+    # arrange
+    t = Task.create(task_source="manual:test")
+
+    # act
+    t.recalculate_from_subtasks([SubTaskStatus.COMPLETE, SubTaskStatus.WAITING])
+
+    # assert
+    assert t.status is TaskStatus.IN_PROGRESS
+    assert t.job_completed is None
+
+
 def test_recalculate_all_complete_marks_complete() -> None:
     # arrange
     t = Task.create(task_source="manual:test")

tests/orchestration/test_task_orchestrator.py

@@ -263,9 +263,32 @@ def test_run_subtasks_isolates_a_failing_item_and_continues(
     assert harness.tasks.get(task.id).status is TaskStatus.FAILED
 
 
-def test_cascade_short_circuits_once_task_already_failed(harness: Harness) -> None:
-    """Once the Task is FAILED, completing another SubTask leaves it FAILED — the
-    terminal state is not recomputed away."""
+def test_completing_a_rerun_failed_subtask_unfails_the_task(
+    harness: Harness,
+) -> None:
+    """A failed batch is re-run by re-sending its own inputs; when the re-run
+    completes, the parent Task recomputes to COMPLETE (ADR-0055)."""
+    # arrange — two children: one failed (task FAILED), the other complete
+    task, failed_batch = harness.orchestrator.create_task_with_subtask(
+        task_source="manual:test"
+    )
+    healthy_batch = harness.orchestrator.create_child_subtask(task.id)
+    harness.orchestrator.complete_subtask(healthy_batch.id)
+    harness.orchestrator.fail_subtask(failed_batch.id, RuntimeError("boom"))
+    assert harness.tasks.get(task.id).status is TaskStatus.FAILED
+
+    # act — the re-run of the failed batch
+    harness.orchestrator.run_subtask(failed_batch.id, work=lambda: "fixed")
+
+    # assert — no failed children remain, so the task un-fails to COMPLETE
+    task_after = harness.tasks.get(task.id)
+    assert task_after.status is TaskStatus.COMPLETE
+    assert task_after.job_completed is not None
+
+
+def test_task_stays_failed_while_a_failed_subtask_remains(harness: Harness) -> None:
+    """Completing a *sibling* does not un-fail the Task — only re-running the
+    failed SubTask itself can (see the re-run test above)."""
     # arrange — two children; fail the first so the task is FAILED
     task, coordinator = harness.orchestrator.create_task_with_subtask(
         task_source="manual:test"

tests/utilities/aws_lambda/test_task_handler.py

@@ -5,10 +5,11 @@ from typing import Any
 from uuid import UUID
 
 import pytest
-from sqlalchemy import Engine
+from sqlalchemy import Engine, text
 from sqlmodel import Session
 
 from domain.tasks.subtasks import SubTaskStatus
+from domain.tasks.subtasks import SubTaskFailure
 from domain.tasks.tasks import Source
 from orchestration.task_orchestrator import TaskOrchestrator
 from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
@@ -48,12 +49,8 @@ def test_task_handler_records_cloudwatch_url_on_subtask(
 ) -> None:
     # arrange
     monkeypatch.setenv("AWS_REGION", "eu-west-2")
-    monkeypatch.setenv(
-        "AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e"
-    )
-    monkeypatch.setenv(
-        "AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123"
-    )
+    monkeypatch.setenv("AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e")
+    monkeypatch.setenv("AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123")
 
     @task_handler(
         task_source="modelling_e2e",
@@ -91,7 +88,10 @@ def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true(
         pass_task_orchestrator=True,
     )
     def _handler(
-        body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID
+        body: dict[str, Any],
+        context: Any,
+        orchestrator: TaskOrchestrator,
+        task_id: UUID,
     ) -> None:
         received.append((orchestrator, task_id))
 
@@ -117,11 +117,7 @@ def test_task_handler_reports_an_ordinarily_failing_record_for_redelivery(
     def handler(body: dict[str, Any], context: Any) -> None:
         raise RuntimeError("transient failure")
 
-    event = {
-        "Records": [
-            {"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}
-        ]
-    }
+    event = {"Records": [{"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}]}
 
     # Act
     result = handler(event, context=None)
@@ -144,11 +140,7 @@ def test_task_handler_does_not_requeue_a_record_failing_non_retriably(
     def handler(body: dict[str, Any], context: Any) -> None:
         raise NonRetriableTaskError("job logged but write-back failed")
 
-    event = {
-        "Records": [
-            {"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}
-        ]
-    }
+    event = {"Records": [{"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}]}
 
     # Act
     result = handler(event, context=None)
@@ -161,6 +153,100 @@ def test_task_handler_does_not_requeue_a_record_failing_non_retriably(
     assert failed.outputs == {"error": "job logged but write-back failed"}
 
 
+def test_task_handler_attaches_to_a_supplied_task_and_subtask(
+    harness: Harness, db_engine: Engine
+) -> None:
+    """A message carrying task_id + subtask_id (a Modelling Run batch,
+    ADR-0055) runs under the distributor's pre-created sub_task — no new Task
+    or SubTask rows are created."""
+    # arrange — the distributor's pre-created task + batch sub_task
+    task, subtask = harness.orchestrator.create_task_with_subtask(
+        task_source="app:modelling_run", inputs={"property_ids": [1, 2]}
+    )
+    received: list[UUID] = []
+
+    @task_handler(
+        task_source="modelling_e2e",
+        source=Source.PROPERTY,
+        orchestrator_cm=harness.factory,
+        pass_task_orchestrator=True,
+    )
+    def handler(
+        body: dict[str, Any],
+        context: Any,
+        orchestrator: TaskOrchestrator,
+        task_id: UUID,
+    ) -> None:
+        received.append(task_id)
+
+    event = {
+        "Records": [
+            {
+                "messageId": "m-1",
+                "body": (
+                    f'{{"task_id": "{task.id}", "subtask_id": "{subtask.id}", '
+                    f'"property_ids": [1, 2]}}'
+                ),
+            }
+        ]
+    }
+
+    # act
+    result = handler(event, context=None)
+
+    # assert — ran under the supplied ids, sub_task completed, nothing created
+    assert received == [task.id]
+    assert result["tasks"] == [{"task_id": str(task.id), "subtask_id": str(subtask.id)}]
+    assert harness.subtasks.get(subtask.id).status.value == "complete"
+    with db_engine.connect() as conn:
+        assert conn.execute(text("SELECT count(*) FROM tasks")).scalar_one() == 1
+        assert conn.execute(text("SELECT count(*) FROM sub_task")).scalar_one() == 1
+
+
+def test_recorded_failure_fails_the_subtask_without_an_sqs_retry(
+    harness: Harness,
+) -> None:
+    """A SubTaskFailure is a database record, not retry fuel (ADR-0055): the
+    sub_task fails with the structured details, but the message is NOT
+    reported to SQS as a batch item failure."""
+    # arrange
+    task, subtask = harness.orchestrator.create_task_with_subtask(
+        task_source="app:modelling_run", inputs={"property_ids": [1, 2]}
+    )
+
+    @task_handler(
+        task_source="modelling_e2e",
+        source=Source.PROPERTY,
+        orchestrator_cm=harness.factory,
+    )
+    def handler(body: dict[str, Any], context: Any) -> None:
+        raise SubTaskFailure(
+            "1 of 2 properties failed",
+            details={"succeeded": 1, "failed": [{"property_id": 2, "error": "x"}]},
+        )
+
+    event = {
+        "Records": [
+            {
+                "messageId": "m-1",
+                "body": (
+                    f'{{"task_id": "{task.id}", "subtask_id": "{subtask.id}", '
+                    f'"property_ids": [1, 2]}}'
+                ),
+            }
+        ]
+    }
+
+    # act
+    result = handler(event, context=None)
+
+    # assert — recorded, not retried
+    assert result["batchItemFailures"] == []
+    failed = harness.subtasks.get(subtask.id)
+    assert failed.status.value == "failed"
+    assert failed.outputs is not None and failed.outputs["succeeded"] == 1
+
+
 def test_task_handler_leaves_cloudwatch_url_unset_outside_lambda(
     harness: Harness, monkeypatch: pytest.MonkeyPatch
 ) -> None:

utilities/aws_lambda/task_handler.py

@@ -9,9 +9,11 @@ import logging
 from contextlib import AbstractContextManager
 from functools import wraps
 from typing import Any, Callable, Optional, cast
+from uuid import UUID
 
 from utilities.aws_lambda.cloud_logs import cloudwatch_url
 from utilities.aws_lambda.default_orchestrator import default_orchestrator
+from domain.tasks.subtasks import SubTaskFailure
 from domain.tasks.tasks import Source
 from orchestration.task_orchestrator import TaskOrchestrator
 
@@ -61,31 +63,40 @@ def task_handler(
 
                 for record in _records(event):
                     body = _parse_body(record)
-                    raw_source_id = body.get(source.value)
-                    source_id = (
-                        str(raw_source_id) if raw_source_id is not None else None
-                    )
 
-                    task, subtask = orchestrator.create_task_with_subtask(
-                        task_source=task_source,
-                        inputs=body,
-                        source=source,
-                        source_id=source_id,
-                    )
+                    # Attach mode (ADR-0055): a Modelling Run batch carries the
+                    # app-owned task_id and its distributor-pre-created
+                    # subtask_id — run under those; create nothing.
+                    supplied = _supplied_ids(body)
+                    source_id: Optional[str] = None
+                    if supplied is not None:
+                        task_id, subtask_id = supplied
+                    else:
+                        raw_source_id = body.get(source.value)
+                        source_id = (
+                            str(raw_source_id) if raw_source_id is not None else None
+                        )
+                        task, subtask = orchestrator.create_task_with_subtask(
+                            task_source=task_source,
+                            inputs=body,
+                            source=source,
+                            source_id=source_id,
+                        )
+                        task_id, subtask_id = task.id, subtask.id
                     task_ids.append(
-                        {"task_id": str(task.id), "subtask_id": str(subtask.id)}
+                        {"task_id": str(task_id), "subtask_id": str(subtask_id)}
                     )
 
                     try:
                         if pass_task_orchestrator:
                             orchestrator.run_subtask(
-                                subtask.id,
-                                work=lambda: func(body, context, orchestrator, task.id),
+                                subtask_id,
+                                work=lambda: func(body, context, orchestrator, task_id),
                                 cloud_logs_url=cloud_logs_url,
                             )
                         else:
                             orchestrator.run_subtask(
-                                subtask.id,
+                                subtask_id,
                                 work=lambda: func(body, context),
                                 cloud_logs_url=cloud_logs_url,
                             )
@@ -100,11 +111,24 @@ def task_handler(
                         )
                         if "Records" not in event:
                             raise
+                    except SubTaskFailure as recorded:
+                        # A recorded failure (ADR-0055): run_subtask has already
+                        # failed the SubTask with the structured details — that
+                        # record IS the outcome. Never returned to SQS for retry;
+                        # recovery is a deliberate re-send of the sub_task's
+                        # own inputs.
+                        logger.warning(
+                            "subtask recorded failure, not retrying "
+                            "(task_source=%s subtask_id=%s): %s",
+                            task_source,
+                            subtask_id,
+                            recorded,
+                        )
                     except Exception:
                         logger.exception(
-                            "subtask failed (task_source=%s source_id=%s)",
+                            "subtask failed (task_source=%s subtask_id=%s)",
                             task_source,
-                            source_id,
+                            subtask_id,
                         )
                         if "Records" in event:
                             message_id = record.get("messageId", "")
@@ -121,6 +145,16 @@ def task_handler(
     return decorator
 
 
+def _supplied_ids(body: dict[str, Any]) -> Optional[tuple[UUID, UUID]]:
+    """The (task_id, subtask_id) an attach-mode message carries, or None for
+    the classic create-your-own-task message."""
+    raw_task_id = body.get("task_id")
+    raw_subtask_id = body.get("subtask_id")
+    if raw_task_id is None or raw_subtask_id is None:
+        return None
+    return UUID(str(raw_task_id)), UUID(str(raw_subtask_id))
+
+
 def _parse_body(record: dict[str, Any]) -> dict[str, Any]:
     raw = record.get("body", record)
     if isinstance(raw, str):

utilities/grouped_batching.py

@@ -0,0 +1,48 @@
+"""Greedy group-preserving batching.
+
+Packs items into batches of at most ``max_batch_size`` without ever splitting
+a group (items sharing a key) across batches; a single group larger than the
+cap becomes its own oversized batch. Shared by the address postcode batcher
+(``domain/addresses/postcode_batching.py``) and the Modelling Run distributor
+(``backend/app/modelling/batching.py``).
+"""
+
+from collections.abc import Callable, Hashable, Iterable, Iterator
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+def iter_grouped_batches(
+    items: Iterable[T],
+    *,
+    key: Callable[[T], Hashable],
+    max_batch_size: int,
+) -> Iterator[list[T]]:
+    if max_batch_size < 1:
+        raise ValueError("max_batch_size must be >= 1")
+
+    groups: dict[Hashable, list[T]] = {}
+    for item in items:
+        groups.setdefault(key(item), []).append(item)
+
+    buffer: list[T] = []
+    for group in groups.values():
+        # Oversize single-key group: flush the buffer first, then dispatch
+        # the group as its own batch.
+        if len(group) >= max_batch_size:
+            if buffer:
+                yield buffer
+                buffer = []
+            yield group
+            continue
+
+        # Adding this group would overflow: flush the buffer before appending.
+        if len(buffer) + len(group) > max_batch_size:
+            yield buffer
+            buffer = []
+
+        buffer.extend(group)
+
+    if buffer:
+        yield buffer