diff --git a/app/services/warmtelink_ingest.py b/app/services/warmtelink_ingest.py
new file mode 100644
index 0000000..bbdd655
--- /dev/null
+++ b/app/services/warmtelink_ingest.py
@@ -0,0 +1,362 @@
+"""Privacy-preserving WarmteLink frame admission and minute sampling.
+
+The serial worker added later owns I/O.  This module deliberately only accepts
+already parsed :class:`P1Telegram` instances (or a parser callable at its
+small convenience entry point), so rejected telegram bytes never enter the
+database or an exception message.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+import re
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.db import get_session_local
+from app.integrations.p1 import IntegrityStatus, P1Channel, P1Telegram, parse_telegram
+from app.models.meter_source import MeterSource, WarmteLinkReading
+from app.services.meter_sources import upsert_discovered_channel
+
+
+@dataclass(frozen=True)
+class _ChannelSample:
+    key: str
+    label: str
+    value: Decimal
+    unit: str
+    device_type: str | None
+    fingerprint: str | None
+    identity: tuple[object, ...]
+
+
+@dataclass(frozen=True)
+class _FrameSnapshot:
+    recorded_at: datetime
+    fingerprint: str | None
+    samples: tuple[_ChannelSample, ...]
+
+
+class WarmteLinkIngestor:
+    """Keep unverifiable candidates isolated by source for one worker lifetime."""
+
+    def __init__(
+        self,
+        *,
+        clock: Callable[[], datetime] | None = None,
+    ) -> None:
+        self._clock = clock or (lambda: datetime.now(UTC))
+        self._previous: dict[int, _FrameSnapshot] = {}
+
+    def ingest(self, session: Session, *, source_id: int, telegram: P1Telegram) -> bool:
+        """Apply one parsed telegram in the caller's transaction.
+
+        Returns whether the frame was admitted.  Callers that own a session
+        must commit on success; :func:`handle_frame` is the failure-contained
+        worker-facing entry point.
+        """
+        source = session.get(MeterSource, source_id)
+        if source is None:
+            raise ValueError("WarmteLink source was not found")
+        if source.kind != "warmtelink_serial":
+            raise ValueError("Source is not a WarmteLink serial source")
+
+        try:
+            integrity = _integrity_status(telegram.integrity)
+        except Exception:
+            self._previous.pop(source_id, None)
+            self._diagnose(source, "WarmteLink frame could not be normalized")
+            return False
+
+        if integrity is IntegrityStatus.INVALID:
+            self._previous.pop(source_id, None)
+            self._diagnose(source, "WarmteLink frame checksum is invalid")
+            return False
+
+        try:
+            snapshot = _snapshot(telegram)
+            fingerprints = _final_fingerprints(snapshot)
+        except Exception:
+            # A parser DTO remains an untrusted boundary.  Do not let an
+            # unnormalizable DTO bridge two otherwise matching candidates.
+            self._previous.pop(source_id, None)
+            self._diagnose(source, "WarmteLink frame could not be normalized")
+            return False
+        if fingerprints is None:
+            self._previous.pop(source_id, None)
+            self._diagnose(source, "WarmteLink frame fingerprint is invalid")
+            return False
+        if integrity is IntegrityStatus.UNVERIFIABLE:
+            previous = self._previous.get(source_id)
+            if previous is None:
+                self._previous[source_id] = snapshot
+                self._diagnose(source, "Awaiting a second matching unverifiable WarmteLink frame")
+                return False
+            reason = _continuity_problem(previous, snapshot)
+            if reason is not None:
+                self._previous[source_id] = snapshot
+                self._diagnose(source, reason)
+                return False
+        try:
+            self._admit(session, source, snapshot, integrity.value, fingerprints)
+        except Exception:
+            # ``ingest`` owns flushes and may be used directly by tests or
+            # future callers.  A failed write must never become a speculative
+            # predecessor for this source.
+            self._previous.pop(source_id, None)
+            raise
+        if integrity is IntegrityStatus.UNVERIFIABLE:
+            # Advance the sliding predecessor only once every database write
+            # for this frame has succeeded.  ``handle_frame`` also clears it
+            # if the caller's later commit fails.
+            self._previous[source_id] = snapshot
+        else:
+            # A verified frame has no need for a speculative predecessor.
+            self._previous.pop(source_id, None)
+        return True
+
+    def handle_frame(
+        self,
+        source_id: int,
+        frame: bytes,
+        *,
+        session_factory: Callable[[], Session] = get_session_local,
+        parser: Callable[[bytes], P1Telegram] = parse_telegram,
+    ) -> bool:
+        """Parse and persist one frame, containing both parse and DB failures.
+
+        A failed write is rolled back before a fresh transaction records only
+        a generic source error.  Consequently no partial latest/history update
+        survives and the following frame may recover normally.
+        """
+        try:
+            telegram = parser(frame)
+        except Exception:
+            self._previous.pop(source_id, None)
+            self._record_error(session_factory, source_id, "WarmteLink frame could not be parsed")
+            return False
+        try:
+            with session_factory() as session:
+                admitted = self.ingest(session, source_id=source_id, telegram=telegram)
+                session.commit()
+                return admitted
+        except Exception:
+            self._previous.pop(source_id, None)
+            self._record_error(session_factory, source_id, "WarmteLink ingest failed")
+            return False
+
+    def _admit(
+        self,
+        session: Session,
+        source: MeterSource,
+        snapshot: _FrameSnapshot,
+        quality: str,
+        fingerprints: tuple[str, ...],
+    ) -> None:
+        received_at = _utc(self._clock())
+        for sample, fingerprint in zip(snapshot.samples, fingerprints, strict=True):
+            channel = upsert_discovered_channel(
+                session,
+                source_id=source.id,
+                channel_key=sample.key,
+                label=sample.label,
+                unit=sample.unit,
+                suggested_commodity=_suggestion(sample.unit),
+                device_type=sample.device_type,
+                fingerprint=fingerprint,
+                latest_value=sample.value,
+                latest_at=snapshot.recorded_at,
+                latest_quality=quality,
+            )
+            session.flush()
+            bucket = snapshot.recorded_at.replace(second=0, microsecond=0)
+            exists = session.scalar(
+                select(WarmteLinkReading.id)
+                .where(
+                    WarmteLinkReading.channel_id == channel.id,
+                    WarmteLinkReading.recorded_at >= bucket,
+                    WarmteLinkReading.recorded_at < bucket + timedelta(minutes=1),
+                )
+                .limit(1)
+            )
+            if exists is None:
+                session.add(
+                    WarmteLinkReading(
+                        channel_id=channel.id,
+                        recorded_at=snapshot.recorded_at,
+                        received_at=received_at,
+                        value=sample.value,
+                        unit=sample.unit,
+                        quality=quality,
+                        equipment_fingerprint=fingerprint,
+                    )
+                )
+        source.status = "online"
+        source.last_seen_at = received_at
+        source.last_error = None
+        source.updated_at = received_at
+
+    def _diagnose(self, source: MeterSource, reason: str) -> None:
+        now = _utc(self._clock())
+        source.status = "error"
+        source.last_error = reason
+        source.updated_at = now
+
+    def _record_error(
+        self, session_factory: Callable[[], Session], source_id: int, message: str) -> None:
+        try:
+            with session_factory() as session:
+                source = session.get(MeterSource, source_id)
+                if source is not None:
+                    self._diagnose(source, message)
+                    session.commit()
+        except Exception:
+            # Error reporting itself must not kill another source worker.
+            return
+
+
+def _snapshot(telegram: P1Telegram) -> _FrameSnapshot:
+    recorded_at = _parse_timestamp(telegram.timestamp)
+    samples = tuple(_sample(channel) for channel in telegram.channels)
+    if not samples:
+        raise ValueError("WarmteLink telegram contains no cumulative channels")
+    return _FrameSnapshot(recorded_at, telegram.equipment_fingerprint, samples)
+
+
+def _sample(channel: P1Channel) -> _ChannelSample:
+    profile = _canonical_channel_profile(channel.number)
+    if channel.device_type != profile.device_type:
+        raise ValueError("WarmteLink channel device type is not canonical")
+    if len(channel.readings) != 1:
+        raise ValueError("WarmteLink channel has no unambiguous cumulative reading")
+    reading = channel.readings[0]
+    if reading.code != profile.reading_code or reading.value is None or reading.unit != profile.raw_unit:
+        raise ValueError("WarmteLink cumulative reading is incomplete")
+    # Parser annotations are not a trust boundary: fake or future parser DTOs
+    # must not put a float (including NaN) or a non-finite Decimal into the
+    # per-source unverifiable candidate state.  Do not coerce here: accepting
+    # another numeric type would make the persistence and continuity paths
+    # disagree about the cumulative-value contract.
+    if not isinstance(reading.value, Decimal) or not reading.value.is_finite():
+        raise ValueError("WarmteLink cumulative reading is not a finite Decimal")
+    fingerprint = channel.equipment_fingerprint
+    return _ChannelSample(
+        key=profile.key,
+        label=profile.label,
+        value=reading.value,
+        unit=profile.unit,
+        device_type=profile.device_type,
+        fingerprint=fingerprint,
+        identity=(channel.number, profile.device_type, fingerprint, profile.reading_code, profile.unit),
+    )
+
+
+@dataclass(frozen=True)
+class _CanonicalChannelProfile:
+    key: str
+    label: str
+    device_type: str
+    reading_code: str
+    raw_unit: str
+    unit: str
+
+
+_CANONICAL_CHANNELS = {
+    1: _CanonicalChannelProfile(
+        key="channel-1",
+        label="WarmteLink channel 1",
+        device_type="006",
+        reading_code="0-1:24.2.1",
+        raw_unit="m3",
+        unit="m³",
+    ),
+    2: _CanonicalChannelProfile(
+        key="channel-2",
+        label="WarmteLink channel 2",
+        device_type="012",
+        reading_code="0-2:24.2.1",
+        raw_unit="GJ",
+        unit="GJ",
+    ),
+}
+
+
+def _canonical_channel_profile(number: object) -> _CanonicalChannelProfile:
+    # Do not format or coerce parser supplied channel numbers: that could turn
+    # an arbitrary object into an identity key before it is rejected.
+    if type(number) is not int:
+        raise ValueError("WarmteLink channel number is not canonical")
+    try:
+        return _CANONICAL_CHANNELS[number]
+    except KeyError as exc:
+        raise ValueError("WarmteLink channel is not supported by the profile") from exc
+
+
+def _integrity_status(value: object) -> IntegrityStatus:
+    """Require a real parser integrity enum, never a look-alike value object."""
+    if not isinstance(value, IntegrityStatus):
+        raise ValueError("WarmteLink integrity status is not canonical")
+    return value
+
+
+def _continuity_problem(previous: _FrameSnapshot, current: _FrameSnapshot) -> str | None:
+    if current.recorded_at <= previous.recorded_at:
+        return "Unverifiable WarmteLink timestamp is not strictly increasing"
+    if current.recorded_at - previous.recorded_at != timedelta(seconds=10):
+        return "Unverifiable WarmteLink frame cadence is not 10 seconds"
+    if current.fingerprint != previous.fingerprint:
+        return "WarmteLink equipment metadata changed"
+    if tuple(sample.identity for sample in current.samples) != tuple(sample.identity for sample in previous.samples):
+        return "WarmteLink channel metadata changed or channel set changed"
+    old_values = {sample.key: sample.value for sample in previous.samples}
+    if any(sample.value < old_values[sample.key] for sample in current.samples):
+        return "WarmteLink cumulative value decreased"
+    return None
+
+
+_FINGERPRINT_PATTERN = re.compile(r"[0-9a-f]{64}")
+
+
+def _final_fingerprints(snapshot: _FrameSnapshot) -> tuple[str, ...] | None:
+    """Return only canonical SHA-256 hexdigests safe to persist.
+
+    A parser DTO is an untrusted boundary: even a field named ``fingerprint``
+    can contain a raw equipment identifier.  Validate the complete DTO before
+    choosing persisted values: the top-level value and every channel value
+    must independently be canonical hashes.  This deliberately does not use
+    a top-level fallback for an absent or malformed channel fingerprint.
+    """
+    values = (snapshot.fingerprint, *(sample.fingerprint for sample in snapshot.samples))
+    if any(not _is_canonical_fingerprint(value) for value in values):
+        return None
+    return tuple(sample.fingerprint for sample in snapshot.samples if sample.fingerprint is not None)
+
+
+def _is_canonical_fingerprint(value: str | None) -> bool:
+    return value is not None and _FINGERPRINT_PATTERN.fullmatch(value) is not None
+
+
+def _parse_timestamp(value: str | None) -> datetime:
+    if value is None or len(value) != 13 or value[-1] not in {"S", "W"} or not value[:-1].isdigit():
+        raise ValueError("WarmteLink timestamp is unavailable")
+    naive = datetime.strptime(value[:-1], "%y%m%d%H%M%S")
+    # The telegram's S/W marker is evidence of the intended Amsterdam offset.
+    offset = 2 if value[-1] == "S" else 1
+    return (naive.replace(tzinfo=UTC) - timedelta(hours=offset)).replace(tzinfo=UTC)
+
+
+def _suggestion(unit: str) -> str | None:
+    return {"GJ": "heating", "m³": "hot_water"}.get(unit)
+
+
+def _canonical_unit(unit: str) -> str:
+    """Map the P1 spelling of cubic metres to the source-profile unit."""
+    return "m³" if unit == "m3" else unit
+
+
+def _utc(value: datetime) -> datetime:
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md
index 7d4b2e4..411f100 100644
--- a/docs/design/m8-warmtelink-energy.md
+++ b/docs/design/m8-warmtelink-energy.md
@@ -615,7 +615,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 
 ### M8-T09 — WarmteLink 质量接纳、发现与分钟采样 [structural]
 
-- **Status**: `todo`
+- **Status**: `done`
 - **Depends**: M8-T08
 - **Context**: 把 parser 输出变成可审计的 latest/history；这里锁住最关键的 unverifiable 接纳策略。
 
diff --git a/tests/test_warmtelink_ingest.py b/tests/test_warmtelink_ingest.py
new file mode 100644
index 0000000..48092a2
--- /dev/null
+++ b/tests/test_warmtelink_ingest.py
@@ -0,0 +1,620 @@
+"""Behaviour tests for the WarmteLink per-source admission state machine."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import UTC, datetime
+from decimal import Decimal
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, func, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.integrations.p1 import IntegrityStatus, ObisField, P1Channel, P1Telegram, dsmr_crc16, parse_telegram
+from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
+from app.services.warmtelink_ingest import WarmteLinkIngestor
+
+
+def _config(url: str) -> Config:
+    config = Config("alembic_app.ini")
+    config.set_main_option("sqlalchemy.url", url)
+    return config
+
+
+@pytest.fixture()
+def database(tmp_path: Path):
+    url = f"sqlite:///{tmp_path / 'warmtelink_ingest.db'}"
+    command.upgrade(_config(url), "head")
+    engine = create_engine(url, connect_args={"check_same_thread": False})
+    factory = sessionmaker(bind=engine, autoflush=False, class_=Session)
+    with factory() as session:
+        for name in ("one", "two"):
+            now = datetime(2026, 8, 22, tzinfo=UTC)
+            session.add(
+                MeterSource(
+                    name=name,
+                    kind="warmtelink_serial",
+                    enabled=True,
+                    config={"path": f"/dev/{name}", "baudrate": 115200, "data_bits": 7, "parity": "N", "stop_bits": 1},
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+        session.commit()
+        ids = list(session.scalars(select(MeterSource.id).order_by(MeterSource.id)))
+    yield engine, factory, ids[-2:]
+    engine.dispose()
+
+
+def _telegram(
+    *,
+    second: int,
+    minute: int = 0,
+    integrity: IntegrityStatus = IntegrityStatus.UNVERIFIABLE,
+    water: str = "5.900",
+    heat: str = "0.017",
+    channels: tuple[int, ...] = (1, 2),
+    device_type: str = "006",
+    fingerprint: str | None = "a" * 64,
+    channel_fingerprint: str | None = None,
+    channel_fingerprints: dict[int, str | None] | None = None,
+) -> P1Telegram:
+    fields: list[ObisField] = []
+    parsed_channels: list[P1Channel] = []
+    for number in channels:
+        value, unit = (water, "m3") if number == 1 else (heat, "GJ")
+        code = f"0-{number}:24.2.1"
+        field = ObisField(code, (value + "*" + unit,), Decimal(value), unit)
+        fields.append(field)
+        parsed_channels.append(
+            P1Channel(
+                number,
+                device_type if number == 1 else "012",
+                (
+                    channel_fingerprints[number]
+                    if channel_fingerprints is not None
+                    else fingerprint if channel_fingerprint is None else channel_fingerprint
+                ),
+                (field,),
+            )
+        )
+    return P1Telegram(
+        100,
+        integrity,
+        "fixture",
+        f"26082212{minute:02d}{second:02d}S",
+        fingerprint,
+        tuple(fields),
+        tuple(parsed_channels),
+    )
+
+
+def _counts(factory) -> tuple[int, int]:
+    with factory() as session:
+        return (
+            # Revision 16 supplies one DSMR channel; this task must not touch it.
+            int(session.scalar(select(func.count(MeterSourceChannel.id))) or 0) - 1,
+            int(session.scalar(select(func.count(WarmteLinkReading.id))) or 0),
+        )
+
+
+def _all_counts(factory) -> tuple[int, int, int]:
+    channels, readings = _counts(factory)
+    with factory() as session:
+        bindings = int(session.scalar(select(func.count(MeterSourceBinding.id))) or 0)
+    return channels, readings, bindings
+
+
+def _fixed_clock() -> datetime:
+    return datetime(2026, 8, 22, 10, tzinfo=UTC)
+
+
+def test_unverifiable_requires_two_continuous_frames_and_preserves_true_metadata(database):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
+        session.commit()
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.902"))
+        session.commit()
+    assert _counts(factory) == (2, 2)
+    with factory() as session:
+        channels = list(
+            session.scalars(
+                select(MeterSourceChannel)
+                .where(MeterSourceChannel.source_id == source_id)
+                .order_by(MeterSourceChannel.channel_key)
+            )
+        )
+        assert [(item.channel_key, item.unit, item.device_type, item.latest_quality) for item in channels] == [
+            ("channel-1", "m³", "006", "unverifiable"),
+            ("channel-2", "GJ", "012", "unverifiable"),
+        ]
+        assert all(item.fingerprint == "a" * 64 for item in channels)
+        assert channels[0].latest_value == Decimal("5.902")
+
+
+def test_unverifiable_cadence_requires_exactly_ten_seconds_and_restarts_after_gaps(database):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, water="5.901"))
+        session.commit()
+    assert _counts(factory) == (2, 2)
+
+    restarted = WarmteLinkIngestor()
+    with factory() as session:
+        assert not restarted.ingest(session, source_id=source_id, telegram=_telegram(second=0, water="5.902"))
+        # 20 seconds misses one expected frame; it is only a new candidate.
+        assert not restarted.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.903"))
+        assert restarted.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.904"))
+        # A larger gap likewise needs its own new matching successor.
+        assert not restarted.ingest(session, source_id=source_id, telegram=_telegram(second=0, minute=2, water="5.905"))
+        assert restarted.ingest(session, source_id=source_id, telegram=_telegram(second=10, minute=2, water="5.906"))
+        session.commit()
+    assert _counts(factory) == (2, 4)
+
+
+@pytest.mark.parametrize(
+    "first, second",
+    [
+        (_telegram(second=10), _telegram(second=9)),
+        (_telegram(second=0), _telegram(second=30)),
+        (_telegram(second=10), _telegram(second=20, water="5.899")),
+        (_telegram(second=10), _telegram(second=20, device_type="007")),
+        (_telegram(second=10), _telegram(second=20, channels=(1,))),
+        (_telegram(second=10, channels=(1,)), _telegram(second=20)),
+    ],
+)
+def test_unverifiable_discontinuities_restart_two_frame_confirmation(database, first, second):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=first)
+        assert not ingest.ingest(session, source_id=source_id, telegram=second)
+        session.commit()
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.status == "error" and source.last_error is not None
+
+
+def test_discontinuity_candidate_needs_a_new_matching_successor(database):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, channels=(1,)))
+        # The restored channel set is a fresh candidate, not an admission.
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20))
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.901"))
+        session.commit()
+    assert _counts(factory) == (2, 2)
+
+
+@pytest.mark.parametrize(
+    "first, discontinuous, successor",
+    [
+        (_telegram(second=0), _telegram(second=10, water="5.899"), _telegram(second=20, water="5.900")),
+        (_telegram(second=0), _telegram(second=10, fingerprint="b" * 64), _telegram(second=20, fingerprint="b" * 64)),
+        (_telegram(second=0), _telegram(second=10, channels=(1,)), _telegram(second=20, channels=(1,))),
+        (_telegram(second=0, channels=(1,)), _telegram(second=10), _telegram(second=20)),
+    ],
+)
+def test_each_metadata_or_channel_discontinuity_requires_its_own_two_frames(
+    database, first, discontinuous, successor
+):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=first)
+        # The discontinuous frame becomes a new first candidate, never an admission.
+        assert not ingest.ingest(session, source_id=source_id, telegram=discontinuous)
+        assert ingest.ingest(session, source_id=source_id, telegram=successor)
+        session.commit()
+
+
+def test_parser_failure_clears_only_its_source_candidate_and_requires_two_new_frames(database):
+    _engine, factory, (one, two) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
+        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
+        session.commit()
+
+    def parse_failure(_frame: bytes) -> P1Telegram:
+        raise ValueError("raw telegram must never be reported")
+
+    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=parse_failure)
+    with factory() as session:
+        # One's candidate was cleared; two's independent candidate survives.
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
+        session.commit()
+    assert _counts(factory) == (4, 4)
+
+
+def _normalization_failure(kind: str) -> P1Telegram:
+    telegram = _telegram(second=10)
+    if kind == "no-channels":
+        return replace(telegram, channels=())
+    first, *rest = telegram.channels
+    if kind == "empty-readings":
+        return replace(telegram, channels=(replace(first, readings=()), *rest))
+    if kind == "ambiguous-readings":
+        return replace(telegram, channels=(replace(first, readings=(first.readings[0],) * 2), *rest))
+    if kind == "incomplete-reading":
+        return replace(
+            telegram,
+            channels=(replace(first, readings=(replace(first.readings[0], value=None),)), *rest),
+        )
+    invalid_values: dict[str, object] = {
+        "decimal-nan": Decimal("NaN"),
+        "decimal-positive-infinity": Decimal("Infinity"),
+        "decimal-negative-infinity": Decimal("-Infinity"),
+        "float": 5.9,
+        "string": "private-invalid-cumulative-value",
+    }
+    if kind in invalid_values:
+        return replace(
+            telegram,
+            channels=(
+                replace(first, readings=(replace(first.readings[0], value=invalid_values[kind]),)),
+                *rest,
+            ),
+        )
+    if kind == "invalid-timestamp":
+        return replace(telegram, timestamp="invalid-time")
+    raise AssertionError(f"unexpected normalization failure kind: {kind}")
+
+
+@pytest.mark.parametrize(
+    "kind",
+    [
+        "no-channels",
+        "empty-readings",
+        "ambiguous-readings",
+        "incomplete-reading",
+        "decimal-nan",
+        "decimal-positive-infinity",
+        "decimal-negative-infinity",
+        "float",
+        "string",
+        "invalid-timestamp",
+    ],
+)
+def test_dto_normalization_failure_clears_only_its_source_candidate_and_requires_two_new_frames(
+    database, kind
+):
+    _engine, factory, (one, two) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
+        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
+        session.commit()
+
+    failed = _normalization_failure(kind)
+    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=lambda _raw: failed)
+    # The rejected DTO creates no discovery/latest/history rows, and source two
+    # retains its independent first candidate.
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, one)
+        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
+        assert "private" not in (source.last_error or "")
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
+        session.commit()
+    assert _counts(factory) == (4, 4)
+
+
+def test_valid_invalid_sampling_replay_restart_and_source_isolation(database):
+    _engine, factory, (one, two) = database
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
+    with factory() as session:
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=0, integrity=IntegrityStatus.VALID))
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=10, integrity=IntegrityStatus.VALID, water="5.901"))
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=20, integrity=IntegrityStatus.INVALID))
+        # Source two cannot consume source one's candidate.
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=30))
+        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=40))
+        session.commit()
+    assert _counts(factory) == (2, 2)  # candidates are never discovered; first minute only once
+    # A new process sees the minute bucket in DB and does not create duplicates.
+    restarted = WarmteLinkIngestor(clock=_fixed_clock)
+    with factory() as session:
+        assert restarted.ingest(session, source_id=one, telegram=_telegram(second=50, integrity=IntegrityStatus.VALID))
+        session.commit()
+    assert _counts(factory) == (2, 2)
+
+
+def test_history_keeps_first_utc_minute_sample_and_creates_next_minute(database):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
+    with factory() as session:
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.VALID))
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, integrity=IntegrityStatus.VALID, water="5.901"))
+        assert ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, minute=1, integrity=IntegrityStatus.VALID, water="5.902"),
+        )
+        session.commit()
+    with factory() as session:
+        readings = list(session.scalars(select(WarmteLinkReading).order_by(WarmteLinkReading.recorded_at)))
+        assert len(readings) == 4
+        assert readings[0].recorded_at.second == 0
+        assert readings[2].recorded_at.minute == 1 and readings[2].recorded_at.second == 0
+        channel = session.scalar(
+            select(MeterSourceChannel).where(
+                MeterSourceChannel.source_id == source_id,
+                MeterSourceChannel.channel_key == "channel-1",
+            )
+        )
+        assert channel is not None and channel.latest_value == Decimal("5.902")
+
+
+def test_handle_frame_rolls_back_database_error_marks_source_and_recovers(database, monkeypatch):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor()
+    telegram = _telegram(second=0, integrity=IntegrityStatus.VALID)
+    import app.services.warmtelink_ingest as module
+
+    original = module.upsert_discovered_channel
+
+    def broken(*_args, **_kwargs):
+        raise RuntimeError("database unavailable")
+
+    monkeypatch.setattr(module, "upsert_discovered_channel", broken)
+    assert not ingest.handle_frame(source_id, b"private telegram", session_factory=factory, parser=lambda _raw: telegram)
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.status == "error"
+        assert "private" not in (source.last_error or "")
+    monkeypatch.setattr(module, "upsert_discovered_channel", original)
+    assert ingest.handle_frame(source_id, b"ignored", session_factory=factory, parser=lambda _raw: telegram)
+    assert _counts(factory) == (2, 2)
+
+
+def test_second_channel_failure_rolls_back_first_channel_latest_and_history(database, monkeypatch):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor()
+    import app.services.warmtelink_ingest as module
+
+    original = module.upsert_discovered_channel
+
+    def broken_on_second(*args, **kwargs):
+        if kwargs["channel_key"] == "channel-2":
+            raise RuntimeError("second channel failed")
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(module, "upsert_discovered_channel", broken_on_second)
+    assert not ingest.handle_frame(
+        source_id,
+        b"private",
+        session_factory=factory,
+        parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID),
+    )
+    assert _counts(factory) == (0, 0)
+    monkeypatch.setattr(module, "upsert_discovered_channel", original)
+    assert ingest.handle_frame(
+        source_id,
+        b"private",
+        session_factory=factory,
+        parser=lambda _raw: _telegram(second=10, integrity=IntegrityStatus.VALID),
+    )
+    assert _counts(factory) == (2, 2)
+
+
+def test_unverifiable_database_failure_clears_only_its_source_sliding_candidate(database, monkeypatch):
+    _engine, factory, (one, two) = database
+    ingest = WarmteLinkIngestor()
+    import app.services.warmtelink_ingest as module
+
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
+        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
+        session.commit()
+
+    original = module.upsert_discovered_channel
+    monkeypatch.setattr(module, "upsert_discovered_channel", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError()))
+    assert not ingest.handle_frame(
+        one, b"private", session_factory=factory, parser=lambda _raw: _telegram(second=10, water="5.901")
+    )
+    monkeypatch.setattr(module, "upsert_discovered_channel", original)
+    assert _all_counts(factory) == (0, 0, 0)
+
+    with factory() as session:
+        # One needs a new pair after its failed write; two retains its own candidate.
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
+        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=30, water="5.903"))
+        session.commit()
+    assert _all_counts(factory) == (4, 4, 0)
+
+
+def _valid_crc_frame(*, device_type: str = "006") -> bytes:
+    body = (
+        b"/WARMTE\r\n"
+        b"0-0:1.0.0(260822120000S)\r\n"
+        b"0-0:96.1.1(WARMTE-REDACTED)\r\n"
+        + f"0-1:24.1.0({device_type})\r\n".encode()
+        + b"0-1:96.1.0(KAM-REDACTED)\r\n"
+        + b"0-1:24.2.1(260822120000S)(5.900*m3)\r\n"
+        + b"0-2:24.1.0(012)\r\n"
+        + b"0-2:96.1.0(KAM-REDACTED)\r\n"
+        + b"0-2:24.2.1(260822120000S)(0.017*GJ)\r\n"
+    )
+    payload = body + b"!"
+    return payload + f"{dsmr_crc16(payload):04X}".encode() + b"\r\n"
+
+
+def test_non_enum_integrity_and_fake_metadata_reject_whole_frame_per_source(database):
+    _engine, factory, (one, two) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
+        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
+        session.commit()
+
+    forged_quality = replace(_telegram(second=10, integrity=IntegrityStatus.VALID), integrity=SimpleNamespace(value="valid"))
+    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=lambda _raw: forged_quality)
+    mixed = replace(_telegram(second=10), channels=(replace(_telegram(second=10).channels[0], device_type="raw-device"),))
+    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=lambda _raw: mixed)
+    assert _all_counts(factory) == (0, 0, 0)
+
+    with factory() as session:
+        source = session.get(MeterSource, one)
+        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
+        # Both malformed frames clear one, while two's candidate is untouched.
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.901"))
+        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=30, water="5.902"))
+        session.commit()
+    assert _all_counts(factory) == (4, 4, 0)
+
+
+def test_valid_crc_parser_frame_with_raw_device_type_is_rejected_without_leakage(database):
+    _engine, factory, (one, two) = database
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
+        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
+        session.commit()
+
+    raw_identifier = "serial-number-which-must-not-persist"
+    assert parse_telegram(_valid_crc_frame(device_type=raw_identifier)).integrity is IntegrityStatus.VALID
+    assert not ingest.handle_frame(one, _valid_crc_frame(device_type=raw_identifier), session_factory=factory)
+    assert _all_counts(factory) == (0, 0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, one)
+        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
+        assert raw_identifier not in (source.last_error or "")
+        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
+        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
+        session.commit()
+    assert _all_counts(factory) == (4, 4, 0)
+
+
+@pytest.mark.parametrize(
+    "fingerprint, channel_fingerprint",
+    [
+        (None, None),
+        ("", ""),
+        ("a" * 63, "a" * 63),
+        ("g" * 64, "g" * 64),
+        ("serial-number-which-must-not-persist", "serial-number-which-must-not-persist"),
+    ],
+)
+def test_invalid_fingerprint_fails_closed_without_persisting_identifiers(
+    database, fingerprint, channel_fingerprint
+):
+    _engine, factory, (source_id, _) = database
+    raw_identifier = "serial-number-which-must-not-persist"
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.INVALID))
+        assert not ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(
+                second=10,
+                integrity=IntegrityStatus.VALID,
+                fingerprint=fingerprint,
+                channel_fingerprint=channel_fingerprint,
+            ),
+        )
+        session.commit()
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.last_error == "WarmteLink frame fingerprint is invalid"
+        assert raw_identifier not in (source.last_error or "")
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, integrity=IntegrityStatus.VALID))
+        session.commit()
+        rendered = " ".join(str(row) for row in session.execute(select(WarmteLinkReading)).all())
+        assert raw_identifier not in rendered
+        assert all("raw" not in column.name for column in WarmteLinkReading.__table__.columns)
+
+
+@pytest.mark.parametrize(
+    "invalid_fingerprint",
+    [None, "", "a" * 63, "serial-number-which-must-not-persist"],
+    ids=["none", "empty", "wrong-length", "raw-like"],
+)
+@pytest.mark.parametrize("invalid_channel", [1, 2])
+def test_invalid_channel_fingerprint_rejects_whole_frame_and_clears_candidate(
+    database, invalid_fingerprint, invalid_channel
+):
+    _engine, factory, (source_id, _) = database
+    raw_identifier = "serial-number-which-must-not-persist"
+    valid = "a" * 64
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        # Establish a candidate, then prove an invalid channel clears only it.
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
+        assert not ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(
+                second=10,
+                fingerprint=valid,
+                channel_fingerprints={1: valid, 2: valid} | {invalid_channel: invalid_fingerprint},
+            ),
+        )
+        session.commit()
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.last_error == "WarmteLink frame fingerprint is invalid"
+        assert raw_identifier not in (source.last_error or "")
+        # The first legal frame is a fresh candidate; only its successor admits.
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.901"))
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.902"))
+        session.commit()
+    assert _counts(factory) == (2, 2)
+
+
+@pytest.mark.parametrize(
+    "invalid_fingerprint",
+    [None, "", "a" * 63, "serial-number-which-must-not-persist"],
+    ids=["none", "empty", "wrong-length", "raw-like"],
+)
+def test_invalid_top_fingerprint_rejects_whole_frame_and_clears_candidate(database, invalid_fingerprint):
+    _engine, factory, (source_id, _) = database
+    raw_identifier = "serial-number-which-must-not-persist"
+    valid = "a" * 64
+    ingest = WarmteLinkIngestor()
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
+        assert not ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(
+                second=10,
+                fingerprint=invalid_fingerprint,
+                channel_fingerprints={1: valid, 2: valid},
+            ),
+        )
+        session.commit()
+    assert _counts(factory) == (0, 0)
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.last_error == "WarmteLink frame fingerprint is invalid"
+        assert raw_identifier not in (source.last_error or "")
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.901"))
+        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.902"))
+        session.commit()
+    assert _counts(factory) == (2, 2)