M8-R04: interpret WarmteLink timestamps as Amsterdam wall time

app/services/warmtelink_ingest.py

@@ -13,6 +13,7 @@ from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from decimal import Decimal
 import re
+from zoneinfo import ZoneInfo
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -52,7 +53,14 @@ class WarmteLinkIngestor:
         self._clock = clock or (lambda: datetime.now(UTC))
         self._previous: dict[int, _FrameSnapshot] = {}
 
-    def ingest(self, session: Session, *, source_id: int, telegram: P1Telegram) -> bool:
+    def ingest(
+        self,
+        session: Session,
+        *,
+        source_id: int,
+        telegram: P1Telegram,
+        received_at: datetime | None = None,
+    ) -> bool:
         """Apply one parsed telegram in the caller's transaction.
 
         Returns whether the frame was admitted.  Callers that own a session
@@ -65,44 +73,47 @@ class WarmteLinkIngestor:
         if source.kind != "warmtelink_serial":
             raise ValueError("Source is not a WarmteLink serial source")
 
+        received_at = _utc(self._clock()) if received_at is None else _utc(received_at)
         try:
             integrity = _integrity_status(telegram.integrity)
         except Exception:
             self._previous.pop(source_id, None)
-            self._diagnose(source, "WarmteLink frame could not be normalized")
+            self._diagnose(source, "WarmteLink frame could not be normalized", now=received_at)
             return False
 
         if integrity is IntegrityStatus.INVALID:
             self._previous.pop(source_id, None)
-            self._diagnose(source, "WarmteLink frame checksum is invalid")
+            self._diagnose(source, "WarmteLink frame checksum is invalid", now=received_at)
             return False
 
         try:
-            snapshot = _snapshot(telegram)
+            snapshot = _snapshot(telegram, received_at)
             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")
+            self._diagnose(source, "WarmteLink frame could not be normalized", now=received_at)
             return False
         if fingerprints is None:
             self._previous.pop(source_id, None)
-            self._diagnose(source, "WarmteLink frame fingerprint is invalid")
+            self._diagnose(source, "WarmteLink frame fingerprint is invalid", now=received_at)
             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")
+                self._diagnose(
+                    source, "Awaiting a second matching unverifiable WarmteLink frame", now=received_at
+                )
                 return False
             reason = _continuity_problem(previous, snapshot)
             if reason is not None:
                 self._previous[source_id] = snapshot
-                self._diagnose(source, reason)
+                self._diagnose(source, reason, now=received_at)
                 return False
         try:
-            self._admit(session, source, snapshot, integrity.value, fingerprints)
+            self._admit(session, source, snapshot, integrity.value, fingerprints, received_at)
         except Exception:
             # ``ingest`` owns flushes and may be used directly by tests or
             # future callers.  A failed write must never become a speculative
@@ -133,20 +144,25 @@ class WarmteLinkIngestor:
         a generic source error.  Consequently no partial latest/history update
         survives and the following frame may recover normally.
         """
+        received_at = _utc(self._clock())
         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")
+            self._record_error(
+                session_factory, source_id, "WarmteLink frame could not be parsed", now=received_at
+            )
             return False
         try:
             with session_factory() as session:
-                admitted = self.ingest(session, source_id=source_id, telegram=telegram)
+                admitted = self.ingest(
+                    session, source_id=source_id, telegram=telegram, received_at=received_at
+                )
                 session.commit()
                 return admitted
         except Exception:
             self._previous.pop(source_id, None)
-            self._record_error(session_factory, source_id, "WarmteLink ingest failed")
+            self._record_error(session_factory, source_id, "WarmteLink ingest failed", now=received_at)
             return False
 
     def _admit(
@@ -156,8 +172,8 @@ class WarmteLinkIngestor:
         snapshot: _FrameSnapshot,
         quality: str,
         fingerprints: tuple[str, ...],
+        received_at: datetime,
     ) -> None:
-        received_at = _utc(self._clock())
         for sample, fingerprint in zip(snapshot.samples, fingerprints, strict=True):
             channel = upsert_discovered_channel(
                 session,
@@ -200,27 +216,33 @@ class WarmteLinkIngestor:
         source.last_error = None
         source.updated_at = received_at
 
-    def _diagnose(self, source: MeterSource, reason: str) -> None:
-        now = _utc(self._clock())
+    def _diagnose(self, source: MeterSource, reason: str, *, now: datetime | None = None) -> None:
+        now = _utc(self._clock()) if now is None else now
         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:
+        self,
+        session_factory: Callable[[], Session],
+        source_id: int,
+        message: str,
+        *,
+        now: datetime,
+    ) -> None:
         try:
             with session_factory() as session:
                 source = session.get(MeterSource, source_id)
                 if source is not None:
-                    self._diagnose(source, message)
+                    self._diagnose(source, message, now=now)
                     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)
+def _snapshot(telegram: P1Telegram, received_at: datetime) -> _FrameSnapshot:
+    recorded_at = _parse_timestamp(telegram.timestamp, received_at)
     samples = tuple(_sample(channel) for channel in telegram.channels)
     if not samples:
         raise ValueError("WarmteLink telegram contains no cumulative channels")
@@ -340,13 +362,30 @@ 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:
+_AMSTERDAM = ZoneInfo("Europe/Amsterdam")
+_MAX_CLOCK_SKEW = timedelta(minutes=5)
+
+
+def _parse_timestamp(value: str | None, received_at: datetime) -> 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)
+    # The S/W marker is only advisory: deployed devices have emitted W while
+    # on CEST.  Validate both folds by a UTC round trip, which rejects spring
+    # gaps and leaves one (ordinary) or two (fall-back) real instants.
+    candidates: list[datetime] = []
+    for fold in (0, 1):
+        candidate = naive.replace(tzinfo=_AMSTERDAM, fold=fold).astimezone(UTC)
+        local = candidate.astimezone(_AMSTERDAM)
+        if local.replace(tzinfo=None) == naive and local.fold == fold and candidate not in candidates:
+            candidates.append(candidate)
+    if not candidates:
+        raise ValueError("WarmteLink timestamp is unavailable")
+    received_at = _utc(received_at)
+    recorded_at = min(candidates, key=lambda candidate: abs(candidate - received_at))
+    if abs(recorded_at - received_at) > _MAX_CLOCK_SKEW:
+        raise ValueError("WarmteLink timestamp is unavailable")
+    return recorded_at
 
 
 def _suggestion(unit: str) -> str | None:

tests/test_warmtelink_ingest.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import replace
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from decimal import Decimal
 from pathlib import Path
 from types import SimpleNamespace
@@ -62,6 +62,7 @@ def _telegram(
     fingerprint: str | None = "a" * 64,
     channel_fingerprint: str | None = None,
     channel_fingerprints: dict[int, str | None] | None = None,
+    timestamp: str | None = None,
 ) -> P1Telegram:
     fields: list[ObisField] = []
     parsed_channels: list[P1Channel] = []
@@ -86,7 +87,7 @@ def _telegram(
         100,
         integrity,
         "fixture",
-        f"26082212{minute:02d}{second:02d}S",
+        timestamp or f"26082212{minute:02d}{second:02d}S",
         fingerprint,
         tuple(fields),
         tuple(parsed_channels),
@@ -113,6 +114,260 @@ def _fixed_clock() -> datetime:
     return datetime(2026, 8, 22, 10, tzinfo=UTC)
 
 
+def _clock_at(value: datetime):
+    return lambda: value
+
+
+class _AdvancingClock:
+    def __init__(self, initial: datetime) -> None:
+        self._next = initial
+        self.calls = 0
+
+    def __call__(self) -> datetime:
+        value = self._next
+        self._next += timedelta(minutes=1)
+        self.calls += 1
+        return value
+
+
+def _as_utc(value: datetime) -> datetime:
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+
+
+def test_summer_wall_clock_ignores_incorrect_w_marker(database):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 23, 21, 40, tzinfo=UTC)
+    ingest = WarmteLinkIngestor(clock=_clock_at(received_at))
+    with factory() as session:
+        assert ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="260823234000W"),
+        )
+        session.commit()
+    with factory() as session:
+        reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id))
+        assert reading is not None
+        assert _as_utc(reading.recorded_at) == datetime(2026, 8, 23, 21, 40, tzinfo=UTC)
+
+
+def test_winter_wall_clock_uses_amsterdam_standard_time_even_with_s_marker(database):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 1, 23, 22, 40, tzinfo=UTC)
+    ingest = WarmteLinkIngestor(clock=_clock_at(received_at))
+    with factory() as session:
+        assert ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="260123234000S"),
+        )
+        session.commit()
+    with factory() as session:
+        reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id))
+        assert reading is not None
+        assert _as_utc(reading.recorded_at) == datetime(2026, 1, 23, 22, 40, tzinfo=UTC)
+
+
+@pytest.mark.parametrize(
+    ("received_at", "expected"),
+    [
+        (datetime(2026, 10, 25, 0, 30, tzinfo=UTC), datetime(2026, 10, 25, 0, 30, tzinfo=UTC)),
+        (datetime(2026, 10, 25, 1, 30, tzinfo=UTC), datetime(2026, 10, 25, 1, 30, tzinfo=UTC)),
+    ],
+)
+def test_fall_back_fold_is_selected_from_received_at(database, received_at, expected):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor(clock=_clock_at(received_at))
+    with factory() as session:
+        assert ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="261025023000W"),
+        )
+        session.commit()
+    with factory() as session:
+        reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id))
+        assert reading is not None and _as_utc(reading.recorded_at) == expected
+
+
+@pytest.mark.parametrize("timestamp", ["260329023000S", "260823234000X"])
+def test_nonexistent_or_invalid_marker_timestamp_fails_closed_with_generic_diagnostic(database, timestamp):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor(clock=_clock_at(datetime(2026, 3, 29, 1, 30, tzinfo=UTC)))
+    with factory() as session:
+        assert not ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp=timestamp),
+        )
+        session.commit()
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
+    assert _counts(factory) == (0, 0)
+
+
+def test_timestamp_beyond_clock_skew_fails_closed_with_generic_diagnostic(database):
+    _engine, factory, (source_id, _) = database
+    ingest = WarmteLinkIngestor(clock=_clock_at(datetime(2026, 8, 23, 20, tzinfo=UTC)))
+    with factory() as session:
+        assert not ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="260823234000W"),
+        )
+        session.commit()
+        source = session.get(MeterSource, source_id)
+        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
+    assert _counts(factory) == (0, 0)
+
+
+@pytest.mark.parametrize("timestamp", ["260823234000X", "260823234600W"])
+def test_post_clock_rejections_reuse_received_at_for_diagnostic(database, timestamp):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 23, 21, 40, tzinfo=UTC)
+    clock = _AdvancingClock(received_at)
+    ingest = WarmteLinkIngestor(clock=clock)
+    with factory() as session:
+        assert not ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp=timestamp),
+        )
+        session.commit()
+        source = session.get(MeterSource, source_id)
+        assert source is not None
+        assert _as_utc(source.updated_at) == received_at
+    assert clock.calls == 1
+
+
+def test_first_unverifiable_candidate_reuses_received_at_for_diagnostic(database):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 22, 10, tzinfo=UTC)
+    clock = _AdvancingClock(received_at)
+    ingest = WarmteLinkIngestor(clock=clock)
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
+        session.commit()
+        source = session.get(MeterSource, source_id)
+        assert source is not None
+        assert _as_utc(source.updated_at) == received_at
+    assert clock.calls == 1
+
+
+def test_accepted_frame_uses_received_at_for_all_source_timestamps(database):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 22, 10, tzinfo=UTC)
+    clock = _AdvancingClock(received_at)
+    ingest = WarmteLinkIngestor(clock=clock)
+    with factory() as session:
+        assert ingest.ingest(
+            session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.VALID)
+        )
+        session.commit()
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id))
+        assert source is not None and reading is not None
+        assert _as_utc(reading.received_at) == received_at
+        assert _as_utc(source.last_seen_at) == received_at
+        assert _as_utc(source.updated_at) == received_at
+    assert clock.calls == 1
+
+
+@pytest.mark.parametrize(
+    "path",
+    ["parser", "integrity", "normalization", "fingerprint", "candidate", "continuity", "accepted"],
+)
+def test_handle_frame_samples_the_receive_clock_once_for_each_admission_path(database, path):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 22, 10, tzinfo=UTC)
+    clock = _AdvancingClock(received_at)
+    ingest = WarmteLinkIngestor(clock=clock)
+
+    if path == "parser":
+        assert not ingest.handle_frame(
+            source_id,
+            b"private",
+            session_factory=factory,
+            parser=lambda _raw: (_ for _ in ()).throw(ValueError()),
+        )
+        expected_at = received_at
+    elif path == "integrity":
+        assert not ingest.handle_frame(
+            source_id,
+            b"private",
+            session_factory=factory,
+            parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.INVALID),
+        )
+        expected_at = received_at
+    elif path == "normalization":
+        assert not ingest.handle_frame(
+            source_id,
+            b"private",
+            session_factory=factory,
+            parser=lambda _raw: _telegram(second=0, timestamp="invalid-time"),
+        )
+        expected_at = received_at
+    elif path == "fingerprint":
+        assert not ingest.handle_frame(
+            source_id,
+            b"private",
+            session_factory=factory,
+            parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID, fingerprint="invalid"),
+        )
+        expected_at = received_at
+    elif path == "candidate":
+        assert not ingest.handle_frame(
+            source_id, b"private", session_factory=factory, parser=lambda _raw: _telegram(second=0)
+        )
+        expected_at = received_at
+    elif path == "continuity":
+        assert not ingest.handle_frame(
+            source_id, b"private", session_factory=factory, parser=lambda _raw: _telegram(second=0)
+        )
+        assert not ingest.handle_frame(
+            source_id,
+            b"private",
+            session_factory=factory,
+            parser=lambda _raw: _telegram(second=20, water="5.901"),
+        )
+        expected_at = received_at + timedelta(minutes=1)
+    else:
+        assert ingest.handle_frame(
+            source_id,
+            b"private",
+            session_factory=factory,
+            parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID),
+        )
+        expected_at = received_at
+
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None
+        assert _as_utc(source.updated_at) == expected_at
+    assert clock.calls == (2 if path == "continuity" else 1)
+
+
+def test_unverifiable_summer_w_frames_keep_utc_continuity(database):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 23, 21, 40, 10, tzinfo=UTC)
+    ingest = WarmteLinkIngestor(clock=_clock_at(received_at))
+    with factory() as session:
+        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, timestamp="260823234000W"))
+        assert ingest.ingest(
+            session,
+            source_id=source_id,
+            telegram=_telegram(second=10, water="5.901", timestamp="260823234010W"),
+        )
+        session.commit()
+    with factory() as session:
+        readings = list(session.scalars(select(WarmteLinkReading).order_by(WarmteLinkReading.id)))
+        assert [_as_utc(reading.recorded_at) for reading in readings] == [
+            datetime(2026, 8, 23, 21, 40, 10, tzinfo=UTC),
+            datetime(2026, 8, 23, 21, 40, 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)
@@ -143,14 +398,14 @@ def test_unverifiable_requires_two_continuous_frames_and_preserves_true_metadata
 
 def test_unverifiable_cadence_requires_exactly_ten_seconds_and_restarts_after_gaps(database):
     _engine, factory, (source_id, _) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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()
+    restarted = WarmteLinkIngestor(clock=_fixed_clock)
     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.
@@ -176,7 +431,7 @@ def test_unverifiable_cadence_requires_exactly_ten_seconds_and_restarts_after_ga
 )
 def test_unverifiable_discontinuities_restart_two_frame_confirmation(database, first, second):
     _engine, factory, (source_id, _) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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)
@@ -189,7 +444,7 @@ def test_unverifiable_discontinuities_restart_two_frame_confirmation(database, f
 
 def test_discontinuity_candidate_needs_a_new_matching_successor(database):
     _engine, factory, (source_id, _) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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,)))
@@ -213,7 +468,7 @@ def test_each_metadata_or_channel_discontinuity_requires_its_own_two_frames(
     database, first, discontinuous, successor
 ):
     _engine, factory, (source_id, _) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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.
@@ -224,7 +479,7 @@ def test_each_metadata_or_channel_discontinuity_requires_its_own_two_frames(
 
 def test_parser_failure_clears_only_its_source_candidate_and_requires_two_new_frames(database):
     _engine, factory, (one, two) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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))
@@ -296,7 +551,7 @@ def test_dto_normalization_failure_clears_only_its_source_candidate_and_requires
     database, kind
 ):
     _engine, factory, (one, two) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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))
@@ -366,7 +621,9 @@ def test_history_keeps_first_utc_minute_sample_and_creates_next_minute(database)
 
 def test_handle_frame_rolls_back_database_error_marks_source_and_recovers(database, monkeypatch):
     _engine, factory, (source_id, _) = database
-    ingest = WarmteLinkIngestor()
+    received_at = datetime(2026, 8, 22, 10, tzinfo=UTC)
+    clock = _AdvancingClock(received_at)
+    ingest = WarmteLinkIngestor(clock=clock)
     telegram = _telegram(second=0, integrity=IntegrityStatus.VALID)
     import app.services.warmtelink_ingest as module
 
@@ -382,14 +639,51 @@ def test_handle_frame_rolls_back_database_error_marks_source_and_recovers(databa
         source = session.get(MeterSource, source_id)
         assert source is not None and source.status == "error"
         assert "private" not in (source.last_error or "")
+        assert _as_utc(source.updated_at) == received_at
+    assert clock.calls == 1
     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_handle_frame_commit_failure_reuses_its_receive_clock_for_error_diagnostic(database, monkeypatch):
+    _engine, factory, (source_id, _) = database
+    received_at = datetime(2026, 8, 22, 10, tzinfo=UTC)
+    clock = _AdvancingClock(received_at)
+    ingest = WarmteLinkIngestor(clock=clock)
+    original_commit = Session.commit
+    sessions_created = 0
+
+    def session_factory():
+        nonlocal sessions_created
+        session = factory()
+        session.info["fail_commit"] = sessions_created == 0
+        sessions_created += 1
+        return session
+
+    def fail_initial_commit(self):
+        if self.info.get("fail_commit"):
+            raise RuntimeError("database commit failed")
+        return original_commit(self)
+
+    monkeypatch.setattr(Session, "commit", fail_initial_commit)
+    assert not ingest.handle_frame(
+        source_id,
+        b"private",
+        session_factory=session_factory,
+        parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID),
+    )
+    with factory() as session:
+        source = session.get(MeterSource, source_id)
+        assert source is not None
+        assert source.status == "error"
+        assert _as_utc(source.updated_at) == received_at
+    assert clock.calls == 1
+
+
 def test_second_channel_failure_rolls_back_first_channel_latest_and_history(database, monkeypatch):
     _engine, factory, (source_id, _) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     import app.services.warmtelink_ingest as module
 
     original = module.upsert_discovered_channel
@@ -419,7 +713,7 @@ def test_second_channel_failure_rolls_back_first_channel_latest_and_history(data
 
 def test_unverifiable_database_failure_clears_only_its_source_sliding_candidate(database, monkeypatch):
     _engine, factory, (one, two) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     import app.services.warmtelink_ingest as module
 
     with factory() as session:
@@ -462,7 +756,7 @@ def _valid_crc_frame(*, device_type: str = "006") -> bytes:
 
 def test_non_enum_integrity_and_fake_metadata_reject_whole_frame_per_source(database):
     _engine, factory, (one, two) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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))
@@ -487,7 +781,7 @@ def test_non_enum_integrity_and_fake_metadata_reject_whole_frame_per_source(data
 
 def test_valid_crc_parser_frame_with_raw_device_type_is_rejected_without_leakage(database):
     _engine, factory, (one, two) = database
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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))
@@ -523,7 +817,7 @@ def test_invalid_fingerprint_fails_closed_without_persisting_identifiers(
 ):
     _engine, factory, (source_id, _) = database
     raw_identifier = "serial-number-which-must-not-persist"
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     with factory() as session:
         assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.INVALID))
         assert not ingest.ingest(
@@ -561,7 +855,7 @@ def test_invalid_channel_fingerprint_rejects_whole_frame_and_clears_candidate(
     _engine, factory, (source_id, _) = database
     raw_identifier = "serial-number-which-must-not-persist"
     valid = "a" * 64
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     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))
@@ -596,7 +890,7 @@ def test_invalid_top_fingerprint_rejects_whole_frame_and_clears_candidate(databa
     _engine, factory, (source_id, _) = database
     raw_identifier = "serial-number-which-must-not-persist"
     valid = "a" * 64
-    ingest = WarmteLinkIngestor()
+    ingest = WarmteLinkIngestor(clock=_fixed_clock)
     with factory() as session:
         assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
         assert not ingest.ingest(