M8-T15: add thermal meter cost engine

app/main.py

@@ -2,6 +2,7 @@ import logging
 import os
 from contextlib import asynccontextmanager
 from pathlib import Path
+from typing import Callable
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import FileResponse
@@ -38,6 +39,7 @@ from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SE
 from app.services.ha_discovery import publish_discovery, publish_states
 from app.services.tibber_prices import refresh_prices
 from app.services.energy_cost import compute_closed_periods
+from app.services.meter_cost import compute_closed_periods as compute_closed_meter_cost_periods
 from app.services.warmtelink_worker import warmtelink_worker_manager
 from app.services.timezone import local_tz
 from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db
@@ -112,22 +114,44 @@ def _run_scheduled_energy_cost() -> None:
     does not crash the scheduler or affect the other background jobs.
     """
     session_local = get_session_local()
-    session: Session = session_local()
+
+    def run_scope(label: str, operation: Callable[[Session], None]) -> None:
+        """Run one best-effort scope in an isolated transaction/session."""
+        session: Session | None = None
         try:
-        compute_closed_periods(session)
+            session = session_local()
-        # After billing periods are computed, push fresh energy-cost state values
+            operation(session)
-        # to MQTT/HA.  publish_states is internally guarded by _should_publish
+        except Exception:
-        # (MQTT disabled / not connected → no-op), so this never raises due to
+            logger.exception("_run_scheduled_energy_cost: %s failed", label)
-        # unconfigured MQTT and does not block the billing job.
+            if session is not None:
                 try:
-            from app.services.ha_discovery import publish_states
+                    session.rollback()
-            publish_states(session)
                 except Exception:
-            logger.exception("_run_scheduled_energy_cost: publish_states failed (non-fatal)")
+                    # A failed cleanup must not replace the operation/factory
-    except Exception:
+                    # error or prevent the following independent scope.
-        logger.exception("_run_scheduled_energy_cost: unexpected error")
+                    logger.exception("_run_scheduled_energy_cost: %s rollback failed", label)
         finally:
+            if session is not None:
+                try:
                     session.close()
+                except Exception:
+                    # Sessions are intentionally isolated; close failures are
+                    # diagnostic only and must remain best-effort too.
+                    logger.exception("_run_scheduled_energy_cost: %s close failed", label)
+
+    # Electricity, thermal and HA publishing must not share failed transaction
+    # state or accidentally commit each other's partially-flushed changes.
+    run_scope("electricity computation", compute_closed_periods)
+    run_scope("thermal computation", compute_closed_meter_cost_periods)
+
+    def publish(session: Session) -> None:
+        # publish_states is internally guarded by _should_publish (MQTT
+        # disabled / disconnected -> no-op), but gets a clean Session anyway.
+        from app.services.ha_discovery import publish_states
+
+        publish_states(session)
+
+    run_scope("publish_states (non-fatal)", publish)
 
 
 def _run_scheduled_ha_state_publish() -> None:

app/services/meter_cost.py

@@ -0,0 +1,288 @@
+"""Thermal (WarmteLink) 15-minute cost ledger.
+
+This module deliberately does not share the electricity ledger: thermal has
+two independently-bound cumulative domains and Decimal database columns.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, date, datetime, time, timedelta
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.energy import Meter, MeterCostPeriod
+from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
+from app.services.contracts import active_contract_version_at, active_contract_versions
+from app.services.energy_cost import floor_to_quarter
+from app.services import timezone as timezone_service
+
+logger = logging.getLogger(__name__)
+
+_PERIOD = timedelta(minutes=15)
+_FRESHNESS = timedelta(seconds=120)
+_LIMITS = {"heating": Decimal("0.1"), "hot_water": Decimal("1")}
+_ACCEPTED_QUALITIES = {"valid", "unverifiable"}
+_SETTLEMENT_TIME = time(1, 5)
+
+
+def _utc(value: datetime) -> datetime:
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+
+
+def _decimal(value: Any) -> Decimal:
+    return value if isinstance(value, Decimal) else Decimal(str(value))
+
+
+def _existing(session: Session, commodity: str, start: datetime) -> MeterCostPeriod | None:
+    return session.execute(
+        select(MeterCostPeriod).where(
+            MeterCostPeriod.commodity == commodity, MeterCostPeriod.period_start == start
+        )
+    ).scalar_one_or_none()
+
+
+def _meter_at(session: Session, commodity: str, instant: datetime) -> Meter | None:
+    meters = session.execute(
+        select(Meter).where(
+            Meter.commodity == commodity,
+            Meter.started_at <= instant,
+            (Meter.ended_at.is_(None)) | (Meter.ended_at > instant),
+        )
+    ).scalars().all()
+    return meters[0] if len(meters) == 1 else None
+
+
+def _binding_at(
+    session: Session, meter: Meter, instant: datetime
+) -> tuple[MeterSourceBinding, MeterSourceChannel] | None:
+    rows = session.execute(
+        select(MeterSourceBinding, MeterSourceChannel)
+        .join(MeterSourceChannel, MeterSourceChannel.id == MeterSourceBinding.channel_id)
+        .join(MeterSource, MeterSource.id == MeterSourceChannel.source_id)
+        .where(
+            MeterSourceBinding.meter_id == meter.id,
+            MeterSourceBinding.started_at <= instant,
+            (MeterSourceBinding.ended_at.is_(None)) | (MeterSourceBinding.ended_at > instant),
+            MeterSource.kind == "warmtelink_serial",
+        )
+    ).all()
+    return rows[0] if len(rows) == 1 else None
+
+
+def _reading_at(
+    session: Session,
+    channel: MeterSourceChannel,
+    meter: Meter,
+    binding: MeterSourceBinding,
+    target: datetime,
+) -> WarmteLinkReading | None:
+    """Choose the nearest accepted reading inside this cumulative domain.
+
+    Freshness alone is insufficient: a frame immediately before a meter or
+    source hand-off belongs to a different cumulative register and must never
+    be used as the other side of a delta.
+    """
+    window_start, window_end = target - _FRESHNESS, target + _FRESHNESS
+    rows = session.execute(
+        select(WarmteLinkReading)
+        .where(
+            WarmteLinkReading.channel_id == channel.id,
+            WarmteLinkReading.recorded_at >= window_start,
+            WarmteLinkReading.recorded_at <= window_end,
+        )
+    ).scalars().all()
+    domain_start = max(_utc(meter.started_at), _utc(binding.started_at))
+    domain_ends = (meter.ended_at, binding.ended_at)
+    domain_end = min((_utc(value) for value in domain_ends if value is not None), default=None)
+    accepted = [
+        row for row in rows
+        if row.quality in _ACCEPTED_QUALITIES
+        and _utc(row.recorded_at) >= domain_start
+        and (domain_end is None or _utc(row.recorded_at) < domain_end)
+    ]
+    if not accepted:
+        return None
+    return min(
+        accepted,
+        key=lambda row: (abs((_utc(row.recorded_at) - target).total_seconds()), _utc(row.recorded_at)),
+    )
+
+
+def _degrade(
+    session: Session,
+    commodity: str,
+    start: datetime,
+    end: datetime,
+    existing: MeterCostPeriod | None,
+    reason: str,
+    meter_id: int | None = None,
+    binding_id: int | None = None,
+) -> None:
+    now = datetime.now(UTC)
+    fields = dict(
+        period_end=end, meter_id=meter_id, source_binding_id=binding_id,
+        contract_version_id=None, quantity=Decimal("0"), cost=Decimal("0"), currency="EUR",
+        cost_breakdown={}, pricing_snapshot={}, quality="invalid", degraded=True,
+        degraded_reason=reason, updated_at=now,
+    )
+    if existing is None:
+        session.add(MeterCostPeriod(commodity=commodity, period_start=start, created_at=now, **fields))
+    else:
+        for key, value in fields.items():
+            setattr(existing, key, value)
+
+
+def compute_period(
+    session: Session, commodity: str, period_start: datetime, *, overwrite: bool = False
+) -> bool:
+    """Compute one closed thermal period, recording every unsafe input as degraded."""
+    if commodity not in _LIMITS:
+        raise ValueError("commodity must be heating or hot_water")
+    start = floor_to_quarter(_utc(period_start))
+    end = start + _PERIOD
+    existing = _existing(session, commodity, start)
+    if existing is not None and not existing.degraded and not overwrite:
+        return False
+
+    meter0, meter1 = _meter_at(session, commodity, start), _meter_at(session, commodity, end)
+    if meter0 is None:
+        _degrade(session, commodity, start, end, existing, "missing_or_ambiguous_meter")
+        return True
+    if meter1 is None or meter1.id != meter0.id:
+        _degrade(session, commodity, start, end, existing, "cross_meter_epoch", meter0.id)
+        return True
+    bound0, bound1 = _binding_at(session, meter0, start), _binding_at(session, meter1, end)
+    if bound0 is None or bound1 is None:
+        _degrade(session, commodity, start, end, existing, "missing_or_ambiguous_binding", meter0.id)
+        return True
+    binding, channel = bound0
+    if bound1[0].id != binding.id or bound1[1].id != channel.id:
+        _degrade(session, commodity, start, end, existing, "cross_source_binding", meter0.id, binding.id)
+        return True
+    first = _reading_at(session, channel, meter0, binding, start)
+    last = _reading_at(session, channel, meter0, binding, end)
+    if first is None or last is None:
+        _degrade(session, commodity, start, end, existing, "missing_stale_or_invalid_reading", meter0.id, binding.id)
+        return True
+    delta = _decimal(last.value) - _decimal(first.value)
+    if delta < 0:
+        _degrade(session, commodity, start, end, existing, "negative_delta", meter0.id, binding.id)
+        return True
+    if delta > _LIMITS[commodity]:
+        _degrade(session, commodity, start, end, existing, "delta_limit_exceeded", meter0.id, binding.id)
+        return True
+    version = active_contract_version_at(session, start, scope="thermal")
+    if version is None:
+        _degrade(session, commodity, start, end, existing, "missing_contract", meter0.id, binding.id)
+        return True
+
+    values = {key: _decimal(value) for key, value in version.values["variable"].items()}
+    if commodity == "heating":
+        breakdown = {"heating": delta * values["heating"]}
+    else:
+        breakdown = {
+            key: delta * values[key]
+            for key in ("hot_water_heating", "hot_water", "hot_water_tax")
+        }
+    cost = sum(breakdown.values(), Decimal("0"))
+    now = datetime.now(UTC)
+    fields = dict(
+        period_end=end, meter_id=meter0.id, source_binding_id=binding.id,
+        contract_version_id=version.id, quantity=delta, cost=cost, currency=version.contract.currency,
+        cost_breakdown=breakdown, pricing_snapshot=dict(version.values),
+        quality="valid" if first.quality == last.quality == "valid" else "unverifiable",
+        degraded=False, degraded_reason=None, updated_at=now,
+    )
+    if existing is None:
+        session.add(MeterCostPeriod(commodity=commodity, period_start=start, created_at=now, **fields))
+    else:
+        for key, value in fields.items():
+            setattr(existing, key, value)
+    return True
+
+
+def compute_closed_periods(session: Session, *, now: datetime | None = None) -> int:
+    """Retry incomplete thermal rows and fill recent closed periods without touching good rows."""
+    now = _utc(now or datetime.now(UTC))
+    first = floor_to_quarter(now - timedelta(days=7))
+    written = 0
+    cursor = first
+    while cursor + _PERIOD <= now:
+        for commodity in ("heating", "hot_water"):
+            if compute_period(session, commodity, cursor):
+                written += 1
+        cursor += _PERIOD
+    session.commit()
+    return written
+
+
+def recompute_range(session: Session, start: datetime, end: datetime) -> int:
+    cursor, end = floor_to_quarter(_utc(start)), _utc(end)
+    now, written = datetime.now(UTC), 0
+    while cursor < end:
+        if cursor + _PERIOD <= now:
+            for commodity in ("heating", "hot_water"):
+                if compute_period(session, commodity, cursor, overwrite=True):
+                    written += 1
+        cursor += _PERIOD
+    session.commit()
+    return written
+
+
+def _settled_end_date(now: datetime) -> date:
+    local_now = timezone_service.to_local(now)
+    return local_now.date() if local_now.timetz().replace(tzinfo=None) >= _SETTLEMENT_TIME else local_now.date() - timedelta(days=1)
+
+
+def summarize(session: Session, start: datetime, end: datetime, *, now: datetime | None = None) -> dict[str, Any]:
+    """Return thermal variable/fixed totals; standing is charged once per contract/day."""
+    start, end = _utc(start), _utc(end)
+    rows = session.execute(select(MeterCostPeriod).where(
+        MeterCostPeriod.period_start >= start, MeterCostPeriod.period_start < end
+    )).scalars().all()
+    good = [row for row in rows if not row.degraded]
+    variable = sum((_decimal(row.cost) for row in good), Decimal("0"))
+    breakdown: dict[str, Decimal] = {key: Decimal("0") for key in (
+        "heating", "hot_water_heating", "hot_water", "hot_water_tax")}
+    for row in good:
+        for key, value in row.cost_breakdown.items():
+            breakdown[key] = breakdown.get(key, Decimal("0")) + _decimal(value)
+
+    # A summary is half-open.  ``end`` at local midnight has no overlap with
+    # that next local date, and an empty/reversed range owns no standing day.
+    final_day = timezone_service.local_date(end - timedelta(microseconds=1))
+    final_day = min(final_day, _settled_end_date(now or datetime.now(UTC)))
+    day = timezone_service.local_date(start)
+    fixed = Decimal("0")
+    versions = active_contract_versions(session, scope="thermal")
+    while start < end and day <= final_day:
+        day_start = datetime.combine(day, time.min, tzinfo=timezone_service.local_tz()).astimezone(UTC)
+        next_day_start = datetime.combine(
+            day + timedelta(days=1), time.min, tzinfo=timezone_service.local_tz()
+        ).astimezone(UTC)
+        local_day_seconds = Decimal(str((next_day_start - day_start).total_seconds()))
+        # A rate revision part-way through a local date is attributable only
+        # to its effective interval.  This preserves one contract-level daily
+        # charge while correctly handling first-version and intra-day changes.
+        for version in versions:
+            segment_start = max(day_start, _utc(version.effective_from))
+            version_end = _utc(version.effective_to) if version.effective_to is not None else next_day_start
+            segment_end = min(next_day_start, version_end)
+            if segment_start >= segment_end:
+                continue
+            values = version.values["standing"]
+            annual = sum((_decimal(values.get(key, "0")) for key in (
+                "heating_network", "metering", "delivery_set", "hot_water_network", "other"
+            )), Decimal("0"))
+            fraction = Decimal(str((segment_end - segment_start).total_seconds())) / local_day_seconds
+            fixed += annual / Decimal("365") * fraction
+        day += timedelta(days=1)
+    return {
+        "currency": good[0].currency if good else "EUR", "variable_cost": variable,
+        "fixed_cost": fixed, "total_cost": variable + fixed, "breakdown": breakdown,
+        "period_count": len(good), "degraded_count": len(rows) - len(good),
+    }

docs/design/m8-warmtelink-energy.md

@@ -850,7 +850,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 
 ### M8-T15 — Thermal 15 分钟成本引擎与调度 [structural]
 
-- **Status**: `todo`
+- **Status**: `done`
 - **Depends**: M8-T14
 - **Context**: 基于两个独立累计 Meter 生成可审计 variable 账本，并在 summary 层只计一次固定费。
 

tests/test_energy_cost.py

@@ -32,7 +32,7 @@ from pathlib import Path
 import pytest
 from alembic import command
 from alembic.config import Config
-from sqlalchemy import create_engine, select
+from sqlalchemy import create_engine, event, select
 from sqlalchemy.orm import Session
 
 from app.integrations.pricing.strategies import PeriodDeltas  # noqa: F401
@@ -57,6 +57,28 @@ from app.services.energy_cost import (
 from app.services import timezone as tz_module
 
 
+def test_scheduler_isolates_electricity_and_thermal_cost_scopes(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A thermal/electricity failure must not suppress the other ledger or HA publish."""
+    from app import main
+
+    calls: list[str] = []
+
+    class DummySession:
+        def rollback(self) -> None:
+            calls.append("rollback")
+
+        def close(self) -> None:
+            calls.append("close")
+
+    monkeypatch.setattr(main, "get_session_local", lambda: lambda: DummySession())
+    monkeypatch.setattr(main, "compute_closed_periods", lambda _session: (_ for _ in ()).throw(ValueError()))
+    monkeypatch.setattr(main, "compute_closed_meter_cost_periods", lambda _session: calls.append("thermal"))
+    from app.services import ha_discovery
+    monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: calls.append("publish"))
+    main._run_scheduled_energy_cost()
+    assert calls == ["rollback", "close", "thermal", "close", "publish", "close"]
+
+
 # ---------------------------------------------------------------------------
 # Fixtures
 # ---------------------------------------------------------------------------
@@ -82,6 +104,163 @@ def energy_db(tmp_path: Path):
     engine.dispose()
 
 
+@pytest.mark.parametrize("failing_scope", ["electricity", "thermal"])
+def test_scheduler_uses_clean_sessions_after_real_flush_failure(
+    energy_db: Session, monkeypatch: pytest.MonkeyPatch, failing_scope: str
+) -> None:
+    """A DB failed-transaction in either ledger cannot poison the next scope or HA."""
+    from app import main
+    from app.services import ha_discovery
+
+    assert energy_db.bind is not None
+    sessions: list[Session] = []
+    calls: list[str] = []
+
+    def factory() -> Session:
+        session = Session(energy_db.bind)
+        sessions.append(session)
+        return session
+
+    def flush_failure(session: Session) -> None:
+        session.add(EnergyContract())
+        session.flush()
+
+    def successful(name: str):
+        def operation(session: Session) -> None:
+            session.execute(select(EnergyContract)).all()
+            calls.append(name)
+
+        return operation
+
+    monkeypatch.setattr(main, "get_session_local", lambda: factory)
+    monkeypatch.setattr(
+        main, "compute_closed_periods",
+        flush_failure if failing_scope == "electricity" else successful("electricity"),
+    )
+    monkeypatch.setattr(
+        main, "compute_closed_meter_cost_periods",
+        flush_failure if failing_scope == "thermal" else successful("thermal"),
+    )
+    monkeypatch.setattr(ha_discovery, "publish_states", lambda session: successful("ha")(session))
+
+    main._run_scheduled_energy_cost()
+
+    assert calls == (["thermal", "ha"] if failing_scope == "electricity" else ["electricity", "ha"])
+    assert len({id(session) for session in sessions}) == 3
+
+
+@pytest.mark.parametrize("failing_scope", ["electricity", "thermal"])
+def test_scheduler_isolates_real_commit_failure_and_rolls_back_write(
+    energy_db: Session, monkeypatch: pytest.MonkeyPatch, failing_scope: str
+) -> None:
+    """A commit exception leaves no partial write and the later scopes still run."""
+    from app import main
+    from app.services import ha_discovery
+
+    assert energy_db.bind is not None
+    sessions: list[Session] = []
+    calls: list[str] = []
+
+    def factory() -> Session:
+        session = Session(energy_db.bind)
+        sessions.append(session)
+        if len(sessions) == (1 if failing_scope == "electricity" else 2):
+            event.listen(session, "before_commit", lambda _session: (_ for _ in ()).throw(RuntimeError("commit")))
+        return session
+
+    def write(name: str):
+        def operation(session: Session) -> None:
+            now = datetime.now(UTC)
+            session.add(EnergyContract(
+                name=name, kind="manual", scope="electricity", active=False, currency="EUR",
+                created_at=now, updated_at=now,
+            ))
+            session.commit()
+            calls.append(name)
+
+        return operation
+
+    monkeypatch.setattr(main, "get_session_local", lambda: factory)
+    monkeypatch.setattr(main, "compute_closed_periods", write("electricity"))
+    monkeypatch.setattr(main, "compute_closed_meter_cost_periods", write("thermal"))
+    monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: calls.append("ha"))
+    main._run_scheduled_energy_cost()
+
+    assert calls == (["thermal", "ha"] if failing_scope == "electricity" else ["electricity", "ha"])
+    persisted = Session(energy_db.bind)
+    try:
+        assert persisted.scalars(select(EnergyContract.name)).all() == (
+            ["thermal"] if failing_scope == "electricity" else ["electricity"]
+        )
+    finally:
+        persisted.close()
+
+
+@pytest.mark.parametrize("failure", ["factory", "rollback", "close"])
+def test_scheduler_lifecycle_failures_are_non_fatal_and_keep_later_scopes_independent(
+    monkeypatch: pytest.MonkeyPatch, failure: str
+) -> None:
+    """Factory and cleanup errors are logged locally, never allowed to escape."""
+    from app import main
+    from app.services import ha_discovery
+
+    calls: list[str] = []
+    created = 0
+
+    class DummySession:
+        def rollback(self) -> None:
+            calls.append("rollback")
+            if failure == "rollback":
+                raise RuntimeError("rollback")
+
+        def close(self) -> None:
+            calls.append("close")
+            if failure == "close":
+                raise RuntimeError("close")
+
+    def factory() -> DummySession:
+        nonlocal created
+        created += 1
+        if failure == "factory" and created == 1:
+            raise RuntimeError("factory")
+        return DummySession()
+
+    def electricity(_session: DummySession) -> None:
+        if failure in {"rollback", "close"}:
+            raise RuntimeError("operation")
+        calls.append("electricity")
+
+    monkeypatch.setattr(main, "get_session_local", lambda: factory)
+    monkeypatch.setattr(main, "compute_closed_periods", electricity)
+    monkeypatch.setattr(main, "compute_closed_meter_cost_periods", lambda _session: calls.append("thermal"))
+    monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: calls.append("ha"))
+    main._run_scheduled_energy_cost()
+
+    assert "thermal" in calls and "ha" in calls
+    assert created == 3
+
+
+def test_scheduler_swallows_ha_failure_after_two_successful_scopes(monkeypatch: pytest.MonkeyPatch) -> None:
+    from app import main
+    from app.services import ha_discovery
+
+    calls: list[str] = []
+
+    class DummySession:
+        def rollback(self) -> None:
+            calls.append("rollback")
+
+        def close(self) -> None:
+            calls.append("close")
+
+    monkeypatch.setattr(main, "get_session_local", lambda: lambda: DummySession())
+    monkeypatch.setattr(main, "compute_closed_periods", lambda _session: calls.append("electricity"))
+    monkeypatch.setattr(main, "compute_closed_meter_cost_periods", lambda _session: calls.append("thermal"))
+    monkeypatch.setattr(ha_discovery, "publish_states", lambda _session: (_ for _ in ()).throw(RuntimeError("ha")))
+    main._run_scheduled_energy_cost()
+    assert calls == ["electricity", "close", "thermal", "close", "rollback", "close"]
+
+
 # ---------------------------------------------------------------------------
 # Data-builder helpers
 # ---------------------------------------------------------------------------

tests/test_meter_cost.py

@@ -0,0 +1,402 @@
+"""M8-T15 tests for the binding-aware thermal ledger."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+from app.models.energy import EnergyContract, EnergyContractVersion, Meter, MeterCostPeriod
+from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
+from app.services import timezone as timezone_service
+from app.services.meter_cost import compute_closed_periods, compute_period, recompute_range, summarize
+
+
+@pytest.fixture()
+def db(tmp_path: Path) -> Session:
+    url = f"sqlite:///{tmp_path / 'meter_cost.db'}"
+    cfg = Config("alembic_app.ini")
+    cfg.set_main_option("sqlalchemy.url", url)
+    command.upgrade(cfg, "head")
+    engine = create_engine(url)
+    session = Session(engine)
+    yield session
+    session.close()
+    engine.dispose()
+
+
+T0 = datetime(2026, 6, 23, 10, tzinfo=UTC)
+
+VALUES = {
+    "variable": {"heating": "20.0", "hot_water_heating": "4.0", "hot_water": "2.0", "hot_water_tax": "1.0"},
+    "standing": {"heating_network": "365", "metering": "73", "delivery_set": "0", "hot_water_network": "0", "other": "0"},
+}
+
+
+def _contract(session: Session, *, values: dict = VALUES, start: datetime = T0 - timedelta(days=1)) -> EnergyContractVersion:
+    now = datetime.now(UTC)
+    contract = EnergyContract(name="thermal", kind="district_heating", scope="thermal", active=True,
+                              currency="EUR", created_at=now, updated_at=now)
+    session.add(contract)
+    session.flush()
+    version = EnergyContractVersion(contract_id=contract.id, effective_from=start, values=values, created_at=now)
+    session.add(version)
+    session.flush()
+    return version
+
+
+def _domain(session: Session, commodity: str, *, start: datetime = T0 - timedelta(days=1), end: datetime | None = None):
+    now = datetime.now(UTC)
+    meter = Meter(label=commodity, commodity=commodity, started_at=start, ended_at=end,
+                  reason="initial", created_at=now)
+    source = MeterSource(name=f"{commodity}-source", kind="warmtelink_serial", enabled=True, config={},
+                         status="online", created_at=now, updated_at=now)
+    session.add_all((meter, source))
+    session.flush()
+    unit = "GJ" if commodity == "heating" else "m3"
+    channel = MeterSourceChannel(source_id=source.id, channel_key=commodity, label=commodity,
+                                 suggested_commodity=commodity, unit=unit, latest_quality="valid",
+                                 created_at=now, updated_at=now)
+    session.add(channel)
+    session.flush()
+    binding = MeterSourceBinding(meter_id=meter.id, channel_id=channel.id, started_at=start, ended_at=end,
+                                 created_at=now, updated_at=now)
+    session.add(binding)
+    session.flush()
+    return meter, channel, binding
+
+
+def _reading(session: Session, channel: MeterSourceChannel, at: datetime, value: str, quality: str = "valid") -> None:
+    session.add(WarmteLinkReading(channel_id=channel.id, recorded_at=at, received_at=at, value=Decimal(value),
+                                  unit=channel.unit, quality=quality, equipment_fingerprint="test"))
+
+
+def _setup_good(session: Session) -> None:
+    _contract(session)
+    _, heating, _ = _domain(session, "heating")
+    _, water, _ = _domain(session, "hot_water")
+    _reading(session, heating, T0, "10.000")
+    _reading(session, heating, T0 + timedelta(minutes=15), "10.050")
+    _reading(session, water, T0, "20.000")
+    _reading(session, water, T0 + timedelta(minutes=15), "20.200")
+    session.commit()
+
+
+def _row(session: Session, commodity: str) -> MeterCostPeriod:
+    return session.execute(select(MeterCostPeriod).where(MeterCostPeriod.commodity == commodity)).scalar_one()
+
+
+def _row_at(session: Session, commodity: str, start: datetime) -> MeterCostPeriod:
+    return session.execute(select(MeterCostPeriod).where(
+        MeterCostPeriod.commodity == commodity, MeterCostPeriod.period_start == start
+    )).scalar_one()
+
+
+def _version(
+    session: Session, contract: EnergyContract, *, start: datetime, values: dict
+) -> EnergyContractVersion:
+    version = EnergyContractVersion(
+        contract_id=contract.id, effective_from=start, values=values, created_at=datetime.now(UTC)
+    )
+    session.add(version)
+    session.flush()
+    return version
+
+
+def test_two_commodity_decimal_breakdown_and_scheduler_idempotency(db: Session) -> None:
+    _setup_good(db)
+    assert compute_period(db, "heating", T0)
+    assert compute_period(db, "hot_water", T0)
+    db.commit()
+    heating, water = _row(db, "heating"), _row(db, "hot_water")
+    assert heating.quantity == Decimal("0.050000")
+    assert heating.cost == Decimal("1.000000000")
+    assert water.cost == Decimal("1.400000000")
+    assert {key: Decimal(value) for key, value in water.cost_breakdown.items()} == {
+        "hot_water_heating": Decimal("0.8"), "hot_water": Decimal("0.4"), "hot_water_tax": Decimal("0.2")
+    }
+    assert not compute_period(db, "heating", T0)
+
+
+@pytest.mark.parametrize("commodity,start,end,reason", [
+    ("heating", "10", "9.99", "negative_delta"),
+    ("heating", "10", "10.101", "delta_limit_exceeded"),
+    ("hot_water", "10", "11.001", "delta_limit_exceeded"),
+])
+def test_bad_deltas_are_degraded(db: Session, commodity: str, start: str, end: str, reason: str) -> None:
+    _contract(db)
+    _, channel, _ = _domain(db, commodity)
+    _reading(db, channel, T0, start)
+    _reading(db, channel, T0 + timedelta(minutes=15), end)
+    db.commit()
+    compute_period(db, commodity, T0)
+    db.commit()
+    row = _row(db, commodity)
+    assert row.degraded and row.degraded_reason == reason and row.contract_version_id is None
+
+
+def test_freshness_quality_binding_and_contract_fail_closed(db: Session) -> None:
+    _contract(db)
+    meter, channel, binding = _domain(db, "heating")
+    _reading(db, channel, T0 - timedelta(seconds=121), "1")
+    _reading(db, channel, T0 + timedelta(minutes=15), "1.01")
+    db.commit()
+    compute_period(db, "heating", T0)
+    assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
+    # Recompute sees invalid quality independently at the end boundary.
+    db.query(WarmteLinkReading).delete()
+    _reading(db, channel, T0, "1")
+    _reading(db, channel, T0 + timedelta(minutes=15), "1.01", "invalid")
+    db.commit()
+    recompute_range(db, T0, T0 + timedelta(minutes=15))
+    assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
+    binding.ended_at = T0 + timedelta(minutes=15)
+    db.commit()
+    recompute_range(db, T0, T0 + timedelta(minutes=15))
+    assert _row(db, "heating").degraded_reason == "missing_or_ambiguous_binding"
+    assert meter.id is not None
+
+
+@pytest.mark.parametrize(
+    ("first_quality", "last_quality", "accepted", "period_quality"),
+    [
+        ("valid", "valid", True, "valid"),
+        ("valid", "unverifiable", True, "unverifiable"),
+        ("unverifiable", "valid", True, "unverifiable"),
+        ("unverifiable", "unverifiable", True, "unverifiable"),
+        ("invalid", "valid", False, "invalid"),
+        ("valid", "invalid", False, "invalid"),
+    ],
+)
+def test_accepted_reading_quality_is_preserved_without_promotion(
+    db: Session, first_quality: str, last_quality: str, accepted: bool, period_quality: str
+) -> None:
+    _contract(db)
+    _, channel, _ = _domain(db, "heating")
+    _reading(db, channel, T0, "10", first_quality)
+    _reading(db, channel, T0 + timedelta(minutes=15), "10.05", last_quality)
+    db.commit()
+    assert compute_period(db, "heating", T0)
+    db.commit()
+    row = _row(db, "heating")
+    assert row.degraded is not accepted
+    assert row.quality == period_quality
+    if not accepted:
+        assert row.degraded_reason == "missing_stale_or_invalid_reading"
+
+
+def test_readings_must_be_inside_binding_and_meter_windows(db: Session) -> None:
+    _contract(db)
+    meter, channel, binding = _domain(db, "heating", start=T0)
+    _reading(db, channel, T0 - timedelta(seconds=60), "10")
+    _reading(db, channel, T0 + timedelta(minutes=15), "10.05")
+    db.commit()
+    compute_period(db, "heating", T0)
+    assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
+
+    # A candidate after a closed binding is equally outside the cumulative domain.
+    binding.started_at = T0 - timedelta(days=1)
+    binding.ended_at = T0 + timedelta(minutes=15, seconds=30)
+    meter.started_at = T0 - timedelta(days=1)
+    db.query(WarmteLinkReading).delete()
+    _reading(db, channel, T0, "10")
+    _reading(db, channel, T0 + timedelta(minutes=15, seconds=60), "10.05")
+    db.commit()
+    recompute_range(db, T0, T0 + timedelta(minutes=15))
+    assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
+
+
+def test_freshness_is_independent_and_inclusive_with_nearest_candidate(db: Session) -> None:
+    _contract(db)
+    _, channel, _ = _domain(db, "heating")
+    _reading(db, channel, T0 - timedelta(seconds=120), "10")
+    _reading(db, channel, T0 + timedelta(minutes=15, seconds=120), "10.05")
+    db.commit()
+    compute_period(db, "heating", T0)
+    db.commit()
+    row = _row(db, "heating")
+    assert not row.degraded and row.quantity == Decimal("0.050000")
+
+
+def test_closed_scheduler_retries_degraded_but_not_normal(db: Session) -> None:
+    _contract(db)
+    _, channel, _ = _domain(db, "heating")
+    # The first closed scheduler pass records a degraded row.  Supplying the
+    # missing boundary inputs later must let its next pass repair that row.
+    db.commit()
+    assert compute_closed_periods(db, now=T0 + timedelta(minutes=16)) > 0
+    assert _row_at(db, "heating", T0).degraded
+    _reading(db, channel, T0, "10")
+    _reading(db, channel, T0 + timedelta(minutes=15), "10.05")
+    db.commit()
+    assert compute_closed_periods(db, now=T0 + timedelta(minutes=16)) > 0
+    normal = _row_at(db, "heating", T0)
+    assert not normal.degraded
+    frozen = (normal.cost, dict(normal.pricing_snapshot), normal.updated_at)
+    end_reading = db.execute(select(WarmteLinkReading).where(
+        WarmteLinkReading.channel_id == channel.id,
+        WarmteLinkReading.recorded_at == T0 + timedelta(minutes=15),
+    )).scalar_one()
+    end_reading.value = Decimal("10.090")
+    version = db.get(EnergyContractVersion, normal.contract_version_id)
+    assert version is not None
+    version.values = {**version.values, "variable": {**version.values["variable"], "heating": "99"}}
+    db.commit()
+    compute_closed_periods(db, now=T0 + timedelta(minutes=16))
+    db.expire_all()
+    unchanged = db.execute(select(MeterCostPeriod).where(MeterCostPeriod.id == normal.id)).scalar_one()
+    assert (unchanged.cost, unchanged.pricing_snapshot, unchanged.updated_at) == frozen
+
+
+def test_explicit_recompute_overwrites_and_source_switch_degrades(db: Session) -> None:
+    _setup_good(db)
+    compute_period(db, "heating", T0)
+    db.commit()
+    row = _row(db, "heating")
+    old_cost = row.cost
+    channel = db.get(MeterSourceChannel, row.source_binding.channel_id)
+    end_reading = db.execute(select(WarmteLinkReading).where(
+        WarmteLinkReading.channel_id == channel.id,
+        WarmteLinkReading.recorded_at == T0 + timedelta(minutes=15),
+    )).scalar_one()
+    end_reading.value = Decimal("10.080")
+    db.commit()
+    assert recompute_range(db, T0, T0 + timedelta(minutes=15)) == 2
+    assert _row(db, "heating").cost != old_cost
+    # A hand-off precisely at the right boundary cannot form a single domain.
+    binding = _row(db, "heating").source_binding
+    binding.ended_at = T0 + timedelta(minutes=15)
+    db.commit()
+    recompute_range(db, T0, T0 + timedelta(minutes=15))
+    assert _row(db, "heating").degraded_reason == "missing_or_ambiguous_binding"
+
+
+def test_real_source_binding_handoff_never_crosses_cumulative_registers(db: Session) -> None:
+    _contract(db)
+    meter, old_channel, old_binding = _domain(db, "heating")
+    handoff = T0 + timedelta(minutes=15)
+    old_binding.ended_at = handoff
+    new_meter, new_channel, new_binding = _domain(db, "heating", start=handoff, end=handoff)
+    # Keep the same Meter: this is a source/channel/binding handoff, not a meter swap.
+    new_binding.meter_id = meter.id
+    new_binding.ended_at = None
+    assert new_meter.id != meter.id
+    _reading(db, old_channel, T0, "1000")
+    _reading(db, old_channel, handoff - timedelta(seconds=1), "1000.01")
+    _reading(db, new_channel, handoff, "7")
+    _reading(db, new_channel, handoff + timedelta(seconds=1), "7.01")
+    db.commit()
+    compute_period(db, "heating", T0)
+    assert _row(db, "heating").degraded_reason == "cross_source_binding"
+
+
+def test_real_meter_epoch_handoff_never_crosses_cumulative_registers(db: Session) -> None:
+    _contract(db)
+    handoff = T0 + timedelta(minutes=15)
+    _, old_channel, _ = _domain(db, "heating", end=handoff)
+    _, new_channel, _ = _domain(db, "heating", start=handoff)
+    _reading(db, old_channel, T0, "1000")
+    _reading(db, old_channel, handoff - timedelta(seconds=1), "1000.01")
+    _reading(db, new_channel, handoff, "2")
+    _reading(db, new_channel, handoff + timedelta(seconds=1), "2.01")
+    db.commit()
+    compute_period(db, "heating", T0)
+    assert _row(db, "heating").degraded_reason == "cross_meter_epoch"
+
+
+def test_summary_fixed_once_per_contract_day_and_dst(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
+    _setup_good(db)
+    compute_period(db, "heating", T0)
+    compute_period(db, "hot_water", T0)
+    db.commit()
+    monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
+    result = summarize(db, T0, T0 + timedelta(days=1), now=datetime(2026, 6, 24, 1, 6, tzinfo=UTC))
+    assert result["variable_cost"] == Decimal("2.400000000")
+    assert result["fixed_cost"] == Decimal("2.4")  # two local days, each charged once not per commodity
+    # DST local day is charged once too, despite being 23 hours long.
+    db.execute(select(EnergyContractVersion)).scalar_one().effective_from = datetime(2026, 3, 1, tzinfo=UTC)
+    db.commit()
+    dst_start = datetime(2026, 3, 28, 23, tzinfo=UTC)
+    dst = summarize(db, dst_start, dst_start + timedelta(days=2), now=datetime(2026, 3, 31, tzinfo=UTC))
+    assert dst["fixed_cost"] == Decimal("3.6")
+
+
+def test_summary_is_half_open_at_local_midnight_and_settlement(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
+    _contract(db)
+    monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
+    local = ZoneInfo("Europe/Amsterdam")
+    start = datetime(2026, 6, 24, 0, tzinfo=local).astimezone(UTC)
+    end = datetime(2026, 6, 25, 0, tzinfo=local).astimezone(UTC)
+    assert summarize(db, start, start, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("0")
+    assert summarize(db, end, start, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("0")
+    before = datetime(2026, 6, 24, 1, 4, 59, tzinfo=local).astimezone(UTC)
+    after = datetime(2026, 6, 24, 1, 5, tzinfo=local).astimezone(UTC)
+    assert summarize(db, start, end, now=before)["fixed_cost"] == Decimal("0")
+    assert summarize(db, start, end, now=after)["fixed_cost"] == Decimal("1.2")
+
+
+def test_summary_versions_cover_first_day_intra_day_and_cross_day(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
+    local = ZoneInfo("Europe/Amsterdam")
+    day = datetime(2026, 6, 24, tzinfo=local)
+    first = _contract(db, start=(day + timedelta(hours=6)).astimezone(UTC))
+    first.values = {**VALUES, "standing": {**VALUES["standing"], "heating_network": "365"}}
+    second_values = {**VALUES, "standing": {**VALUES["standing"], "heating_network": "730"}}
+    switch = (day + timedelta(hours=18)).astimezone(UTC)
+    first.effective_to = switch
+    _version(db, first.contract, start=switch, values=second_values)
+    db.commit()
+    start = day.astimezone(UTC)
+    next_day = (day + timedelta(days=1)).astimezone(UTC)
+    following = (day + timedelta(days=2)).astimezone(UTC)
+    # Initial-version day only owns 18h: V1 owns 12h, V2 6h; on the following
+    # day V2 owns the full daily fixed amount.
+    assert summarize(db, start, next_day, now=following + timedelta(hours=2))["fixed_cost"] == Decimal("1.15")
+    assert summarize(db, start, following, now=following + timedelta(hours=2))["fixed_cost"] == Decimal("3.35")
+
+
+def test_normal_row_has_full_audit_snapshot_and_single_commodity_scheduler(db: Session) -> None:
+    version = _contract(db)
+    meter, channel, binding = _domain(db, "heating")
+    _reading(db, channel, T0, "10")
+    _reading(db, channel, T0 + timedelta(minutes=15), "10.05")
+    db.commit()
+    compute_closed_periods(db, now=T0 + timedelta(minutes=16))
+    row = _row_at(db, "heating", T0)
+    assert (row.meter_id, row.source_binding_id, row.contract_version_id) == (meter.id, binding.id, version.id)
+    assert row.pricing_snapshot == VALUES
+    assert row.currency == "EUR" and {
+        key: Decimal(value) for key, value in row.cost_breakdown.items()
+    } == {"heating": Decimal("1.0")}
+    assert not row.degraded
+    assert _row_at(db, "hot_water", T0).degraded
+
+
+@pytest.mark.parametrize(("commodity", "end_value"), [("heating", "10.1"), ("hot_water", "11")])
+def test_delta_limit_is_inclusive_at_exact_boundary(db: Session, commodity: str, end_value: str) -> None:
+    _contract(db)
+    _, channel, _ = _domain(db, commodity)
+    _reading(db, channel, T0, "10")
+    _reading(db, channel, T0 + timedelta(minutes=15), end_value)
+    db.commit()
+    compute_period(db, commodity, T0)
+    assert not _row(db, commodity).degraded
+
+
+@pytest.mark.parametrize("day", [datetime(2026, 3, 29), datetime(2026, 10, 25)])
+def test_summary_dst_local_midnights_charge_one_daily_rate(
+    db: Session, monkeypatch: pytest.MonkeyPatch, day: datetime
+) -> None:
+    _contract(db, start=datetime(2026, 1, 1, tzinfo=UTC))
+    monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
+    local = ZoneInfo("Europe/Amsterdam")
+    start = day.replace(tzinfo=local).astimezone(UTC)
+    end = (day + timedelta(days=1)).replace(tzinfo=local).astimezone(UTC)
+    assert summarize(db, start, end, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("1.2")