diff --git a/README.md b/README.md
index 1e7fdff..40a5d65 100644
--- a/README.md
+++ b/README.md
@@ -142,6 +142,31 @@ Setting the signal stops subscribed connection attempts and closes their
 temporary UDP sockets. It does not alter an already established session.
 Interrupted attempts raise `SessionClosedError`.
 
+Some Samsung OCF-PKI firmware can retain a half-open DTLS peer when a
+handshake stops immediately after the cookie exchange. A caller using a
+`SamsungServerProfile` and a fixed non-zero local UDP port can opt in to the
+narrow cleanup path:
+
+```python
+from smartthings_local.errors import HandshakePeerCleanupError
+
+try:
+    sess.connect(timeout=8.0, cleanup_hvr_peer=True)
+except HandshakePeerCleanupError:
+    # Apply the device-specific settle delay, then retry under caller policy.
+    schedule_connection_retry()
+```
+
+Cleanup is sent only after the bounded transcript contains at least two
+complete epoch-zero ClientHello messages and every received record is a
+complete epoch-zero HelloVerifyRequest. A malformed, fragmented, mixed, or
+oversized transcript remains an ordinary `SessionTimeoutError`. On the exact
+HVR-only shape, the session sends one epoch-zero fatal `handshake_failure`
+alert, closes the temporary socket, and raises `HandshakePeerCleanupError`.
+The package never sleeps or retries automatically, so the caller retains the
+overall recovery budget and can use the settle interval validated for its
+device. Cancellation and backend failures never trigger the alert.
+
 Hosts that stop network work before their blocking executor drains can use the
 session's two-phase shutdown. `quiesce_for_close()` is terminal: it interrupts
 an in-progress handshake, wakes pending requests and notification refetches,
@@ -338,8 +363,32 @@ sess = DtlsCoapSession("192.0.2.100", 5684, auth=auth)
 `ServerCertificateAuth` is for a server-authenticated channel that does not
 send a client certificate, such as the initial DTLS carrier used by
 manufacturer-certificate OTM. It still verifies the CA chain, exact selected
-subject role, and pinned certificate UUID. It does not learn an identity from
-the first endpoint it reaches and cannot be combined with client credentials.
+subject role, and pinned certificate UUID. It cannot be combined with client
+credentials.
+
+An explicit first-use workflow may need to authenticate the Samsung hardware
+certificate before its subject UUID is known. Use the discovery profile only
+for that bounded step:
+
+```python
+server_profile = SamsungServerProfile.discover_device(
+    additional_ca_pem=additional_samsung_ca_pem,
+)
+auth = ServerCertificateAuth(server_profile=server_profile)
+sess = DtlsCoapSession("192.0.2.100", 5684, auth=auth)
+sess.connect()
+certificate_uuid = sess.server_certificate_identity
+```
+
+The discovery profile still verifies the CA chain and the complete selected
+Samsung subject role before `connect()` exposes the non-zero certificate UUID.
+It does not trust an arbitrary first certificate, and neither the immutable
+profile nor its provider retains the learned identity. The caller must bind
+that UUID to independently authenticated device evidence, such as `/oic/d`
+read over the same authenticated session, before persisting it. Subsequent
+connections should use `bound_device()` with that verified binding. The
+certificate UUID and the OCF device UUID are separate identities and must not
+be assumed equal.
 
 This API deliberately does not discover, mint, authorize, provision, rotate,
 or persist credentials, and it performs no ownership transfer or OCF security
@@ -417,6 +466,7 @@ callers can keep catching the built-in types used by earlier releases:
 | `AuthenticationError` | `authentication` | `ConnectionError` |
 | `AuthorizationError` | `authorization` | `PermissionError` |
 | `SessionTimeoutError` | `timeout` | `TimeoutError` |
+| `HandshakePeerCleanupError` | `handshake_peer_cleanup` | `TimeoutError` |
 | `SessionClosedError` | `session_closed` | `ConnectionError` |
 | `MalformedMessageError` | `malformed_message` | `ValueError` |
 | `BlockwiseError` | `blockwise` | `ConnectionError` |
diff --git a/smartthings_local/errors.py b/smartthings_local/errors.py
index 334728b..9f3b5ca 100644
--- a/smartthings_local/errors.py
+++ b/smartthings_local/errors.py
@@ -5,6 +5,7 @@
     'AuthorizationError',
     'BlockwiseError',
     'EndpointError',
+    'HandshakePeerCleanupError',
     'MalformedMessageError',
     'ObserveError',
     'ProbeError',
@@ -77,6 +78,16 @@ class SessionTimeoutError(SmartThingsLocalError, TimeoutError):
     message = 'session operation timed out'
 
 
+class HandshakePeerCleanupError(SessionTimeoutError):
+    """A cleanup alert was sent for an HVR-only half-open DTLS peer.
+
+    The caller may apply its device-specific settle delay and retry policy.
+    """
+
+    code = 'handshake_peer_cleanup'
+    message = 'handshake peer cleanup was sent'
+
+
 class SessionClosedError(SessionError):
     """An operation was attempted on a closed session."""
 
diff --git a/smartthings_local/protocol/auth.py b/smartthings_local/protocol/auth.py
index 14e87a1..2a14357 100644
--- a/smartthings_local/protocol/auth.py
+++ b/smartthings_local/protocol/auth.py
@@ -93,6 +93,19 @@ def __init__(
         role: SamsungServerRole = SamsungServerRole.HOME_APPLIANCE,
         additional_ca_pem: str | None = None,
     ) -> None:
+        parsed_identity = self._parse_expected_identity(
+            expected_certificate_identity
+        )
+        self._initialize(
+            expected_certificate_identity=parsed_identity,
+            role=role,
+            additional_ca_pem=additional_ca_pem,
+        )
+
+    @staticmethod
+    def _parse_expected_identity(
+        expected_certificate_identity: UUID | str,
+    ) -> UUID:
         if type(expected_certificate_identity) is UUID:
             parsed_identity = expected_certificate_identity
         elif type(expected_certificate_identity) is str:
@@ -117,6 +130,15 @@ def __init__(
                 "expected_certificate_identity must be a canonical "
                 "non-zero UUID"
             )
+        return parsed_identity
+
+    def _initialize(
+        self,
+        *,
+        expected_certificate_identity: UUID | None,
+        role: SamsungServerRole,
+        additional_ca_pem: str | None,
+    ) -> None:
         if type(role) is not SamsungServerRole:
             raise TypeError("role must be a SamsungServerRole")
 
@@ -176,7 +198,7 @@ def __init__(
         object.__setattr__(
             self,
             "_expected_certificate_identity",
-            parsed_identity,
+            expected_certificate_identity,
         )
         object.__setattr__(self, "_role", role)
         object.__setattr__(self, "_additional_ca_certificates", certificates)
@@ -202,6 +224,27 @@ def bound_device(
             additional_ca_pem=additional_ca_pem,
         )
 
+    @classmethod
+    def discover_device(
+        cls,
+        *,
+        role: SamsungServerRole = SamsungServerRole.HOME_APPLIANCE,
+        additional_ca_pem: str | None = None,
+    ) -> SamsungServerProfile:
+        """Verify a Samsung hardware leaf before learning its certificate UUID.
+
+        This profile is intended only for an explicit first-use workflow. The
+        caller must bind the returned session identity to independently
+        authenticated device evidence before persisting it.
+        """
+        profile = object.__new__(cls)
+        profile._initialize(
+            expected_certificate_identity=None,
+            role=role,
+            additional_ca_pem=additional_ca_pem,
+        )
+        return profile
+
     def __repr__(self) -> str:
         """Return a representation without device or trust-chain details."""
         return "SamsungServerProfile()"
@@ -273,6 +316,14 @@ def _verify_peer(
             return False
         if depth > 0:
             return True
+        identity = self._certificate_identity(certificate)
+        return identity is not None and (
+            self._expected_certificate_identity is None
+            or identity == self._expected_certificate_identity
+        )
+
+    def _certificate_identity(self, certificate) -> UUID | None:
+        """Extract a UUID only from the selected Samsung subject role."""
         try:
             with warnings.catch_warnings():
                 # pyOpenSSL deprecates this API in favor of cryptography, but
@@ -291,7 +342,7 @@ def _verify_peer(
             crypto.Error,
         ):
             logger.warning("Unable to parse Samsung server certificate subject")
-            return False
+            return None
 
         common_names = [value for name, value in components if name == "CN"]
         organizational_units = [
@@ -308,13 +359,12 @@ def _verify_peer(
             or organizations != ["Samsung Electronics"]
             or countries != ["KR"]
         ):
-            return False
+            return None
         match = _SAMSUNG_SERVER_CN_RE.fullmatch(common_names[0])
-        return (
-            match is not None
-            and UUID(match.group("device_identity"))
-            == self._expected_certificate_identity
-        )
+        if match is None:
+            return None
+        identity = UUID(match.group("device_identity"))
+        return identity if identity.int != 0 else None
 
 
 def _configure_certificate_server(
@@ -362,6 +412,13 @@ def configure_context(self, context: SSL.Context) -> None:
         """Verify the selected server profile without loading client material."""
         _configure_certificate_server(context, self._server_profile)
 
+    def _authenticated_server_identity(self, connection) -> UUID | None:
+        certificate = connection.get_peer_certificate()
+        identity = self._server_profile._certificate_identity(certificate)
+        if identity is None:
+            raise ValueError("verified server identity was unavailable")
+        return identity
+
 
 class CertificateAuth:
     """Certificate authentication loaded from files or in-memory PEM data.
@@ -486,6 +543,15 @@ def configure_context(self, context: SSL.Context) -> None:
             context.use_privatekey_file(self._private_key_path)
             context.check_privatekey()
 
+    def _authenticated_server_identity(self, connection) -> UUID | None:
+        if self._server_profile is None:
+            return None
+        certificate = connection.get_peer_certificate()
+        identity = self._server_profile._certificate_identity(certificate)
+        if identity is None:
+            raise ValueError("verified server identity was unavailable")
+        return identity
+
 
 class PskAuth:
     """DTLS authentication using an existing OCF PSK credential.
diff --git a/smartthings_local/protocol/dtls_handshake.py b/smartthings_local/protocol/dtls_handshake.py
index 12fa30a..ec5be7b 100644
--- a/smartthings_local/protocol/dtls_handshake.py
+++ b/smartthings_local/protocol/dtls_handshake.py
@@ -12,6 +12,163 @@
 
 _HANDSHAKE_POLL_S = 0.5
 _MAX_DATAGRAM_SIZE = 65535
+_DTLS_RECORD_HEADER_BYTES = 13
+_DTLS_CONTENT_TYPE_ALERT = 21
+_DTLS_CONTENT_TYPE_HANDSHAKE = 22
+_DTLS_HANDSHAKE_CLIENT_HELLO = 1
+_DTLS_HANDSHAKE_HELLO_VERIFY_REQUEST = 3
+_DTLS_ALERT_LEVEL_FATAL = 2
+_DTLS_ALERT_HANDSHAKE_FAILURE = 40
+_DTLS_EPOCH_ZERO = b'\x00\x00'
+_DTLS_VERSIONS = frozenset((b'\xfe\xff', b'\xfe\xfd'))
+_MAX_CLEANUP_TRANSCRIPT_RECORDS = 32
+
+
+def _complete_epoch_zero_handshake_types(record):
+    """Return complete handshake message types from one strict DTLS record."""
+    if (
+        len(record) < _DTLS_RECORD_HEADER_BYTES
+        or record[0] != _DTLS_CONTENT_TYPE_HANDSHAKE
+        or record[1:3] not in _DTLS_VERSIONS
+        or record[3:5] != _DTLS_EPOCH_ZERO
+        or int.from_bytes(record[11:13], 'big')
+        != len(record) - _DTLS_RECORD_HEADER_BYTES
+    ):
+        return None
+
+    payload = record[_DTLS_RECORD_HEADER_BYTES:]
+    offset = 0
+    message_types = []
+    while offset < len(payload):
+        if len(payload) - offset < 12:
+            return None
+        message_length = int.from_bytes(payload[offset + 1:offset + 4], 'big')
+        fragment_offset = int.from_bytes(payload[offset + 6:offset + 9], 'big')
+        fragment_length = int.from_bytes(payload[offset + 9:offset + 12], 'big')
+        end = offset + 12 + fragment_length
+        if (
+            fragment_offset != 0
+            or fragment_length != message_length
+            or end > len(payload)
+        ):
+            return None
+        message_types.append(payload[offset])
+        offset = end
+    return tuple(message_types) if message_types else None
+
+
+def _strict_dtls_records(datagram):
+    """Split one complete DTLS datagram or reject all of it."""
+    offset = 0
+    records = []
+    while offset < len(datagram):
+        if len(datagram) - offset < _DTLS_RECORD_HEADER_BYTES:
+            return None
+        payload_length = int.from_bytes(datagram[offset + 11:offset + 13], 'big')
+        end = offset + _DTLS_RECORD_HEADER_BYTES + payload_length
+        if end > len(datagram):
+            return None
+        records.append(datagram[offset:end])
+        if len(records) > _MAX_CLEANUP_TRANSCRIPT_RECORDS:
+            return None
+        offset = end
+    return tuple(records) if records else None
+
+
+class _HvrPeerCleanupTranscript:
+    """Retain only bounded metadata needed for the HVR cleanup decision."""
+
+    __slots__ = (
+        '_invalid',
+        '_last_client_hello_header',
+        '_received_datagrams',
+        '_received_records',
+        '_sent_client_hello_after_hvr',
+        '_sent_client_hellos',
+    )
+
+    def __init__(self):
+        self._invalid = False
+        self._last_client_hello_header = None
+        self._received_datagrams = 0
+        self._received_records = 0
+        self._sent_client_hello_after_hvr = False
+        self._sent_client_hellos = 0
+
+    def record_sent(self, record):
+        """Record one complete epoch-zero ClientHello without its payload."""
+        message_types = _complete_epoch_zero_handshake_types(record)
+        if message_types != (_DTLS_HANDSHAKE_CLIENT_HELLO,):
+            self._invalid = True
+            return
+        self._sent_client_hellos += 1
+        if self._sent_client_hellos > _MAX_CLEANUP_TRANSCRIPT_RECORDS:
+            self._invalid = True
+            return
+        if self._received_records:
+            self._sent_client_hello_after_hvr = True
+        header = record[:_DTLS_RECORD_HEADER_BYTES]
+        previous = self._last_client_hello_header
+        if previous is None or header[5:11] >= previous[5:11]:
+            self._last_client_hello_header = header
+
+    def record_received(self, datagram):
+        """Record only whether a bounded datagram contains complete HVRs."""
+        self._received_datagrams += 1
+        if self._received_datagrams > _MAX_CLEANUP_TRANSCRIPT_RECORDS:
+            self._invalid = True
+            return
+        records = _strict_dtls_records(datagram)
+        if records is None:
+            self._invalid = True
+            return
+        if (
+            self._received_records + len(records)
+            > _MAX_CLEANUP_TRANSCRIPT_RECORDS
+        ):
+            self._invalid = True
+            return
+        for record in records:
+            message_types = _complete_epoch_zero_handshake_types(record)
+            if (
+                message_types is None
+                or any(
+                    message_type != _DTLS_HANDSHAKE_HELLO_VERIFY_REQUEST
+                    for message_type in message_types
+                )
+            ):
+                self._invalid = True
+                return
+            self._received_records += 1
+
+    def cleanup_alert(self):
+        """Build one epoch-zero alert only for an exact HVR-only transcript."""
+        header = self._last_client_hello_header
+        if (
+            self._invalid
+            or self._sent_client_hellos < 2
+            or not self._sent_client_hello_after_hvr
+            or self._received_datagrams < 1
+            or self._received_records < 1
+            or header is None
+        ):
+            return None
+        sequence = int.from_bytes(header[5:11], 'big')
+        if sequence >= (1 << 48) - 1:
+            return None
+        return (
+            bytes((_DTLS_CONTENT_TYPE_ALERT,))
+            + header[1:3]
+            + _DTLS_EPOCH_ZERO
+            + (sequence + 1).to_bytes(6, 'big')
+            + b'\x00\x02'
+            + bytes(
+                (
+                    _DTLS_ALERT_LEVEL_FATAL,
+                    _DTLS_ALERT_HANDSHAKE_FAILURE,
+                )
+            )
+        )
 
 
 class _HandshakeCancelled(Exception):
@@ -25,6 +182,7 @@ def _drive_dtls_handshake(
     deadline: float,
     retries: int | None = None,
     on_datagram: Callable[[bytes], None] | None = None,
+    on_record_sent: Callable[[bytes], None] | None = None,
     wake_socket=None,
 ) -> bool:
     """Drive one memory-BIO DTLS handshake up to a monotonic deadline.
@@ -55,6 +213,8 @@ def _drive_dtls_handshake(
             for record in split_dtls(output):
                 if sock.send(record) != len(record):
                     raise OSError("incomplete UDP send")
+                if on_record_sent is not None:
+                    on_record_sent(record)
 
         remaining = deadline - time.monotonic()
         if remaining <= 0:
diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py
index e73e8ce..1875205 100644
--- a/smartthings_local/protocol/dtls_session.py
+++ b/smartthings_local/protocol/dtls_session.py
@@ -34,12 +34,15 @@
 import time
 from collections.abc import Mapping
 from dataclasses import dataclass
+from uuid import UUID
 
 from OpenSSL import SSL
 
 from ..errors import (
+    AuthenticationError,
     BlockwiseError,
     EndpointError,
+    HandshakePeerCleanupError,
     MalformedMessageError,
     SessionClosedError,
     SessionError,
@@ -52,6 +55,8 @@
 from .auth import (
     AuthenticationProvider,
     CertificateAuth,
+    SamsungServerProfile,
+    ServerCertificateAuth,
 )
 from .coap import (
     ACCEPT,
@@ -86,6 +91,7 @@
 )
 from .dtls_handshake import (
     _HANDSHAKE_POLL_S,
+    _HvrPeerCleanupTranscript,
     _drive_dtls_handshake,
     _HandshakeCancelled,
 )
@@ -462,6 +468,10 @@ def __init__(self, host, port, cert_path=None, key_path=None, *,
         self.conn = None
         self.dest = None
         self.endpoint = None
+        # Populated only when a SamsungServerProfile verified the peer. The
+        # UUID comes from the authenticated hardware-certificate subject; it
+        # is not necessarily the OCF device UUID returned by /oic/d.
+        self.server_certificate_identity = None
 
         # Terminal session shutdown has its own wake signal so
         # quiesce_for_close() can interrupt connect() even when the caller did
@@ -559,6 +569,7 @@ def connect(
         *,
         timeout: float | None = None,
         cancel: ConnectCancellation | None = None,
+        cleanup_hvr_peer: bool = False,
     ):
         """Perform a cancellable DTLS handshake using a monotonic deadline.
 
@@ -568,12 +579,34 @@ def connect(
         Once OpenSSL reports completion, that completed session is retained
         even if the call returns just after the deadline. A
         ``ConnectCancellation`` wakes the network wait immediately and does not
-        alter an already established session.
+        alter an already established session. ``cleanup_hvr_peer`` is an
+        opt-in recovery signal for a Samsung-profiled handshake that times out
+        after only the DTLS cookie exchange. It requires a fixed local port,
+        emits at most one standards-based alert, and leaves retry policy to the
+        caller.
         """
         handshake_timeout = _validate_handshake_timeout(
             timeout, self.HANDSHAKE_TIMEOUT_S)
         if cancel is not None and not isinstance(cancel, ConnectCancellation):
             raise TypeError("cancel must be a ConnectCancellation or None")
+        if type(cleanup_hvr_peer) is not bool:
+            raise TypeError("cleanup_hvr_peer must be a bool")
+        if cleanup_hvr_peer and (
+            type(self.auth) not in (CertificateAuth, ServerCertificateAuth)
+            or type(getattr(self.auth, "_server_profile", None))
+            is not SamsungServerProfile
+        ):
+            raise ValueError(
+                "HVR peer cleanup requires a Samsung server profile"
+            )
+        if cleanup_hvr_peer and (
+            isinstance(self.local_port, bool)
+            or not isinstance(self.local_port, int)
+            or not 1 <= self.local_port <= 65535
+        ):
+            raise ValueError(
+                "HVR peer cleanup requires a fixed non-zero local port"
+            )
         if self._lifecycle_cancel.is_set() or \
                 (cancel is not None and cancel.is_set()):
             raise SessionClosedError()
@@ -638,6 +671,9 @@ def connect(
         cancelled = False
         interrupted = False
         completed = False
+        cleanup_transcript = (
+            _HvrPeerCleanupTranscript() if cleanup_hvr_peer else None
+        )
         try:
             try:
                 completed = _drive_dtls_handshake(
@@ -649,6 +685,16 @@ def connect(
                         if wake_subscription is not None
                         else None
                     ),
+                    on_datagram=(
+                        cleanup_transcript.record_received
+                        if cleanup_transcript is not None
+                        else None
+                    ),
+                    on_record_sent=(
+                        cleanup_transcript.record_sent
+                        if cleanup_transcript is not None
+                        else None
+                    ),
                 )
             except _HandshakeCancelled:
                 cancelled = True
@@ -675,9 +721,49 @@ def connect(
             sock.close()
             raise EndpointError() from OSError('UDP handshake I/O failed')
         if not completed:
+            cleanup_sent = False
+            if cleanup_transcript is not None:
+                cleanup_alert = cleanup_transcript.cleanup_alert()
+                if cleanup_alert is not None:
+                    try:
+                        # The handshake deadline has expired. Keep this single
+                        # advisory datagram non-blocking so cleanup cannot
+                        # extend the caller's timeout budget.
+                        sock.settimeout(0.0)
+                        cleanup_sent = (
+                            sock.send(cleanup_alert) == len(cleanup_alert)
+                        )
+                    except Exception:
+                        pass
             sock.close()
+            if cleanup_sent:
+                raise HandshakePeerCleanupError()
             raise SessionTimeoutError()
 
+        try:
+            identity_reader = getattr(
+                self.auth,
+                "_authenticated_server_identity",
+                None,
+            )
+            if identity_reader is not None:
+                server_certificate_identity = identity_reader(conn)
+                if server_certificate_identity is not None and (
+                    type(server_certificate_identity) is not UUID
+                    or server_certificate_identity.int == 0
+                ):
+                    raise ValueError(
+                        "verified server identity was unavailable"
+                    )
+            else:
+                server_certificate_identity = None
+        except Exception:
+            self._send_close_notify(conn, sock)
+            sock.close()
+            raise AuthenticationError() from ConnectionError(
+                "verified server identity was unavailable"
+            )
+
         with self._lifecycle_lock:
             if self._lifecycle_cancel.is_set():
                 sock.close()
@@ -686,6 +772,7 @@ def connect(
             self.conn = conn
             self.dest = dest
             self.endpoint = endpoint
+            self.server_certificate_identity = server_certificate_identity
             self._stop.clear()
 
     def start_reader(self):
diff --git a/tests/test_certificate_profiles.py b/tests/test_certificate_profiles.py
index 91ba1fd..684b1cd 100644
--- a/tests/test_certificate_profiles.py
+++ b/tests/test_certificate_profiles.py
@@ -453,6 +453,56 @@ def test_profile_accepts_only_canonical_nonzero_identity():
             SamsungServerProfile.bound_device(invalid)
 
 
+def test_discovery_profile_verifies_role_before_exposing_identity():
+    first_chain = _make_generated_chain(_IDENTITY)
+    second_chain = _make_generated_chain(_OTHER_IDENTITY)
+    zero_chain = _make_generated_chain(UUID(int=0))
+    wrong_role = _make_generated_chain(
+        _IDENTITY,
+        organizational_unit="OCF VD Device",
+    )
+    profile = SamsungServerProfile.discover_device()
+
+    assert repr(profile) == "SamsungServerProfile()"
+    assert profile._verify_peer(None, first_chain.leaf, 0, 0, True) is True
+    assert profile._verify_peer(None, second_chain.leaf, 0, 0, True) is True
+    assert profile._verify_peer(None, zero_chain.leaf, 0, 0, True) is False
+    assert profile._verify_peer(None, wrong_role.leaf, 0, 0, True) is False
+
+    for invalid in ("OCF HA Device", None, object()):
+        with pytest.raises(TypeError, match="SamsungServerRole"):
+            SamsungServerProfile.discover_device(role=invalid)
+
+
+def test_discovery_profile_is_reusable_and_does_not_retain_peer_identity():
+    first_chain = _make_generated_chain(_IDENTITY)
+    second_chain = _make_generated_chain(_OTHER_IDENTITY)
+    profile = SamsungServerProfile.discover_device()
+    provider = ServerCertificateAuth(server_profile=profile)
+
+    first = provider._authenticated_server_identity(
+        SimpleNamespace(get_peer_certificate=lambda: first_chain.leaf)
+    )
+    second = provider._authenticated_server_identity(
+        SimpleNamespace(get_peer_certificate=lambda: second_chain.leaf)
+    )
+
+    assert first == _IDENTITY
+    assert second == _OTHER_IDENTITY
+    assert not hasattr(profile, "certificate_identity")
+    assert str(_IDENTITY) not in repr(profile)
+    assert str(_OTHER_IDENTITY) not in repr(profile)
+
+    certificate_provider = CertificateAuth.from_memory(
+        first_chain.certificate_pem,
+        first_chain.private_key_pem,
+        server_profile=profile,
+    )
+    assert certificate_provider._authenticated_server_identity(
+        SimpleNamespace(get_peer_certificate=lambda: second_chain.leaf)
+    ) == _OTHER_IDENTITY
+
+
 def test_profile_roles_are_explicit_and_fail_closed():
     home_chain = _make_generated_chain(_IDENTITY)
     video_chain = _make_generated_chain(
@@ -894,6 +944,7 @@ def set_options(self, options):
 
 def test_profile_identity_verification_rejects_malformed_subjects(caplog):
     profile = SamsungServerProfile.bound_device(_IDENTITY)
+    discovery_profile = SamsungServerProfile.discover_device()
     wrong_role = _make_generated_chain(
         _IDENTITY,
         organizational_unit="Unexpected Device",
@@ -934,6 +985,10 @@ def test_profile_identity_verification_rejects_malformed_subjects(caplog):
     caplog.clear()
     with caplog.at_level(logging.WARNING, logger=auth_module.__name__):
         assert profile._verify_peer(None, object(), 0, 0, True) is False
+        assert (
+            discovery_profile._verify_peer(None, object(), 0, 0, True)
+            is False
+        )
         assert (
             profile._verify_peer(
                 None,
@@ -947,6 +1002,7 @@ def test_profile_identity_verification_rejects_malformed_subjects(caplog):
     assert caplog.messages == [
         "Unable to parse Samsung server certificate subject",
         "Unable to parse Samsung server certificate subject",
+        "Unable to parse Samsung server certificate subject",
     ]
 
     caplog.clear()
diff --git a/tests/test_errors.py b/tests/test_errors.py
index 9567dcb..7e1365a 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -8,6 +8,7 @@
     AuthorizationError,
     BlockwiseError,
     EndpointError,
+    HandshakePeerCleanupError,
     MalformedMessageError,
     ObserveError,
     ProbeError,
@@ -27,6 +28,7 @@
     AuthenticationError,
     AuthorizationError,
     SessionTimeoutError,
+    HandshakePeerCleanupError,
     SessionClosedError,
     MalformedMessageError,
     BlockwiseError,
@@ -43,6 +45,7 @@
         (AuthenticationError, ConnectionError),
         (AuthorizationError, PermissionError),
         (SessionTimeoutError, TimeoutError),
+        (HandshakePeerCleanupError, TimeoutError),
         (SessionClosedError, ConnectionError),
         (MalformedMessageError, ValueError),
         (BlockwiseError, ConnectionError),
diff --git a/tests/test_hvr_peer_cleanup.py b/tests/test_hvr_peer_cleanup.py
new file mode 100644
index 0000000..a5003d3
--- /dev/null
+++ b/tests/test_hvr_peer_cleanup.py
@@ -0,0 +1,355 @@
+"""Exact transcript and session contract for opt-in HVR peer cleanup."""
+
+from __future__ import annotations
+
+import socket
+
+import pytest
+
+from smartthings_local.errors import (
+    HandshakePeerCleanupError,
+    SessionClosedError,
+    SessionTimeoutError,
+)
+from smartthings_local.protocol import dtls_session
+from smartthings_local.protocol.auth import (
+    SamsungServerProfile,
+    ServerCertificateAuth,
+)
+from smartthings_local.protocol.dtls_session import DtlsCoapSession
+from smartthings_local.protocol.endpoint import ResolvedUdpEndpoint
+
+
+def _handshake_record(
+    message_type,
+    *,
+    record_sequence,
+    body=b'',
+    version=b'\xfe\xfd',
+    epoch=0,
+    fragment_offset=0,
+    fragment_length=None,
+):
+    if fragment_length is None:
+        fragment_length = len(body)
+    message = (
+        bytes((message_type,))
+        + len(body).to_bytes(3, 'big')
+        + b'\x00\x00'
+        + fragment_offset.to_bytes(3, 'big')
+        + fragment_length.to_bytes(3, 'big')
+        + body
+    )
+    return (
+        b'\x16'
+        + version
+        + epoch.to_bytes(2, 'big')
+        + record_sequence.to_bytes(6, 'big')
+        + len(message).to_bytes(2, 'big')
+        + message
+    )
+
+
+def _eligible_transcript(*, marker=b''):
+    transcript = dtls_session._HvrPeerCleanupTranscript()
+    transcript.record_sent(
+        _handshake_record(1, record_sequence=0, body=b'first')
+    )
+    transcript.record_received(
+        _handshake_record(3, record_sequence=0, body=marker)
+    )
+    transcript.record_sent(
+        _handshake_record(1, record_sequence=1, body=b'cookie')
+    )
+    return transcript
+
+
+def test_exact_hvr_only_transcript_builds_standard_epoch_zero_alert():
+    transcript = _eligible_transcript()
+
+    assert transcript.cleanup_alert() == (
+        b'\x15\xfe\xfd'
+        + b'\x00\x00'
+        + (2).to_bytes(6, 'big')
+        + b'\x00\x02\x02\x28'
+    )
+
+
+def test_transcript_retains_metadata_without_handshake_payload():
+    marker = b'private-cookie-material'
+    transcript = _eligible_transcript(marker=marker)
+
+    with pytest.raises(TypeError):
+        vars(transcript)
+    retained = tuple(
+        getattr(transcript, name)
+        for name in transcript.__slots__
+    )
+    assert all(marker not in value for value in retained if isinstance(value, bytes))
+
+
+@pytest.mark.parametrize(
+    'mutate',
+    (
+        lambda transcript: setattr(transcript, '_sent_client_hellos', 1),
+        lambda transcript: setattr(transcript, '_received_datagrams', 0),
+        lambda transcript: setattr(transcript, '_received_records', 0),
+        lambda transcript: setattr(
+            transcript,
+            '_sent_client_hello_after_hvr',
+            False,
+        ),
+        lambda transcript: setattr(transcript, '_invalid', True),
+        lambda transcript: setattr(
+            transcript,
+            '_last_client_hello_header',
+            None,
+        ),
+        lambda transcript: setattr(
+            transcript,
+            '_last_client_hello_header',
+            transcript._last_client_hello_header[:5]
+            + ((1 << 48) - 1).to_bytes(6, 'big')
+            + transcript._last_client_hello_header[11:],
+        ),
+    ),
+)
+def test_cleanup_alert_fails_closed_when_required_metadata_is_missing(
+    mutate,
+):
+    transcript = _eligible_transcript()
+    mutate(transcript)
+
+    assert transcript.cleanup_alert() is None
+
+
+@pytest.mark.parametrize(
+    'record',
+    (
+        _handshake_record(2, record_sequence=0),
+        _handshake_record(3, record_sequence=0, epoch=1),
+        _handshake_record(
+            3,
+            record_sequence=0,
+            body=b'fragment',
+            fragment_offset=1,
+        ),
+        b'\x15\xfe\xfd\x00\x00' + b'\x00' * 8,
+        b'\x16\xfe\xfd\x00\x00' + b'\x00' * 7,
+    ),
+)
+def test_non_hvr_or_malformed_inbound_record_disables_cleanup(record):
+    transcript = _eligible_transcript()
+    transcript.record_received(record)
+
+    assert transcript.cleanup_alert() is None
+
+
+def test_mixed_hvr_and_non_hvr_datagram_disables_cleanup():
+    transcript = _eligible_transcript()
+    transcript.record_received(
+        _handshake_record(3, record_sequence=1)
+        + _handshake_record(2, record_sequence=2)
+    )
+
+    assert transcript.cleanup_alert() is None
+
+
+def test_non_client_hello_outbound_record_disables_cleanup():
+    transcript = _eligible_transcript()
+    transcript.record_sent(_handshake_record(11, record_sequence=2))
+
+    assert transcript.cleanup_alert() is None
+
+
+def test_transcript_record_bound_fails_closed():
+    transcript = _eligible_transcript()
+    for sequence in range(32):
+        transcript.record_received(
+            _handshake_record(3, record_sequence=sequence)
+        )
+
+    assert transcript.cleanup_alert() is None
+
+
+class _Connection:
+    def set_connect_state(self):
+        return None
+
+    def set_ciphertext_mtu(self, _mtu):
+        return None
+
+
+class _Socket:
+    def __init__(self, send_result=None, send_error=None):
+        self.closed = False
+        self.send_error = send_error
+        self.send_result = send_result
+        self.sent = []
+        self.timeouts = []
+
+    def send(self, data):
+        if self.send_error is not None:
+            raise self.send_error
+        self.sent.append(data)
+        return len(data) if self.send_result is None else self.send_result
+
+    def close(self):
+        self.closed = True
+
+    def settimeout(self, timeout):
+        self.timeouts.append(timeout)
+
+
+def _profiled_session(*, local_port=49745):
+    profile = SamsungServerProfile.discover_device()
+    return DtlsCoapSession(
+        'device.example',
+        5684,
+        auth=ServerCertificateAuth(server_profile=profile),
+        local_port=local_port,
+    )
+
+
+def _install_connect(monkeypatch, drive, *, udp_socket=None):
+    udp_socket = udp_socket or _Socket()
+    endpoint = ResolvedUdpEndpoint(
+        socket.AF_INET,
+        ('192.0.2.10', 5684),
+    )
+    monkeypatch.setattr(dtls_session.SSL, 'Context', lambda *_args: object())
+    monkeypatch.setattr(
+        dtls_session.SSL,
+        'Connection',
+        lambda *_args: _Connection(),
+    )
+    monkeypatch.setattr(
+        ServerCertificateAuth,
+        'configure_context',
+        lambda self, context: None,
+    )
+    monkeypatch.setattr(
+        dtls_session,
+        'open_host_filtered_udp_socket',
+        lambda *_args, **_kwargs: (udp_socket, endpoint),
+    )
+    monkeypatch.setattr(dtls_session, '_drive_dtls_handshake', drive)
+    return udp_socket
+
+
+def _hvr_timeout_driver(
+    _connection,
+    _socket,
+    *,
+    on_datagram,
+    on_record_sent,
+    **_kwargs,
+):
+    on_record_sent(_handshake_record(1, record_sequence=0, body=b'first'))
+    on_datagram(_handshake_record(3, record_sequence=0, body=b'cookie'))
+    on_record_sent(_handshake_record(1, record_sequence=1, body=b'cookie'))
+    return False
+
+
+def test_connect_sends_one_cleanup_alert_and_returns_retry_to_caller(
+    monkeypatch,
+):
+    udp_socket = _install_connect(monkeypatch, _hvr_timeout_driver)
+    session = _profiled_session()
+
+    with pytest.raises(HandshakePeerCleanupError) as captured:
+        session.connect(timeout=1.0, cleanup_hvr_peer=True)
+
+    assert captured.value.code == 'handshake_peer_cleanup'
+    assert isinstance(captured.value, SessionTimeoutError)
+    assert udp_socket.sent == [
+        b'\x15\xfe\xfd'
+        + b'\x00\x00'
+        + (2).to_bytes(6, 'big')
+        + b'\x00\x02\x02\x28'
+    ]
+    assert udp_socket.timeouts == [0.0]
+    assert udp_socket.closed
+    assert session.sock is None
+    assert session.conn is None
+
+
+@pytest.mark.parametrize(
+    'udp_socket',
+    (
+        _Socket(send_result=0),
+        _Socket(send_error=OSError('synthetic send failure')),
+    ),
+)
+def test_cleanup_send_failure_remains_an_ordinary_timeout(
+    monkeypatch,
+    udp_socket,
+):
+    _install_connect(monkeypatch, _hvr_timeout_driver, udp_socket=udp_socket)
+
+    with pytest.raises(SessionTimeoutError) as captured:
+        _profiled_session().connect(timeout=1.0, cleanup_hvr_peer=True)
+
+    assert type(captured.value) is SessionTimeoutError
+    assert udp_socket.closed
+
+
+def test_non_hvr_timeout_never_sends_cleanup_alert(monkeypatch):
+    def drive(
+        _connection,
+        _socket,
+        *,
+        on_datagram,
+        on_record_sent,
+        **_kwargs,
+    ):
+        on_record_sent(_handshake_record(1, record_sequence=0))
+        on_datagram(_handshake_record(2, record_sequence=0))
+        on_record_sent(_handshake_record(1, record_sequence=1))
+        return False
+
+    udp_socket = _install_connect(monkeypatch, drive)
+
+    with pytest.raises(SessionTimeoutError) as captured:
+        _profiled_session().connect(timeout=1.0, cleanup_hvr_peer=True)
+
+    assert type(captured.value) is SessionTimeoutError
+    assert udp_socket.sent == []
+
+
+def test_cancelled_hvr_handshake_never_sends_cleanup_alert(monkeypatch):
+    def drive(*args, **kwargs):
+        _hvr_timeout_driver(*args, **kwargs)
+        raise dtls_session._HandshakeCancelled()
+
+    udp_socket = _install_connect(monkeypatch, drive)
+
+    with pytest.raises(SessionClosedError):
+        _profiled_session().connect(timeout=1.0, cleanup_hvr_peer=True)
+
+    assert udp_socket.sent == []
+    assert udp_socket.closed
+
+
+@pytest.mark.parametrize('value', (None, 0, True, -1, 65536))
+def test_cleanup_requires_fixed_nonzero_local_port(value):
+    with pytest.raises(ValueError, match='fixed non-zero local port'):
+        _profiled_session(local_port=value).connect(cleanup_hvr_peer=True)
+
+
+def test_cleanup_requires_builtin_bool():
+    with pytest.raises(TypeError, match='must be a bool'):
+        _profiled_session().connect(cleanup_hvr_peer=1)
+
+
+def test_cleanup_requires_real_profiled_certificate_provider():
+    session = DtlsCoapSession(
+        'device.example',
+        5684,
+        cert_path='/synthetic/client.pem',
+        key_path='/synthetic/client.key',
+        local_port=49745,
+    )
+
+    with pytest.raises(ValueError, match='Samsung server profile'):
+        session.connect(cleanup_hvr_peer=True)
diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py
index 8358466..0396ac3 100644
--- a/tests/test_public_api_contract.py
+++ b/tests/test_public_api_contract.py
@@ -125,6 +125,20 @@ def test_samsung_server_profile_is_public_and_explicitly_bound():
     assert parameters["additional_ca_pem"].kind is inspect.Parameter.KEYWORD_ONLY
     assert parameters["additional_ca_pem"].default is None
 
+    discovery_parameters = inspect.signature(
+        SamsungServerProfile.discover_device
+    ).parameters
+    assert list(discovery_parameters) == ["role", "additional_ca_pem"]
+    assert all(
+        parameter.kind is inspect.Parameter.KEYWORD_ONLY
+        for parameter in discovery_parameters.values()
+    )
+    assert (
+        discovery_parameters["role"].default
+        is SamsungServerRole.HOME_APPLIANCE
+    )
+    assert discovery_parameters["additional_ca_pem"].default is None
+
 
 def test_server_certificate_auth_is_a_public_authentication_provider():
     profile = SamsungServerProfile.bound_device(
@@ -139,6 +153,7 @@ def test_server_certificate_auth_is_a_public_authentication_provider():
     assert session.key_path is None
     assert session.cert_pem is None
     assert session.key_pem is None
+    assert session.server_certificate_identity is None
     parameters = inspect.signature(ServerCertificateAuth).parameters
     assert list(parameters) == ["server_profile"]
     assert parameters["server_profile"].kind is inspect.Parameter.KEYWORD_ONLY
@@ -206,6 +221,11 @@ def test_dtls_session_keeps_current_consumer_methods():
     ]
     assert connect_cancel.kind is inspect.Parameter.KEYWORD_ONLY
     assert connect_cancel.default is None
+    connect_cleanup = inspect.signature(DtlsCoapSession.connect).parameters[
+        "cleanup_hvr_peer"
+    ]
+    assert connect_cleanup.kind is inspect.Parameter.KEYWORD_ONLY
+    assert connect_cleanup.default is False
     assert callable(ConnectCancellation().set)
     _assert_compatible_signature(
         DtlsCoapSession.quiesce_for_close,
diff --git a/tests/test_session_connect_deadline.py b/tests/test_session_connect_deadline.py
index c7c1f3b..bcb98d7 100644
--- a/tests/test_session_connect_deadline.py
+++ b/tests/test_session_connect_deadline.py
@@ -3,11 +3,12 @@
 from __future__ import annotations
 
 import socket
+from uuid import UUID
 
 import pytest
 from OpenSSL import SSL
 
-from smartthings_local.errors import SessionTimeoutError
+from smartthings_local.errors import AuthenticationError, SessionTimeoutError
 from smartthings_local.protocol import dtls_session
 from smartthings_local.protocol.dtls_session import DtlsCoapSession
 from smartthings_local.protocol.endpoint import ResolvedUdpEndpoint
@@ -34,6 +35,17 @@ def configure_context(self, _context):
             self.clock.advance(self.configure_delay)
 
 
+class _IdentityAuth(_Auth):
+    def __init__(self, identity):
+        super().__init__()
+        self.identity = identity
+
+    def _authenticated_server_identity(self, _connection):
+        if isinstance(self.identity, Exception):
+            raise self.identity
+        return self.identity
+
+
 class _Connection:
     def __init__(self, outcomes=None, outputs=None, timer=None):
         self.outcomes = list(outcomes or ())
@@ -300,9 +312,58 @@ def test_successful_handshake_preserves_connected_session_state(monkeypatch):
     assert session.sock is sock
     assert session.endpoint is endpoint
     assert session.dest == endpoint.sockaddr
+    assert session.server_certificate_identity is None
+    assert not sock.closed
+
+
+def test_successful_profiled_handshake_publishes_server_identity(monkeypatch):
+    clock = _Clock()
+    identity = UUID(bytes=b"\xab" * 16)
+    connection, sock, endpoint, _open_calls = _install_handshake(
+        monkeypatch,
+        clock,
+        outcomes=("want-read", "success"),
+        inbound=(b"synthetic server flight",),
+    )
+    session = _session(_IdentityAuth(identity))
+
+    session.connect(timeout=1.0)
+
+    assert session.server_certificate_identity == identity
+    assert session.conn is connection
+    assert session.sock is sock
+    assert session.endpoint is endpoint
     assert not sock.closed
 
 
+@pytest.mark.parametrize(
+    "identity",
+    (ValueError("missing identity"), "not-a-uuid", UUID(int=0)),
+)
+def test_profiled_handshake_fails_closed_without_valid_server_identity(
+    monkeypatch,
+    identity,
+):
+    clock = _Clock()
+    _connection, sock, _endpoint, _open_calls = _install_handshake(
+        monkeypatch,
+        clock,
+        outcomes=("want-read", "success"),
+        inbound=(b"synthetic server flight",),
+    )
+    session = _session(_IdentityAuth(identity))
+
+    with pytest.raises(AuthenticationError) as captured:
+        session.connect(timeout=1.0)
+
+    assert captured.value.code == "authentication"
+    assert str(identity) not in str(captured.value)
+    assert session.server_certificate_identity is None
+    assert session.conn is None
+    assert session.sock is None
+    assert sock.closed
+
+
 def test_connect_services_openssl_retransmit_timer(monkeypatch):
     clock = _Clock()
     outbound = b"\x16\xfe\xfd" + b"\x00" * 8 + b"\x00\x01x"
@@ -322,6 +383,31 @@ def test_connect_services_openssl_retransmit_timer(monkeypatch):
     assert connection.bio_writes == [b"synthetic server flight"]
 
 
+def test_handshake_driver_reports_each_successfully_sent_record(monkeypatch):
+    clock = _Clock()
+    first = b"\x16\xfe\xfd" + b"\x00" * 8 + b"\x00\x01a"
+    second = b"\x16\xfe\xfd" + b"\x00" * 7 + b"\x01\x00\x01b"
+    connection, sock, _endpoint, _open_calls = _install_handshake(
+        monkeypatch,
+        clock,
+        outcomes=("want-read", "success"),
+        outputs=(first + second,),
+        inbound=(b"synthetic server flight",),
+    )
+    sent_records = []
+
+    completed = dtls_session._drive_dtls_handshake(
+        connection,
+        sock,
+        deadline=clock.now + 1.0,
+        on_record_sent=sent_records.append,
+    )
+
+    assert completed is True
+    assert sock.sent == [first, second]
+    assert sent_records == [first, second]
+
+
 @pytest.mark.parametrize("success_delay", (0.1, 0.2))
 def test_handshake_success_at_or_after_deadline_is_retained(
     monkeypatch,