M8-T03: adopt DSMR history into meter sources

alembic_app/versions/20260822_16_dsmr_source_adoption.py

@@ -0,0 +1,269 @@
+"""adopt historical DSMR rows into the source and binding model
+
+Revision ID: 20260822_16_dsmr_source_adoption
+Revises: 20260822_15_meter_sources
+Create Date: 2026-08-22 00:00:00.000000
+
+The upgrade is deliberately data-preserving: it creates one migration-owned
+DSMR source/channel, moves the telegram identifier to ``telegram_id``, and
+audits every reading and cost row before committing.  Old ``app_config`` rows,
+payload JSON, and cost snapshots are never deleted or rewritten.
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "20260822_16_dsmr_source_adoption"
+down_revision: Union[str, None] = "20260822_15_meter_sources"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _as_bool(value: str | None) -> bool:
+    return value is not None and value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _iso_now() -> str:
+    return datetime.now(tz=timezone.utc).replace(tzinfo=None).isoformat(sep=" ")
+
+
+def _config(connection: sa.Connection) -> dict[str, object]:
+    rows = connection.execute(sa.text("SELECT key, value FROM app_config")).all()
+    values = {str(key): str(value) for key, value in rows}
+    # An unconfigured historical DSMR installation needs a disabled identity,
+    # not guessed connection details.  Preserve every legacy value we model
+    # when any legacy DSMR/MQTT configuration was explicitly present.
+    legacy_keys = {
+        "MQTT_BROKER_HOST", "MQTT_BROKER_PORT", "MQTT_USERNAME", "MQTT_PASSWORD",
+        "MQTT_TLS_ENABLED", "DSMR_MQTT_TOPIC", "DSMR_TARIFF_TOPIC", "DSMR_SAMPLE_INTERVAL_S",
+    }
+    if not legacy_keys & values.keys():
+        return {}
+    return {
+        "broker_host": values.get("MQTT_BROKER_HOST", ""),
+        "broker_port": int(values.get("MQTT_BROKER_PORT", "1883")),
+        "username": values.get("MQTT_USERNAME", ""),
+        "password": values.get("MQTT_PASSWORD", ""),
+        "tls_enabled": _as_bool(values.get("MQTT_TLS_ENABLED")),
+        "topic": values.get("DSMR_MQTT_TOPIC", "dsmr/json"),
+        "tariff_topic": values.get("DSMR_TARIFF_TOPIC", "dsmr/meter-stats/electricity_tariff"),
+        "sample_interval_s": int(values.get("DSMR_SAMPLE_INTERVAL_S", "10")),
+    }
+
+
+def _count(connection: sa.Connection, table: str) -> int:
+    return int(connection.execute(sa.text(f"SELECT COUNT(*) FROM {table}")).scalar_one())
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    readings_before = _count(connection, "dsmr_reading")
+    costs_before = _count(connection, "energy_cost_period")
+    sources_before = _count(connection, "meter_source")
+    channels_before = _count(connection, "meter_source_channel")
+    bindings_before = _count(connection, "meter_source_binding")
+    now = _iso_now()
+
+    # A source exists even without historical configuration/readings.  It stays
+    # disabled unless the old explicit DSMR switch was enabled, so no broker or
+    # topic is guessed at runtime.
+    config = _config(connection)
+    source_result = connection.execute(
+        sa.text(
+            "INSERT INTO meter_source "
+            "(uuid, name, kind, enabled, config, status, last_seen_at, last_error, created_at, updated_at) "
+            "VALUES (:uuid, :name, 'dsmr_mqtt', :enabled, :config, 'unknown', NULL, NULL, :now, :now)"
+        ),
+        {
+            "uuid": str(uuid.uuid4()),
+            "name": "Migrated DSMR source",
+            "enabled": _as_bool(
+                connection.execute(
+                    sa.text("SELECT value FROM app_config WHERE key = 'DSMR_INGEST_ENABLED'")
+                ).scalar_one_or_none()
+            ),
+            "config": __import__("json").dumps(config),
+            "now": now,
+        },
+    )
+    source_id = source_result.lastrowid
+    if source_id is None:
+        raise RuntimeError("DSMR source adoption failed to create a source")
+    channel_result = connection.execute(
+        sa.text(
+            "INSERT INTO meter_source_channel "
+            "(uuid, source_id, channel_key, label, suggested_commodity, unit, device_type, fingerprint, "
+            "latest_value, latest_at, latest_quality, created_at, updated_at) "
+            "VALUES (:uuid, :source_id, 'electricity-total', 'DSMR electricity total', 'electricity', "
+            "'kWh', NULL, NULL, NULL, NULL, NULL, :now, :now)"
+        ),
+        {"uuid": str(uuid.uuid4()), "source_id": source_id, "now": now},
+    )
+    channel_id = channel_result.lastrowid
+    if channel_id is None:
+        raise RuntimeError("DSMR source adoption failed to create an electricity channel")
+    if _count(connection, "meter_source") != sources_before + 1:
+        raise RuntimeError("DSMR source adoption source row-count audit failed")
+    if _count(connection, "meter_source_channel") != channels_before + 1:
+        raise RuntimeError("DSMR source adoption channel row-count audit failed")
+
+    # Rename/add while nullable, back-fill all rows, then make the FK non-null
+    # and replace the legacy timestamp-only uniqueness in a SQLite batch rebuild.
+    with op.batch_alter_table("dsmr_reading", schema=None) as batch_op:
+        batch_op.alter_column("source_id", new_column_name="telegram_id")
+        batch_op.add_column(sa.Column("meter_source_id", sa.Integer(), nullable=True))
+    connection.execute(
+        sa.text("UPDATE dsmr_reading SET meter_source_id = :source_id WHERE meter_source_id IS NULL"),
+        {"source_id": source_id},
+    )
+    with op.batch_alter_table("dsmr_reading", schema=None) as batch_op:
+        batch_op.drop_constraint("uq_dsmr_reading_recorded_at", type_="unique")
+        batch_op.alter_column("meter_source_id", existing_type=sa.Integer(), nullable=False)
+        batch_op.create_foreign_key(
+            "fk_dsmr_reading_meter_source_id", "meter_source", ["meter_source_id"], ["id"],
+            ondelete="RESTRICT",
+        )
+        batch_op.create_unique_constraint(
+            "uq_dsmr_reading_source_recorded_at", ["meter_source_id", "recorded_at"]
+        )
+        batch_op.create_index("ix_dsmr_reading_meter_source_id", ["meter_source_id"])
+    adopted_readings = int(
+        connection.execute(
+            sa.text("SELECT COUNT(*) FROM dsmr_reading WHERE meter_source_id = :source_id"),
+            {"source_id": source_id},
+        ).scalar_one()
+    )
+    if adopted_readings != readings_before:
+        raise RuntimeError("DSMR source adoption reading source audit failed")
+
+    # Bind each electricity meter only where it overlaps the actual DSMR data.
+    data_window = connection.execute(
+        sa.text("SELECT MIN(recorded_at), MAX(recorded_at) FROM dsmr_reading")
+    ).one()
+    expected_binding_count = 0
+    if data_window[0] is not None:
+        meters = connection.execute(
+            sa.text(
+                "SELECT id, started_at, ended_at FROM meter WHERE commodity = 'electricity' "
+                "ORDER BY started_at, id"
+            )
+        ).all()
+        for meter_id, started_at, ended_at in meters:
+            # Intersect [meter start, meter end) with the inclusive historical
+            # samples.  A closed boundary at the final sample remains valid for
+            # the preceding interval; an empty intersection gets no fake binding.
+            if started_at > data_window[1] or (ended_at is not None and ended_at <= data_window[0]):
+                continue
+            expected_binding_count += 1
+            binding_start = max(started_at, data_window[0])
+            binding_end = ended_at
+            connection.execute(
+                sa.text(
+                    "INSERT INTO meter_source_binding "
+                    "(uuid, meter_id, channel_id, started_at, ended_at, created_at, updated_at) "
+                    "VALUES (:uuid, :meter_id, :channel_id, :started_at, :ended_at, :now, :now)"
+                ),
+                {
+                    "uuid": str(uuid.uuid4()), "meter_id": meter_id, "channel_id": channel_id,
+                    "started_at": binding_start, "ended_at": binding_end, "now": now,
+                },
+            )
+
+    # A cost period may be linked only if exactly one binding covers both its
+    # start and end.  Historical boundary/unknown rows remain auditable but are
+    # explicitly degraded instead of being silently attributed to a current meter.
+    periods = connection.execute(
+        sa.text("SELECT id, meter_id, period_start, degraded FROM energy_cost_period")
+    ).all()
+    resolvable_normal_periods: dict[int, int] = {}
+    unresolved_period_ids: set[int] = set()
+    for period_id, meter_id, period_start, degraded_before in periods:
+        candidates = []
+        if meter_id is not None:
+            candidates = connection.execute(
+                sa.text(
+                    "SELECT id FROM meter_source_binding "
+                    "WHERE meter_id = :meter_id AND started_at <= :start "
+                    "AND (ended_at IS NULL OR julianday(ended_at) > julianday(:start, '+15 minutes'))"
+                ),
+                {"meter_id": meter_id, "start": period_start},
+            ).all()
+        if len(candidates) == 1:
+            if not degraded_before:
+                resolvable_normal_periods[period_id] = candidates[0][0]
+            connection.execute(
+                sa.text("UPDATE energy_cost_period SET source_binding_id = :binding_id WHERE id = :id"),
+                {"binding_id": candidates[0][0], "id": period_id},
+            )
+        else:
+            unresolved_period_ids.add(period_id)
+            connection.execute(
+                sa.text(
+                    "UPDATE energy_cost_period SET degraded = 1, source_binding_id = NULL WHERE id = :id"
+                ),
+                {"id": period_id},
+            )
+
+    readings_after = _count(connection, "dsmr_reading")
+    costs_after = _count(connection, "energy_cost_period")
+    if readings_after != readings_before or costs_after != costs_before:
+        raise RuntimeError("DSMR source adoption row-count audit failed")
+    if _count(connection, "meter_source_binding") != bindings_before + expected_binding_count:
+        raise RuntimeError("DSMR source adoption binding row-count audit failed")
+    if int(
+        connection.execute(
+            sa.text("SELECT COUNT(*) FROM meter_source_binding WHERE channel_id = :channel_id"),
+            {"channel_id": channel_id},
+        ).scalar_one()
+    ) != expected_binding_count:
+        raise RuntimeError("DSMR source adoption binding channel audit failed")
+    for period_id, binding_id in resolvable_normal_periods.items():
+        bound, degraded = connection.execute(
+            sa.text("SELECT source_binding_id, degraded FROM energy_cost_period WHERE id = :id"),
+            {"id": period_id},
+        ).one()
+        if bound != binding_id or degraded:
+            raise RuntimeError("DSMR source adoption resolvable cost audit failed")
+    if unresolved_period_ids:
+        unresolved_count = int(
+            connection.execute(
+                sa.text(
+                    "SELECT COUNT(*) FROM energy_cost_period "
+                    "WHERE id IN :period_ids AND (degraded != 1 OR source_binding_id IS NOT NULL)"
+                ).bindparams(sa.bindparam("period_ids", expanding=True)),
+                {"period_ids": list(unresolved_period_ids)},
+            ).scalar_one()
+        )
+        if unresolved_count:
+            raise RuntimeError("DSMR source adoption unresolved cost audit failed")
+    orphan_rows = connection.execute(sa.text("PRAGMA foreign_key_check")).all()
+    if orphan_rows:
+        raise RuntimeError("DSMR source adoption foreign-key audit failed")
+    normal_unbound = int(
+        connection.execute(
+            sa.text("SELECT COUNT(*) FROM energy_cost_period WHERE degraded = 0 AND source_binding_id IS NULL")
+        ).scalar_one()
+    )
+    if normal_unbound:
+        raise RuntimeError(f"DSMR source adoption left {normal_unbound} normal cost period(s) unbound")
+    if _count(connection, "meter_source") < 1:
+        raise RuntimeError("DSMR source adoption source audit failed")
+
+
+def downgrade() -> None:
+    # Schema-only downgrade for isolated test databases.  It intentionally does
+    # not delete migration-created source/channel/binding rows.
+    with op.batch_alter_table("dsmr_reading", schema=None) as batch_op:
+        batch_op.drop_index("ix_dsmr_reading_meter_source_id")
+        batch_op.drop_constraint("uq_dsmr_reading_source_recorded_at", type_="unique")
+        batch_op.drop_constraint("fk_dsmr_reading_meter_source_id", type_="foreignkey")
+        batch_op.drop_column("meter_source_id")
+        batch_op.alter_column("telegram_id", new_column_name="source_id")
+        batch_op.create_unique_constraint("uq_dsmr_reading_recorded_at", ["recorded_at"])

app/models/energy.py

@@ -13,8 +13,8 @@ from __future__ import annotations
 
 import uuid as _uuid
 from datetime import datetime
-from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String
-from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, UniqueConstraint, event, text
+from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym
 from sqlalchemy.types import JSON
 
 from app.db import Base
@@ -97,8 +97,9 @@ class DsmrReading(Base):
     (that field overflows and must be manually reset to zero — a known DSMR
     quirk — so relying on it for uniqueness risks silently dropping new data).
     The table's own autoincrement ``id`` PK is the stable internal identity, and
-    ``recorded_at`` (the telegram timestamp) is the UNIQUE de-duplication key: a
-    single P1 meter emits exactly one telegram per timestamp.
+    ``(meter_source_id, recorded_at)`` is the UNIQUE de-duplication key: each
+    configured P1 source emits at most one telegram per timestamp, while
+    different sources may legitimately emit at the same instant.
 
     ``recorded_at`` is a real column (not inside the payload) so time-range
     queries are efficient.  The entire telegram frame is stored verbatim in
@@ -110,21 +111,44 @@ class DsmrReading(Base):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
 
-    # UTC timestamp of the sample — real column, UNIQUE (telegram-id-independent
-    # idempotency key).  The unique index also serves time-range queries.
-    recorded_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), nullable=False, unique=True
-    )
+    # UTC timestamp of the sample.  Idempotency is per configured source, so
+    # distinct P1 sources may legitimately emit at the same instant.
+    recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
 
     # Telegram's own id (DSMR Reader assigns it).  Stored only as a reference /
     # debugging aid — NOT used for uniqueness or idempotency (it overflows and
     # gets reset to zero).  Nullable because some DSMR sources may not emit one.
-    source_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    telegram_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    # Compatibility for the pre-M8 ingest implementation.  This is an ORM
+    # alias only; the physical database column is ``telegram_id``.
+    source_id = synonym("telegram_id")
+
+    # The configured source is the durable identity of the cumulative reading
+    # stream.  It is non-null after the revision-16 historical adoption.
+    meter_source_id: Mapped[int] = mapped_column(
+        ForeignKey("meter_source.id", ondelete="RESTRICT"), nullable=False, index=True
+    )
 
     # Full telegram frame as a JSON object; values are typically JSON strings
     # (e.g. "20915.154") — callers must cast to Decimal before arithmetic.
     payload: Mapped[dict] = mapped_column(JSON, nullable=False)
 
+    __table_args__ = (
+        UniqueConstraint(
+            "meter_source_id", "recorded_at", name="uq_dsmr_reading_source_recorded_at"
+        ),
+    )
+
+
+@event.listens_for(DsmrReading, "before_insert")
+def _supply_legacy_dsmr_source(_mapper, connection, target: DsmrReading) -> None:
+    """Keep the pre-T04 single-source writer working during the schema handoff."""
+    if target.meter_source_id is None:
+        target.meter_source_id = connection.execute(
+            text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt' ORDER BY id LIMIT 1")
+        ).scalar_one()
+
 
 class EnergyContract(Base):
     """Contract head: a named energy contract with a chosen pricing strategy.

docs/design/m8-warmtelink-energy.md

@@ -363,7 +363,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 
 ### M8-T03 — 将 DSMR 历史迁入 Source / Binding [structural]
 
-- **Status**: `todo`
+- **Status**: `done`
 - **Depends**: M8-T02
 - **Context**: 先把既有电力链路安全迁到统一模型，之后才能让 runtime 和计费真正按 source 工作。
 
@@ -373,6 +373,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 - `modify  scripts/app_db_adopt.py`
 - `create  tests/test_dsmr_source_migration.py`
 - `modify  tests/test_energy_models.py`
+- `modify  tests/test_meter_sources.py`
 
 **Steps**
 1. 把 ORM `DsmrReading.source_id` 重命名为 nullable `telegram_id`，新增 non-null
@@ -385,6 +386,8 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 5. 用 revision 14 的历史 fixture 覆盖单 Meter、多次换表、跨界成本、无读数 Meter 和无旧 config。
 6. 把 `APP_BASELINE_REVISION` 同步到 revision 16；所有升级 fixture 仅在 `tmp_path` 中构造，不读取
    真实 app DB 或 volume。
+7. 把 T01 自身的 revision 14→15 schema 对账测试固定升级到明确 revision 15，不以可继续前进的
+   `head` 作为 T01 终点；T03 的新 fixture 单独负责 revision 14/15→16 历史回填对账。
 
 **Out of scope / 不要碰**
 - 不删除旧 DSMR `app_config` 行，不改变 MQTT subscription，不改 API 响应。
@@ -395,11 +398,13 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 - [ ] 两个 source 可在同一 timestamp 各存一条 DSMR reading；同 source 重复 timestamp 被拒绝。
 - [ ] `telegram_id` 不参与幂等唯一键，旧 telegram id 完整保留。
 - [ ] `APP_BASELINE_REVISION` 等于唯一 revision 16 head，升级重复运行幂等。
+- [ ] T01 schema-only fixture 固定停在 revision 15，T03 fixture 到 revision 16，二者职责不随 head 漂移。
 - [ ] 历史升级 fixture、空库升级、`pytest`、`ruff check .` 全绿。
 
 **Reviewer checklist**
 - 对账是否在 migration 中真实执行，而不只是测试断言；失败能否原子回滚。
 - baseline 常量是否随 revision 16 同步，fixture 是否完全隔离于真实生产路径。
+- 前序 migration 测试是否使用明确 revision 边界，而非把历史阶段误写为永久 `head`。
 - 是否存在“把所有历史强绑当前 Meter/source”的静默错误或任何 destructive cleanup。
 
 ### M8-T04 — DSMR runtime 改为多 Source 配置 [structural]

scripts/app_db_adopt.py

@@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path:
 
 from app.config import get_settings
 
-APP_BASELINE_REVISION = "20260822_15_meter_sources"
+APP_BASELINE_REVISION = "20260822_16_dsmr_source_adoption"
 
 
 class AppDatabaseAdoptionError(RuntimeError):

tests/test_dsmr_source_migration.py

@@ -0,0 +1,285 @@
+"""Isolated revision-14/15 fixtures for the DSMR source-adoption migration."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+import sqlalchemy.exc
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, event, inspect, text
+
+
+def _config(database_url: str) -> Config:
+    config = Config("alembic_app.ini")
+    config.set_main_option("sqlalchemy.url", database_url)
+    return config
+
+
+def _engine(database_url: str):
+    engine = create_engine(database_url, connect_args={"check_same_thread": False})
+
+    @event.listens_for(engine, "connect")
+    def _foreign_keys(connection, _record) -> None:
+        connection.execute("PRAGMA foreign_keys=ON")
+
+    return engine
+
+
+def _insert_meter(connection, label: str, started: datetime, ended: datetime | None) -> int:
+    connection.execute(
+        text(
+            "INSERT INTO meter (uuid, label, commodity, started_at, ended_at, reason, note, created_at) "
+            "VALUES (:uuid, :label, 'electricity', :started, :ended, 'initial', NULL, :started)"
+        ),
+        {"uuid": f"{label:0<8}-0000-4000-8000-000000000000", "label": label,
+         "started": started, "ended": ended},
+    )
+    return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
+
+
+def _insert_contract_version(connection, start: datetime) -> int:
+    connection.execute(
+        text(
+            "INSERT INTO energy_contract (name, kind, active, currency, created_at, updated_at) "
+            "VALUES ('Historic contract', 'manual', 1, 'EUR', :at, :at)"
+        ),
+        {"at": start},
+    )
+    contract_id = int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
+    connection.execute(
+        text(
+            "INSERT INTO energy_contract_version "
+            "(contract_id, effective_from, effective_to, \"values\", created_at) "
+            "VALUES (:contract_id, :at, NULL, :values, :at)"
+        ),
+        {"contract_id": contract_id, "at": start, "values": json.dumps({"historic": True})},
+    )
+    return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
+
+
+def _insert_cost(
+    connection,
+    period_start: datetime,
+    meter_id: int,
+    contract_version_id: int,
+    sequence: int,
+) -> None:
+    amount = 2.5 + sequence
+    connection.execute(
+        text(
+            "INSERT INTO energy_cost_period "
+            "(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, export_revenue, net_cost, "
+            "currency, pricing, contract_version_id, meter_id, degraded, computed_at) "
+            "VALUES (:start, :d1, :d2, :r1, :r2, :import_cost, :export_revenue, :net_cost, "
+            "'EUR', :pricing, :contract_version_id, :meter, 0, :computed_at)"
+        ),
+        {
+            "start": period_start,
+            "d1": 1.0 + sequence,
+            "d2": 2.0 + sequence,
+            "r1": 3.0 + sequence,
+            "r2": 4.0 + sequence,
+            "import_cost": amount,
+            "export_revenue": 0.25 + sequence,
+            "net_cost": amount - (0.25 + sequence),
+            "pricing": json.dumps({"historic": True, "sequence": sequence}),
+            "contract_version_id": contract_version_id,
+            "meter": meter_id,
+            "computed_at": period_start + timedelta(seconds=sequence),
+        },
+    )
+
+
+def test_populated_revision_14_adopts_dsmr_history_at_revision_16(tmp_path: Path):
+    database_url = f"sqlite:///{tmp_path / 'revision_14_history.db'}"
+    config = _config(database_url)
+    command.upgrade(config, "20260625_14_meter_uuid")
+    start = datetime(2026, 8, 1, tzinfo=timezone.utc)
+    engine = _engine(database_url)
+    try:
+        with engine.begin() as connection:
+            connection.execute(
+                text("INSERT INTO app_config (key, value, updated_at) VALUES (:key, :value, :at)"),
+                [
+                    {"key": "DSMR_INGEST_ENABLED", "value": "true", "at": start},
+                    {"key": "DSMR_MQTT_TOPIC", "value": "historic/dsmr", "at": start},
+                    {"key": "DSMR_TARIFF_TOPIC", "value": "historic/tariff", "at": start},
+                    {"key": "DSMR_SAMPLE_INTERVAL_S", "value": "15", "at": start},
+                    {"key": "MQTT_BROKER_HOST", "value": "mqtt.example.invalid", "at": start},
+                    {"key": "MQTT_BROKER_PORT", "value": "1884", "at": start},
+                    {"key": "MQTT_USERNAME", "value": "historic-user", "at": start},
+                    {"key": "MQTT_PASSWORD", "value": "historic-password", "at": start},
+                    {"key": "MQTT_TLS_ENABLED", "value": "true", "at": start},
+                    {"key": "UNRELATED_CONFIG", "value": "untouched", "at": start},
+                ],
+            )
+            config_before = dict(connection.execute(text("SELECT key, value FROM app_config")).all())
+            for offset, telegram_id in ((0, 77), (20, 78), (40, 77)):
+                connection.execute(
+                    text("INSERT INTO dsmr_reading (recorded_at, source_id, payload) VALUES (:at, :id, :payload)"),
+                    {"at": start + timedelta(minutes=offset), "id": telegram_id,
+                     "payload": json.dumps({"id": telegram_id, "keep": f"payload-{offset}"})},
+                )
+            first = _insert_meter(connection, "meterone", start - timedelta(hours=1), start + timedelta(minutes=20))
+            second = _insert_meter(connection, "metertwo", start + timedelta(minutes=20), start + timedelta(minutes=40))
+            third = _insert_meter(connection, "meterthree", start + timedelta(minutes=40), None)
+            _insert_meter(connection, "nodata", start + timedelta(days=1), None)
+            contract_version = _insert_contract_version(connection, start - timedelta(days=1))
+            # One normal period per epoch plus a period ending exactly at each
+            # replacement boundary.  Meter/binding intervals are half-open, so
+            # the latter must remain unbound/degraded.
+            _insert_cost(connection, start, first, contract_version, 0)
+            _insert_cost(connection, start + timedelta(minutes=5), first, contract_version, 1)
+            _insert_cost(connection, start + timedelta(minutes=21), second, contract_version, 2)
+            _insert_cost(connection, start + timedelta(minutes=25), second, contract_version, 3)
+            _insert_cost(connection, start + timedelta(minutes=41), third, contract_version, 4)
+            cost_before = connection.execute(
+                text(
+                    "SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
+                    "export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, "
+                    "degraded, computed_at FROM energy_cost_period ORDER BY period_start"
+                )
+            ).mappings().all()
+    finally:
+        engine.dispose()
+
+    command.upgrade(config, "20260822_16_dsmr_source_adoption")
+    engine = _engine(database_url)
+    try:
+        with engine.connect() as connection:
+            assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == (
+                "20260822_16_dsmr_source_adoption"
+            )
+            assert connection.execute(text("SELECT COUNT(*) FROM dsmr_reading")).scalar_one() == 3
+            source = connection.execute(
+                text("SELECT id, enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
+            ).one()
+            assert source.enabled == 1
+            assert json.loads(source.config) == {
+                "broker_host": "mqtt.example.invalid", "broker_port": 1884,
+                "username": "historic-user", "password": "historic-password", "tls_enabled": True,
+                "topic": "historic/dsmr", "tariff_topic": "historic/tariff", "sample_interval_s": 15,
+            }
+            assert connection.execute(text("SELECT value FROM app_config WHERE key = 'DSMR_MQTT_TOPIC'")).scalar_one() == "historic/dsmr"
+            assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == config_before
+            assert connection.execute(text("SELECT group_concat(telegram_id) FROM dsmr_reading")).scalar_one() == "77,78,77"
+            assert connection.execute(text("SELECT payload FROM dsmr_reading ORDER BY recorded_at")).scalars().all() == [
+                json.dumps({"id": 77, "keep": "payload-0"}),
+                json.dumps({"id": 78, "keep": "payload-20"}),
+                json.dumps({"id": 77, "keep": "payload-40"}),
+            ]
+            assert connection.execute(text("SELECT COUNT(*) FROM meter_source_binding")).scalar_one() == 3
+            assert connection.execute(
+                text("SELECT COUNT(*) FROM meter_source_binding WHERE meter_id = (SELECT id FROM meter WHERE label = 'nodata')")
+            ).scalar_one() == 0
+            periods = connection.execute(
+                text(
+                    "SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
+                    "export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, "
+                    "degraded, computed_at, source_binding_id FROM energy_cost_period ORDER BY period_start"
+                )
+            ).mappings().all()
+            assert [
+                {key: value for key, value in period.items() if key not in {"degraded", "source_binding_id"}}
+                for period in periods
+            ] == [
+                {key: value for key, value in period.items() if key != "degraded"}
+                for period in cost_before
+            ]
+            assert [(period["degraded"], period["source_binding_id"] is not None) for period in periods] == [
+                (0, True), (1, False), (0, True), (1, False), (0, True)
+            ]
+            assert [period["period_start"] for period in periods if period["source_binding_id"] is None] == [
+                (start + timedelta(minutes=5)).isoformat(sep=" "),
+                (start + timedelta(minutes=25)).isoformat(sep=" "),
+            ]
+            bound_meter_ids = connection.execute(
+                text(
+                    "SELECT binding.meter_id FROM energy_cost_period AS period "
+                    "LEFT JOIN meter_source_binding AS binding "
+                    "ON binding.id = period.source_binding_id ORDER BY period.period_start"
+                )
+            ).scalars().all()
+            assert bound_meter_ids == [first, None, second, None, third]
+            assert connection.execute(text("PRAGMA foreign_key_check")).all() == []
+    finally:
+        engine.dispose()
+
+
+def test_dsmr_source_timestamp_uniqueness_allows_two_sources(tmp_path: Path):
+    database_url = f"sqlite:///{tmp_path / 'two_sources.db'}"
+    config = _config(database_url)
+    command.upgrade(config, "20260822_16_dsmr_source_adoption")
+    engine = _engine(database_url)
+    timestamp = datetime(2026, 8, 1, tzinfo=timezone.utc)
+    try:
+        with engine.begin() as connection:
+            first_source = connection.execute(text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt'")).scalar_one()
+            assert connection.execute(
+                text("SELECT enabled FROM meter_source WHERE id = :id"), {"id": first_source}
+            ).scalar_one() == 0
+            connection.execute(
+                text(
+                    "INSERT INTO meter_source (uuid, name, kind, enabled, config, status, created_at, updated_at) "
+                    "VALUES ('22222222-2222-4222-8222-222222222222', 'Second DSMR', 'dsmr_mqtt', 0, '{}', "
+                    "'unknown', :at, :at)"
+                ), {"at": timestamp},
+            )
+            second_source = connection.execute(text("SELECT last_insert_rowid()")).scalar_one()
+            for source_id in (first_source, second_source):
+                connection.execute(
+                    text(
+                        "INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) "
+                        "VALUES (:at, 9, :source, '{}')"
+                    ), {"at": timestamp, "source": source_id},
+                )
+            with pytest.raises(sqlalchemy.exc.IntegrityError):
+                connection.execute(
+                    text(
+                        "INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) "
+                        "VALUES (:at, 10, :source, '{}')"
+                    ), {"at": timestamp, "source": first_source},
+                )
+    finally:
+        engine.dispose()
+    inspector = inspect(create_engine(database_url))
+    assert ("meter_source_id", "recorded_at") in {
+        tuple(item["column_names"]) for item in inspector.get_unique_constraints("dsmr_reading")
+    }
+
+
+def test_revision_15_without_legacy_dsmr_config_creates_unconfigured_source(tmp_path: Path):
+    database_url = f"sqlite:///{tmp_path / 'revision_15_no_dsmr_config.db'}"
+    config = _config(database_url)
+    command.upgrade(config, "20260822_15_meter_sources")
+    engine = _engine(database_url)
+    try:
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    "INSERT INTO app_config (key, value, updated_at) "
+                    "VALUES ('UNRELATED_CONFIG', 'untouched', :at)"
+                ),
+                {"at": datetime(2026, 8, 1, tzinfo=timezone.utc)},
+            )
+    finally:
+        engine.dispose()
+
+    command.upgrade(config, "20260822_16_dsmr_source_adoption")
+    engine = _engine(database_url)
+    try:
+        with engine.connect() as connection:
+            source = connection.execute(
+                text("SELECT enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
+            ).one()
+            assert source.enabled == 0
+            assert json.loads(source.config) == {}
+            assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == {
+                "UNRELATED_CONFIG": "untouched"
+            }
+    finally:
+        engine.dispose()

tests/test_energy_models.py

@@ -21,7 +21,7 @@ import pytest
 import sqlalchemy.exc
 from alembic import command
 from alembic.config import Config
-from sqlalchemy import create_engine, event as sa_event, inspect, text
+from sqlalchemy import UniqueConstraint, create_engine, event as sa_event, inspect, text
 from sqlalchemy.orm import Session
 
 from app.db import Base
@@ -114,31 +114,42 @@ def test_energy_tables_exist_after_upgrade(energy_db):
 
 
 def test_dsmr_reading_columns(energy_db):
-    """dsmr_reading must have id, recorded_at (NOT NULL), source_id (nullable), payload (NOT NULL)."""
+    """dsmr_reading stores its telegram id separately from its source identity."""
     inspector = inspect(energy_db)
     columns = {col["name"]: col for col in inspector.get_columns("dsmr_reading")}
 
     assert "id" in columns and not columns["id"]["nullable"]
     assert "recorded_at" in columns and not columns["recorded_at"]["nullable"]
-    assert "source_id" in columns and columns["source_id"]["nullable"]
+    assert "telegram_id" in columns and columns["telegram_id"]["nullable"]
+    assert "meter_source_id" in columns and not columns["meter_source_id"]["nullable"]
     assert "payload" in columns and not columns["payload"]["nullable"]
 
 
-def test_dsmr_reading_recorded_at_unique(energy_db):
-    """dsmr_reading.recorded_at is the UNIQUE de-dup key (telegram-id-independent)."""
+def test_dsmr_reading_source_timestamp_unique(energy_db):
+    """DSMR de-duplication is unique per configured source and timestamp."""
     inspector = inspect(energy_db)
     unique_constraints = inspector.get_unique_constraints("dsmr_reading")
-    unique_cols = [col for uc in unique_constraints for col in uc["column_names"]]
-    assert "recorded_at" in unique_cols, "recorded_at must have a unique constraint"
+    assert ("meter_source_id", "recorded_at") in {
+        tuple(uc["column_names"]) for uc in unique_constraints
+    }
+    foreign_keys = {
+        tuple(foreign_key["constrained_columns"]): foreign_key
+        for foreign_key in inspector.get_foreign_keys("dsmr_reading")
+    }
+    assert foreign_keys[("meter_source_id",)]["referred_table"] == "meter_source"
+    assert foreign_keys[("meter_source_id",)]["options"]["ondelete"] == "RESTRICT"
+    assert "ix_dsmr_reading_meter_source_id" in {
+        index["name"] for index in inspector.get_indexes("dsmr_reading")
+    }
 
 
-def test_dsmr_reading_source_id_not_unique(energy_db):
-    """dsmr_reading.source_id (telegram id) must NOT be unique — it overflows/resets,
+def test_dsmr_reading_telegram_id_not_unique(energy_db):
+    """dsmr_reading.telegram_id must NOT be unique — it overflows/resets,
     so it is kept only as a reference value and never relied on for dedup."""
     inspector = inspect(energy_db)
     unique_constraints = inspector.get_unique_constraints("dsmr_reading")
     unique_cols = [col for uc in unique_constraints for col in uc["column_names"]]
-    assert "source_id" not in unique_cols, "source_id must NOT have a unique constraint"
+    assert "telegram_id" not in unique_cols, "telegram_id must NOT have a unique constraint"
 
 
 def test_energy_contract_columns(energy_db):
@@ -410,12 +421,15 @@ def test_energy_cost_period_meter_id_fk_ondelete_restrict():
     )
 
 
-def test_dsmr_reading_recorded_at_unique_in_metadata():
-    """DsmrReading.recorded_at must be the unique de-dup key in ORM metadata,
-    and source_id must NOT be unique (decoupled from the telegram id)."""
+def test_dsmr_reading_source_timestamp_unique_in_metadata():
+    """DsmrReading de-duplicates by source/timestamp, never telegram id."""
     table = Base.metadata.tables["dsmr_reading"]
-    assert table.columns["recorded_at"].unique, "recorded_at must be declared unique"
-    assert not table.columns["source_id"].unique, "source_id must NOT be unique"
+    assert not table.columns["telegram_id"].unique, "telegram_id must NOT be unique"
+    assert any(
+        tuple(constraint.columns.keys()) == ("meter_source_id", "recorded_at")
+        for constraint in table.constraints
+        if isinstance(constraint, UniqueConstraint)
+    )
 
 
 def test_tibber_price_starts_at_unique_in_metadata():

tests/test_meter_sources.py

@@ -139,7 +139,9 @@ def test_populated_revision_14_upgrades_to_meter_source_head_with_audit(tmp_path
     finally:
         engine.dispose()
 
-    command.upgrade(config, "head")
+    # This T01 fixture intentionally audits the schema-only revision 15.
+    # Revision 16 has its own DSMR-history adoption fixture.
+    command.upgrade(config, "20260822_15_meter_sources")
 
     engine = _engine_with_foreign_keys(database_url)
     try:
@@ -187,7 +189,7 @@ def test_populated_revision_14_upgrades_to_meter_source_head_with_audit(tmp_path
         assert cost_fks["contract_version_id"]["referred_table"] == "energy_contract_version"
         assert cost_fks["source_binding_id"]["referred_table"] == "meter_source_binding"
 
-        command.upgrade(config, "head")
+        command.upgrade(config, "20260822_15_meter_sources")
         assert {
             table_name: engine.connect().execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar_one()
             for table_name in before_counts