feat(modelling): domain Scenario + ScenarioPostgresRepository (#1157)

Slice 1 of the #1157 build. The FE creates a Scenario and passes only
its id to the pipeline; the Modelling stage reads it back here.

- domain/modelling/scenario.py: thin `Scenario(id, goal, goal_value,
  budget, is_default)` — the slice the stage uses today (goal/budget for
  the Optimiser later; is_default drives plan.is_default). No phases
  (ADR-0005); legacy file-path/aggregate columns not modelled.
- infrastructure/postgres/scenario_table.py: `ScenarioRow` SQLModel
  mirror of the live `scenario` table (ADR-0017), declaring only the
  read columns; goal mapped as its string value.
- ScenarioPostgresRepository.get_many(scenario_ids) -> list[Scenario]:
  bulk read, input-order-preserving, raises on a missing id.

The method shape lives on the concrete repo for now; it is promoted to
an @abstractmethod on the port when the real orchestrator is wired and
the bare-stub instantiations retire (keeps the stubbed Modelling wiring
composing meanwhile). 2 tests, pyright strict clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

domain/modelling/scenario.py

@@ -0,0 +1,27 @@
+"""Scenario — the named retrofit brief the Modelling stage scores against.
+
+Built by a user in the scenario-builder UI and persisted before any modelling
+fires; the pipeline is handed only its id and reads it back via a
+`ScenarioRepository`. This is the thin slice the Modelling stage uses today:
+the goal + budget that the Optimiser will consume (#1160) and `is_default`
+(which drives `plan.is_default`). The legacy file-path / portfolio-aggregate
+columns are not modelled. Carries no phases — multi-phase is deferred
+(ADR-0005). See CONTEXT.md.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class Scenario:
+    """A retrofit brief: its goal, optional budget, and whether it is the
+    Property's default Scenario. `goal` / `goal_value` are the lodged target
+    (e.g. "INCREASING_EPC" → band "C"); carried for the Optimiser, not yet
+    enforced."""
+
+    id: int
+    goal: str
+    goal_value: str
+    budget: Optional[float]
+    is_default: bool

infrastructure/postgres/scenario_table.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import ClassVar, Optional
+
+from sqlmodel import Field, SQLModel
+
+from domain.modelling.scenario import Scenario
+
+
+class ScenarioRow(SQLModel, table=True):
+    """SQLModel mirror of the live ``scenario`` table (ADR-0017).
+
+    Declares only the columns the Modelling stage reads — the legacy
+    file-path columns (`trigger_file_path`, `exclusions`, …) and the
+    portfolio-level aggregates are left to the legacy SQLAlchemy model
+    (`backend/app/db/models/recommendations.py::ScenarioModel`), which still
+    owns the live reads. The physical table is the shared contract; this
+    mirror is read-only from the rebuild's side.
+
+    `goal` is a Postgres enum in production; mapped here as its string value
+    (the Modelling stage does not yet branch on it — #1160).
+    """
+
+    __tablename__: ClassVar[str] = "scenario"  # pyright: ignore[reportIncompatibleVariableOverride]
+
+    id: Optional[int] = Field(default=None, primary_key=True)
+    goal: str
+    goal_value: str
+    budget: Optional[float] = Field(default=None)
+    is_default: bool = Field(default=False)
+
+    def to_domain(self) -> Scenario:
+        if self.id is None:
+            raise ValueError("scenario row has no id")
+        return Scenario(
+            id=self.id,
+            goal=self.goal,
+            goal_value=self.goal_value,
+            budget=self.budget,
+            is_default=self.is_default,
+        )

repositories/scenario/scenario_postgres_repository.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from sqlmodel import Session, col, select
+
+from domain.modelling.scenario import Scenario
+from infrastructure.postgres.scenario_table import ScenarioRow
+from repositories.scenario.scenario_repository import ScenarioRepository
+
+
+class ScenarioPostgresRepository(ScenarioRepository):
+    """Reads the live ``scenario`` table (via the ``ScenarioRow`` mirror) and
+    maps each row to the thin domain ``Scenario`` the Modelling stage uses
+    (ADR-0017). The legacy file-path / aggregate columns are not read."""
+
+    def __init__(self, session: Session) -> None:
+        self._session = session
+
+    def get_many(self, scenario_ids: list[int]) -> list[Scenario]:
+        rows = self._session.exec(
+            select(ScenarioRow).where(col(ScenarioRow.id).in_(scenario_ids))
+        ).all()
+        by_id: dict[int, ScenarioRow] = {
+            row.id: row for row in rows if row.id is not None
+        }
+        scenarios: list[Scenario] = []
+        for scenario_id in scenario_ids:
+            row = by_id.get(scenario_id)
+            if row is None:
+                raise ValueError(f"no scenario with id {scenario_id}")
+            scenarios.append(row.to_domain())
+        return scenarios

repositories/scenario/scenario_repository.py

@@ -4,11 +4,16 @@ from abc import ABC
 
 
 class ScenarioRepository(ABC):
-    """Loads the Scenarios (and Scenario Snapshots) the Modelling stage scores
-    a Property against.
+    """Loads the Scenarios the Modelling stage scores a Property against.
 
-    Seam only at this stage (#1136): the method shape is deferred to the
-    Modelling per-service grill, where Scenario / Scenario Phase / Scenario
-    Snapshot are designed (CONTEXT.md). Declared now so the pipeline can be
-    composed end-to-end with Modelling stubbed.
+    The FE creates a Scenario in the scenario-builder and passes only its id
+    to the pipeline (#1130); the orchestrator reads it back through this port
+    at modelling time.
+
+    The concrete method shape is ``get_many(scenario_ids) -> list[Scenario]``
+    (bulk read by id, load-whole per ADR-0012), implemented by
+    ``ScenarioPostgresRepository``. It is promoted to an ``@abstractmethod``
+    here when the real ``ModellingOrchestrator`` is wired and the bare-stub
+    instantiations are retired (#1157 orchestrator slice) — until then the port
+    stays instantiable so the stubbed Modelling wiring composes.
     """

tests/repositories/scenario/test_scenario_postgres_repository.py

@@ -0,0 +1,82 @@
+"""Behaviour of the Postgres-backed ScenarioRepository: reading the Scenarios
+the Modelling stage scores a Property against, off the live ``scenario`` table.
+
+The FE creates a Scenario in the scenario-builder and passes its id to the
+pipeline (#1130); the orchestrator reads it back here at modelling time. Only
+the fields modelling uses are mapped — goal / goal_value / budget / is_default;
+the legacy file-path columns are ignored. See CONTEXT.md (Scenario) and
+ADR-0017.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from domain.modelling.scenario import Scenario
+from infrastructure.postgres.scenario_table import ScenarioRow
+from repositories.scenario.scenario_postgres_repository import (
+    ScenarioPostgresRepository,
+)
+
+
+def test_get_many_maps_live_scenario_rows_to_domain_in_input_order(
+    db_engine: Engine,
+) -> None:
+    # Arrange
+    with Session(db_engine) as session:
+        session.add(
+            ScenarioRow(
+                id=7,
+                goal="INCREASING_EPC",
+                goal_value="C",
+                budget=15000.0,
+                is_default=True,
+            )
+        )
+        session.add(
+            ScenarioRow(
+                id=9,
+                goal="INCREASING_EPC",
+                goal_value="B",
+                budget=None,
+                is_default=False,
+            )
+        )
+        session.commit()
+
+    # Act
+    with Session(db_engine) as session:
+        scenarios: list[Scenario] = ScenarioPostgresRepository(session).get_many(
+            [9, 7]
+        )
+
+    # Assert
+    assert [s.id for s in scenarios] == [9, 7]  # input order preserved
+    assert scenarios[0] == Scenario(
+        id=9, goal="INCREASING_EPC", goal_value="B", budget=None, is_default=False
+    )
+    assert scenarios[1] == Scenario(
+        id=7,
+        goal="INCREASING_EPC",
+        goal_value="C",
+        budget=15000.0,
+        is_default=True,
+    )
+
+
+def test_get_many_raises_when_a_scenario_id_is_missing(db_engine: Engine) -> None:
+    # Arrange
+    with Session(db_engine) as session:
+        session.add(
+            ScenarioRow(
+                id=7, goal="INCREASING_EPC", goal_value="C", is_default=True
+            )
+        )
+        session.commit()
+
+    # Act / Assert
+    with Session(db_engine) as session:
+        with pytest.raises(ValueError):
+            ScenarioPostgresRepository(session).get_many([7, 404])