Separate Observe registration answers from pushes

README.md

@@ -87,6 +87,25 @@ Optional `on_observe_pending`, `on_legacy_notification`, and `on_observe_error`
 constructor callbacks let consumers keep that compatibility path distinct from
 confirmed RFC notifications and ordinary polling.
 
+`on_observe_delivery` replaces `on_notification` for consumers that need the
+whole relation context rather than `(href, payload)`. It receives one
+`ObserveDelivery`, whose `registration` field separates the server's answer to
+the register CON from a change the server chose to send, and whose `query`
+completes the relation identity when one href carries several query-qualified
+relations. `sequence` is the Observe option value, or `None` on the optionless
+responses some firmware sends. Setting it suppresses `on_notification` and
+`on_legacy_notification`, so representations are delivered once:
+
+```python
+def on_delivery(delivery):
+    if delivery.registration:
+        seed(delivery.href, delivery.payload)   # answered because we asked
+    else:
+        record_push(delivery.href, delivery.payload)
+
+sess = DtlsCoapSession(..., on_observe_delivery=on_delivery)
+```
+
 Periodic renewal can target only the relations that need it; unrelated
 observations remain active. Existing query variants are preserved unless the
 caller supplies an explicit replacement:
@@ -740,7 +759,9 @@ There are two parallel paths between the appliance and the app over the local Co
 
 In normal operation both happen at once: an OBSERVE notification arrives first, the cache absorbs it, and the next-poll timer for that resource is reset. In an air-gapped LAN the app keeps working. Only the worst-case freshness changes (from ~100 ms with push to ≤1 s on hot-tier resources via polling). Reads, writes, and HA entities behave identically.
 
-Which path is doing the work is visible in Home Assistant. The bridge publishes per-appliance diagnostic entities including **Push Active** (on while OBSERVE is firing), **Last Update Source** (`observe` / `poll` / `sweep` / `optimistic`), **Last OBSERVE Age**, **Poll Max RTT**, **Slow Polls (window)**, **Poll Errors (window)**, and **Stalest Resource Age**, all under each device's Diagnostic section.
+Which path is doing the work is visible in Home Assistant. The bridge publishes per-appliance diagnostic entities including **Push Active** (on while OBSERVE is firing), **Last Update Source** (`observe` / `observe-register` / `poll` / `sweep` / `optimistic`), **Last OBSERVE Age**, **Poll Max RTT**, **Slow Polls (window)**, **Poll Errors (window)**, and **Stalest Resource Age**, all under each device's Diagnostic section.
+
+**Push Active** counts only what the appliance sent of its own accord. A device answers every OBSERVE register CON with the current representation, and that answer reaches the notification callback exactly as a spontaneous notification does. The bridge reads `ObserveDelivery.registration` to tell them apart and records the answer as `observe-register`, so an appliance with no route to Samsung's cloud reads offline instead of going online for the window after every connect.
 
 ---
 

mqtt_demo/bridge.py

@@ -197,23 +197,35 @@ def _on_cache_change(self, changed: bool, source: str) -> None:
                 self._last_observe_change_ts = self.last_change_ts
         self.maybe_publish_state()
 
-    def _on_notification(self, href, payload_bytes):
-        """Reader-thread callback for OBSERVE notifications. Large
-        resources (oven /mode/vs/0 ~9KB) arrive truncated with Block2.M=1
-        and we use cbor-decode failure as the partial signal."""
+    def _on_observe_delivery(self, delivery) -> None:
+        """Reader-thread callback for everything arriving on an OBSERVE
+        relation.
+
+        The session answers the question the bridge cannot: whether this
+        representation is the device's reply to our register CON or a
+        change it chose to send (#41). Only the latter is push.
+        """
+        self._on_notification(
+            delivery.href, delivery.payload,
+            source=('observe-register' if delivery.registration
+                    else 'observe'))
+
+    def _on_notification(self, href, payload_bytes, source='observe'):
+        """Large resources (oven /mode/vs/0 ~9KB) arrive truncated with
+        Block2.M=1 and we use cbor-decode failure as the partial signal."""
         if not payload_bytes:
-            self._schedule_fetchback(href)
+            self._schedule_fetchback(href, source=source)
             return
         try:
             rep = cbor2.loads(payload_bytes)
         except Exception:
-            self._schedule_fetchback(href)
+            self._schedule_fetchback(href, source=source)
             return
         if not isinstance(rep, dict):
             return
         if DEBUG_BRIDGE:
             self._debug_log_rep(href, rep)
-        self.cache.apply_rep(href, rep, source='observe')
+        self.cache.apply_rep(href, rep, source=source)
 
     def _debug_log_rep(self, href, rep):
         if href == '/mode/vs/0' and isinstance(rep, dict):
@@ -223,18 +235,20 @@ def _debug_log_rep(self, href, rep):
         elif href in ('/operational/state/vs/0', '/oven/vs/0', '/power/vs/0'):
             self.log.info("REP %s = %r", href, rep)
 
-    def _schedule_fetchback(self, href, delay_s: float = 0.0):
+    def _schedule_fetchback(self, href, delay_s: float = 0.0,
+                            source: str = 'observe'):
         with self._fetch_lock:
             gen = self._fetch_gen.get(href, 0) + 1
             self._fetch_gen[href] = gen
         threading.Thread(
             target=self._fetch_back,
-            args=(href, delay_s, gen),
+            args=(href, delay_s, gen, source),
             daemon=True,
             name=f'fetch{href}',
         ).start()
 
-    def _fetch_back(self, href, delay_s: float, gen: int):
+    def _fetch_back(self, href, delay_s: float, gen: int,
+                    source: str = 'observe'):
         if delay_s > 0 and self.stop.wait(delay_s):
             return
         with self._fetch_lock:
@@ -261,7 +275,7 @@ def _fetch_back(self, href, delay_s: float, gen: int):
             self.log.warning("fetchback %s cbor: %s", href, e)
             return
         if isinstance(rep, dict):
-            self.cache.apply_rep(href, rep, source='observe')
+            self.cache.apply_rep(href, rep, source=source)
 
     def _retag_logger_with_serial(self):
         if self._serial is not None:
@@ -347,7 +361,7 @@ def session_once(self):
             self.app.ip, port,
             cert_path=self.shared.CERT_PATH,
             key_path=self.shared.KEY_PATH,
-            on_notification=self._on_notification,
+            on_observe_delivery=self._on_observe_delivery,
             local_port=DTLS_LOCAL_PORT_BASE + self.app.index,
             write_max_attempts=self.shared.WRITE_MAX_ATTEMPTS,
         )

smartthings_local/protocol/dtls_session.py

@@ -164,6 +164,62 @@ class _EtagChanged(Exception):
     transfer, so the blocks in hand are from two different versions."""
 
 
+def _openssl_error_reasons(error):
+    """The reason strings OpenSSL recorded, and nothing else.
+
+    A handshake that dies at the TLS layer raises ``SSL.Error``, and the
+    only thing that says why is the alert inside it. That detail cannot
+    go into the raised ``SessionError``: the errors module deliberately
+    refuses arbitrary detail, because backend errors elsewhere can carry
+    remote endpoints, local paths or credential metadata. So it goes to
+    the local log instead, narrowed to the reason strings.
+
+    Those are protocol vocabulary -- ``tlsv1 alert unknown ca``,
+    ``sslv3 alert handshake failure``, ``Unexpected EOF`` -- and name
+    the failure without naming the peer.
+    """
+    reasons = []
+    first = error.args[0] if error.args else None
+    if isinstance(first, (list, tuple)):
+        for entry in first:
+            if isinstance(entry, (list, tuple)) and entry:
+                reasons.append(str(entry[-1]))
+            elif isinstance(entry, str):
+                reasons.append(entry)
+    else:
+        reasons.extend(arg for arg in error.args if isinstance(arg, str))
+    return ', '.join(r for r in reasons if r) or type(error).__name__
+
+
+@dataclass(frozen=True, slots=True)
+class ObserveDelivery:
+    """One representation delivered on an Observe relation.
+
+    ``registration`` separates the server's answer to the register CON
+    from a change the server chose to send. RFC 7641 §3.2 makes the
+    first response on the token the answer to the registration, and a
+    consumer that treats it as a push reports a device as pushing when
+    it has only replied to being asked. The session is the only layer
+    that can tell them apart, since the token, the Message ID and the
+    Observe option are all resolved here and none of them reach a
+    caller.
+
+    ``query`` completes the relation identity: the same href can carry
+    several query-qualified relations, each registering separately.
+
+    ``sequence`` is the Observe option value (§3.4), or ``None`` on the
+    optionless responses some Samsung firmware sends. ``legacy`` marks a
+    relation promoted to that optionless path.
+    """
+
+    href: str
+    payload: bytes
+    query: tuple[str, ...] = ()
+    registration: bool = False
+    sequence: int | None = None
+    legacy: bool = False
+
+
 @dataclass(slots=True)
 class _MidExchange:
     """One pending request, indexed independently by token and MID."""
@@ -414,7 +470,8 @@ def __init__(self, host, port, cert_path=None, key_path=None, *,
                  auth: AuthenticationProvider | None = None,
                  on_legacy_notification=None,
                  on_observe_pending=None,
-                 on_observe_error=None):
+                 on_observe_error=None,
+                 on_observe_delivery=None):
         file_supplied = cert_path is not None or key_path is not None
         memory_supplied = cert_pem is not None or key_pem is not None
         if auth is not None and (file_supplied or memory_supplied):
@@ -448,6 +505,11 @@ def __init__(self, host, port, cert_path=None, key_path=None, *,
         self.on_legacy_notification = on_legacy_notification
         self.on_observe_pending = on_observe_pending
         self.on_observe_error = on_observe_error
+        # fn(ObserveDelivery). Takes precedence over on_notification and
+        # on_legacy_notification, which carry only (href, payload) and so
+        # cannot express which delivery answered the register CON, nor
+        # which query-qualified relation it belongs to.
+        self.on_observe_delivery = on_observe_delivery
         self.mtu = mtu
         self._min_req_interval = 1.0 / rate_limit_rps
         self._write_max_attempts = max(1, int(write_max_attempts))
@@ -698,8 +760,12 @@ def connect(
                 )
             except _HandshakeCancelled:
                 cancelled = True
-            except SSL.Error:
+            except SSL.Error as e:
                 backend_failed = True
+                # The alert is the whole diagnosis and the raised error
+                # is redacted by contract, so record it here or lose it.
+                logger.warning("dtls handshake failed at the TLS layer: %s",
+                               _openssl_error_reasons(e))
             except OSError:
                 io_failed = True
         finally:
@@ -1039,6 +1105,24 @@ def _clear_observe_relations(self):
             self._legacy_observe_mids.clear()
             self._observe_sequences.clear()
 
+    def _observe_sequence_for(self, href, query):
+        """The Observe value last recorded for one relation, if any.
+
+        A refetched representation is delivered after its triggering
+        notification has already been ordered, so the sequence to report
+        is the one that ordering recorded. Optionless relations have
+        none.
+        """
+        with self._state_lock:
+            for tok, observed_href in self._observe_tokens.items():
+                if observed_href != href or \
+                        self._observe_queries.get(tok, ()) != query:
+                    continue
+                recorded = self._observe_sequences.get(tok)
+                if recorded is not None:
+                    return recorded[0]
+        return None
+
     def _observe_relation_active(self, href, query, legacy):
         """Return whether one relation still owns this callback identity."""
         with self._state_lock:
@@ -1316,6 +1400,8 @@ def _dispatch_coap(self, datagram):
                 value for number, value in ropts if number == OBSERVE
             ]
             legacy = False
+            registration = False
+            sequence = None
             if observe_values:
                 if len(observe_values) != 1 or len(observe_values[0]) > 3:
                     logger.debug("observe %s: malformed Observe option", href)
@@ -1327,6 +1413,12 @@ def _dispatch_coap(self, datagram):
                     if not self._observe_sequence_is_fresh(
                             previous, sequence, received_at):
                         return
+                    # Nothing recorded for this token yet, so this is the
+                    # first response it has carried: the answer to the
+                    # register CON. Retiring a token on unsubscribe or
+                    # refresh clears the entry, so a re-registration is
+                    # recognised as one without any caller bookkeeping.
+                    registration = previous is None
                     self._observe_sequences[tok] = (sequence, received_at)
                     self._observe_plain_response_mids.pop(tok, None)
                     self._legacy_observe_tokens.discard(tok)
@@ -1352,6 +1444,11 @@ def _dispatch_coap(self, datagram):
                         self._legacy_observe_mids[tok] = mid
                         legacy = True
                 if pending:
+                    # The optionless equivalent of the branch above: the
+                    # first plain 2.05 on this token answers the register
+                    # CON, even though the relation stays probationary
+                    # until a later different-MID packet promotes it.
+                    registration = True
                     logger.debug(
                         "observe %s: probationary 2.05 without Observe option",
                         href,
@@ -1377,7 +1474,13 @@ def _dispatch_coap(self, datagram):
             # block past the first) goes to the refetch worker instead.
             if blockwise_refetch:
                 self._queue_refetch(
-                    href, tuple(observe_query), legacy=legacy)
+                    href, tuple(observe_query), legacy=legacy,
+                    registration=registration)
+                return
+            if self._deliver_observation(
+                    href, payload, tuple(observe_query),
+                    registration=registration, sequence=sequence,
+                    legacy=legacy):
                 return
             cb = (
                 self.on_legacy_notification
@@ -1405,7 +1508,28 @@ def _log_refetch(msg, *args):
         without also turning on every per-block retransmit line."""
         (logger.info if DEBUG_BRIDGE else logger.debug)(msg, *args)
 
-    def _queue_refetch(self, href, query=(), *, legacy=False):
+    def _deliver_observation(self, href, payload, query, *,
+                             registration, sequence, legacy):
+        """Hand one observation to the rich callback, if one is set.
+
+        Returns True when it took the delivery, so the two call sites
+        fall through to the (href, payload) callbacks only when no
+        consumer asked for the full relation context.
+        """
+        cb = self.on_observe_delivery
+        if cb is None:
+            return False
+        try:
+            cb(ObserveDelivery(
+                href=href, payload=payload, query=tuple(query),
+                registration=registration, sequence=sequence,
+                legacy=legacy))
+        except Exception as e:
+            logger.warning("observe delivery callback %s: %s", href, e)
+        return True
+
+    def _queue_refetch(self, href, query=(), *, legacy=False,
+                       registration=False):
         """Queue a blockwise notification for re-reading.
 
         Called from the reader thread, so it must not block: _dispatch_coap
@@ -1422,7 +1546,10 @@ def _queue_refetch(self, href, query=(), *, legacy=False):
                     href, len(self._refetch_pending))
                 return
             self._refetch_seq += 1
-            self._refetch_pending[key] = self._refetch_seq
+            # Latest wins, and a real change superseding the registration
+            # answer means what finally gets delivered is that change.
+            self._refetch_pending[key] = (self._refetch_seq,
+                                          bool(registration))
             self._refetch_cond.notify()
         self._start_refetch_worker()
 
@@ -1452,9 +1579,10 @@ def _refetch_loop(self):
                     self._refetch_cond.wait(1.0)
                 if not self._refetch_pending:
                     return
-                key, seq = next(iter(self._refetch_pending.items()))
+                key, (seq, registration) = next(
+                    iter(self._refetch_pending.items()))
                 del self._refetch_pending[key]
-            self._refetch_one(key, seq)
+            self._refetch_one(key, seq, registration)
 
     def _refetch_alive(self):
         """False once the session is closing or the reader has died. A
@@ -1464,7 +1592,7 @@ def _refetch_alive(self):
             return False
         return self._reader_thread is None or self._reader_running.is_set()
 
-    def _refetch_one(self, key, seq):
+    def _refetch_one(self, key, seq, registration=False):
         """Re-read one href from block 0 and deliver it if it is still
         the freshest thing we know about that resource."""
         href, query, legacy = key
@@ -1488,7 +1616,7 @@ def _refetch_one(self, key, seq):
         with self._refetch_cond:
             # A newer notification landed while we were reading. That one
             # has its own refetch queued, so this result is already stale.
-            if self._refetch_pending.get(key, 0) > seq:
+            if self._refetch_pending.get(key, (0, False))[0] > seq:
                 self._log_refetch(
                     "refetch %s tok=%s blocks=%d bytes=%d superseded",
                     href, tok.hex(), blocks, len(payload))
@@ -1497,6 +1625,11 @@ def _refetch_one(self, key, seq):
             return
         self._log_refetch("refetch %s tok=%s blocks=%d bytes=%d ok",
                           href, tok.hex(), blocks, len(payload))
+        if self._deliver_observation(
+                href, payload, query, registration=registration,
+                sequence=self._observe_sequence_for(href, query),
+                legacy=legacy):
+            return
         cb = (
             self.on_legacy_notification
             if legacy and self.on_legacy_notification is not None