feat(modelling): run_one console entrypoint for DB-less inspection

Slice 3. `harness.console.run_one(epc, goal_band=...)` wires the full
AraFirstRunPipeline against in-memory fakes — no Postgres, no network —
runs one property, prints the sense-check table, and returns the Plan
for interactive poking from a REPL at the worktree root. Defaults to the
committed harness sample catalogue.

Refactors the slice-1 integration test to delegate to run_one (dropping
~70 lines of duplicated wiring + the now-unused test catalogue fixture),
so it exercises the shipped entrypoint rather than a parallel copy. The
new console test covers run_one's print/return contract.

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

harness/console.py

@@ -0,0 +1,157 @@
+"""Run one property through the full First Run pipeline with no database.
+
+The interactive inspection entrypoint: hand it an `EpcPropertyData` (e.g.
+`EpcPropertyDataMapper.from_api_response(json)`), and it wires the whole
+`AraFirstRunPipeline` (Ingestion -> Baseline -> Modelling) against in-memory
+fakes — no Postgres, no network — runs it, prints the sense-check table, and
+returns the `Plan` for further poking.
+
+Dev tooling, not deployed: it reuses the in-memory test fakes, so run it from a
+REPL at the worktree root::
+
+    from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+    from harness.console import run_one
+    plan = run_one(EpcPropertyDataMapper.from_api_response(my_api_json), goal_band="C")
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Optional
+
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from domain.geospatial.coordinates import Coordinates
+from domain.modelling.plan import Plan
+from domain.modelling.scenario import Scenario
+from domain.property.property import Property, PropertyIdentity
+from domain.property_baseline.rebaseliner import StubRebaseliner
+from domain.sap10_calculator.calculator import Sap10Calculator
+from harness.plan_table import format_plan_table
+from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
+from orchestration.ingestion_orchestrator import IngestionOrchestrator
+from orchestration.modelling_orchestrator import ModellingOrchestrator
+from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
+from repositories.fuel_rates.fuel_rates_static_file_repository import (
+    FuelRatesStaticFileRepository,
+)
+from repositories.geospatial.geospatial_repository import GeospatialRepository
+from repositories.product.product_json_repository import ProductJsonRepository
+from tests.orchestration.fakes import (
+    FakeEpcRepo,
+    FakePlanRepository,
+    FakePropertyRepo,
+    FakeScenarioRepository,
+    FakeUnitOfWork,
+)
+
+DEFAULT_CATALOGUE = Path(__file__).resolve().parent / "sample_catalogue.json"
+
+_PROPERTY_ID = 1
+_SCENARIO_ID = 7
+_PORTFOLIO_ID = 1
+
+
+@dataclass
+class _Command:
+    portfolio_id: int
+    property_ids: list[int]
+    scenario_ids: list[int]
+
+
+class _FetcherReturning:
+    def __init__(self, epc: EpcPropertyData) -> None:
+        self._epc = epc
+
+    def get_by_uprn(self, uprn: int) -> Optional[EpcPropertyData]:
+        return self._epc
+
+
+class _NoCoordinates(GeospatialRepository):
+    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
+        return None  # skip the solar leg
+
+
+class _UnusedSolarFetcher:
+    def get_building_insights(
+        self, longitude: float, latitude: float
+    ) -> dict[str, Any]:  # pragma: no cover
+        return {}
+
+
+def run_one(
+    epc: EpcPropertyData,
+    *,
+    goal_band: str = "C",
+    catalogue_path: Path = DEFAULT_CATALOGUE,
+    print_table: bool = True,
+) -> Plan:
+    """Run ``epc`` through the full First Run pipeline with no database and
+    return its Plan for the default Increasing-EPC Scenario targeting
+    ``goal_band``. Prints the sense-check table unless ``print_table`` is False.
+
+    ``epc`` must carry lodged recorded-performance + the RHI block (a real lodged
+    EPC does) so the Baseline stage can run."""
+    epc_repo = FakeEpcRepo()
+    plan_repo = FakePlanRepository()
+    property_repo = FakePropertyRepo(
+        {
+            _PROPERTY_ID: Property(
+                identity=PropertyIdentity(
+                    portfolio_id=_PORTFOLIO_ID,
+                    postcode="A0 0AA",
+                    address="1 Some Street",
+                    uprn=12345,
+                )
+            )
+        },
+        epc_repo=epc_repo,
+    )
+    unit = FakeUnitOfWork(
+        property=property_repo,
+        epc=epc_repo,
+        scenario=FakeScenarioRepository(
+            {
+                _SCENARIO_ID: Scenario(
+                    id=_SCENARIO_ID,
+                    goal="Increasing EPC",
+                    goal_value=goal_band,
+                    budget=None,
+                    is_default=True,
+                )
+            }
+        ),
+        product=ProductJsonRepository(catalogue_path),
+        plan=plan_repo,
+    )
+
+    pipeline = AraFirstRunPipeline(
+        ingestion=IngestionOrchestrator(
+            unit_of_work=lambda: unit,
+            epc_fetcher=_FetcherReturning(epc),
+            geospatial_repo=_NoCoordinates(),
+            solar_fetcher=_UnusedSolarFetcher(),
+        ),
+        baseline=PropertyBaselineOrchestrator(
+            unit_of_work=lambda: unit,
+            rebaseliner=StubRebaseliner(),
+            fuel_rates=FuelRatesStaticFileRepository(),
+        ),
+        modelling=ModellingOrchestrator(
+            unit_of_work=lambda: unit,
+            calculator=Sap10Calculator(),
+            fuel_rates=FuelRatesStaticFileRepository(),
+        ),
+    )
+    pipeline.run(
+        _Command(
+            portfolio_id=_PORTFOLIO_ID,
+            property_ids=[_PROPERTY_ID],
+            scenario_ids=[_SCENARIO_ID],
+        )
+    )
+
+    plan = plan_repo.saved[(_PROPERTY_ID, _SCENARIO_ID)]
+    if print_table:
+        print("\n" + format_plan_table(plan))
+    return plan

tests/harness/test_console.py

@@ -0,0 +1,42 @@
+"""The one-property console entrypoint for interactive sense-checking."""
+
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from datatypes.epc.domain.epc import Epc
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from harness.console import run_one
+from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
+    build_epc as _build_uninsulated_cavity_and_floor_epc,
+)
+
+
+def _uninsulated_lodged_epc() -> EpcPropertyData:
+    epc = _build_uninsulated_cavity_and_floor_epc()
+    return dataclasses.replace(
+        epc,
+        energy_rating_current=57,
+        current_energy_efficiency_band=Epc.D,
+        co2_emissions_current=3.0,
+        energy_consumption_current=300,
+    )
+
+
+def test_run_one_returns_a_plan_and_prints_the_table(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    # Arrange
+    epc: EpcPropertyData = _uninsulated_lodged_epc()
+
+    # Act — run one property end-to-end with no database, against the default
+    # sample catalogue.
+    plan = run_one(epc, goal_band="C")
+
+    # Assert — a multi-measure Plan came back, and its sense-check table printed.
+    assert len(plan.measures) >= 1
+    printed: str = capsys.readouterr().out
+    assert "Plan  SAP" in printed
+    assert "cavity_wall_insulation" in printed

tests/orchestration/test_first_run_without_database.py

@@ -1,82 +1,29 @@
-"""First Run end-to-end with NO database — in-memory fakes only.
+"""First Run end-to-end with NO database, via the harness console entrypoint.
 
-The same `AraFirstRunPipeline` the Postgres integration test drives, but wired
-against a `FakeUnitOfWork` instead of a `PostgresUnitOfWork`: Ingestion ->
-Baseline -> Modelling run start-to-finish, hand off through in-memory repos, and
-produce an inspectable multi-measure Plan without a `Session` ever being opened.
-This is the harness the owner runs to sense-check recommendations interactively.
+`harness.console.run_one` wires the full AraFirstRunPipeline (Ingestion ->
+Baseline -> Modelling) against in-memory fakes. This proves the whole flow runs
+start-to-finish with no Session ever opened and yields a multi-measure Plan;
+`tests/harness/test_console.py` covers the entrypoint's print/return contract.
 """
 
 from __future__ import annotations
 
 import dataclasses
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Optional
 
 from datatypes.epc.domain.epc import Epc
 from datatypes.epc.domain.epc_property_data import EpcPropertyData
-from domain.modelling.scenario import Scenario
-from domain.property.property import Property, PropertyIdentity
-from domain.property_baseline.rebaseliner import StubRebaseliner
-from domain.sap10_calculator.calculator import Sap10Calculator
-from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
-from orchestration.ingestion_orchestrator import IngestionOrchestrator
-from orchestration.modelling_orchestrator import ModellingOrchestrator
-from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
-from repositories.fuel_rates.fuel_rates_static_file_repository import (
-    FuelRatesStaticFileRepository,
-)
-from repositories.geospatial.geospatial_repository import GeospatialRepository
-from harness.plan_table import format_plan_table
-from repositories.product.product_json_repository import ProductJsonRepository
+from harness.console import run_one
 from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
     build_epc as _build_uninsulated_cavity_and_floor_epc,
 )
-from tests.orchestration.fakes import (
-    FakeEpcRepo,
-    FakePlanRepository,
-    FakePropertyRepo,
-    FakeScenarioRepository,
-    FakeUnitOfWork,
-)
-
-_CATALOGUE = Path(__file__).resolve().parent / "fixtures/product_catalogue.json"
-
-
-@dataclass
-class _Command:
-    portfolio_id: int
-    property_ids: list[int]
-    scenario_ids: list[int]
-
-
-class _FetcherReturning:
-    def __init__(self, epc: EpcPropertyData) -> None:
-        self._epc = epc
-
-    def get_by_uprn(self, uprn: int) -> Optional[EpcPropertyData]:
-        return self._epc
-
-
-class _NoCoordinates(GeospatialRepository):
-    def coordinates_for(self, uprn: int):  # type: ignore[no-untyped-def]
-        return None  # skip the solar leg
-
-
-class _UnusedSolarFetcher:
-    def get_building_insights(
-        self, longitude: float, latitude: float
-    ) -> dict[str, Any]:  # pragma: no cover
-        return {}
 
 
 def _uninsulated_lodged_epc() -> EpcPropertyData:
     # 000490: an uninsulated cavity wall + suspended floor (loft already 300mm),
     # so the wall + floor Generators fire and the ventilation Dependency follows.
     # The calculator fixture carries no lodged recorded-performance, so we fill it
-    # in (as a real lodged EPC would) — it already carries the RHI block — so the
-    # Baseline stage can run inside the full pipeline.
+    # in (it already carries the RHI block) so the Baseline stage can run inside
+    # the full pipeline.
     epc = _build_uninsulated_cavity_and_floor_epc()
     return dataclasses.replace(
         epc,
@@ -88,66 +35,12 @@ def _uninsulated_lodged_epc() -> EpcPropertyData:
 
 
 def test_first_run_produces_a_multi_measure_plan_without_a_database() -> None:
-    # Arrange — an in-memory Property (no EPC yet; Ingestion supplies it), a
-    # default Increasing-EPC Scenario, and a file-backed product catalogue.
-    epc_repo = FakeEpcRepo()
-    plan_repo = FakePlanRepository()
-    property_repo = FakePropertyRepo(
-        {
-            10: Property(
-                identity=PropertyIdentity(
-                    portfolio_id=1,
-                    postcode="A0 0AA",
-                    address="1 Some Street",
-                    uprn=12345,
-                )
-            )
-        },
-        epc_repo=epc_repo,
-    )
-    unit: FakeUnitOfWork = FakeUnitOfWork(
-        property=property_repo,
-        epc=epc_repo,
-        scenario=FakeScenarioRepository(
-            {
-                7: Scenario(
-                    id=7,
-                    goal="Increasing EPC",
-                    goal_value="C",
-                    budget=None,
-                    is_default=True,
-                )
-            }
-        ),
-        product=ProductJsonRepository(_CATALOGUE),
-        plan=plan_repo,
-    )
-
-    pipeline = AraFirstRunPipeline(
-        ingestion=IngestionOrchestrator(
-            unit_of_work=lambda: unit,
-            epc_fetcher=_FetcherReturning(_uninsulated_lodged_epc()),
-            geospatial_repo=_NoCoordinates(),
-            solar_fetcher=_UnusedSolarFetcher(),
-        ),
-        baseline=PropertyBaselineOrchestrator(
-            unit_of_work=lambda: unit,
-            rebaseliner=StubRebaseliner(),
-            fuel_rates=FuelRatesStaticFileRepository(),
-        ),
-        modelling=ModellingOrchestrator(
-            unit_of_work=lambda: unit,
-            calculator=Sap10Calculator(),
-            fuel_rates=FuelRatesStaticFileRepository(),
-        ),
-    )
+    # Arrange
+    epc: EpcPropertyData = _uninsulated_lodged_epc()
 
     # Act — the whole First Run, no Session ever opened.
-    pipeline.run(_Command(portfolio_id=1, property_ids=[10], scenario_ids=[7]))
+    plan = run_one(epc, goal_band="C", print_table=False)
 
-    # Assert — a Plan was persisted in memory for (property 10, scenario 7),
-    # with at least one Plan Measure and a post-retrofit SAP no worse than baseline.
-    plan = plan_repo.saved[(10, 7)]
-    print("\n" + format_plan_table(plan))  # visible under `pytest -s` for sense-checking
+    # Assert — a multi-measure Plan that improves on the baseline SAP.
     assert len(plan.measures) >= 1
     assert plan.post_sap_continuous >= plan.baseline.sap_continuous