diff --git a/.github/workflows/deploy_terraform.yml b/.github/workflows/deploy_terraform.yml index b9a4227ab..a8b23bf9e 100644 --- a/.github/workflows/deploy_terraform.yml +++ b/.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 diff --git a/CONTEXT.md b/CONTEXT.md index aa2199323..caaf3029e 100644 --- a/CONTEXT.md +++ b/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 ``, ` 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() diff --git a/applications/modelling_e2e/modelling_e2e_trigger_body.py b/applications/modelling_e2e/modelling_e2e_trigger_body.py index f9c92513d..a234ca06c 100644 --- a/applications/modelling_e2e/modelling_e2e_trigger_body.py +++ b/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 diff --git a/backend/app/config.py b/backend/app/config.py index df6a32d1d..af79b0106 100644 --- a/backend/app/config.py +++ b/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" diff --git a/backend/app/db/models/tasks.py b/backend/app/db/models/tasks.py index db1b7c04f..373a16b33 100644 --- a/backend/app/db/models/tasks.py +++ b/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 diff --git a/backend/app/main.py b/backend/app/main.py index 0dd9e489b..4b4aaaa70 100644 --- a/backend/app/main.py +++ b/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 diff --git a/backend/app/modelling/__init__.py b/backend/app/modelling/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/modelling/batching.py b/backend/app/modelling/batching.py new file mode 100644 index 000000000..785855c41 --- /dev/null +++ b/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 + ) + ) diff --git a/backend/app/modelling/property_filters.py b/backend/app/modelling/property_filters.py new file mode 100644 index 000000000..44d8f19eb --- /dev/null +++ b/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 + } diff --git a/backend/app/modelling/router.py b/backend/app/modelling/router.py new file mode 100644 index 000000000..168973392 --- /dev/null +++ b/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"} diff --git a/backend/app/modelling/run_tasks.py b/backend/app/modelling/run_tasks.py new file mode 100644 index 000000000..7d7021b04 --- /dev/null +++ b/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() diff --git a/backend/app/modelling/schemas.py b/backend/app/modelling/schemas.py new file mode 100644 index 000000000..272f5ab4d --- /dev/null +++ b/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() diff --git a/deployment/terraform/lambda/fast-api/main.tf b/deployment/terraform/lambda/fast-api/main.tf index 696f46650..26e154d6a 100644 --- a/deployment/terraform/lambda/fast-api/main.tf +++ b/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 diff --git a/docs/adr/0055-modelling-run-batches-attach-to-the-app-owned-task.md b/docs/adr/0055-modelling-run-batches-attach-to-the-app-owned-task.md new file mode 100644 index 000000000..9420e6d23 --- /dev/null +++ b/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`. diff --git a/docs/adr/0056-modelling-run-filter-resolution-is-a-shared-contract-with-the-app-preview.md b/docs/adr/0056-modelling-run-filter-resolution-is-a-shared-contract-with-the-app-preview.md new file mode 100644 index 000000000..9e13a5732 --- /dev/null +++ b/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. diff --git a/domain/addresses/postcode_batching.py b/domain/addresses/postcode_batching.py index ca4cd7529..7de6a5b43 100644 --- a/domain/addresses/postcode_batching.py +++ b/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) diff --git a/domain/tasks/subtasks.py b/domain/tasks/subtasks.py index bd49a6ec6..a0e2f985c 100644 --- a/domain/tasks/subtasks.py +++ b/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) diff --git a/domain/tasks/tasks.py b/domain/tasks/tasks.py index ead253ab6..992ae8d71 100644 --- a/domain/tasks/tasks.py +++ b/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: diff --git a/infrastructure/postgres/property_table.py b/infrastructure/postgres/property_table.py index 56d0b2fa7..e62ac958e 100644 --- a/infrastructure/postgres/property_table.py +++ b/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) diff --git a/orchestration/task_orchestrator.py b/orchestration/task_orchestrator.py index 15915d9f1..58cb22bac 100644 --- a/orchestration/task_orchestrator.py +++ b/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) diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index 80c44aeeb..f2e33187b 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/backend/__init__.py b/tests/backend/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/backend/app/__init__.py b/tests/backend/app/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/backend/app/modelling/__init__.py b/tests/backend/app/modelling/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/backend/app/modelling/test_batching.py b/tests/backend/app/modelling/test_batching.py new file mode 100644 index 000000000..635f80e0b --- /dev/null +++ b/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] diff --git a/tests/backend/app/modelling/test_property_filters.py b/tests/backend/app/modelling/test_property_filters.py new file mode 100644 index 000000000..dda558547 --- /dev/null +++ b/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] diff --git a/tests/backend/app/modelling/test_trigger_run_router.py b/tests/backend/app/modelling/test_trigger_run_router.py new file mode 100644 index 000000000..30ebb17b7 --- /dev/null +++ b/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"] diff --git a/tests/domain/tasks/test_subtasks.py b/tests/domain/tasks/test_subtasks.py index 8cee44960..8d000ba96 100644 --- a/tests/domain/tasks/test_subtasks.py +++ b/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()) diff --git a/tests/domain/tasks/test_tasks.py b/tests/domain/tasks/test_tasks.py index ba82412b1..ff30a3671 100644 --- a/tests/domain/tasks/test_tasks.py +++ b/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") diff --git a/tests/orchestration/test_task_orchestrator.py b/tests/orchestration/test_task_orchestrator.py index 8767c8c15..fb0cd8e6c 100644 --- a/tests/orchestration/test_task_orchestrator.py +++ b/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" diff --git a/tests/utilities/aws_lambda/test_task_handler.py b/tests/utilities/aws_lambda/test_task_handler.py index 10d82979d..1d9ad84cd 100644 --- a/tests/utilities/aws_lambda/test_task_handler.py +++ b/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: diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index 0298f3814..de420eb12 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/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): diff --git a/utilities/grouped_batching.py b/utilities/grouped_batching.py new file mode 100644 index 000000000..b5e42b3f9 --- /dev/null +++ b/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