feat(core): switch from Python THP implementation to Rust-based one

diff --git a/core/.changelog.d/6442.changed b/core/.changelog.d/6442.changed
new file mode 100644
index 00000000..ccbdc8f9
--- /dev/null
+++ b/core/.changelog.d/6442.changed
@@ -0,0 +1 @@
+[T3W1] Switch to optimized THP implementation.
diff --git a/core/SConscript.firmware b/core/SConscript.firmware
index 32a844ea..f87288cb 100644
--- a/core/SConscript.firmware
+++ b/core/SConscript.firmware
@@ -940,6 +940,8 @@ if UI_PERFORMANCE_OVERLAY:
     features.append('ui_performance_overlay')
 if not PRODUCTION:
     features.append('dev_keys')
+if THP:
+    features.append('thp')
 
 
 rust = tools.add_rust_lib(
diff --git a/core/SConscript.unix b/core/SConscript.unix
index 05c3d563..04febee5 100644
--- a/core/SConscript.unix
+++ b/core/SConscript.unix
@@ -964,6 +964,8 @@ if EVERYTHING:
     features.append('universal_fw')
 if not PRODUCTION:
     features.append('dev_keys')
+if THP:
+    features.append('thp')
 
 rust = tools.add_rust_lib(
     env=env,
diff --git a/core/embed/rust/src/thp/micropython.rs b/core/embed/rust/src/thp/micropython.rs
index 1636a46f..21a76d68 100644
--- a/core/embed/rust/src/thp/micropython.rs
+++ b/core/embed/rust/src/thp/micropython.rs
@@ -417,7 +417,6 @@ pub static mp_module_trezorthp: Module = obj_module! {
     ///     - An integer: Lower 16 bits contain channel id, upper 16 bits contain buffer size hint in 8-byte blocks.
     ///       The event loop should call the `packet_in_channel()` function for this interface and if
     ///       the size hint is non-zero, then the receive buffer needs to be at least as large.
-    ///       If such buffer cannot be obtained, `channel_close()` should be called.
     ///       If buffer is in use by another channel, `send_transport_busy()` should be called.
     ///     """
     Qstr::MP_QSTR_packet_in => obj_fn_3!(thp_packet_in).as_obj(),
diff --git a/core/embed/upymod/qstrdefsport.h b/core/embed/upymod/qstrdefsport.h
index 374a539b..5bd6af27 100644
--- a/core/embed/upymod/qstrdefsport.h
+++ b/core/embed/upymod/qstrdefsport.h
@@ -430,16 +430,12 @@ Q(writers)
 #if USE_THP
 Q(ThpMessageType)
 Q(ThpPairingMethod)
-Q(alternating_bit_protocol)
 Q(apps.thp)
 Q(apps.thp.credential_manager)
 Q(apps.thp.pairing)
 Q(cache_thp)
 Q(cache_thp_keys)
 Q(channel)
-Q(channel_manager)
-Q(checksum)
-Q(control_byte)
 Q(cpace)
 Q(credential_manager)
 Q(crypto)
@@ -457,11 +453,7 @@ Q(thp)
 Q(trezor.enums.ThpMessageType)
 Q(trezor.enums.ThpPairingMethod)
 Q(trezor.wire.thp)
-Q(trezor.wire.thp.alternating_bit_protocol)
 Q(trezor.wire.thp.channel)
-Q(trezor.wire.thp.channel_manager)
-Q(trezor.wire.thp.checksum)
-Q(trezor.wire.thp.control_byte)
 Q(trezor.wire.thp.cpace)
 Q(trezor.wire.thp.crypto)
 Q(trezor.wire.thp.interface_context)
@@ -472,9 +464,7 @@ Q(trezor.wire.thp.received_message_handler)
 Q(trezor.wire.thp.session_context)
 Q(trezor.wire.thp.session_manager)
 Q(trezor.wire.thp.ui)
-Q(trezor.wire.thp.writer)
 Q(ui)
-Q(writer)
 #endif
 
 #if !BITCOIN_ONLY
diff --git a/core/embed/upymod/rustmods.c b/core/embed/upymod/rustmods.c
index c5511039..99d119b9 100644
--- a/core/embed/upymod/rustmods.c
+++ b/core/embed/upymod/rustmods.c
@@ -36,7 +36,7 @@ MP_REGISTER_MODULE(MP_QSTR_trezortranslate, mp_module_trezortranslate);
 MP_REGISTER_MODULE(MP_QSTR_trezorble, mp_module_trezorble);
 #endif
 
-#if 0
+#ifdef USE_THP
 MP_REGISTER_MODULE(MP_QSTR_trezorthp, mp_module_trezorthp);
 #endif
 
diff --git a/core/mocks/generated/trezorthp.pyi b/core/mocks/generated/trezorthp.pyi
index 4fd4cde9..3cf6f042 100644
--- a/core/mocks/generated/trezorthp.pyi
+++ b/core/mocks/generated/trezorthp.pyi
@@ -35,7 +35,6 @@ def packet_in(iface_num: int, packet_buffer: AnyBytes, credential_verify_fn: Cal
     - An integer: Lower 16 bits contain channel id, upper 16 bits contain buffer size hint in 8-byte blocks.
       The event loop should call the `packet_in_channel()` function for this interface and if
       the size hint is non-zero, then the receive buffer needs to be at least as large.
-      If such buffer cannot be obtained, `channel_close()` should be called.
       If buffer is in use by another channel, `send_transport_busy()` should be called.
     """
 
diff --git a/core/src/apps/debug/__init__.py b/core/src/apps/debug/__init__.py
index 7c6ecb23..81ac185f 100644
--- a/core/src/apps/debug/__init__.py
+++ b/core/src/apps/debug/__init__.py
@@ -311,7 +311,7 @@ if __debug__:
         from trezor.messages import DebugLinkPairingInfo
 
         return DebugLinkPairingInfo(
-            channel_id=ctx.channel_id,
+            channel_id=ctx.channel_ctx.channel_id_bytes(),
             handshake_hash=ctx.channel_ctx.get_handshake_hash(),
             code_entry_code=ctx.code_code_entry,
             code_qr_code=ctx.code_qr_code,
diff --git a/core/src/apps/evolu/get_delegated_identity_key.py b/core/src/apps/evolu/get_delegated_identity_key.py
index 3482ac1b..68da38bb 100644
--- a/core/src/apps/evolu/get_delegated_identity_key.py
+++ b/core/src/apps/evolu/get_delegated_identity_key.py
@@ -69,9 +69,7 @@ if utils.USE_THP:
         if msg.thp_credential is None:
             raise DataError("THP credential must be provided when THP is enabled")
         credential_received = decode_credential(msg.thp_credential)
-        host_static_public_key = (
-            get_channel_context().channel_cache.get_host_static_public_key()
-        )
+        host_static_public_key = get_channel_context().get_host_static_public_key()
         if not validate_credential(credential_received, host_static_public_key):
             raise DataError("Invalid credential")
 
diff --git a/core/src/apps/thp/credential_manager.py b/core/src/apps/thp/credential_manager.py
index 497e52d0..233529bc 100644
--- a/core/src/apps/thp/credential_manager.py
+++ b/core/src/apps/thp/credential_manager.py
@@ -5,9 +5,12 @@ from trezor.crypto import hmac
 from trezor.messages import (
     ThpAuthenticatedCredentialData,
     ThpCredentialMetadata,
+    ThpHandshakeCompletionReqNoisePayload,
     ThpPairingCredential,
 )
+from trezor.utils import truncate_utf8
 from trezor.wire.message_handler import wrap_protobuf_load
+from trezorthp import MAX_CREDENTIAL_LEN
 
 if TYPE_CHECKING:
     from buffer_types import AnyBytes
@@ -53,6 +56,20 @@ def issue_credential(
     Issue a pairing credential binded to the provided host static public key
     and credential metadata.
     """
+    # Truncate app_name and host_name to fit MAX_CREDENTIAL_LEN
+    if (
+        len(credential_metadata.app_name.encode())
+        + len(credential_metadata.host_name.encode())
+        > 80
+    ):
+        credential_metadata.host_name = truncate_utf8(credential_metadata.host_name, 40)
+    if (
+        len(credential_metadata.app_name.encode())
+        + len(credential_metadata.host_name.encode())
+        > 80
+    ):
+        credential_metadata.app_name = truncate_utf8(credential_metadata.app_name, 40)
+
     cred_auth_key = derive_cred_auth_key()
     proto_msg = ThpAuthenticatedCredentialData(
         host_static_public_key=host_static_public_key,
@@ -63,9 +80,19 @@ def issue_credential(
 
     proto_msg = ThpPairingCredential(cred_metadata=credential_metadata, mac=mac)
     credential_raw = _encode_message_into_new_buffer(proto_msg)
+    if len(credential_raw) > MAX_CREDENTIAL_LEN:
+        raise ValueError("Credential too long")
     return credential_raw
 
 
+def unwrap_credential(encoded_noise_payload: AnyBytes) -> AnyBytes | None:
+    expected_type = protobuf.type_for_name("ThpHandshakeCompletionReqNoisePayload")
+    msg = wrap_protobuf_load(encoded_noise_payload, expected_type)
+    if not ThpHandshakeCompletionReqNoisePayload.is_type_of(msg):
+        raise TypeError
+    return msg.host_pairing_credential
+
+
 def decode_credential(
     encoded_pairing_credential_message: AnyBytes,
 ) -> ThpPairingCredential:
@@ -74,7 +101,8 @@ def decode_credential(
     """
     expected_type = protobuf.type_for_name("ThpPairingCredential")
     credential = wrap_protobuf_load(encoded_pairing_credential_message, expected_type)
-    assert ThpPairingCredential.is_type_of(credential)
+    if not ThpPairingCredential.is_type_of(credential):
+        raise TypeError
     return credential
 
 
diff --git a/core/src/apps/thp/pairing.py b/core/src/apps/thp/pairing.py
index 1e81a461..41409b39 100644
--- a/core/src/apps/thp/pairing.py
+++ b/core/src/apps/thp/pairing.py
@@ -144,8 +144,7 @@ async def handle_pairing_request(
             # Should raise UnexpectedMessageException
             result = await ctx.show_pairing_method_screen()
         except UnexpectedMessageException as e:
-            if (raw_response := e.msg) is None:
-                raise  # propagate stale channel preemption
+            raw_response = e.msg
             req_type = protobuf.type_for_wire(
                 ctx.message_type_enum_name, raw_response.type
             )
@@ -191,7 +190,7 @@ async def handle_credential_phase(
     if credential is not None:
         autoconnect = is_credential_autoconnect(credential)
         if not autoconnect:
-            autoconnect = ctx.channel_ctx.is_channel_to_replace()
+            autoconnect = ctx.channel_ctx.is_autoconnected()
         if credential.cred_metadata is not None:
             ctx.host_name = credential.cred_metadata.host_name
             ctx.app_name = credential.cred_metadata.app_name
@@ -404,13 +403,10 @@ async def _handle_credential_request(
         # Cannot ask for autoconnect=True credential directly after pairing
         if ctx.channel_ctx.credential is None:
             raise DataError("Cannot ask for autoconnect credential after pairing")
-        from storage.cache_common import CHANNEL_HOST_STATIC_PUBKEY
 
         from .credential_manager import validate_credential
 
-        host_static_public_key = ctx.channel_ctx.channel_cache.get(
-            CHANNEL_HOST_STATIC_PUBKEY
-        )
+        host_static_public_key = ctx.channel_ctx.get_host_static_public_key()
 
         if not host_static_public_key or not validate_credential(
             credential=ctx.channel_ctx.credential,
@@ -454,7 +450,7 @@ async def _handle_end_request(
 
 
 async def _end_pairing(ctx: PairingContext) -> ThpEndResponse:
-    ctx.channel_ctx.replace_old_channels_with_the_same_host_public_key()
+    ctx.channel_ctx.end_pairing_and_replace()
     ctx.channel_ctx.set_channel_state(ChannelState.ENCRYPTED_TRANSPORT)
     return ThpEndResponse()
 
diff --git a/core/src/storage/cache_codec.py b/core/src/storage/cache_codec.py
index 06d9f46f..c1d07801 100644
--- a/core/src/storage/cache_codec.py
+++ b/core/src/storage/cache_codec.py
@@ -13,7 +13,7 @@ if TYPE_CHECKING:
 
 
 _MAX_SESSIONS_COUNT = const(10)
-SESSION_ID_LENGTH = const(32)
+_SESSION_ID_LENGTH = const(32)
 
 
 class SessionCache(DataCache):
@@ -23,7 +23,7 @@ class SessionCache(DataCache):
     """
 
     def __init__(self) -> None:
-        self.session_id = bytearray(SESSION_ID_LENGTH)
+        self.session_id = bytearray(_SESSION_ID_LENGTH)
         if utils.BITCOIN_ONLY:
             self.fields = (
                 64,  # APP_COMMON_SEED
@@ -50,7 +50,7 @@ class SessionCache(DataCache):
 
         # generate a new session id if we don't have it yet
         if not self.session_id:
-            self.session_id[:] = random.bytes(SESSION_ID_LENGTH)
+            self.session_id[:] = random.bytes(_SESSION_ID_LENGTH)
         # export it as immutable bytes
         return bytes(self.session_id)
 
@@ -94,7 +94,7 @@ def start_session(received_session_id: AnyBytes | None = None) -> AnyBytes:
 
     if (
         received_session_id is not None
-        and len(received_session_id) != SESSION_ID_LENGTH
+        and len(received_session_id) != _SESSION_ID_LENGTH
     ):
         # Prevent the caller from setting received_session_id=b"" and finding a cleared
         # session. More generally, short-circuit the session id search, because we know
diff --git a/core/src/storage/cache_thp.py b/core/src/storage/cache_thp.py
index 256db94f..40bc9236 100644
--- a/core/src/storage/cache_thp.py
+++ b/core/src/storage/cache_thp.py
@@ -3,10 +3,8 @@ from micropython import const
 from typing import TYPE_CHECKING
 
 from storage.cache_common import (
-    CHANNEL_HOST_STATIC_PUBKEY,
     CHANNEL_ID,
-    CHANNEL_STATE,
-    CHANNEL_SYNC,
+    LAST_USAGE,
     SESSION_ID,
     SESSION_STATE,
     DataCache,
@@ -14,82 +12,19 @@ from storage.cache_common import (
 
 if TYPE_CHECKING:
     from buffer_types import AnyBytes
-    from typing import Sequence
 
 # THP specific constants
-_MAX_CHANNELS_COUNT = const(10)
 _MAX_SESSIONS_COUNT = const(20)
 
 _CHANNEL_ID_LENGTH = const(2)
-SESSION_ID_LENGTH = const(1)
-KEY_LENGTH = const(32)
-TAG_LENGTH = const(16)
+_SESSION_ID_LENGTH = const(1)
 
 _UNALLOCATED_STATE = const(0)
 _ALLOCATED_STATE = const(1)
 _SEEDLESS_STATE = const(2)
 
-_MAX_CHANNEL_ID = const(0xFFEF)
-# Channel IDs from 0xFFF0 to 0xFFFE, and 0x0000, are reserved for future use
-BROADCAST_CHANNEL_ID = const(0xFFFF)
-
-
-class ThpDataCache(DataCache):
-
-    def __init__(self) -> None:
-        self.last_usage = 0
-        super().__init__()
-
-    @property
-    def channel_id(self) -> bytes:
-        return self.get(CHANNEL_ID) or b""
-
-    def clear(self) -> None:
-        self.last_usage = 0
-        super().clear()
-
-
-class ChannelCache(ThpDataCache):
 
-    def __init__(self) -> None:
-        self.fields = (
-            2,  # CHANNEL_ID
-            1,  # CHANNEL_STATE
-            1,  # CHANNEL_IFACE
-            1,  # CHANNEL_SYNC
-            32,  # CHANNEL_HANDSHAKE_HASH
-            32,  # CHANNEL_KEY_RECEIVE
-            32,  # CHANNEL_KEY_SEND
-            8,  # CHANNEL_NONCE_RECEIVE
-            8,  # CHANNEL_NONCE_SEND
-            32,  # CHANNEL_HOST_STATIC_PUBKEY
-            2,  # CHANNEL_ACK_LATENCY_MS
-        )
-        super().__init__()
-        self.set_int(CHANNEL_SYNC, 0x80)
-
-    @property
-    def sync(self) -> int:
-        # can_send_bit | sync_receive_bit | sync_send_bit | ack_piggybacking | rfu(4)
-        return self.get_int(CHANNEL_SYNC) or 0x00
-
-    @sync.setter
-    def sync(self, value: int) -> None:
-        self.set_int(CHANNEL_SYNC, value)
-
-    def set_host_static_public_key(self, key: memoryview) -> None:
-        if len(key) != KEY_LENGTH:
-            raise ValueError  # Invalid key length
-        self.set(CHANNEL_HOST_STATIC_PUBKEY, key)
-
-    def get_host_static_public_key(self) -> bytes:
-        key = self.get(CHANNEL_HOST_STATIC_PUBKEY)
-        if key is None:
-            raise ValueError  # Host static public key is not set in the channel cache.
-        return key
-
-
-class SessionThpCache(ThpDataCache):
+class SessionThpCache(DataCache):
     def __init__(self) -> None:
         from trezor import utils
 
@@ -98,6 +33,7 @@ class SessionThpCache(ThpDataCache):
                 2,  # CHANNEL_ID
                 1,  # SESSION_ID
                 1,  # SESSION_STATE
+                4,  # LAST_USAGE
                 64,  # APP_COMMON_SEED
                 2,  # APP_COMMON_AUTHORIZATION_TYPE
                 128,  # APP_COMMON_AUTHORIZATION_DATA
@@ -108,6 +44,7 @@ class SessionThpCache(ThpDataCache):
                 2,  # CHANNEL_ID
                 1,  # SESSION_ID
                 1,  # SESSION_STATE
+                4,  # LAST_USAGE
                 64,  # APP_COMMON_SEED
                 2,  # APP_COMMON_AUTHORIZATION_TYPE
                 128,  # APP_COMMON_AUTHORIZATION_DATA
@@ -123,81 +60,46 @@ class SessionThpCache(ThpDataCache):
     def session_id(self) -> bytes:
         return self.get(SESSION_ID) or b""
 
+    @property
+    def channel_id(self) -> bytes:
+        return self.get(CHANNEL_ID) or b""
+
+    @property
+    def last_usage(self) -> int:
+        return self.get_int(LAST_USAGE) or 0
+
     def clear(self) -> None:
         super().clear()
 
 
-_CHANNELS: list[ChannelCache] = []
 _SESSIONS: list[SessionThpCache] = []
-cid_counter: int = 0
 
 # Last-used counter
 _usage_counter = 0
 
 
 def initialize() -> None:
-    global cid_counter
-
-    for _ in range(_MAX_CHANNELS_COUNT):
-        _CHANNELS.append(ChannelCache())
     for _ in range(_MAX_SESSIONS_COUNT):
         _SESSIONS.append(SessionThpCache())
 
-    for channel in _CHANNELS:
-        channel.clear()
     for session in _SESSIONS:
         session.clear()
 
-    from trezorcrypto import random
-
-    cid_counter = random.uniform(_MAX_CHANNEL_ID)
-
-
-def get_new_channel() -> ChannelCache:
-
-    new_cid = get_next_channel_id()
-    index = _get_next_channel_index()
-
-    # clear sessions from replaced channel
-    if (
-        _CHANNELS[index].get_int(CHANNEL_STATE, _UNALLOCATED_STATE)
-        != _UNALLOCATED_STATE
-    ):
-        old_cid = _CHANNELS[index].channel_id
-        clear_sessions_with_channel_id(old_cid)
-
-    _CHANNELS[index] = ChannelCache()
-    _CHANNELS[index].set(CHANNEL_ID, new_cid)
-    _CHANNELS[index].last_usage = _get_usage_counter_and_increment()
-    _CHANNELS[index].set_int(CHANNEL_STATE, _UNALLOCATED_STATE)
-    return _CHANNELS[index]
-
 
 def update_channel_last_used(channel_id: AnyBytes) -> None:
-    for channel in _CHANNELS:
-        if channel.channel_id == channel_id:
-            channel.last_usage = _get_usage_counter_and_increment()
-            return
+    from trezorthp import channel_update_last_usage
+
+    channel_update_last_usage(int.from_bytes(channel_id, "big"))
 
 
 def update_session_last_used(channel_id: AnyBytes, session_id: AnyBytes) -> None:
+    update_channel_last_used(channel_id)  # update channel even if session is seedless
     for session in _SESSIONS:
         if session.channel_id == channel_id and session.session_id == session_id:
-            session.last_usage = _get_usage_counter_and_increment()
-            update_channel_last_used(channel_id)
+            session.set_int(LAST_USAGE, _get_usage_counter_and_increment())
             return
 
 
-def find_allocated_channel(cid: int) -> ChannelCache | None:
-    for channel in _CHANNELS:
-        state = channel.get_int(CHANNEL_STATE, _UNALLOCATED_STATE)
-        if state == _UNALLOCATED_STATE:
-            continue
-        if channel.get_int(CHANNEL_ID) == cid:
-            return channel
-    return None
-
-
 def get_allocated_session(
     channel_id: bytes, session_id: bytes
 ) -> SessionThpCache | None:
@@ -214,7 +116,7 @@ def get_allocated_session_index(channel_id: bytes, session_id: bytes) -> int | N
 
     Raises `Exception` if either channel_id or session_id has an invalid length.
     """
-    if len(channel_id) != _CHANNEL_ID_LENGTH or len(session_id) != SESSION_ID_LENGTH:
+    if len(channel_id) != _CHANNEL_ID_LENGTH or len(session_id) != _SESSION_ID_LENGTH:
         raise ValueError("At least one of arguments has invalid length")
 
     for i in range(_MAX_SESSIONS_COUNT):
@@ -235,70 +137,25 @@ def is_seedless_session(session_cache: SessionThpCache) -> bool:
     return session_cache.get_int(SESSION_STATE, _UNALLOCATED_STATE) == _SEEDLESS_STATE
 
 
-def create_or_replace_session(
-    channel: ChannelCache, session_id: bytes
-) -> SessionThpCache:
-    index = get_allocated_session_index(channel.channel_id, session_id)
+def create_or_replace_session(channel_id: bytes, session_id: bytes) -> SessionThpCache:
+    index = get_allocated_session_index(channel_id, session_id)
     if index is None:
         index = _get_next_session_index()
 
     _SESSIONS[index] = SessionThpCache()
-    _SESSIONS[index].set(CHANNEL_ID, channel.channel_id)
+    _SESSIONS[index].set(CHANNEL_ID, channel_id)
     _SESSIONS[index].set(SESSION_ID, session_id)
-    _SESSIONS[index].last_usage = _get_usage_counter_and_increment()
-    channel.last_usage = (
-        _get_usage_counter_and_increment()
-    )  # increment also use of the channel so it does not get replaced
+    _SESSIONS[index].set_int(LAST_USAGE, _get_usage_counter_and_increment())
+    update_channel_last_used(channel_id)
 
     _SESSIONS[index].set_int(SESSION_STATE, _ALLOCATED_STATE)
     return _SESSIONS[index]
 
 
-def _migrate_sessions(old_channel: ChannelCache, new_channel: ChannelCache) -> None:
+def migrate_sessions(old_channel_id: bytes, new_channel_id: bytes) -> None:
     for session in _SESSIONS:
-        if session.channel_id == old_channel.channel_id:
-            session.set(CHANNEL_ID, new_channel.channel_id)
-
-
-def _replace_channel(old_channel: ChannelCache, new_channel: ChannelCache) -> None:
-    _migrate_sessions(old_channel, new_channel)
-    old_channel.clear()
-
-
-def conditionally_replace_channel(
-    new_channel: ChannelCache, required_state: int, required_key: int
-) -> bool:
-    """Replaces "old channel" cache entry with a `new_channel` if two conditions are met:
-
-    1. The "old channel" is in a state `required_state`
-    2. The "old channel" has the same value for `required_key` as the `new_channel`
-
-
-    Returns: bool - whether any channel was replaced.
-    """
-    was_any_channel_replaced: bool = False
-    for channel in _CHANNELS:
-        if channel.channel_id == new_channel.channel_id:
-            continue
-        if channel.get_int(CHANNEL_STATE) == required_state and channel.get(
-            required_key
-        ) == new_channel.get(required_key):
-            _replace_channel(channel, new_channel)
-            was_any_channel_replaced = True
-    return was_any_channel_replaced
-
-
-def is_there_a_channel_to_replace(
-    new_channel: ChannelCache, required_state: int, required_key: int
-) -> bool:
-    for channel in _CHANNELS:
-        if channel.channel_id == new_channel.channel_id:
-            continue
-        if channel.get_int(CHANNEL_STATE) == required_state and channel.get(
-            required_key
-        ) == new_channel.get(required_key):
-            return True
-    return False
+        if session.channel_id == old_channel_id:
+            session.set(CHANNEL_ID, new_channel_id)
 
 
 def _get_usage_counter_and_increment() -> int:
@@ -307,28 +164,11 @@ def _get_usage_counter_and_increment() -> int:
     return _usage_counter
 
 
-def _get_next_channel_index() -> int:
-    idx = _get_unallocated_channel_index()
-    if idx is not None:
-        return idx
-    return _get_least_recently_used_item(_CHANNELS, max_count=_MAX_CHANNELS_COUNT)
-
-
 def _get_next_session_index() -> int:
     idx = _get_unallocated_session_index()
     if idx is not None:
         return idx
-    return _get_least_recently_used_item(_SESSIONS, max_count=_MAX_SESSIONS_COUNT)
-
-
-def _get_unallocated_channel_index() -> int | None:
-    for i in range(_MAX_CHANNELS_COUNT):
-        if (
-            _CHANNELS[i].get_int(CHANNEL_STATE, _UNALLOCATED_STATE)
-            == _UNALLOCATED_STATE
-        ):
-            return i
-    return None
+    return _get_least_recently_used_session()
 
 
 def _get_unallocated_session_index() -> int | None:
@@ -338,31 +178,12 @@ def _get_unallocated_session_index() -> int | None:
     return None
 
 
-def get_next_channel_id() -> bytes:
-    global cid_counter
-    while True:
-        cid_counter += 1
-        if cid_counter > _MAX_CHANNEL_ID:
-            cid_counter = 1
-        if _is_cid_unique():
-            break
-    return cid_counter.to_bytes(_CHANNEL_ID_LENGTH, "big")
-
-
-def _is_cid_unique() -> bool:
-    cid_counter_bytes = cid_counter.to_bytes(_CHANNEL_ID_LENGTH, "big")
-    for channel in _CHANNELS:
-        if channel.channel_id == cid_counter_bytes:
-            return False
-    return True
-
-
-def _get_least_recently_used_item(list: Sequence[ThpDataCache], max_count: int) -> int:
+def _get_least_recently_used_session() -> int:
     lru_counter = _usage_counter + 1
     lru_item_index = 0
-    for i in range(max_count):
-        if list[i].last_usage < lru_counter:
-            lru_counter = list[i].last_usage
+    for i in range(_MAX_SESSIONS_COUNT):
+        if _SESSIONS[i].last_usage < lru_counter:
+            lru_counter = _SESSIONS[i].last_usage
             lru_item_index = i
     return lru_item_index
 
@@ -382,6 +203,14 @@ def clear_sessions_with_channel_id(channel_id: bytes) -> None:
             session.clear()
 
 
+def clear_sessions_without_channel() -> None:
+    from trezorthp import channel_is_open
+
+    for session in _SESSIONS:
+        if not channel_is_open(int.from_bytes(session.channel_id, "big")):
+            session.clear()
+
+
 def clear_session(session: SessionThpCache) -> None:
     for s in _SESSIONS:
         if s.channel_id == session.channel_id and s.session_id == session.session_id:
@@ -391,16 +220,18 @@ def clear_session(session: SessionThpCache) -> None:
 def clear_all() -> None:
     for session in _SESSIONS:
         session.clear()
-    for channel in _CHANNELS:
-        channel.clear()
+
+    from trezorthp import channel_close_all
+
+    channel_close_all()
 
 
 def clear_all_except_one_session_keys(excluded: tuple[AnyBytes, AnyBytes]) -> None:
     cid, sid = excluded
 
-    for channel in _CHANNELS:
-        if channel.channel_id != cid:
-            channel.clear()
+    from trezorthp import channel_close_all
+
+    channel_close_all(exclude_channel_id=int.from_bytes(cid, "big"))
 
     for session in _SESSIONS:
         if session.channel_id != cid or session.session_id != sid:
@@ -408,4 +239,4 @@ def clear_all_except_one_session_keys(excluded: tuple[AnyBytes, AnyBytes]) -> No
         else:
             s_last_usage = session.last_usage
             session.clear()
-            session.last_usage = s_last_usage
+            session.set_int(LAST_USAGE, s_last_usage)
diff --git a/core/src/storage/cache_thp_keys.py b/core/src/storage/cache_thp_keys.py
index 8744a227..6d1eae0f 100644
--- a/core/src/storage/cache_thp_keys.py
+++ b/core/src/storage/cache_thp_keys.py
@@ -3,29 +3,17 @@ from micropython import const
 from trezor import utils
 
 if utils.USE_THP:
-    # Cache keys for THP channel
-    CHANNEL_ID = const(0)
-    CHANNEL_STATE = const(1)
-    CHANNEL_IFACE = const(2)
-    CHANNEL_SYNC = const(3)
-    CHANNEL_HANDSHAKE_HASH = const(4)
-    CHANNEL_KEY_RECEIVE = const(5)
-    CHANNEL_KEY_SEND = const(6)
-    CHANNEL_NONCE_RECEIVE = const(7)
-    CHANNEL_NONCE_SEND = const(8)
-    CHANNEL_HOST_STATIC_PUBKEY = const(9)
-    CHANNEL_ACK_LATENCY_MS = const(10)
-
     # Cache keys for THP session
-    # CHANNEL_ID = const(0)
+    CHANNEL_ID = const(0)
     SESSION_ID = const(1)
     SESSION_STATE = const(2)
-    APP_COMMON_SEED = const(3)
-    APP_COMMON_AUTHORIZATION_TYPE = const(4)
-    APP_COMMON_AUTHORIZATION_DATA = const(5)
-    APP_COMMON_NONCE = const(6)
+    LAST_USAGE = const(3)
+    APP_COMMON_SEED = const(4)
+    APP_COMMON_AUTHORIZATION_TYPE = const(5)
+    APP_COMMON_AUTHORIZATION_DATA = const(6)
+    APP_COMMON_NONCE = const(7)
     if not utils.BITCOIN_ONLY:
-        APP_COMMON_DERIVE_CARDANO = const(7)
-        APP_CARDANO_ICARUS_SECRET = const(8)
-        APP_CARDANO_ICARUS_TREZOR_SECRET = const(9)
-        APP_MONERO_LIVE_REFRESH = const(10)
+        APP_COMMON_DERIVE_CARDANO = const(8)
+        APP_CARDANO_ICARUS_SECRET = const(9)
+        APP_CARDANO_ICARUS_TREZOR_SECRET = const(10)
+        APP_MONERO_LIVE_REFRESH = const(11)
diff --git a/core/src/trezor/loop.py b/core/src/trezor/loop.py
index 5dbb34c8..95ba235a 100644
--- a/core/src/trezor/loop.py
+++ b/core/src/trezor/loop.py
@@ -238,8 +238,9 @@ class sleep(Syscall[int]):
 class wait(Syscall[T]):
     """
     Pause current task, and resume only after a message on `msg_iface` is
-    received.  Messages are received either from an USB interface, or the
-    touch display.  Result value is a tuple of message values.
+    received.  Messages are received either from an USB/BLE interface,
+    or the touch display, or a physical button.  Result value is a tuple of
+    message values.
 
     Example:
 
diff --git a/core/src/trezor/wire/__init__.py b/core/src/trezor/wire/__init__.py
index 3a330f54..46e1489f 100644
--- a/core/src/trezor/wire/__init__.py
+++ b/core/src/trezor/wire/__init__.py
@@ -53,6 +53,7 @@ if TYPE_CHECKING:
     from typing import Any, Callable, Coroutine, Generic, Type, TypeVar
 
     from trezor.wire.thp.channel import Channel
+    from trezor.wire.thp.interface_context import InterfaceContext
 
     T = TypeVar("T")
     Msg = TypeVar("Msg", bound=protobuf.MessageType)
@@ -109,15 +110,17 @@ if utils.USE_THP:
     THP_BUFFERS_PROVIDER = Provider((ThpBuffer(), ThpBuffer()))
 
     if __debug__:
-        _THP_CHANNELS = []
+        _THP_IFACES: list[InterfaceContext] = []
 
         def find_thp_channel(channel_id: AnyBytes) -> Channel | None:
-            """Used by `DebugLinkGetPairingInfo` (only for tests)."""
-            key = int.from_bytes(channel_id, "big")
-            for channels in _THP_CHANNELS:
-                result = channels.get(key)
-                if result is not None:
-                    return result
+            """Used by `DebugLinkGetPairingInfo` (only for tests). Currently only
+            works with channels that have active workflow (e.g. pairing)."""
+            for ifctx in _THP_IFACES:
+                if (
+                    ifctx.active_channel
+                    and ifctx.active_channel.channel_id_bytes() == channel_id
+                ):
+                    return ifctx.active_channel
             return None
 
     def setup(*ifaces: WireInterface) -> None:
@@ -127,15 +130,23 @@ if utils.USE_THP:
     async def handle_session_thp(*ifaces: WireInterface) -> None:
         ctx = ThpContext(*ifaces)
         if __debug__:
-            _THP_CHANNELS.extend(iface_ctx._channels for iface_ctx in ctx._iface_ctxs)
+            _THP_IFACES[:] = ctx._iface_ctxs
 
         try:
-            while (channel := await ctx.get_next_message()) is None:
-                # wait until a new channel is established (on any interface)
-                pass
+            # wait until channel activity (on any interface)
+            channel = await ctx.get_active_channel()
 
+            # at this point channel has valid message waiting
+            # process messages until it returns do_not_restart=False
             while await received_message_handler.handle_received_message(channel):
-                pass
+                if __debug__:
+                    log.debug(
+                        __name__,
+                        "Skipping THP session restart on channel %04x",
+                        channel.channel_id,
+                        iface=channel.iface,
+                    )
+
         finally:
             if __debug__:
                 log.debug(__name__, "Finished THP session: %s", ifaces)
@@ -145,6 +156,8 @@ if utils.USE_THP:
                 import apps.debug
 
                 await apps.debug.close_session()
+            # Send out any queued messages.
+            await ctx.close()
             loop.clear()
 
 else:
diff --git a/core/src/trezor/wire/context.py b/core/src/trezor/wire/context.py
index 7f68b3d1..3a42990f 100644
--- a/core/src/trezor/wire/context.py
+++ b/core/src/trezor/wire/context.py
@@ -42,11 +42,9 @@ class UnexpectedMessageException(Exception):
 
     Utility exception to inform the session handler that the current workflow
     should be aborted and a new one started as if `msg` was the first message.
-
-    If `msg` is `None`, the event loop should be restarted.
     """
 
-    def __init__(self, msg: Message | None) -> None:
+    def __init__(self, msg: Message) -> None:
         super().__init__()
         self.msg = msg
 
@@ -155,7 +153,7 @@ def try_get_ctx_ids() -> tuple[AnyBytes, AnyBytes] | None:
         except NoWireContext:
             return None
         if isinstance(ctx, GenericSessionContext):
-            ids = (ctx.channel_id, ctx.session_id.to_bytes(1, "big"))
+            ids = (ctx.channel_id.to_bytes(2, "big"), ctx.session_id.to_bytes(1, "big"))
     return ids
 
 
diff --git a/core/src/trezor/wire/message_handler.py b/core/src/trezor/wire/message_handler.py
index f97590d8..6b7eba6b 100644
--- a/core/src/trezor/wire/message_handler.py
+++ b/core/src/trezor/wire/message_handler.py
@@ -70,11 +70,10 @@ async def handle_single_message(ctx: Context, msg: Message) -> bool:
         except Exception:
             msg_type = f"{msg.type} - unknown message type"
         if utils.USE_THP:
-            cid = utils.hexlify_if_bytes(ctx.channel_id)
             log.info(
                 __name__,
-                "(cid: %s) received message: %s",
-                cid,
+                "(cid: %04x) received message: %s",
+                ctx.channel_id,
                 msg_type,
                 iface=ctx.iface,
             )
diff --git a/core/src/trezor/wire/protocol_common.py b/core/src/trezor/wire/protocol_common.py
index 8f4be82a..c460ae76 100644
--- a/core/src/trezor/wire/protocol_common.py
+++ b/core/src/trezor/wire/protocol_common.py
@@ -53,12 +53,12 @@ class Context:
     single Bluetooth connection, etc.).
     """
 
-    channel_id: AnyBytes
+    channel_id: int
 
     def __init__(
         self,
         iface: WireInterface,
-        channel_id: AnyBytes | None = None,
+        channel_id: int | None = None,
         message_type_enum_name: str = "MessageType",
     ) -> None:
         self.iface: WireInterface = iface
@@ -299,17 +299,19 @@ class ContinueOnErrors(ButtonRequestHandler):
                     # All is well, continue handling ButtonRequests.
                     success = True
                     return
-                except UnexpectedMessageException as exc:
-                    # in case of THP channel preemption, `msg` is not set.
-                    # TRANSPORT_BUSY error has been already sent by `InterfaceContext.handle_packet()`.
-                    if exc.msg:
-                        from trezor.enums import FailureType
-                        from trezor.messages import Failure
-
-                        # notify the host that the device cannot be preempted
-                        await self.ctx.write(
-                            Failure(code=FailureType.InProgress, message=self.msg)
-                        )
+                except UnexpectedMessageException:
+                    from trezor.enums import FailureType
+                    from trezor.messages import Failure
+
+                    # notify the host that the device cannot be preempted
+                    await self.ctx.write(
+                        Failure(code=FailureType.InProgress, message=self.msg)
+                    )
+                    # continue receiving messages
+                except ChannelPreemptedException:
+                    # TRANSPORT_BUSY error has been already sent by
+                    # `InterfaceContext.read_packet_for_channel()`.
+                    pass
                     # continue receiving messages
                 except Exception as exc:
                     if __debug__:
@@ -334,3 +336,10 @@ class ContinueOnErrors(ButtonRequestHandler):
 
 class WireError(Exception):
     pass
+
+
+class ChannelPreemptedException(Exception):
+    """THP uses this exception to free up resources taken by potentially stuck channel.
+    Raising this exception should restart the event loop."""
+
+    pass
diff --git a/core/src/trezor/wire/thp/__init__.py b/core/src/trezor/wire/thp/__init__.py
index 0945178f..25542e11 100644
--- a/core/src/trezor/wire/thp/__init__.py
+++ b/core/src/trezor/wire/thp/__init__.py
@@ -1,71 +1,29 @@
-import ustruct
-from micropython import const
 from typing import TYPE_CHECKING
 
-from storage.cache_thp import BROADCAST_CHANNEL_ID
 from trezor import protobuf, utils
 from trezor.enums import ThpPairingMethod
 from trezor.messages import ThpDeviceProperties
+from trezorthp import MAX_DEVICE_PROPERTIES_LEN
 
 from ..protocol_common import WireError
 
 if TYPE_CHECKING:
     from buffer_types import AnyBytes
     from enum import IntEnum
-    from typing import Iterable
 
     from trezor.wire import WireInterface
-    from typing_extensions import Self
 else:
     IntEnum = object
 
-CODEC_V1 = const(0x3F)
 
-HANDSHAKE_INIT_REQ = const(0x00)
-HANDSHAKE_INIT_RES = const(0x01)
-HANDSHAKE_COMP_REQ = const(0x02)
-HANDSHAKE_COMP_RES = const(0x03)
-ENCRYPTED = const(0x04)
-
-ACK_MESSAGE = const(0x20)
-CHANNEL_ALLOCATION_REQ = const(0x40)
-_CHANNEL_ALLOCATION_RES = const(0x41)
-_ERROR = const(0x42)
-PING = const(0x43)
-_PONG = const(0x44)
-
-CONTINUATION_PACKET = const(0x80)
-
-
-class ThpError(WireError):
-    pass
-
-
-class ThpDecryptionError(ThpError):
-    pass
-
-
-class ThpDeviceLockedError(ThpError):
-    pass
-
-
-class ThpUnallocatedSessionError(ThpError):
+class ThpUnallocatedSessionError(WireError):
 
     def __init__(self, session_id: int) -> None:
         self.session_id = session_id
 
 
-class ThpErrorType(IntEnum):
-    TRANSPORT_BUSY = 1
-    UNALLOCATED_CHANNEL = 2
-    DECRYPTION_FAILED = 3
-    DEVICE_LOCKED = 5
-
-
+# Only a subset, handshake states are not visible to python
 class ChannelState(IntEnum):
-    UNALLOCATED = 0
-    TH1 = 1
-    TH2 = 2
     TP0 = 3
     TP1 = 4
     TP2 = 5
@@ -73,7 +31,6 @@ class ChannelState(IntEnum):
     TP4 = 7
     TC1 = 8
     ENCRYPTED_TRANSPORT = 9
-    INVALIDATED = 10
 
 
 class SessionState(IntEnum):
@@ -82,97 +39,6 @@ class SessionState(IntEnum):
     SEEDLESS = 2
 
 
-class PacketHeader:
-    INIT_FORMAT = ">BHH"
-    CONT_FORMAT = ">BH"
-
-    INIT_LENGTH = ustruct.calcsize(INIT_FORMAT)
-    CONT_LENGTH = ustruct.calcsize(CONT_FORMAT)
-
-    def __init__(self, ctrl_byte: int, cid: int, length: int) -> None:
-        self.ctrl_byte = ctrl_byte
-        self.cid = cid
-        self.length = length
-
-    def to_bytes(self) -> bytes:
-        return ustruct.pack(self.INIT_FORMAT, self.ctrl_byte, self.cid, self.length)
-
-    def pack_to_init_buffer(self, buffer: bytearray, buffer_offset: int = 0) -> None:
-        """
-        Packs header information in the form of **intial** packet
-        into the provided buffer.
-        """
-        ustruct.pack_into(
-            self.INIT_FORMAT,
-            buffer,
-            buffer_offset,
-            self.ctrl_byte,
-            self.cid,
-            self.length,
-        )
-
-    def pack_to_cont_buffer(self, buffer: bytearray, buffer_offset: int = 0) -> None:
-        """
-        Packs header information in the form of **continuation** packet header
-        into the provided buffer.
-        """
-        ustruct.pack_into(
-            self.CONT_FORMAT, buffer, buffer_offset, CONTINUATION_PACKET, self.cid
-        )
-
-    def fragment_payload(
-        self, packet_size: int, *items: AnyBytes
-    ) -> Iterable[AnyBytes]:
-        """Fragment payload into THP transport packets."""
-        packet = bytearray(packet_size)
-        self.pack_to_init_buffer(packet)
-
-        buf = memoryview(packet)[self.INIT_LENGTH :]
-        buf_offset = 0
-        should_zero_pad = False
-
-        for item in items:
-            item_offset = 0
-            while item_offset < len(item):
-                n = utils.memcpy(buf, buf_offset, item, item_offset)
-                buf_offset += n
-                item_offset += n
-
-                if buf_offset == len(buf):
-                    should_zero_pad = True
-                    yield packet  # packet is full - send to the host
-                    self.pack_to_cont_buffer(packet)
-                    buf = memoryview(packet)[self.CONT_LENGTH :]
-                    buf_offset = 0
-
-        if buf_offset > 0:
-            # send last packet (pad with zeroes if needed)
-            if should_zero_pad:
-                utils.memzero(buf[buf_offset:])
-            yield packet
-
-    @classmethod
-    def get_error_header(cls, cid: int, length: int) -> Self:
-        """
-        Returns header for protocol-level error messages.
-        """
-        return cls(_ERROR, cid, length)
-
-    @classmethod
-    def get_channel_allocation_response_header(cls, length: int) -> Self:
-        """
-        Returns header for allocation response handshake message.
-        """
-        return cls(_CHANNEL_ALLOCATION_RES, BROADCAST_CHANNEL_ID, length)
-
-    @classmethod
-    def get_pong_header(cls, length: int) -> Self:
-        """
-        Returns header for pong message.
-        """
-        return cls(_PONG, BROADCAST_CHANNEL_ID, length)
-
-
 _DEFAULT_ENABLED_PAIRING_METHODS = [
     # TODO: Add pairing methods https://github.com/trezor/trezor-firmware/issues/6036
     ThpPairingMethod.CodeEntry
@@ -205,6 +71,8 @@ def _get_device_properties(iface: WireInterface) -> ThpDeviceProperties:
         | (int(utils.unit_btconly() or False) << 8)
         | ((utils.unit_packaging() or 0) << 16)
     )
+    # NOTE: if adding a variable-length field, make sure the encoded message
+    # fits MAX_DEVICE_PROPERTIES_LEN
     return ThpDeviceProperties(
         pairing_methods=get_enabled_pairing_methods(iface),
         internal_model=utils.INTERNAL_MODEL,
@@ -219,22 +87,5 @@ def get_encoded_device_properties(iface: WireInterface) -> AnyBytes:
     length = protobuf.encoded_length(props)
     encoded_properties = bytearray(length)
     protobuf.encode(encoded_properties, props)
+    utils.ensure(length <= MAX_DEVICE_PROPERTIES_LEN)
     return encoded_properties
-
-
-def get_channel_allocation_response(
-    nonce: AnyBytes, new_cid: AnyBytes, iface: WireInterface
-) -> bytes:
-    props_msg = get_encoded_device_properties(iface)
-    return bytes(nonce) + bytes(new_cid) + props_msg
-
-
-if __debug__:
-
-    def state_to_str(state: int) -> str:
-        name = {
-            v: k for k, v in ChannelState.__dict__.items() if not k.startswith("__")
-        }.get(state)
-        if name is not None:
-            return name
-        return "UNKNOWN_STATE"
diff --git a/core/src/trezor/wire/thp/alternating_bit_protocol.py b/core/src/trezor/wire/thp/alternating_bit_protocol.py
deleted file mode 100644
index 98ebe71f..00000000
--- a/core/src/trezor/wire/thp/alternating_bit_protocol.py
+++ /dev/null
@@ -1,109 +0,0 @@
-from storage.cache_thp import ChannelCache
-
-
-def is_ack_valid(cache: ChannelCache, ack_bit: int) -> bool:
-    """
-    Checks if:
-    - an ACK message is expected
-    - the received ACK message acknowledges correct sequence number (bit)
-    """
-    if not _is_ack_expected(cache):
-        return False
-
-    if not _has_ack_correct_sync_bit(cache, ack_bit):
-        return False
-
-    return True
-
-
-def _is_ack_expected(cache: ChannelCache) -> bool:
-    is_expected: bool = not is_sending_allowed(cache)
-    return is_expected
-
-
-def _has_ack_correct_sync_bit(cache: ChannelCache, sync_bit: int) -> bool:
-    is_correct: bool = get_send_seq_bit(cache) == sync_bit
-    return is_correct
-
-
-def has_msg_correct_seq_bit(cache: ChannelCache, sync_bit: int) -> bool:
-    return sync_bit == get_expected_receive_seq_bit(cache)
-
-
-def is_sending_allowed(cache: ChannelCache) -> bool:
-    """
-    Checks whether sending a message in the provided channel is allowed.
-
-    Note: Sending a message in a channel before receipt of ACK message for the previously
-    sent message (in the channel) is prohibited, as it can lead to desynchronization.
-    """
-    return bool(cache.sync & 0x80)
-
-
-def get_send_ack_bit(cache: ChannelCache) -> int:
-    """
-    Returns the sequential number (bit) of the last message successfully received on this channel.
-    """
-    return 1 - get_expected_receive_seq_bit(cache)
-
-
-def get_send_seq_bit(cache: ChannelCache) -> int:
-    """
-    Returns the sequential number (bit) of the next message to be sent
-    in the provided channel.
-    """
-    return (cache.sync & 0x20) >> 5
-
-
-def get_expected_receive_seq_bit(cache: ChannelCache) -> int:
-    """
-    Returns the (expected) sequential number (bit) of the next message
-    to be received in the provided channel.
-    """
-    return (cache.sync & 0x40) >> 6
-
-
-def set_sending_allowed(cache: ChannelCache, sending_allowed: bool) -> None:
-    """
-    Set the flag whether sending a message in this channel is allowed or not.
-    """
-    cache.sync &= 0x7F
-    if sending_allowed:
-        cache.sync |= 0x80
-
-
-def set_expected_receive_seq_bit(cache: ChannelCache, seq_bit: int) -> None:
-    """
-    Set the expected sequential number (bit) of the next message to be received
-    in the provided channel
-    """
-    assert seq_bit in (0, 1)
-
-    # set second bit to "seq_bit" value
-    cache.sync &= 0xBF
-    if seq_bit:
-        cache.sync |= 0x40
-
-
-def _set_send_seq_bit(cache: ChannelCache, seq_bit: int) -> None:
-    assert seq_bit in (0, 1)
-    # set third bit to "seq_bit" value
-    cache.sync &= 0xDF
-    if seq_bit:
-        cache.sync |= 0x20
-
-
-def set_send_seq_bit_to_opposite(cache: ChannelCache) -> None:
-    """
-    Set the sequential bit of the "next message to be send" to the opposite value,
-    i.e. 1 -> 0 and 0 -> 1
-    """
-    _set_send_seq_bit(cache=cache, seq_bit=1 - get_send_seq_bit(cache))
-
-
-def is_ack_piggybacking_allowed(cache: ChannelCache) -> bool:
-    return bool(cache.sync & 0x10)
-
-
-def allow_ack_piggybacking(cache: ChannelCache) -> None:
-    cache.sync |= 0x10
diff --git a/core/src/trezor/wire/thp/channel.py b/core/src/trezor/wire/thp/channel.py
index a1526a80..a627f29d 100644
--- a/core/src/trezor/wire/thp/channel.py
+++ b/core/src/trezor/wire/thp/channel.py
@@ -1,47 +1,22 @@
-import ustruct
-import utime
 from micropython import const
 from typing import TYPE_CHECKING
 
-from storage.cache_common import (
-    CHANNEL_ACK_LATENCY_MS,
-    CHANNEL_HANDSHAKE_HASH,
-    CHANNEL_HOST_STATIC_PUBKEY,
-    CHANNEL_IFACE,
-    CHANNEL_KEY_RECEIVE,
-    CHANNEL_KEY_SEND,
-    CHANNEL_NONCE_RECEIVE,
-    CHANNEL_NONCE_SEND,
-    CHANNEL_STATE,
-)
-from storage.cache_thp import (
-    SESSION_ID_LENGTH,
-    TAG_LENGTH,
-    ChannelCache,
-    clear_sessions_with_channel_id,
-    conditionally_replace_channel,
-    is_there_a_channel_to_replace,
-)
-from trezor import protobuf, utils, workflow
-from trezor.loop import Timeout, race, sleep
-from trezor.wire.context import UnexpectedMessageException
+import trezorthp
+from storage.cache_thp import clear_sessions_with_channel_id, migrate_sessions
+from trezor import loop, protobuf, utils, workflow
 
+from apps.thp.credential_manager import decode_credential, unwrap_credential
+
+from ..errors import DataError
 from ..protocol_common import Message
-from . import ACK_MESSAGE, ENCRYPTED, ChannelState, PacketHeader, ThpDecryptionError
-from . import alternating_bit_protocol as ABP
-from . import control_byte, crypto, memory_manager
-from .checksum import CHECKSUM_LENGTH, is_valid
-from .writer import MESSAGE_TYPE_LENGTH
+from . import ChannelState, memory_manager
 
 if __debug__:
     from trezor import log
-    from trezor.utils import hexlify_if_bytes
-
-    from . import state_to_str
 
 if TYPE_CHECKING:
     from buffer_types import AnyBuffer, AnyBytes
-    from typing import Any, Awaitable, Callable
+    from typing import Any
 
     from trezor.messages import ThpPairingCredential
     from trezor.wire import WireInterface
@@ -52,106 +27,13 @@ if TYPE_CHECKING:
     from .session_context import GenericSessionContext
 
 
-_MAX_RETRANSMISSION_COUNT = const(50)
-_MIN_RETRANSMISSION_COUNT = const(2)
-
-# Stop retransmission if writes are blocked - e.g. due to USB flow control.
-# It allows restarting the event loop to handle other THP channels.
-_WRITE_TIMEOUT_MS = const(5_000)
-_WRITE_TIMEOUT = sleep(_WRITE_TIMEOUT_MS)
-
-# Preempt a stale channel if another channel becomes active and we allowed enough time for the host to respond.
-# It allows interrupting a "stuck" THP workflow using a different channel on the same interface.
-_PREEMPT_TIMEOUT_MS = const(1_000)
-
-EMPTY_ACK_PAYLOAD = memoryview(b"")
-
 _TRACE = const(False)
 
+TREZOR_STATE_UNPAIRED = const(0x00)
+TREZOR_STATE_PAIRED = const(0x01)
+TREZOR_STATE_PAIRED_AUTOCONNECT = const(0x02)
 
-class Reassembler:
-    def __init__(self, read_buf: ThpBuffer) -> None:
-        self.thp_read_buf = read_buf
-        self.reset()
-
-    def reset(self, message: memoryview | None = None) -> None:
-        self.bytes_read: int = 0
-        self.buffer_len: int = 0
-        self.message = message
-
-    def handle_packet(self, packet: memoryview) -> bool:
-        """
-        Process current packet, returning `True` when a valid message is reassembled.
-        The parsed message can retrieved via the `message` field (if it's not `None`).
-        In case of a checksum error or if the reassembly is not over, return `False`.
-        """
-        ctrl_byte = packet[0]
-        if control_byte.is_continuation(ctrl_byte):
-            if not self.bytes_read:
-                # ignore unexpected continuation packets
-                return False
-
-            buffer = self.thp_read_buf.get(self.buffer_len)
-            if buffer is None:
-                # Failed to get the buffer
-                return False
-            self._buffer_packet_data(buffer, packet, PacketHeader.CONT_LENGTH)
-        else:
-            self.reset()
-            _, _, payload_length = ustruct.unpack(PacketHeader.INIT_FORMAT, packet)
-            self.buffer_len = payload_length + PacketHeader.INIT_LENGTH
-
-            buffer = self.thp_read_buf.get(self.buffer_len)
-            if buffer is None:
-                # Failed to get the buffer
-                return False
-            self._buffer_packet_data(buffer, packet, 0)
-
-        assert len(buffer) == self.buffer_len
-        if self.bytes_read < self.buffer_len:
-            return False
-
-        if self.bytes_read > self.buffer_len:
-            if __debug__:
-                log.warning(
-                    __name__,
-                    "Reassembled %d bytes, %d expected",
-                    self.bytes_read,
-                    self.buffer_len,
-                )
-            self.reset()
-            return False
-
-        if not is_checksum_valid(buffer):
-            return False
-
-        assert self.message is None
-        self.message = buffer
-        return True
-
-    def _buffer_packet_data(
-        self, payload_buffer: memoryview, packet: memoryview, offset: int
-    ) -> None:
-        self.bytes_read += utils.memcpy(payload_buffer, self.bytes_read, packet, offset)
-
-
-def is_checksum_valid(buffer: memoryview) -> bool:
-    """
-    Returns `True` if the checksum is valid, otherwise returns `False`.
-    """
-    if is_valid(buffer[-CHECKSUM_LENGTH:], buffer[:-CHECKSUM_LENGTH]):
-        return True
-    # ignore invalid payloads
-    if __debug__:
-        log.warning("Invalid payload checksum: %s", utils.hexlify_if_bytes(buffer))
-    return False
-
-
-class ChannelPreemptedException(UnexpectedMessageException):
-    """Raising this exception should restart the event loop."""
-
-    def __init__(self) -> None:
-        super().__init__(msg=None)
+EMPTY_BUFFER = memoryview(b"")
 
 
 class Channel:
@@ -161,251 +43,161 @@ class Channel:
 
     def __init__(
         self,
-        channel_cache: ChannelCache,
-        ctx: InterfaceContext,
+        channel_id: int,
+        iface_ctx: InterfaceContext,
         buffers: tuple[ThpBuffer, ThpBuffer],
     ) -> None:
-        assert ctx._iface.iface_num() == channel_cache.get_int(CHANNEL_IFACE)
-
         # Channel properties
-        self.channel_id: bytes = channel_cache.channel_id
-        self.iface_ctx: InterfaceContext = ctx
-        self.read_buf, self.write_buf = buffers
-        if __debug__ and _TRACE:
-            self._log("channel initialization")
-        self.channel_cache: ChannelCache = channel_cache
+        self.channel_id = channel_id
+        self.iface_ctx: InterfaceContext = iface_ctx
+        self.receive_buf_src, self.send_buf_src = buffers
+
+        # Used by read loop to wake up context.read()
+        self.incoming_box: loop.mailbox[None | Exception] = loop.mailbox()
+        # Used by read loop to wake up context.write()
+        self.ack_box: loop.mailbox[None | Exception] = loop.mailbox()
+
+        # Conditions used to pause read_loop
+        self.expecting_message = False
+        self.expecting_ack = False
+
+        # Current send buffer, or None if not sending a message
+        self.send_buffer: memoryview | None = None
+        # Current receive buffer, or None if not receiving a message
+        self.receive_buffer: memoryview | None = None
+
+        self._info = trezorthp.channel_info(channel_id)
+        self.state = {
+            TREZOR_STATE_UNPAIRED: ChannelState.TP0,
+            TREZOR_STATE_PAIRED: ChannelState.TC1,
+            TREZOR_STATE_PAIRED_AUTOCONNECT: ChannelState.TC1,
+            None: ChannelState.ENCRYPTED_TRANSPORT,
+        }[self._info.pairing_state]
 
         # Shared variables
         self.sessions: dict[int, GenericSessionContext] = {}
-        self.reassembler = Reassembler(self.read_buf)
-        self.last_write_ms: int = utime.ticks_ms()
 
         # Temporary objects
-        self.credential: ThpPairingCredential | None = None
         self.connection_context: PairingContext | None = None
+        self.credential: ThpPairingCredential | None = None
+        try:
+            if self._info.credential and (
+                inner := unwrap_credential(self._info.credential)
+            ):
+                self.credential = decode_credential(inner)
+        except DataError as exc:
+            if __debug__:
+                log.exception(__name__, exc, iface=self.iface)
+
+    def channel_id_bytes(self) -> bytes:
+        return self.channel_id.to_bytes(2, "big")
 
     @property
     def iface(self) -> WireInterface:
         return self.iface_ctx._iface
 
-    def clear(self) -> None:
-        clear_sessions_with_channel_id(self.channel_id)
-        self.channel_cache.clear()
-
-    # ACCESS TO CHANNEL_DATA
+    def clear(self, exc: Exception | None = None) -> None:
+        """
+        Close a channel, delete associated sessions, optionally kill task.
+        """
+        if __debug__:
+            self._log("closing channel")
+        clear_sessions_with_channel_id(self.channel_id_bytes())
+        trezorthp.channel_close(self.channel_id)
+        if exc is not None:
+            self.kill(exc)
 
-    def get_channel_id_int(self) -> int:
-        return int.from_bytes(self.channel_id, "big")
+    def kill(self, exc: Exception) -> None:
+        """
+        Inject an exception into task waiting on read()/write().
+        """
+        if __debug__:
+            self._log(f"killing task (exception: {exc.__class__.__name__})")
+        self.expecting_message = False
+        self.expecting_ack = False
+        self.incoming_box.put(exc, replace=True)
+        self.ack_box.put(exc, replace=True)
 
-    def get_channel_state(self) -> int:
-        state = self.channel_cache.get_int(
-            CHANNEL_STATE, default=ChannelState.UNALLOCATED
-        )
-        assert isinstance(state, int)
-        if __debug__ and _TRACE:
-            self._log("get_channel_state: ", state_to_str(state))
-        return state
+    # ACCESS TO CHANNEL_DATA
 
     def get_handshake_hash(self) -> bytes:
-        h = self.channel_cache.get(CHANNEL_HANDSHAKE_HASH)
-        assert h is not None
-        return h
+        assert self._info.handshake_hash is not None
+        return self._info.handshake_hash
 
-    def set_channel_state(self, state: ChannelState) -> None:
-        self.channel_cache.set_int(CHANNEL_STATE, state)
-        if __debug__ and _TRACE:
-            self._log("set_channel_state: ", state_to_str(state))
+    def get_host_static_public_key(self) -> bytes:
+        assert self._info.host_static_public_key is not None
+        return self._info.host_static_public_key
 
-    def replace_old_channels_with_the_same_host_public_key(self) -> None:
-        was_any_replaced = conditionally_replace_channel(
-            new_channel=self.channel_cache,
-            required_state=ChannelState.ENCRYPTED_TRANSPORT,
-            required_key=CHANNEL_HOST_STATIC_PUBKEY,
-        )
-        if was_any_replaced:
-            # In case a channel was replaced, close all running workflows
-            workflow.close_others()
-        if __debug__ and _TRACE:
-            self._log("Was any channel replaced? ", str(was_any_replaced))
-
-    def is_channel_to_replace(self) -> bool:
-        return is_there_a_channel_to_replace(
-            new_channel=self.channel_cache,
-            required_state=ChannelState.ENCRYPTED_TRANSPORT,
-            required_key=CHANNEL_HOST_STATIC_PUBKEY,
-        )
-
-    # READ and DECRYPT
-
-    async def recv_payload(
-        self,
-        expected_ctrl_byte: Callable[[int], bool] | None,
-        timeout_ms: int | None = None,
-    ) -> memoryview:
+    def get_last_write(self) -> int | None:
         """
-        Receive and return a valid THP payload from this channel & its control byte.
-        Also handle ACKs while waiting for the payload.
+        Return milliseconds since channel started sending last message.
+        """
+        try:
+            info = trezorthp.channel_info(self.channel_id)
+            return info.last_write
+        except IndexError:
+            return None
 
-        Raise if the received control byte is an unexpected one.
+    def get_channel_state(self) -> int:
+        return self.state
 
-        If `expected_ctrl_byte` is `None`, returns after the first received ACK.
-        """
+    def set_channel_state(self, state: ChannelState) -> None:
+        if __debug__:
+            self._log(f"set state {state}")
+        self.state = state
 
-        return_after_ack = expected_ctrl_byte is None
-        is_ack_piggybacking_allowed = ABP.is_ack_piggybacking_allowed(
-            self.channel_cache
-        )
+    def is_autoconnected(self) -> bool:
+        return self._info.pairing_state == TREZOR_STATE_PAIRED_AUTOCONNECT
 
-        while True:
-            # Handle an existing message (if already reassembled).
-            # Otherwise, receive and reassemble a new one.
-            msg = await self._get_reassembled_message(timeout_ms=timeout_ms)
-
-            # Synchronization process
-            ctrl_byte = msg[0]
-            payload = msg[PacketHeader.INIT_LENGTH : -CHECKSUM_LENGTH]
-            seq_bit = control_byte.get_seq_bit(ctrl_byte)
-
-            # 1: Handle ACKs
-            if control_byte.is_ack(ctrl_byte):
-                handle_ack(self, control_byte.get_ack_bit(ctrl_byte))
-                if return_after_ack:
-                    assert not payload
-                    return EMPTY_ACK_PAYLOAD
-                continue
-
-            if is_ack_piggybacking_allowed:
-                handle_ack(self, control_byte.get_ack_bit(ctrl_byte))
-                if return_after_ack and ABP.is_sending_allowed(self.channel_cache):
-                    # A valid ACK has been received - keep the payload for the next `recv_payload()` call
-                    self.reassembler.reset(msg)
-                    return EMPTY_ACK_PAYLOAD
-
-            if return_after_ack or not expected_ctrl_byte(ctrl_byte):
-                if __debug__:
-                    self._log(
-                        "Unexpected control byte - ignoring ",
-                        utils.hexlify_if_bytes(msg),
-                        logger=log.warning,
-                    )
-                continue
-
-            # 2: Handle message with unexpected sequential bit
-            if seq_bit != ABP.get_expected_receive_seq_bit(self.channel_cache):
-                if __debug__:
-                    self._log(
-                        "Received message with an unexpected sequential bit",
-                        logger=log.warning,
-                    )
-                await send_ack(self, ack_bit=seq_bit)
-                continue
-
-            # 3: Send ACK in response
-            await send_ack(self, ack_bit=seq_bit)
-
-            ABP.set_expected_receive_seq_bit(self.channel_cache, 1 - seq_bit)
-
-            return payload
-
-    async def _get_reassembled_message(
-        self, timeout_ms: int | None = None
-    ) -> memoryview:
-        """Doesn't block if a message has been already reassembled."""
-        thp_ctx = self.iface_ctx.thp_ctx
-        while self.reassembler.message is None:
-            # receive and reassemble a new message from any THP channel
-            try:
-                channel = await thp_ctx.get_next_message(timeout_ms=timeout_ms)
-                if channel is None:
-                    continue
-            except ChannelPreemptedException:
-                elapsed_ms = utime.ticks_diff(utime.ticks_ms(), self.last_write_ms)
-                # allow preempting channel only after enough time has passed
-                is_stale = elapsed_ms > _PREEMPT_TIMEOUT_MS
-                if __debug__:
-                    self._log(
-                        f"Interrupted channel after {elapsed_ms} ms",
-                        logger=(log.error if is_stale else log.warning),
-                    )
-                if is_stale:
-                    raise
-                continue
-
-            if channel is self:
-                break
-
-            # currently only single-channel sessions are supported during a single event loop run
+    def end_pairing_and_replace(self) -> None:
+        replaced_channel_id = trezorthp.channel_paired(self.channel_id)
+        if replaced_channel_id is not None:
+            migrate_sessions(
+                replaced_channel_id.to_bytes(2, "big"), self.channel_id_bytes()
+            )
+            # In case a channel was replaced, close all running workflows
+            workflow.close_others()
+        self.credential = None
+        if __debug__ and _TRACE:
             self._log(
-                "Ignoring unexpected channel: ",
-                utils.hexlify_if_bytes(channel.channel_id),
-                logger=log.warning,
+                "Was any channel replaced? ", str(replaced_channel_id is not None)
             )
 
-        msg = self.reassembler.message
-        self.reassembler.reset()  # next call will reassemble a new message
-        assert msg is not None
-        return msg
-
-    def reassemble(self, packet: AnyBuffer) -> bool:
+    async def read(self) -> tuple[int, Message]:
         """
-        Process current packet, returning `True` when a valid message is reassembled.
-        The parsed message can retrieved via the `message` field (if it's not `None`).
-        In case of a checksum error or if the reassembly is not over, return `False`.
+        Wait for reassembled message, decrypt it, and return a `(session_id, message)` tuple.
         """
-        if self.get_channel_state() == ChannelState.UNALLOCATED:
-            return False
-        return self.reassembler.handle_packet(memoryview(packet))
-
-    async def decrypt_message(self) -> tuple[int, Message]:
-        """
-        Receive, decrypt and return a `(session_id, message)` tuple.
-        Also handle ACKs while waiting for the message.
-        """
-        payload = await self.recv_payload(control_byte.is_encrypted_transport)
-        self._decrypt_buffer(payload)
-        session_id, message_type = ustruct.unpack(">BH", payload)
+        self.expecting_message = True
+        self.iface_ctx.request_read()
+        await self.incoming_box
+        assert self.receive_buffer is not None
+        try:
+            session_id, message_type, message_bytes = trezorthp.message_out(
+                self.channel_id, self.receive_buffer
+            )
+        except Exception:
+            self.expecting_message = False
+            raise
+        finally:
+            # wake up write loop to send ACKs or DECRYPTION_FAILED
+            self.iface_ctx.request_write()
+        if __debug__ and _TRACE:
+            self._log("message is ready")
         message = Message(
             message_type,
-            payload[SESSION_ID_LENGTH + MESSAGE_TYPE_LENGTH : -TAG_LENGTH],
+            message_bytes,
         )
+        self.receive_buffer = None
         return (session_id, message)
 
-    def _decrypt_buffer(self, payload: memoryview) -> None:
-        noise_buffer = payload[:-TAG_LENGTH]
-        tag = payload[-TAG_LENGTH:]
-
-        key_receive = self.channel_cache.get(CHANNEL_KEY_RECEIVE)
-        nonce_receive = self.channel_cache.get_int(CHANNEL_NONCE_RECEIVE)
-
-        assert key_receive is not None
-        assert nonce_receive is not None
-
-        if __debug__ and _TRACE:
-            self._log("Buffer before decryption: ", hexlify_if_bytes(noise_buffer))
-
-        is_tag_valid = crypto.dec(noise_buffer, tag, key_receive, nonce_receive)
-        if __debug__ and _TRACE:
-            self._log("Buffer after decryption: ", hexlify_if_bytes(noise_buffer))
-
-        self.channel_cache.set_int(CHANNEL_NONCE_RECEIVE, nonce_receive + 1)
-
-        if __debug__ and _TRACE:
-            self._log("Is decrypted tag valid? ", str(is_tag_valid))
-            self._log("Received tag: ", hexlify_if_bytes(tag))
-            self._log("New nonce_receive: ", str((nonce_receive + 1)))
-
-        if not is_tag_valid:
-            raise ThpDecryptionError()
-
-    # WRITE and ENCRYPT
-
     async def write(
         self,
         msg: protobuf.MessageType,
         session_id: int = 0,
     ) -> None:
-        assert ABP.is_sending_allowed(self.channel_cache)
-
+        """
+        Encrypt a message, wait until it is send, wait until ACK is received.
+        """
         if __debug__:
             self._log(
                 f"write message: {msg.MESSAGE_NAME}",
@@ -419,155 +211,70 @@ class Channel:
                     iface=self.iface,
                 )
 
-        msg_size = protobuf.encoded_length(msg)
-        payload_size = SESSION_ID_LENGTH + MESSAGE_TYPE_LENGTH + msg_size
-        length = payload_size + CHECKSUM_LENGTH + TAG_LENGTH + PacketHeader.INIT_LENGTH
-
-        buffer = self.write_buf.get(length)
-        if buffer is None:
-            from trezor import wire
-
-            raise wire.FirmwareError("Failed to get a sufficiently large write buffer.")
-
-        noise_payload_len = memory_manager.encode_into_buffer(buffer, msg, session_id)
-
-        self._encrypt(buffer, noise_payload_len)
-        payload_length = noise_payload_len + TAG_LENGTH
-
-        return await self.write_encrypted_payload(ENCRYPTED, buffer[:payload_length])
-
-    async def write_encrypted_payload(self, ctrl_byte: int, payload: AnyBytes) -> None:
-        assert ABP.is_sending_allowed(self.channel_cache)
-
-        # Construct THP header
-        payload_len = len(payload) + CHECKSUM_LENGTH
-        sync_bit = ABP.get_send_seq_bit(self.channel_cache)
-        ctrl_byte = control_byte.add_seq_bit_to_ctrl_byte(ctrl_byte, sync_bit)
-
-        if ABP.is_ack_piggybacking_allowed(self.channel_cache):
-            ack_bit = ABP.get_send_ack_bit(self.channel_cache)
-            ctrl_byte = control_byte.add_ack_bit_to_ctrl_byte(ctrl_byte, ack_bit)
-
-        header = PacketHeader(ctrl_byte, self.get_channel_id_int(), payload_len)
-
-        ack_latency_ms = self.channel_cache.get_int(CHANNEL_ACK_LATENCY_MS) or 0
-
-        # ACK is needed before sending more data
-        ABP.set_sending_allowed(self.channel_cache, False)
-
-        # allows preempting this channel, if another channel becomes active
-        self.last_write_ms = utime.ticks_ms()
-
-        async def _write_loop() -> None:
-            """
-            Retransmit the payload (with increasing delay), raising `Timeout` in the end.
-
-            This task is spawned concurrently with `_wait_for_ack()` using `loop.race()`,
-            so it will be cancelled when the expected ACK is received.
-            """
-            if __debug__ and _TRACE:
-                self._log(f"Sending {len(payload)} bytes, latency: {ack_latency_ms} ms")
-
-            for i in range(_MAX_RETRANSMISSION_COUNT):
-                # Try to send the payload (split into packets), or raise if transport is blocked
-                await self._write_payload_once(header, payload)
-                # Channel's estimated latency + a variable delay (from 200ms till ~3.52s)
-                delay_ms = ack_latency_ms + round(10300 - 1010000 / (100 + i))
-                await sleep(delay_ms)
-                if __debug__:
-                    log.warning(__name__, "Retransmit after %d ms", delay_ms)
-
-            # restart event loop due to unresponsive channel
-            raise Timeout("THP retransmission timeout")
-
-        async def _wait_for_ack() -> None:
-            """
-            Wait for the expected ACK to be received.
-
-            This task is spawned concurrently with `_write_loop()` using `loop.race()`,
-            so it will be cancelled when retransmission loop is over.
-            """
-            while not ABP.is_sending_allowed(self.channel_cache):
-                # `ABP.set_sending_allowed()` will be called after a valid ACK
-                await self.recv_payload(expected_ctrl_byte=None)
+        self.expecting_message = False
 
         try:
-            # wait and return after receiving an ACK, or raise in case of an unexpected message / retransmission timeout.
-            await race(_wait_for_ack(), _write_loop())
+            buffer_size = memory_manager.buffer_size(msg)
+            self.send_buffer = self.send_buf_src.get(buffer_size)
+            noise_payload_len = memory_manager.encode_into_buffer(
+                self.send_buffer, msg, session_id
+            )
+            trezorthp.message_in(self.channel_id, noise_payload_len, self.send_buffer)
+            self.iface_ctx.request_write()
+            # Might raise Timeout or ChannelPreemptedException.
+            await self.ack_box
         finally:
-            ack_latency_ms = utime.ticks_diff(utime.ticks_ms(), self.last_write_ms)
-            # Limit estimated latency to avoid integer overflows and too long delays
-            ack_latency_ms = max(0, min(800, ack_latency_ms))
-            self.channel_cache.set_int(CHANNEL_ACK_LATENCY_MS, ack_latency_ms)
-
-            # Make sure to use the next `seq_bit` for the next payload
-            ABP.set_send_seq_bit_to_opposite(self.channel_cache)
+            self.send_buffer = None
 
-    async def _write_payload_once(
-        self, header: PacketHeader, payload: AnyBytes
-    ) -> None:
-        """Write the payload and raise if the interface is blocked."""
-        result = await race(
-            self.iface_ctx.write_payload(header, payload), _WRITE_TIMEOUT
+    def read_packet(self, packet_buffer: AnyBytes, buffer_hint: int) -> None:
+        """
+        Called by read_loop() to process incoming packet.
+        """
+        if self.receive_buffer is None or buffer_hint > len(self.receive_buffer):
+            self.receive_buffer = self.receive_buf_src.get(buffer_hint)
+        result = trezorthp.packet_in_channel(
+            self.channel_id, packet_buffer, self.receive_buffer
         )
-        if isinstance(result, int):
-            # Can happen when the USB peer is not reading.
-            raise Timeout("THP write is blocked")
-
-    def _encrypt(self, buffer: AnyBuffer, noise_payload_len: int) -> None:
-        assert len(buffer) >= noise_payload_len + TAG_LENGTH + CHECKSUM_LENGTH
-
-        noise_buffer = memoryview(buffer)[0:noise_payload_len]
-
-        key_send = self.channel_cache.get(CHANNEL_KEY_SEND)
-        nonce_send = self.channel_cache.get_int(CHANNEL_NONCE_SEND)
-
-        assert key_send is not None
-        assert nonce_send is not None
-
-        tag = crypto.enc(noise_buffer, key_send, nonce_send)
-
-        self.channel_cache.set_int(CHANNEL_NONCE_SEND, nonce_send + 1)
-        if __debug__ and _TRACE:
-            self._log("New nonce_send: ", str((nonce_send + 1)))
-
-        buffer[noise_payload_len : noise_payload_len + TAG_LENGTH] = tag
+        if __debug__ and _TRACE and result is not None:
+            self._log(f"packet_in: {result}")
+        if result is trezorthp.ACK or result is trezorthp.MESSAGE_READY_ACK:
+            assert self.expecting_ack
+            self.ack_box.put(None, replace=True)
+            self.expecting_ack = False
+            self.iface_ctx.recompute_timeouts()
+        if result is trezorthp.MESSAGE_READY or result is trezorthp.MESSAGE_READY_ACK:
+            self.incoming_box.put(None, replace=True)
+            self.expecting_message = False
+        elif result == trezorthp.FAILED:
+            # channel is closed now
+            self.kill(trezorthp.ThpError("Channel failed"))
+
+    def write_packet(self, packet: AnyBuffer) -> bool:
+        """
+        Called by write_loop() to send outgoing packets.
+        """
+        try:
+            # If not sending application message, provide empty buffer for ACK.
+            buffer = self.send_buffer or EMPTY_BUFFER
+            res = trezorthp.packet_out_channel(self.channel_id, buffer, packet)
+            if self.send_buffer:
+                self.expecting_ack = True
+                self.iface_ctx.request_read()
+            return res
+        except Exception as e:
+            if __debug__:
+                log.exception(__name__, e, iface=self.iface)
+            self.kill(e)
+            return False
 
     if __debug__:
 
         def _log(self, text_1: str, text_2: str = "", logger: Any = log.debug) -> None:
             logger(
                 __name__,
-                "(cid: %s) %s%s",
-                hexlify_if_bytes(self.channel_id),
+                "(cid: %04x) %s%s",
+                self.channel_id,
                 text_1,
                 text_2,
                 iface=self.iface,
             )
-
-
-def send_ack(channel: Channel, ack_bit: int) -> Awaitable[None]:
-    ctrl_byte = control_byte.add_ack_bit_to_ctrl_byte(ACK_MESSAGE, ack_bit)
-    header = PacketHeader(ctrl_byte, channel.get_channel_id_int(), CHECKSUM_LENGTH)
-    if __debug__ and _TRACE:
-        log.debug(
-            __name__,
-            "Writing ACK message to a channel with cid: %s, ack_bit: %d",
-            hexlify_if_bytes(channel.channel_id),
-            ack_bit,
-            iface=channel.iface,
-        )
-    return channel.iface_ctx.write_payload(header, b"")
-
-
-def handle_ack(ctx: Channel, ack_bit: int) -> None:
-    if not ABP.is_ack_valid(ctx.channel_cache, ack_bit):
-        return
-    # ACK is expected and it has correct sync bit
-    if __debug__ and _TRACE:
-        log.debug(
-            __name__,
-            "Received ACK message with correct ack bit",
-            iface=ctx.iface,
-        )
-    ABP.set_sending_allowed(ctx.channel_cache, True)
diff --git a/core/src/trezor/wire/thp/channel_manager.py b/core/src/trezor/wire/thp/channel_manager.py
deleted file mode 100644
index 56eee54b..00000000
--- a/core/src/trezor/wire/thp/channel_manager.py
+++ /dev/null
@@ -1,21 +0,0 @@
-from typing import TYPE_CHECKING
-
-from storage import cache_thp
-from storage.cache_common import CHANNEL_IFACE, CHANNEL_STATE
-
-from . import ChannelState
-
-if TYPE_CHECKING:
-    from trezorio import WireInterface
-
-    from storage.cache_thp import ChannelCache
-
-
-def create_new_channel(iface: WireInterface) -> ChannelCache:
-    """
-    Creates a new channel for the interface `iface`.
-    """
-    channel_cache: ChannelCache = cache_thp.get_new_channel()
-    channel_cache.set_int(CHANNEL_IFACE, iface.iface_num())
-    channel_cache.set_int(CHANNEL_STATE, ChannelState.TH1)
-    return channel_cache
diff --git a/core/src/trezor/wire/thp/checksum.py b/core/src/trezor/wire/thp/checksum.py
deleted file mode 100644
index 846827d6..00000000
--- a/core/src/trezor/wire/thp/checksum.py
+++ /dev/null
@@ -1,36 +0,0 @@
-from micropython import const
-from typing import TYPE_CHECKING
-
-from trezor.crypto import crc
-
-if TYPE_CHECKING:
-    from buffer_types import AnyBytes
-
-CHECKSUM_LENGTH = const(4)
-
-
-def compute(data: AnyBytes, crc_chain: int = 0) -> bytes:
-    """
-    Returns a CRC-32 checksum of the provided `data`. Allows for for chaining
-    computations over multiple data segments using `crc_chain` (optional).
-    """
-    return crc.crc32(data, crc_chain).to_bytes(CHECKSUM_LENGTH, "big")
-
-
-def compute_int(data: AnyBytes, crc_chain: int = 0) -> int:
-    """
-    Returns a CRC-32 checksum of the provided `data`. Allows for for chaining
-    computations over multiple data segments using `crc_chain` (optional).
-
-    Returns checksum in the form of `int`.
-    """
-    return crc.crc32(data, crc_chain)
-
-
-def is_valid(checksum: AnyBytes, data: AnyBytes) -> bool:
-    """
-    Checks whether the CRC-32 checksum of the `data` is the same
-    as the checksum provided in `checksum`.
-    """
-    data_checksum = compute(data)
-    return checksum == data_checksum
diff --git a/core/src/trezor/wire/thp/control_byte.py b/core/src/trezor/wire/thp/control_byte.py
deleted file mode 100644
index ddd0a8f9..00000000
--- a/core/src/trezor/wire/thp/control_byte.py
+++ /dev/null
@@ -1,57 +0,0 @@
-from micropython import const
-
-from . import (
-    ACK_MESSAGE,
-    CONTINUATION_PACKET,
-    ENCRYPTED,
-    HANDSHAKE_COMP_REQ,
-    HANDSHAKE_INIT_REQ,
-)
-
-_CONTINUATION_PACKET_MASK = const(0x80)
-_ACK_MASK = const(0xF7)
-_DATA_MASK = const(0xE7)
-
-
-def add_seq_bit_to_ctrl_byte(ctrl_byte: int, seq_bit: int) -> int:
-    assert seq_bit in (0, 1)
-    if seq_bit:
-        return ctrl_byte | 0x10
-    else:
-        return ctrl_byte & 0xEF
-
-
-def add_ack_bit_to_ctrl_byte(ctrl_byte: int, ack_bit: int) -> int:
-    assert ack_bit in (0, 1)
-    if ack_bit:
-        return ctrl_byte | 0x08
-    else:
-        return ctrl_byte & 0xF7
-
-
-def get_ack_bit(ctrl_byte: int) -> int:
-    return (ctrl_byte & 0x08) >> 3
-
-
-def get_seq_bit(ctrl_byte: int) -> int:
-    return (ctrl_byte & 0x10) >> 4
-
-
-def is_ack(ctrl_byte: int) -> bool:
-    return ctrl_byte & _ACK_MASK == ACK_MESSAGE
-
-
-def is_continuation(ctrl_byte: int) -> bool:
-    return ctrl_byte & _CONTINUATION_PACKET_MASK == CONTINUATION_PACKET
-
-
-def is_encrypted_transport(ctrl_byte: int) -> bool:
-    return ctrl_byte & _DATA_MASK == ENCRYPTED
-
-
-def is_handshake_init_req(ctrl_byte: int) -> bool:
-    return ctrl_byte & _DATA_MASK == HANDSHAKE_INIT_REQ
-
-
-def is_handshake_comp_req(ctrl_byte: int) -> bool:
-    return ctrl_byte & _DATA_MASK == HANDSHAKE_COMP_REQ
diff --git a/core/src/trezor/wire/thp/crypto.py b/core/src/trezor/wire/thp/crypto.py
index 72572e59..35ad0eca 100644
--- a/core/src/trezor/wire/thp/crypto.py
+++ b/core/src/trezor/wire/thp/crypto.py
@@ -1,210 +1,11 @@
-import ustruct
 from micropython import const
-from trezorcrypto import (
-    AuthenticationError,
-    aesgcm_decrypt,
-    aesgcm_encrypt,
-    bip32,
-    curve25519,
-    hmac,
-)
-from typing import TYPE_CHECKING
+from trezorcrypto import bip32
 
 from storage import device
-from trezor import log, utils
-from trezor.crypto.hashlib import sha256
-from trezor.wire.thp import ThpDecryptionError
-
-if TYPE_CHECKING:
-    from buffer_types import AnyBuffer, AnyBytes
 
 # The HARDENED flag is taken from apps.common.paths
 # It is not imported to save on resources
 HARDENED = const(0x8000_0000)
-PUBKEY_LENGTH = const(32)
-
-_TRACE = const(False)
-
-if __debug__:
-    from trezor.utils import hexlify_if_bytes
-
-
-def enc(buffer: AnyBuffer, key: bytes, nonce: int, auth_data: bytes = b"") -> bytes:
-    """
-    Encrypts the provided `buffer` with AES-GCM (in place).
-    Returns a 16-byte long encryption tag.
-    """
-    if __debug__ and _TRACE:
-        log.debug(__name__, "enc (key: %s, nonce: %d)", hexlify_if_bytes(key), nonce)
-    iv = _get_iv_from_nonce(nonce)
-    aes_ctx = aesgcm_encrypt(key, iv)
-    aes_ctx.auth(auth_data)
-    aes_ctx.encrypt_in_place(buffer)
-    return aes_ctx.finish()
-
-
-def dec(
-    buffer: AnyBuffer,
-    tag: AnyBytes,
-    key: AnyBytes,
-    nonce: int,
-    auth_data: AnyBytes = b"",
-) -> bool:
-    """
-    Decrypts the provided buffer (in place). Returns `True` if the provided authentication `tag` is the same as
-    the tag computed in decryption, otherwise it returns `False`.
-    """
-    iv = _get_iv_from_nonce(nonce)
-    if __debug__ and _TRACE:
-        log.debug(__name__, "dec (key: %s, nonce: %d)", hexlify_if_bytes(key), nonce)
-    aes_ctx = aesgcm_decrypt(key, iv)
-    aes_ctx.auth(auth_data)
-    aes_ctx.decrypt_in_place(buffer)
-    try:
-        aes_ctx.finish(tag)
-    except AuthenticationError:
-        return False
-    return True
-
-
-PROTOCOL_NAME = b"Noise_XX_25519_AESGCM_SHA256\x00\x00\x00\x00"
-IV_1 = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
-IV_2 = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01"
-
-
-class Handshake:
-    """
-    `Handshake` holds (temporary) values and keys that are used during the creation of an encrypted channel.
-    The following values should be saved for future use before disposing of this object:
-    - `h` - handshake hash, can be used to bind other values to the channel
-    - `key_receive` - key for decrypting incoming communication
-    - `key_send` - key for encrypting outgoing communication
-    """
-
-    def __init__(self) -> None:
-        self.trezor_ephemeral_private_key: bytes
-        self.ck: bytes
-        self.k: bytes
-        self.h: bytes
-        self.key_receive: bytes
-        self.key_send: bytes
-
-    def handle_th1_crypto(
-        self,
-        device_properties: AnyBytes,
-        host_ephemeral_public_key: AnyBytes,
-        payload: AnyBytes,
-    ) -> tuple[bytes, bytes, bytes]:
-
-        trezor_static_private_key, trezor_static_public_key = _derive_static_key_pair()
-        self.trezor_ephemeral_private_key = curve25519.generate_secret()
-        trezor_ephemeral_public_key = curve25519.publickey(
-            self.trezor_ephemeral_private_key
-        )
-        self.h = _hash_of_two(PROTOCOL_NAME, device_properties)
-        self.h = _hash_of_two(self.h, host_ephemeral_public_key)
-        self.h = _hash_of_two(self.h, payload)
-        self.h = _hash_of_two(self.h, trezor_ephemeral_public_key)
-        point = curve25519.multiply(
-            self.trezor_ephemeral_private_key, host_ephemeral_public_key
-        )
-        self.ck, self.k = _hkdf(PROTOCOL_NAME, point)
-        mask = _hash_of_two(trezor_static_public_key, trezor_ephemeral_public_key)
-        trezor_masked_static_public_key = curve25519.multiply(
-            mask, trezor_static_public_key
-        )
-        aes_ctx = aesgcm_encrypt(self.k, IV_1)
-        encrypted_trezor_static_public_key = aes_ctx.encrypt(
-            trezor_masked_static_public_key
-        )
-        if __debug__:
-            log.debug(
-                __name__,
-                "th1 - enc (key: %s, nonce: %d, handshake_hash %s)",
-                hexlify_if_bytes(self.k),
-                0,
-                hexlify_if_bytes(self.h),
-            )
-
-        aes_ctx.auth(self.h)
-        tag_to_encrypted_key = aes_ctx.finish()
-        encrypted_trezor_static_public_key = (
-            encrypted_trezor_static_public_key + tag_to_encrypted_key
-        )
-        self.h = _hash_of_two(self.h, encrypted_trezor_static_public_key)
-        point = curve25519.multiply(
-            trezor_static_private_key, host_ephemeral_public_key
-        )
-        self.ck, self.k = _hkdf(self.ck, curve25519.multiply(mask, point))
-        aes_ctx = aesgcm_encrypt(self.k, IV_1)
-        aes_ctx.auth(self.h)
-        tag = aes_ctx.finish()
-        self.h = _hash_of_two(self.h, tag)
-        return (trezor_ephemeral_public_key, encrypted_trezor_static_public_key, tag)
-
-    def handle_th2_crypto(
-        self,
-        encrypted_host_static_public_key: AnyBuffer,
-        encrypted_payload: AnyBuffer,
-    ) -> None:
-
-        aes_ctx = aesgcm_decrypt(self.k, IV_2)
-
-        # The new value of hash `h` MUST be computed before the `encrypted_host_static_public_key` is decrypted.
-        # However, decryption of `encrypted_host_static_public_key` MUST use the previous value of `h` for
-        # authentication of the gcm tag.
-        aes_ctx.auth(self.h)  # Authenticate with the previous value of `h`
-        self.h = _hash_of_two(
-            self.h, encrypted_host_static_public_key
-        )  # Compute new value
-        aes_ctx.decrypt_in_place(
-            memoryview(encrypted_host_static_public_key)[:PUBKEY_LENGTH]
-        )
-        if __debug__:
-            log.debug(
-                __name__, "th2 - dec (key: %s, nonce: %d)", hexlify_if_bytes(self.k), 1
-            )
-        host_static_public_key = memoryview(encrypted_host_static_public_key)[
-            :PUBKEY_LENGTH
-        ]
-        try:
-            aes_ctx.finish(encrypted_host_static_public_key[-16:])
-        except AuthenticationError:
-            raise ThpDecryptionError()
-
-        self.ck, self.k = _hkdf(
-            self.ck,
-            curve25519.multiply(
-                self.trezor_ephemeral_private_key, host_static_public_key
-            ),
-        )
-        aes_ctx = aesgcm_decrypt(self.k, IV_1)
-        aes_ctx.auth(self.h)
-        self.h = _hash_of_two(self.h, memoryview(encrypted_payload))
-        aes_ctx.decrypt_in_place(memoryview(encrypted_payload)[:-16])
-        if __debug__:
-            log.debug(
-                __name__, "th2 - dec (key: %s, nonce: %d)", hexlify_if_bytes(self.k), 0
-            )
-        try:
-            aes_ctx.finish(encrypted_payload[-16:])
-        except AuthenticationError:
-            raise ThpDecryptionError()
-
-        self.key_receive, self.key_send = _hkdf(self.ck, b"")
-        if __debug__:
-            log.debug(
-                __name__,
-                "(key_receive: %s, key_send: %s)",
-                hexlify_if_bytes(self.key_receive),
-                hexlify_if_bytes(self.key_send),
-            )
-
-    def get_handshake_completion_response(self, trezor_state: bytes) -> bytes:
-        aes_ctx = aesgcm_encrypt(self.key_send, IV_1)
-        encrypted_trezor_state = aes_ctx.encrypt(trezor_state)
-        tag = aes_ctx.finish()
-        return encrypted_trezor_state + tag
 
 
 def _derive_static_key_pair() -> tuple[bytes, bytes]:
@@ -220,26 +21,11 @@ def _derive_static_key_pair() -> tuple[bytes, bytes]:
     return trezor_static_private_key, trezor_static_public_key
 
 
+def get_trezor_static_private_key() -> bytes:
+    private_key, _ = _derive_static_key_pair()
+    return private_key
+
+
 def get_trezor_static_public_key() -> bytes:
     _, public_key = _derive_static_key_pair()
     return public_key
-
-
-def _hkdf(chaining_key: bytes, input: bytes) -> tuple[bytes, bytes]:
-    temp_key = hmac(hmac.SHA256, chaining_key, input).digest()
-    output_1 = hmac(hmac.SHA256, temp_key, b"\x01").digest()
-    ctx_output_2 = hmac(hmac.SHA256, temp_key, output_1)
-    ctx_output_2.update(b"\x02")
-    output_2 = ctx_output_2.digest()
-    return (output_1, output_2)
-
-
-def _hash_of_two(part_1: AnyBytes, part_2: AnyBytes) -> bytes:
-    ctx = sha256(part_1)
-    ctx.update(part_2)
-    return ctx.digest()
-
-
-def _get_iv_from_nonce(nonce: int) -> bytes:
-    utils.ensure(nonce <= 0xFFFFFFFFFFFFFFFF, "Nonce overflow, terminate the channel")
-    return ustruct.pack(">4sQ", b"\x00\x00\x00\x00", nonce)
diff --git a/core/src/trezor/wire/thp/interface_context.py b/core/src/trezor/wire/thp/interface_context.py
index 4ce076ab..ea2b54ab 100644
--- a/core/src/trezor/wire/thp/interface_context.py
+++ b/core/src/trezor/wire/thp/interface_context.py
@@ -1,50 +1,43 @@
-import ustruct
 from micropython import const
 from typing import TYPE_CHECKING
 
-from storage.cache_thp import (
-    BROADCAST_CHANNEL_ID,
-    find_allocated_channel,
-    update_channel_last_used,
-)
-from trezor import io, utils
-from trezor.loop import Timeout, race, sleep, wait
-
-from . import (
-    CHANNEL_ALLOCATION_REQ,
-    CODEC_V1,
-    PING,
-    PacketHeader,
-    ThpErrorType,
-    channel_manager,
-    checksum,
-    control_byte,
-    get_channel_allocation_response,
-)
-from .channel import Channel, ChannelPreemptedException
-from .checksum import CHECKSUM_LENGTH
+import trezorthp
+from storage.cache_thp import clear_sessions_without_channel
+from trezor import config, io, loop, utils
+from trezor.loop import race, wait
+
+from ..protocol_common import ChannelPreemptedException
+from . import get_encoded_device_properties
+from .channel import TREZOR_STATE_PAIRED, TREZOR_STATE_UNPAIRED, Channel
+from .crypto import get_trezor_static_private_key
 
 if __debug__:
     from trezor import log
 
-
 if utils.USE_BLE:
     import trezorble as ble
     from trezor.workflow import idle_timer
 
 if TYPE_CHECKING:
-    from buffer_types import AnyBuffer, AnyBytes
+    from buffer_types import AnyBytes
     from trezorio import WireInterface
-    from typing import Awaitable, Generator, Iterable, NoReturn
+    from typing import Any, Generator
+
 
-_BROADCAST_PAYLOAD_LENGTH = const(12)
 _TRACE = const(False)
 
+# Preempt a stale channel if another channel becomes active and we allowed enough time for the host to respond.
+# It allows interrupting a "stuck" THP workflow using a different channel on the same interface.
+_PREEMPT_TIMEOUT_MS = const(1_000)
+
+# Stop retransmission if writes are blocked - e.g. due to USB flow control.
+# It allows restarting the event loop to handle other THP channels.
+_WRITE_TIMEOUT_MS = const(5_000)
+_WRITE_TIMEOUT = loop.sleep(_WRITE_TIMEOUT_MS)
+
+_KEY_REQUIRED_VALS = (trezorthp.KEY_REQUIRED, trezorthp.KEY_REQUIRED_UNLOCK)
 
-# Uses `yield` instead of `await` to avoid allocations.
-def _timeout_after(ms: int) -> Generator[sleep, int, NoReturn]:
-    yield sleep(ms)
-    raise Timeout
+EMPTY_BUFFER = bytearray()
 
 
 class ThpContext:
@@ -54,186 +47,413 @@ class ThpContext:
     """
 
     def __init__(self, *ifaces: WireInterface) -> None:
-        max_packet_len = max(iface.RX_PACKET_LEN for iface in ifaces)
-        self._packet_buf = bytearray(max_packet_len)
-        self._packet_view = memoryview(self._packet_buf)
         self._iface_ctxs = [InterfaceContext(iface, self) for iface in ifaces]
+        self.channel_ready_box: loop.mailbox[None] = loop.mailbox()
+        self.active_channel: Channel | None = None
 
-    async def get_next_message(self, timeout_ms: int | None = None) -> Channel | None:
+    # Blocks until a channel in pairing/credential/transport phase starts receiving data.
+    async def get_active_channel(self) -> Channel:
         """
         Reassemble a valid THP payload from any THP interface, and return its channel.
 
         Also handle THP channel allocation.
         """
-        # wait until one of the channels becomes readable
-        children = (iface_ctx._wait_for_packet() for iface_ctx in self._iface_ctxs)
-        if timeout_ms is None:
-            race_task = race(*children)
-        else:
-            race_task = race(*children, _timeout_after(timeout_ms))
+        await self.channel_ready_box
+        assert self.active_channel is not None
+        return self.active_channel
 
-        iface_ctx, packet_len = await race_task  # will raise on timeout
-        assert packet_len == iface_ctx._iface.RX_PACKET_LEN
+    def preempt_active_channel_if_stale(self) -> None:
+        if not self.active_channel:
+            return
+        last_write_ms = self.active_channel.get_last_write()
+        if last_write_ms is None or last_write_ms > _PREEMPT_TIMEOUT_MS:
+            if __debug__:
+                log.error(
+                    __name__,
+                    f"Interrupted channel {hex(self.active_channel.channel_id)} after {last_write_ms} ms",
+                )
+            self.active_channel.kill(ChannelPreemptedException())
 
-        # read and handle the packet using its `InterfaceContext`
-        iface_ctx._iface.read(self._packet_buf, 0)
-        return await iface_ctx.handle_packet(self._packet_view[:packet_len])
+    async def close(self) -> None:
+        for iface_ctx in self._iface_ctxs:
+            try:
+                await iface_ctx.close()
+            except Exception as exc:
+                if __debug__:
+                    log.exception(__name__, exc)
 
 
 class InterfaceContext:
     """
-    This class handles multi-packet THP payloads from a single interface.
-    It also handles and responds to low-level single packet THP messages, creating new channels if needed.
+    This class shuffles packets between an interface and non-blocking rust/trezor-thp code.
     """
 
     def __init__(self, iface: WireInterface, thp_ctx: ThpContext) -> None:
         self._iface = iface
         self._read = wait(iface.iface_num() | io.POLL_READ)
         self._write = wait(iface.iface_num() | io.POLL_WRITE)
-        self._channels: dict[int, Channel] = {}
+        # Currently only one active channel is allowed in a session. Without session restart
+        # this might become a dict[int, Channel].
+        self.active_channel: Channel | None = None
         self.thp_ctx = thp_ctx
 
-    def _wait_for_packet(self) -> Generator[wait, int, tuple["InterfaceContext", int]]:
-        """Block until this interface is readable.
+        self._read_loop: loop.spawn = loop.spawn(self.read_loop())
+        self._write_loop: loop.spawn = loop.spawn(self.write_loop())
+        self._retrans_loop: loop.spawn = loop.spawn(self.retransmission_loop())
+        self._handshake_key_task: loop.spawn | None = None
 
-        It adapts `loop.wait`, to be used in a `race()` over multiple THP interfaces by `ThpContext.get_next_message()`.
-        """
-        # Uses `yield` instead of `await` to avoid allocations.
-        packet_len = yield self._read
-        if utils.USE_BLE and self._iface is ble.interface:
-            # prevent auto-lock while handling longer workflows on Bluetooth
-            idle_timer.touch()
-        return self, packet_len
+        # Mailboxes used to wake up each loop.
+        self._read_box: loop.mailbox[None] = loop.mailbox()
+        self._write_box: loop.mailbox[None] = loop.mailbox()
+        self._retrans_box: loop.mailbox[None] = loop.mailbox()
+        # Whether the write loop should exit after completing the current iteration.
+        self._write_loop_exit: bool = False
 
-    async def handle_packet(self, packet: AnyBuffer) -> Channel | None:
-        """
-        Reassemble a valid THP payload and return its channel, if reassembly succeeds.
-        Otherwise, returns `None` and should be called again (with the next packet).
+        self._rx_packet_buf = bytearray(iface.RX_PACKET_LEN)
+        self._tx_packet_buf = bytearray(iface.TX_PACKET_LEN)
 
-        Also handle THP channel allocation.
+        # IDs of channels that would like to become active but will get error instead.
+        self.inactive_channels: set[int] = set()
+
+        trezorthp.init(
+            iface.iface_num(),
+            get_encoded_device_properties(iface),
+        )
+
+    async def close(self) -> None:
+        """
+        Shut down THP processing on this interface. Try waiting for the write loop
+        to finish in case it is sending an error to host.
+        """
+        if self._handshake_key_task:
+            self._handshake_key_task.close()
+        self._retrans_loop.close()
+        self._read_loop.close()
+
+        self.request_write(exit_afterwards=True)
+        try:
+            # This should not take forever thanks to _WRITE_TIMEOUT.
+            await self._write_loop
+        finally:
+            self._write_loop.close()
+
+    def read_loop(self) -> Generator[Any, Any, None]:
         """
-        ctrl_byte = _get_ctrl_byte(packet)
-        if ctrl_byte == CODEC_V1:
-            return await self._handle_codec_v1(packet)
+        Waits for incoming packets and stuffs them into rust/trezor-thp for processing.
+        Passes packets to corresponding Channel object if needed. Spawns storage
+        unlocking task if needed for a handshake.
+        The loop is not trying to read packets all the time and may have to be woken up
+        using `request_read()` - please see the documentation for `should_read()`.
+
+        The loop should only ever await the interface or _read_box, any other blocking
+        processing should happen in a different task.
+        """
+        iface = self._iface
+        iface_num = iface.iface_num()
+        verify_fn = self.verify_credential
+        packet_buffer = self._rx_packet_buf
+
+        while True:
+            while not self.should_read():
+                if __debug__ and _TRACE:
+                    log.debug(__name__, "read loop paused", iface=iface)
+                yield self._read_box
+
+            packet_len = yield self._read
+            if utils.USE_BLE and self._iface is ble.interface:
+                # prevent auto-lock while handling longer workflows on Bluetooth
+                idle_timer.touch()
+
+            assert packet_len == self._iface.RX_PACKET_LEN
+
+            self._iface.read(packet_buffer, 0)
+            if __debug__ and _TRACE:
+                log.debug(
+                    __name__,
+                    f"read: {utils.hexlify_if_bytes(packet_buffer)}",
+                    iface=iface,
+                )
 
-        cid = ustruct.unpack(">BH", packet)[1]
-        if cid == BROADCAST_CHANNEL_ID:
-            return await self._handle_broadcast(packet)
+            result = trezorthp.packet_in(iface_num, packet_buffer, verify_fn)
+            if isinstance(result, int):
+                self.read_packet_for_channel(result, packet_buffer)
+                self.clear_closed_sessions()
+                continue
 
-        if (cache := find_allocated_channel(cid)) is None:
-            if not control_byte.is_continuation(_get_ctrl_byte(packet)):
-                await self.write_error(cid, ThpErrorType.UNALLOCATED_CHANNEL)
-            return None
+            if __debug__ and _TRACE and result is not None:
+                log.debug(__name__, f"packet_in: {result}", iface=iface)
+            if result in _KEY_REQUIRED_VALS:
+                self.handle_handshake_key(result == trezorthp.KEY_REQUIRED_UNLOCK)
 
-        if (channel := self._channels.get(cid)) is None:
-            from .. import THP_BUFFERS_PROVIDER
+            # maybe we got ACK, recompute next retransmission timeout
+            self.recompute_timeouts()
+            # wake up write loop in case broadcast/handshake channels have outgoing data
+            self.request_write()
 
-            if (buffers := THP_BUFFERS_PROVIDER.take()) is None:
-                # concurrent payload reassembly is not supported
-                await self.write_error(cid, ThpErrorType.TRANSPORT_BUSY)
-                raise ChannelPreemptedException  # try to preempt the caller (if stale)
+    def should_read(self) -> bool:
+        """
+        We want to avoid the following sequence of events:
+        - workflow triggered by a message has finished,
+        - next message is reassembled before session.py restarts,
+        - session is restarted, receive buffer is lost,
+        - host has to resend message after a delay.
+          - NOTE: trezorlib doesn't resend ChannelAllocationRequest
+
+        To avoid unnecessary delay, interface is only awaited when:
+        - there is no active channel,
+        - a session called `read()` and is expecting a message,
+        - a session called `write()` and is expecting an ACK.
+
+        We can get rid of this logic if we ever get rid of loop restarts.
+        """
+        waiting_for_channel = self.thp_ctx.active_channel is None
+        expecting_message = False
+        expecting_ack = False
+        for ifctx in self.thp_ctx._iface_ctxs:
+            if ifctx.active_channel:
+                expecting_message = (
+                    expecting_message or ifctx.active_channel.expecting_message
+                )
+                expecting_ack = expecting_ack or ifctx.active_channel.expecting_ack
+        if __debug__ and _TRACE:
+            log.debug(
+                __name__,
+                f"should_read: waiting_for_channel:{waiting_for_channel} expecting_message:{expecting_message} expecting_ack:{expecting_ack}",
+                iface=self._iface,
+            )
+        return waiting_for_channel or expecting_message or expecting_ack
 
-            channel = self._channels[cid] = Channel(cache, self, buffers)
+    def read_packet_for_channel(self, result: int, packet_buffer: AnyBytes) -> None:
+        channel_id = result & 0xFFFF
+        buffer_size = (result >> 16) * 8
 
-        if channel.reassemble(packet):
-            if __debug__ and _TRACE and channel.reassembler.message is not None:
-                msg_type = "ACK" if control_byte.is_ack(ctrl_byte) else "message"
-                msg = channel.reassembler.message
-                channel._log(f"reassembled valid {msg_type}: {len(msg)} bytes")
-            update_channel_last_used(channel.channel_id)
-            return channel
+        if self.active_channel is None:
+            from .. import THP_BUFFERS_PROVIDER
 
-    def write_payload(self, header: PacketHeader, payload: AnyBytes) -> Awaitable[None]:
-        checksum_bytes = checksum.compute(
-            payload, checksum.compute_int(header.to_bytes())
-        )
-        return self._write_payload_chunks(header, payload, checksum_bytes)
-
-    def _write_payload_chunks(
-        self, header: PacketHeader, *chunks: AnyBytes
-    ) -> Awaitable[None]:
-        fragments = header.fragment_payload(self._iface.TX_PACKET_LEN, *chunks)
-        return self._write_packets(fragments)
-
-    async def _write_packets(self, fragments: Iterable[AnyBytes]) -> None:
-        packet_len = self._iface.TX_PACKET_LEN
-        for packet in fragments:
-            assert len(packet) == packet_len
-
-            n_written = 0
-            while n_written == 0:
-                await self._write
-                n_written = self._iface.write(packet)
-
-            assert n_written == packet_len
-
-    async def _handle_codec_v1(self, packet: AnyBytes) -> None:
-        # If the received packet is not an initial codec_v1 packet, do not send error message
-        if packet[1:3] == b"##":
-            response = bytearray(self._iface.TX_PACKET_LEN)
-            # Codec_v1 magic constant:
-            # "?##" + Failure message type + msg_size + msg_data (code = "Failure_InvalidProtocol")
-            utils.memcpy(response, 0, b"?##\x00\x03\x00\x00\x00\x02\x08\x11", 0)
-            await self._write_packets([response])
-
-    async def _handle_broadcast(self, packet: AnyBytes) -> None:
-        ctrl_byte, _, payload_length = ustruct.unpack(">BHH", packet)
-
-        packet = packet[: PacketHeader.INIT_LENGTH + payload_length]
-        if not checksum.is_valid(packet[-CHECKSUM_LENGTH:], packet[:-CHECKSUM_LENGTH]):
-            if __debug__:
-                log.debug(
-                    __name__, "Invalid checksum: %s", utils.hexlify_if_bytes(packet)
-                )
+            if buffers := THP_BUFFERS_PROVIDER.take():
+                self.active_channel = Channel(channel_id, self, buffers=buffers)
+                if self.thp_ctx.active_channel is None:
+                    self.thp_ctx.active_channel = self.active_channel
+                    self.thp_ctx.channel_ready_box.put(None, replace=True)
+
+        if self.active_channel is None or self.active_channel.channel_id != channel_id:
+            trezorthp.send_transport_busy(channel_id)
+            self.inactive_channels.add(channel_id)
+            self.request_write()
+            self.thp_ctx.preempt_active_channel_if_stale()
             return
 
-        if payload_length != _BROADCAST_PAYLOAD_LENGTH:
+        try:
+            self.active_channel.read_packet(packet_buffer, buffer_size)
+        except Exception as exc:
             if __debug__:
-                log.debug(
-                    __name__,
-                    "Invalid length in broadcast channel packet: %d",
-                    payload_length,
-                )
-            return
+                log.exception(__name__, exc)
+            self.active_channel.kill(exc)
+            self.active_channel = None
+
+    def write_loop(self) -> Generator[Any, Any, None]:
+        """
+        Loop that queries rust/trezor-thp for outgoing packets and writes them to
+        an interface. When there are no more packets to be sent, awaits _write_box
+        and needs to be poked (using `request_write()`) after more packets are
+        available.
+        The loop should only ever await the interface or _write_box, any other
+        blocking processing should happen in a different task.
+        """
+        iface = self._iface
+
+        while True:
+            yield self._write_box
+            if __debug__ and _TRACE:
+                log.debug(__name__, "write requested", iface=iface)
+            result = yield race(self.write_all_packets(), _WRITE_TIMEOUT)
+            if isinstance(result, int):
+                if self.active_channel:
+                    self.active_channel.kill(trezorthp.ThpError("Write is blocked"))
+            self.clear_closed_sessions()
+            self.recompute_timeouts()
+            if __debug__ and _TRACE:
+                log.debug(__name__, "write done", iface=iface)
+            if self._write_loop_exit:
+                break
+
+    def write_all_packets(self) -> Generator[Any, Any, None]:
+        packet_buffer = self._tx_packet_buf
+        iface_num = self._iface.iface_num()
+        # broadcast and channels doing handshake
+        while trezorthp.packet_out(iface_num, packet_buffer):
+            yield from self.write_packet(packet_buffer)
+        # active channel
+        if self.active_channel:
+            while self.active_channel.write_packet(packet_buffer):
+                yield from self.write_packet(packet_buffer)
+        # transport_busy for currently inactive channels
+        while self.inactive_channels:
+            cid = self.inactive_channels.pop()
+            while trezorthp.packet_out_channel(cid, EMPTY_BUFFER, packet_buffer):
+                yield from self.write_packet(packet_buffer)
+        self.inactive_channels.clear()
+
+    def write_packet(self, packet_buffer: AnyBytes) -> Generator[Any, Any, None]:
+        if __debug__ and _TRACE:
+            log.debug(
+                __name__,
+                f"write: {utils.hexlify_if_bytes(packet_buffer)}",
+                iface=self._iface,
+            )
+        n_written = 0
+        while n_written == 0:
+            yield self._write
+            n_written = self._iface.write(packet_buffer)
 
-        nonce = packet[PacketHeader.INIT_LENGTH : -CHECKSUM_LENGTH]
+        assert n_written == self._iface.TX_PACKET_LEN
 
-        if ctrl_byte == PING:
-            response_header = PacketHeader.get_pong_header(_BROADCAST_PAYLOAD_LENGTH)
-            return await self.write_payload(response_header, nonce)
+    async def retransmission_loop(self) -> None:
+        """
+        Loop for handling THP message retransmission.
+        If an event related to retransmission happens, i.e. message packets are written
+        or an ACK is received, the loop needs to be waken up using recompute_timeouts()
+        to adjust to the new state.
+        """
+        channel_id = None
+        timeout_ms = None
+        iface_num = self._iface.iface_num()
+
+        while True:
+            if timeout_ms is None or channel_id is None:
+                await self._retrans_box
+            else:
+                res = await race(self._retrans_box, loop.sleep(timeout_ms))
+                if isinstance(res, int):
+                    ok = trezorthp.message_retransmit(channel_id)
+                    if ok:
+                        if __debug__:
+                            log.warning(
+                                __name__,
+                                "(cid: %04x) retransmitting message after %s ms",
+                                channel_id,
+                                timeout_ms,
+                                iface=self._iface,
+                            )
+                        self.request_write()
+                    else:
+                        if __debug__:
+                            log.error(
+                                __name__,
+                                "(cid: %04x) retransmission timeout",
+                                channel_id,
+                                iface=self._iface,
+                            )
+                        if (
+                            self.active_channel
+                            and self.active_channel.channel_id == channel_id
+                        ):
+                            self.active_channel.kill(
+                                trezorthp.ThpError("Retransmission timeout")
+                            )
+                        self.clear_closed_sessions()
+
+            res = trezorthp.next_timeout(iface_num)
+            channel_id, timeout_ms = res or (None, None)
+
+    def recompute_timeouts(self) -> None:
+        """
+        Wake up retransmission loop to recompute earliest timeout. Needs to be
+        called after message is written to interface, or an ACK is received.
+        Safe to call even when not necessarry.
+        """
+        self._retrans_box.put(None, replace=True)
+
+    def request_write(self, exit_afterwards: bool = False) -> None:
+        """
+        Wake up write loop after new packets become ready to be written. Safe to
+        call even when no packets are ready to be written.
+        """
+        if exit_afterwards:
+            self._write_loop_exit = True
+        self._write_box.put(None, replace=True)
+
+    def request_read(self) -> None:
+        """
+        Wake up read loop when session expects a message or an ACK. The variables
+        that influence the result of `should_read()` need to be modified beforehand.
 
-        if ctrl_byte != CHANNEL_ALLOCATION_REQ:
+        Read loop is woken up on all interfaces to facilitate channel preemption.
+        """
+        for ifctx in self.thp_ctx._iface_ctxs:
+            ifctx._read_box.put(None, replace=True)
+
+    def handle_handshake_key(self, try_to_unlock: bool) -> None:
+        if config.is_unlocked():
+            trezor_static_privkey = get_trezor_static_private_key()
+            trezorthp.handshake_key(self._iface.iface_num(), trezor_static_privkey)
+        elif not try_to_unlock:
+            trezorthp.handshake_key(self._iface.iface_num(), None)
+        elif self._handshake_key_task is None:
             if __debug__:
                 log.debug(
                     __name__,
-                    "Unexpected ctrl_byte in a broadcast channel packet: %d",
-                    ctrl_byte,
+                    "Static key needed but device is locked, spawning unlock dialog",
+                    iface=self._iface,
                 )
-            return
+            self._handshake_key_task = loop.spawn(self.handshake_unlock())
+        elif __debug__:
+            log.debug(__name__, "Unlock task already running", iface=self._iface)
 
-        channel_cache = channel_manager.create_new_channel(self._iface)
-        response_data = get_channel_allocation_response(
-            nonce, channel_cache.channel_id, self._iface
-        )
-        response_header = PacketHeader.get_channel_allocation_response_header(
-            len(response_data) + CHECKSUM_LENGTH,
+    async def handshake_unlock(self) -> None:
+        try:
+            from trezor import workflow
+
+            from apps.common.lock_manager import unlock_device
+
+            # Register the unlock prompt with the workflow management system
+            # (in order to avoid immediately respawning the lockscreen task)
+            await workflow.spawn(unlock_device())
+            trezor_static_privkey = get_trezor_static_private_key()
+        except Exception as e:
+            if __debug__:
+                log.exception(__name__, e)
+            trezorthp.handshake_key(self._iface.iface_num(), None)
+        else:
+            trezorthp.handshake_key(self._iface.iface_num(), trezor_static_privkey)
+        finally:
+            self.request_write()
+            self._handshake_key_task = None
+
+    def verify_credential(self, host_static_public_key: bytes, payload: bytes) -> int:
+        """
+        Credential verification callback invoked from rust code.
+        Please note calling most trezorthp.* functions will fail because the lock on
+        global state is already held.
+        """
+        from apps.thp.credential_manager import (
+            decode_credential,
+            unwrap_credential,
+            validate_credential,
         )
-        if __debug__:
-            log.debug(
-                __name__,
-                "New channel allocated with id: %s",
-                utils.hexlify_if_bytes(channel_cache.channel_id),
-                iface=self._iface,
+
+        try:
+            encoded_credential = unwrap_credential(payload)
+            if not encoded_credential:
+                return TREZOR_STATE_UNPAIRED
+            credential = decode_credential(encoded_credential)
+            paired = validate_credential(
+                credential,
+                host_static_public_key,
             )
-        await self.write_payload(response_header, response_data)
+            if paired:
+                from trezor.wire.thp.paired_cache import cache_host_info
 
-    def write_error(self, cid: int, err_type: ThpErrorType) -> Awaitable[None]:
-        if __debug__:
-            log.error(__name__, "(cid: %04x) THP error #%d", cid, err_type)
-        msg_data = err_type.to_bytes(1, "big")
-        length = len(msg_data) + CHECKSUM_LENGTH
-        header = PacketHeader.get_error_header(cid, length)
-        return self.write_payload(header, msg_data)
+                cache_host_info(
+                    mac_addr=self.connected_addr(),
+                    host_name=credential.cred_metadata.host_name,
+                    app_name=credential.cred_metadata.app_name,
+                )
+                return TREZOR_STATE_PAIRED
+        except Exception as e:
+            if __debug__:
+                log.exception(__name__, e, iface=self._iface)
+        return TREZOR_STATE_UNPAIRED
 
     def connected_addr(self) -> AnyBytes | None:
         """
@@ -247,6 +467,7 @@ class InterfaceContext:
 
         return None
 
-
-def _get_ctrl_byte(packet: AnyBytes) -> int:
-    return packet[0]
+    def clear_closed_sessions(self) -> None:
+        if not trezorthp.channel_was_closed():
+            return
+        clear_sessions_without_channel()
diff --git a/core/src/trezor/wire/thp/memory_manager.py b/core/src/trezor/wire/thp/memory_manager.py
index 726bfc1c..6b5eba54 100644
--- a/core/src/trezor/wire/thp/memory_manager.py
+++ b/core/src/trezor/wire/thp/memory_manager.py
@@ -1,10 +1,9 @@
 from micropython import const
 from typing import TYPE_CHECKING
+from ustruct import pack_into
 
-from storage.cache_thp import SESSION_ID_LENGTH
-from trezor import protobuf, utils
-
-from .writer import MESSAGE_TYPE_LENGTH
+from trezor import protobuf, wire
+from trezorthp import APP_HEADER_LEN, SEND_BUFFER_OVERHEAD
 
 if TYPE_CHECKING:
     from buffer_types import AnyBuffer
@@ -20,7 +19,7 @@ class ThpBuffer:
     def __init__(self) -> None:
         self.buf = memoryview(bytearray(_PROTOBUF_BUFFER_SIZE))
 
-    def get(self, length: int) -> memoryview | None:
+    def get(self, length: int) -> memoryview:
         assert length >= 0
         if length > len(self.buf):
             if __debug__:
@@ -29,10 +28,14 @@ class ThpBuffer:
                     "Failed to get a buffer - requested length (%d) is too big.",
                     length,
                 )
-            return None
+            raise wire.FirmwareError("Failed to get a sufficiently large buffer")
         return self.buf[:length]
 
 
+def buffer_size(msg: protobuf.MessageType) -> int:
+    return SEND_BUFFER_OVERHEAD + protobuf.encoded_length(msg)
+
+
 def encode_into_buffer(
     buffer: AnyBuffer, msg: protobuf.MessageType, session_id: int
 ) -> int:
@@ -44,33 +47,7 @@ def encode_into_buffer(
     if msg_type is None:
         raise Exception("Message has no wire type.")
 
-    msg_size = protobuf.encoded_length(msg)
-    payload_size = SESSION_ID_LENGTH + MESSAGE_TYPE_LENGTH + msg_size
-
-    _encode_session_into_buffer(memoryview(buffer), session_id)
-    _encode_message_type_into_buffer(memoryview(buffer), msg_type, SESSION_ID_LENGTH)
-    _encode_message_into_buffer(
-        memoryview(buffer), msg, SESSION_ID_LENGTH + MESSAGE_TYPE_LENGTH
-    )
-
-    return payload_size
-
-
-def _encode_session_into_buffer(
-    buffer: AnyBuffer, session_id: int, buffer_offset: int = 0
-) -> None:
-    session_id_bytes = int.to_bytes(session_id, SESSION_ID_LENGTH, "big")
-    utils.memcpy(buffer, buffer_offset, session_id_bytes, 0)
-
-
-def _encode_message_type_into_buffer(
-    buffer: AnyBuffer, message_type: int, offset: int = 0
-) -> None:
-    msg_type_bytes = int.to_bytes(message_type, MESSAGE_TYPE_LENGTH, "big")
-    utils.memcpy(buffer, offset, msg_type_bytes, 0)
-
+    pack_into(">BH", memoryview(buffer)[:APP_HEADER_LEN], 0, session_id, msg_type)
+    msg_size = protobuf.encode(memoryview(buffer)[APP_HEADER_LEN:], msg)
 
-def _encode_message_into_buffer(
-    buffer: AnyBuffer, message: protobuf.MessageType, buffer_offset: int = 0
-) -> None:
-    protobuf.encode(memoryview(buffer[buffer_offset:]), message)
+    return APP_HEADER_LEN + msg_size
diff --git a/core/src/trezor/wire/thp/pairing_context.py b/core/src/trezor/wire/thp/pairing_context.py
index 0d0a2d83..38ccd449 100644
--- a/core/src/trezor/wire/thp/pairing_context.py
+++ b/core/src/trezor/wire/thp/pairing_context.py
@@ -59,7 +59,7 @@ class PairingContext(Context):
                     # If the previous run did not keep an unprocessed message for us,
                     # wait for a new one.
                     try:
-                        _, message = await self.channel_ctx.decrypt_message()
+                        _, message = await self.channel_ctx.read()
                     except protocol_common.WireError as e:
                         if __debug__:
                             log.exception(__name__, e, iface=self.iface)
@@ -106,7 +106,7 @@ class PairingContext(Context):
                 iface=self.iface,
             )
 
-        _, message = await self.channel_ctx.decrypt_message()
+        _, message = await self.channel_ctx.read()
         if not expected_types or message.type not in expected_types:
             from trezor.messages import Cancel
 
diff --git a/core/src/trezor/wire/thp/received_message_handler.py b/core/src/trezor/wire/thp/received_message_handler.py
index 056ebb3d..f513d3a5 100644
--- a/core/src/trezor/wire/thp/received_message_handler.py
+++ b/core/src/trezor/wire/thp/received_message_handler.py
@@ -1,312 +1,60 @@
 from typing import TYPE_CHECKING
 
-from storage.cache_common import (
-    CHANNEL_HANDSHAKE_HASH,
-    CHANNEL_KEY_RECEIVE,
-    CHANNEL_KEY_SEND,
-    CHANNEL_NONCE_RECEIVE,
-    CHANNEL_NONCE_SEND,
-)
-from storage.cache_thp import KEY_LENGTH, TAG_LENGTH, update_session_last_used
-from trezor import config, protobuf, utils
+from storage.cache_thp import update_session_last_used
 from trezor.enums import FailureType
 from trezor.messages import Failure
 
-from .. import message_handler
-from ..errors import DataError
-from . import (
-    HANDSHAKE_COMP_RES,
-    HANDSHAKE_INIT_RES,
-    ChannelState,
-    SessionState,
-    ThpDecryptionError,
-    ThpDeviceLockedError,
-    ThpErrorType,
-    ThpUnallocatedSessionError,
-)
-from . import alternating_bit_protocol as ABP
-from . import control_byte, get_encoded_device_properties, session_manager
-from .crypto import PUBKEY_LENGTH, Handshake
+from . import ChannelState, SessionState, ThpUnallocatedSessionError, session_manager
 from .session_context import SeedlessSessionContext
 
 if TYPE_CHECKING:
-    from buffer_types import AnyBytes
-
-    from trezor.messages import ThpHandshakeCompletionReqNoisePayload
-
     from .channel import Channel
 
 if __debug__:
     from trezor import log
-    from trezor.utils import hexlify_if_bytes
-
-_TREZOR_STATE_UNPAIRED = b"\x00"
-_TREZOR_STATE_PAIRED = b"\x01"
-_TREZOR_STATE_PAIRED_AUTOCONNECT = b"\x02"
 
 
 async def handle_received_message(channel: Channel) -> bool:
     """
     Handle a message received from the channel.
-
-    Returns False if we can restart the event loop.
     """
     try:
-        state = channel.get_channel_state()
-        if state is ChannelState.ENCRYPTED_TRANSPORT:
+        if channel.state == ChannelState.ENCRYPTED_TRANSPORT:
             return await _handle_state_ENCRYPTED_TRANSPORT(channel)
-        elif _is_channel_state_pairing(state):
+        else:
             await _handle_pairing(channel)
-            return False
-        elif state is ChannelState.TH1:
-            await _handle_state_handshake(channel)
-            return channel.get_channel_state() == ChannelState.TC1
-        if __debug__:
-            channel._log("Invalid channel state", logger=log.error)
     except ThpUnallocatedSessionError as e:
         error_message = Failure(code=FailureType.ThpUnallocatedSession)
         await channel.write(error_message, e.session_id)
-    except ThpDecryptionError:
-        await channel.iface_ctx.write_error(
-            channel.get_channel_id_int(), ThpErrorType.DECRYPTION_FAILED
-        )
-        channel.clear()
-    except ThpDeviceLockedError:
-        await channel.iface_ctx.write_error(
-            channel.get_channel_id_int(), ThpErrorType.DEVICE_LOCKED
-        )
-        channel.clear()
     return False
 
 
-async def _handle_thp_during_unlock(channel: Channel) -> None:
-    """
-    Keep handling THP messages while waiting for unlock.
-    It allows preemption and ping/pong handling if the device is soft-locked.
-    """
-    while True:
-        # may raise ChannelPreemptedException if another channel preempts this one
-        msg = await channel._get_reassembled_message()
-        if __debug__:
-            # we don't expect messages from this channel since the handshake is not over
-            channel._log(
-                "drop unexpected message",
-                utils.hexlify_if_bytes(msg),
-                logger=log.warning,
-            )
-
-
-async def _handle_state_handshake(
-    ctx: Channel,
-) -> None:
-    if __debug__:
-        log.debug(__name__, "handle_state_handshake", iface=ctx.iface)
-
-    def _handshake_callback(ctrl_byte: int) -> bool:
-        success = control_byte.is_handshake_init_req(ctrl_byte)
-
-        if success and control_byte.get_ack_bit(ctrl_byte) == 1:
-            # Newer Suite versions will send `handshake_init_req` with a non-zero ACK bit.
-            # The device should not use ACK piggybacking with older Suite versions.
-            ABP.allow_ack_piggybacking(ctx.channel_cache)
-
-        if __debug__:
-            ctx._log(
-                "THP ACK piggybacking = ",
-                str(ABP.is_ack_piggybacking_allowed(ctx.channel_cache)),
-            )
-
-        return success
-
-    payload = await ctx.recv_payload(_handshake_callback)
-
-    if len(payload) != PUBKEY_LENGTH + 1:
-        if __debug__:
-            log.error(
-                __name__,
-                "Message received is not a valid handshake init request: %d bytes",
-                len(payload),
-            )
-        return
-
-    host_ephemeral_public_key = payload[:PUBKEY_LENGTH]
-    # show the PIN keyboard to allow the user to unlock the device
-    try_to_unlock = payload[PUBKEY_LENGTH] & 0x01 == 1
-
-    async def _check_unlocked() -> None:
-        if config.is_unlocked():
-            return
-
-        if try_to_unlock:
-            from trezor import loop, workflow
-
-            from apps.common.lock_manager import unlock_device
-
-            # Register the unlock prompt with the workflow management system
-            # (in order to avoid immediately respawning the lockscreen task)
-            try:
-                unlock = workflow.spawn(unlock_device())
-                handle = _handle_thp_during_unlock(channel=ctx)
-                return await loop.race(unlock, handle)
-            except Exception as e:
-                if __debug__:
-                    log.exception(__name__, e)
-
-        # Fail pairing if still locked
-        raise ThpDeviceLockedError
-
-    await _check_unlocked()
-
-    handshake = Handshake()
-
-    trezor_ephemeral_public_key, encrypted_trezor_static_public_key, tag = (
-        handshake.handle_th1_crypto(
-            get_encoded_device_properties(ctx.iface),
-            host_ephemeral_public_key=host_ephemeral_public_key,
-            payload=payload[PUBKEY_LENGTH:],
-        )
-    )
-
-    if __debug__:
-        log.debug(
-            __name__,
-            "trezor ephemeral public key: %s",
-            hexlify_if_bytes(trezor_ephemeral_public_key),
-            iface=ctx.iface,
-        )
-        log.debug(
-            __name__,
-            "encrypted trezor masked static public key: %s",
-            hexlify_if_bytes(encrypted_trezor_static_public_key),
-            iface=ctx.iface,
-        )
-        log.debug(__name__, "tag: %s", hexlify_if_bytes(tag), iface=ctx.iface)
-
-    payload = trezor_ephemeral_public_key + encrypted_trezor_static_public_key + tag
-
-    # send handshake init response message
-    await ctx.write_encrypted_payload(HANDSHAKE_INIT_RES, payload)
-
-    payload = await ctx.recv_payload(control_byte.is_handshake_comp_req)
-
-    # will be `None` on USB interface, to be ignored by `cache_host_info()`
-    mac_addr: AnyBytes | None = ctx.iface_ctx.connected_addr()
-
-    await _check_unlocked()
-
-    host_encrypted_static_public_key = payload[: KEY_LENGTH + TAG_LENGTH]
-    handshake_completion_request_noise_payload = payload[KEY_LENGTH + TAG_LENGTH :]
-
-    handshake.handle_th2_crypto(
-        host_encrypted_static_public_key, handshake_completion_request_noise_payload
-    )
-
-    ctx.channel_cache.set(CHANNEL_KEY_RECEIVE, handshake.key_receive)
-    ctx.channel_cache.set(CHANNEL_KEY_SEND, handshake.key_send)
-    ctx.channel_cache.set(CHANNEL_HANDSHAKE_HASH, handshake.h)
-    ctx.channel_cache.set_int(CHANNEL_NONCE_RECEIVE, 0)
-    ctx.channel_cache.set_int(CHANNEL_NONCE_SEND, 1)
-
-    buffer = payload[KEY_LENGTH + TAG_LENGTH : -TAG_LENGTH]
-
-    payload_type = protobuf.type_for_name("ThpHandshakeCompletionReqNoisePayload")
-    noise_payload = message_handler.wrap_protobuf_load(buffer, payload_type)
-
-    if TYPE_CHECKING:
-        assert ThpHandshakeCompletionReqNoisePayload.is_type_of(noise_payload)
-
-    if __debug__:
-        log.debug(
-            __name__,
-            "host static public key: %s, noise payload: %s",
-            utils.hexlify_if_bytes(host_encrypted_static_public_key),
-            utils.hexlify_if_bytes(handshake_completion_request_noise_payload),
-            iface=ctx.iface,
-        )
-
-    # key is decoded in handshake._handle_th2_crypto
-    host_static_public_key = host_encrypted_static_public_key[:PUBKEY_LENGTH]
-    ctx.channel_cache.set_host_static_public_key(host_static_public_key)
-
-    paired: bool = False
-    trezor_state = _TREZOR_STATE_UNPAIRED
-
-    if noise_payload.host_pairing_credential is not None:
-        from apps.thp.credential_manager import decode_credential, validate_credential
-
-        try:  # TODO change try-except for something better
-            credential = decode_credential(noise_payload.host_pairing_credential)
-            paired = validate_credential(
-                credential,
-                host_static_public_key,
-            )
-            if paired:
-                from trezor.wire.thp.paired_cache import cache_host_info
-
-                cache_host_info(
-                    mac_addr=mac_addr,
-                    host_name=credential.cred_metadata.host_name,
-                    app_name=credential.cred_metadata.app_name,
-                )
-                trezor_state = _TREZOR_STATE_PAIRED
-                ctx.credential = credential
-                if ctx.is_channel_to_replace():
-                    # When replacing existing channel, user confirmation is not needed
-                    trezor_state = _TREZOR_STATE_PAIRED_AUTOCONNECT
-            else:
-                ctx.credential = None
-        except DataError as e:
-            if __debug__:
-                log.exception(__name__, e, iface=ctx.iface)
-            pass
-
-    # send hanshake completion response
-    response = handshake.get_handshake_completion_response(trezor_state)
-    await ctx.write_encrypted_payload(HANDSHAKE_COMP_RES, response)
-
-    if paired:
-        ctx.set_channel_state(ChannelState.TC1)
-    else:
-        ctx.set_channel_state(ChannelState.TP0)
-
-
-async def _handle_state_ENCRYPTED_TRANSPORT(ctx: Channel) -> bool:
+async def _handle_state_ENCRYPTED_TRANSPORT(channel: Channel) -> bool:
     if __debug__:
-        log.debug(__name__, "handle_state_ENCRYPTED_TRANSPORT", iface=ctx.iface)
-
-    session_id, message = await ctx.decrypt_message()
-    if session_id not in ctx.sessions:
-
-        s = session_manager.get_session_from_cache(ctx, session_id)
+        log.debug(__name__, "handle_state_ENCRYPTED_TRANSPORT", iface=channel.iface)
 
+    session_id, message = await channel.read()
+    if session_id not in channel.sessions:
+        s = session_manager.get_session_from_cache(channel, session_id)
         if s is None:
-            s = SeedlessSessionContext(ctx, session_id)
+            s = SeedlessSessionContext(channel, session_id)
 
-        ctx.sessions[session_id] = s
+        channel.sessions[session_id] = s
 
-    elif ctx.sessions[session_id].get_session_state() is SessionState.UNALLOCATED:
+    elif channel.sessions[session_id].get_session_state() is SessionState.UNALLOCATED:
         raise ThpUnallocatedSessionError(session_id)
 
-    s = ctx.sessions[session_id]
-    update_session_last_used(s.channel_id, (s.session_id).to_bytes(1, "big"))
+    s = channel.sessions[session_id]
+    update_session_last_used(
+        s.channel_id.to_bytes(2, "big"), s.session_id.to_bytes(1, "big")
+    )
     return await s.handle(message)
 
 
-async def _handle_pairing(ctx: Channel) -> None:
+async def _handle_pairing(channel: Channel) -> None:
     from .pairing_context import PairingContext
 
-    ctx.connection_context = PairingContext(ctx)
+    channel.connection_context = PairingContext(channel)
 
-    _session_id, message = await ctx.decrypt_message()
-    await ctx.connection_context.handle(message)
-
-
-def _is_channel_state_pairing(state: int) -> bool:
-    return state in (
-        ChannelState.TP0,
-        ChannelState.TP1,
-        ChannelState.TP2,
-        ChannelState.TP3,
-        ChannelState.TP4,
-        ChannelState.TC1,
-    )
+    _session_id, message = await channel.read()
+    await channel.connection_context.handle(message)
diff --git a/core/src/trezor/wire/thp/session_context.py b/core/src/trezor/wire/thp/session_context.py
index a7800603..44eef824 100644
--- a/core/src/trezor/wire/thp/session_context.py
+++ b/core/src/trezor/wire/thp/session_context.py
@@ -25,7 +25,6 @@ _REPEAT_LOOP = False
 
 if __debug__:
     from trezor import log
-    from trezor.utils import hexlify_if_bytes
 
 
 class GenericSessionContext(Context):
@@ -39,32 +38,33 @@ class GenericSessionContext(Context):
         if __debug__:
             log.debug(
                 __name__,
-                "handle - start (channel_id (bytes): %s, session_id: %d)",
-                hexlify_if_bytes(self.channel_id),
+                "handle - start (channel_id (bytes): %04x, session_id: %d)",
+                self.channel_id,
                 self.session_id,
                 iface=self.iface,
             )
 
+        do_not_restart = False
         while True:
             try:
-                return await handle_single_message(self, message)
+                do_not_restart = await handle_single_message(self, message)
             except protocol_common.WireError as e:
                 if __debug__:
                     log.exception(__name__, e, iface=self.iface)
                 await self.write(failure(e))
             except UnexpectedMessageException as unexpected:
-                if unexpected.msg is not None:
-                    # The workflow was interrupted by an unexpected message. We need to
-                    # process it as if it was a new message...
-                    message = unexpected.msg
-                    continue
+                # The workflow was interrupted by an unexpected message. We need to
+                # process it as if it was a new message...
+                message = unexpected.msg
+                continue
             except Exception as exc:
                 if __debug__:
                     log.exception(__name__, exc, iface=self.iface)
+            return do_not_restart
 
     async def _read_next_message(self) -> Message:
         while True:
-            session_id, message = await self.channel.decrypt_message()
+            session_id, message = await self.channel.read()
             if session_id == self.session_id:
                 return message
             if __debug__:
@@ -130,7 +130,7 @@ class SeedlessSessionContext(GenericSessionContext):
 class SessionContext(GenericSessionContext):
 
     def __init__(self, channel_ctx: Channel, session_cache: SessionThpCache) -> None:
-        if channel_ctx.channel_id != session_cache.channel_id:
+        if channel_ctx.channel_id_bytes() != session_cache.channel_id:
             raise Exception(
                 "The session has different channel id than the provided channel context!"
             )
diff --git a/core/src/trezor/wire/thp/session_manager.py b/core/src/trezor/wire/thp/session_manager.py
index 0543b577..73351069 100644
--- a/core/src/trezor/wire/thp/session_manager.py
+++ b/core/src/trezor/wire/thp/session_manager.py
@@ -17,7 +17,7 @@ def get_new_session_context(
     session_id: int,
 ) -> SessionContext:
     session_cache = cache_thp.create_or_replace_session(
-        channel=channel_ctx.channel_cache,
+        channel_id=channel_ctx.channel_id_bytes(),
         session_id=session_id.to_bytes(1, "big"),
     )
     return SessionContext(channel_ctx, session_cache)
@@ -31,7 +31,7 @@ def get_session_from_cache(
     """
     session_id_bytes = session_id.to_bytes(1, "big")
     session_cache = cache_thp.get_allocated_session(
-        channel_ctx.channel_id, session_id_bytes
+        channel_ctx.channel_id_bytes(), session_id_bytes
     )
     if session_cache is None:
         return None
diff --git a/core/src/trezor/wire/thp/writer.py b/core/src/trezor/wire/thp/writer.py
deleted file mode 100644
index 20316302..00000000
--- a/core/src/trezor/wire/thp/writer.py
+++ /dev/null
@@ -1,4 +0,0 @@
-from micropython import const
-
-MAX_PAYLOAD_LEN = const(60000)
-MESSAGE_TYPE_LENGTH = const(2)
diff --git a/core/tests/test_apps.thp.credential_manager.py b/core/tests/test_apps.thp.credential_manager.py
index f894750d..927189aa 100644
--- a/core/tests/test_apps.thp.credential_manager.py
+++ b/core/tests/test_apps.thp.credential_manager.py
@@ -7,8 +7,10 @@ if utils.USE_THP:
 
     from apps.thp import credential_manager
 
-    def _issue_credential(host_name: str, host_static_public_key: bytes) -> bytes:
-        metadata = ThpCredentialMetadata(host_name=host_name, app_name="APP")
+    def _issue_credential(
+        host_name: str, host_static_public_key: bytes, app_name: str = "APP"
+    ) -> bytes:
+        metadata = ThpCredentialMetadata(host_name=host_name, app_name=app_name)
         return credential_manager.issue_credential(host_static_public_key, metadata)
 
     def _dummy_log(name: str, msg: str, *args):
@@ -71,6 +73,27 @@ class TestTrezorHostProtocolCredentialManager(unittest.TestCase):
             credential_manager.decode_and_validate_credential(cred_4, DUMMY_KEY_1)
         )
 
+    def test_credentials_truncate(self):
+        app_name = "a" * 100
+        host_name = "h" * 100
+
+        cred_long = _issue_credential(host_name, DUMMY_KEY_1, app_name)
+
+        self.assertTrue(len(cred_long) < 200)
+        self.assertTrue(
+            credential_manager.decode_and_validate_credential(cred_long, DUMMY_KEY_1)
+        )
+
+        app_name = "🐧" * 100
+        host_name = "Ř" * 100
+
+        cred_long = _issue_credential(host_name, DUMMY_KEY_1, app_name)
+
+        self.assertTrue(len(cred_long) < 200)
+        self.assertTrue(
+            credential_manager.decode_and_validate_credential(cred_long, DUMMY_KEY_1)
+        )
+
     def test_protobuf_encoding(self):
         """
         If the protobuf encoding of credentials changes in the future, this
diff --git a/core/tests/test_storage.cache.py b/core/tests/test_storage.cache.py
index 8047ae2b..a85845b4 100644
--- a/core/tests/test_storage.cache.py
+++ b/core/tests/test_storage.cache.py
@@ -6,25 +6,24 @@ KEY_BOOL = 0
 if utils.USE_THP:
     import thp_common
     from mock_wire_interface import MockHID
-    from storage import cache, cache_thp
-    from storage.cache_common import CHANNEL_STATE, SESSION_STATE
-    from trezor.wire.thp import ChannelState
+    from storage import cache, cache_thp, cache_thp_keys
+    from storage.cache_common import SESSION_STATE
     from trezor.wire.thp.session_context import SessionContext
 
     _PROTOCOL_CACHE = cache_thp
 
-    KEY = 5
+    KEY = cache_thp_keys.APP_COMMON_SEED
 
 else:
     from mock_storage import mock_storage
-    from storage import cache, cache_codec
+    from storage import cache, cache_codec, cache_codec_keys
     from trezor.messages import EndSession, Initialize
 
     from apps.base import handle_EndSession
 
     _PROTOCOL_CACHE = cache_codec
 
-    KEY = 0
+    KEY = cache_codec_keys.APP_COMMON_SEED
 
     def is_session_started() -> bool:
         return cache_codec.get_active_session() is not None
@@ -41,21 +40,26 @@ class TestStorageCache(TestCaseWithContext):
             self.interface = MockHID()
             cache.clear_all()
 
+        def assertEqualExceptLastUsage(
+            self, x: list[bytearray], y: list[bytearray], msg=""
+        ):
+            skip = cache_thp_keys.LAST_USAGE
+            x1 = [bytearray(b) for i, b in enumerate(x) if i != skip]
+            y1 = [bytearray(b) for i, b in enumerate(y) if i != skip]
+            self.assertEqual(x1, y1, msg)
+
         def test_new_channel_and_session(self):
             channel = thp_common.get_new_channel(self.interface)
 
-            # Assert that channel is created without any sessions
-            self.assertEqual(len(channel.sessions), 0)
-
             cid_1 = channel.channel_id
             session_cache_1 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x01"
+                channel.channel_id_bytes(), b"\x01"
             )
             session_1 = SessionContext(channel, session_cache_1)
             self.assertEqual(session_1.channel_id, cid_1)
 
             session_cache_2 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x02"
+                channel.channel_id_bytes(), b"\x02"
             )
             session_2 = SessionContext(channel, session_cache_2)
             self.assertEqual(session_2.channel_id, cid_1)
@@ -67,7 +71,7 @@ class TestStorageCache(TestCaseWithContext):
             self.assertNotEqual(cid_1, cid_2)
 
             session_cache_3 = cache_thp.create_or_replace_session(
-                channel_2.channel_cache, b"\x01"
+                channel_2.channel_id_bytes(), b"\x01"
             )
             session_3 = SessionContext(channel_2, session_cache_3)
             self.assertEqual(session_3.channel_id, cid_2)
@@ -78,7 +82,10 @@ class TestStorageCache(TestCaseWithContext):
 
             self.assertEqual(cache_thp._SESSIONS[0], session_cache_1)
             self.assertNotEqual(cache_thp._SESSIONS[0], session_cache_2)
-            self.assertEqual(cache_thp._SESSIONS[0].channel_id, session_1.channel_id)
+            self.assertEqual(
+                cache_thp._SESSIONS[0].channel_id,
+                session_1.channel_id.to_bytes(2, "big"),
+            )
 
             # Check that session data IS in cache for created sessions ONLY
             for i in range(3):
@@ -98,45 +105,10 @@ class TestStorageCache(TestCaseWithContext):
                 self.assertEqual(session.last_usage, 0)
                 self.assertFalse(session.is_set(SESSION_STATE))
 
-        def test_channel_capacity_in_cache(self):
-            self.assertTrue(cache_thp._MAX_CHANNELS_COUNT >= 3)
-            channels = []
-            for i in range(cache_thp._MAX_CHANNELS_COUNT):
-                channels.append(thp_common.get_new_channel(self.interface))
-            channel_ids = [channel.channel_cache.channel_id for channel in channels]
-
-            # Assert that each channel_id is unique and that cache and list of channels
-            # have the same "channels" on the same indexes
-            for i in range(len(channel_ids)):
-                self.assertEqual(cache_thp._CHANNELS[i].channel_id, channel_ids[i])
-                for j in range(i + 1, len(channel_ids)):
-                    self.assertNotEqual(channel_ids[i], channel_ids[j])
-
-            # Create a new channel that is over the capacity
-            new_channel = thp_common.get_new_channel(self.interface)
-            for c in channels:
-                self.assertNotEqual(c.channel_id, new_channel.channel_id)
-
-            # Test that the oldest (least used) channel was replaced (_CHANNELS[0])
-            self.assertNotEqual(cache_thp._CHANNELS[0].channel_id, channel_ids[0])
-            self.assertEqual(cache_thp._CHANNELS[0].channel_id, new_channel.channel_id)
-
-            # Update the "last used" value of the second channel in cache (_CHANNELS[1]) and
-            # assert that it is not replaced when creating a new channel
-            cache_thp.update_channel_last_used(channel_ids[1])
-            new_new_channel = thp_common.get_new_channel(self.interface)
-            self.assertEqual(cache_thp._CHANNELS[1].channel_id, channel_ids[1])
-
-            # Assert that it was in fact the _CHANNEL[2] that was replaced
-            self.assertNotEqual(cache_thp._CHANNELS[2].channel_id, channel_ids[2])
-            self.assertEqual(
-                cache_thp._CHANNELS[2].channel_id, new_new_channel.channel_id
-            )
-
         def test_session_capacity_in_cache(self):
             self.assertTrue(cache_thp._MAX_SESSIONS_COUNT >= 4)
-            channel_cache_A = thp_common.get_new_channel(self.interface).channel_cache
-            channel_cache_B = thp_common.get_new_channel(self.interface).channel_cache
+            channel_A = thp_common.get_new_channel(self.interface)
+            channel_B = thp_common.get_new_channel(self.interface)
 
             sesions_A = []
             cid = []
@@ -144,7 +116,7 @@ class TestStorageCache(TestCaseWithContext):
             for i in range(3):
                 sesions_A.append(
                     cache_thp.create_or_replace_session(
-                        channel_cache_A, (i + 1).to_bytes(1, "big")
+                        channel_A.channel_id_bytes(), (i + 1).to_bytes(1, "big")
                     )
                 )
                 cid.append(sesions_A[i].channel_id)
@@ -154,7 +126,7 @@ class TestStorageCache(TestCaseWithContext):
             for i in range(cache_thp._MAX_SESSIONS_COUNT - 3):
                 sessions_B.append(
                     cache_thp.create_or_replace_session(
-                        channel_cache_B, (i + 10).to_bytes(1, "big")
+                        channel_B.channel_id_bytes(), (i + 10).to_bytes(1, "big")
                     )
                 )
 
@@ -165,68 +137,50 @@ class TestStorageCache(TestCaseWithContext):
             for i in range(3, cache_thp._MAX_SESSIONS_COUNT):
                 self.assertEqual(sessions_B[i - 3], cache_thp._SESSIONS[i])
 
-            # Assert that new session replaces the oldest (least used) one (_SESSOIONS[0])
-            new_session = cache_thp.create_or_replace_session(channel_cache_B, b"\xab")
+            # Assert that new session replaces the oldest (least used) one (_SESSIONS[0])
+            new_session = cache_thp.create_or_replace_session(
+                channel_B.channel_id_bytes(), b"\xab"
+            )
             self.assertEqual(new_session, cache_thp._SESSIONS[0])
             self.assertNotEqual(new_session.channel_id, cid[0])
             self.assertNotEqual(new_session.session_id, sid[0])
 
-            # Assert that updating "last used" for session on channel A increases also
-            # the "last usage" of channel A.
-            self.assertTrue(channel_cache_A.last_usage < channel_cache_B.last_usage)
+            # Assert that creating a new session on channel B shifts the "last usage" again
+            # and that _SESSIONS[1] was not replaced, but that _SESSIONS[2] was replaced
             cache_thp.update_session_last_used(
-                channel_cache_A.channel_id, sesions_A[1].session_id
+                channel_A.channel_id_bytes(), sesions_A[1].session_id
             )
-            self.assertTrue(channel_cache_A.last_usage > channel_cache_B.last_usage)
-
             new_new_session = cache_thp.create_or_replace_session(
-                channel_cache_B, b"\xaa"
+                channel_B.channel_id_bytes(), b"\xaa"
             )
-
-            # Assert that creating a new session on channel B shifts the "last usage" again
-            # and that _SESSIONS[1] was not replaced, but that _SESSIONS[2] was replaced
-            self.assertTrue(channel_cache_A.last_usage < channel_cache_B.last_usage)
-            self.assertEqual(sesions_A[1], cache_thp._SESSIONS[1])
+            self.assertEqualExceptLastUsage(sesions_A[1], cache_thp._SESSIONS[1])
             self.assertNotEqual(sesions_A[2], cache_thp._SESSIONS[2])
             self.assertEqual(new_new_session, cache_thp._SESSIONS[2])
 
         def test_clear(self):
             channel_A = thp_common.get_new_channel(self.interface)
             channel_B = thp_common.get_new_channel(self.interface)
-            cid_A = channel_A.channel_id
-            cid_B = channel_B.channel_id
+            cid_A = channel_A.channel_id_bytes()
+            cid_B = channel_B.channel_id_bytes()
             sessions = []
 
             for i in range(3):
                 sessions.append(
                     cache_thp.create_or_replace_session(
-                        channel_A.channel_cache, (i + 1).to_bytes(1, "big")
+                        cid_A, (i + 1).to_bytes(1, "big")
                     )
                 )
                 sessions.append(
                     cache_thp.create_or_replace_session(
-                        channel_B.channel_cache, (i + 10).to_bytes(1, "big")
+                        cid_B, (i + 10).to_bytes(1, "big")
                     )
                 )
 
                 self.assertEqual(cache_thp._SESSIONS[2 * i].channel_id, cid_A)
-                self.assertNotEqual(cache_thp._SESSIONS[2 * i].last_usage, 0)
-
                 self.assertEqual(cache_thp._SESSIONS[2 * i + 1].channel_id, cid_B)
-                self.assertNotEqual(cache_thp._SESSIONS[2 * i + 1].last_usage, 0)
-
-            # Assert that clearing of channel A works
-            self.assertNotEqual(channel_A.channel_cache.channel_id, b"")
-            self.assertNotEqual(channel_A.channel_cache.last_usage, 0)
-            self.assertEqual(channel_A.get_channel_state(), ChannelState.TH1)
-
-            channel_A.clear()
-
-            self.assertEqual(channel_A.channel_cache.channel_id, b"")
-            self.assertEqual(channel_A.channel_cache.last_usage, 0)
-            self.assertEqual(channel_A.get_channel_state(), ChannelState.UNALLOCATED)
 
             # Assert that clearing channel A also cleared all its sessions
+            cache_thp.clear_sessions_with_channel_id(cid_A)
             for i in range(3):
                 self.assertEqual(cache_thp._SESSIONS[2 * i].last_usage, 0)
                 self.assertEqual(cache_thp._SESSIONS[2 * i].channel_id, b"")
@@ -238,25 +192,18 @@ class TestStorageCache(TestCaseWithContext):
             for session in cache_thp._SESSIONS:
                 self.assertEqual(session.last_usage, 0)
                 self.assertEqual(session.channel_id, b"")
-            for channel in cache_thp._CHANNELS:
-                self.assertEqual(channel.channel_id, b"")
-                self.assertEqual(channel.last_usage, 0)
-                self.assertEqual(
-                    channel.get_int(CHANNEL_STATE, ChannelState.UNALLOCATED),
-                    ChannelState.UNALLOCATED,
-                )
 
         def test_get_set(self):
             channel = thp_common.get_new_channel(self.interface)
 
             session_1 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x01"
+                channel.channel_id_bytes(), b"\x01"
             )
             session_1.set(KEY, b"hello")
             self.assertEqual(session_1.get(KEY), b"hello")
 
             session_2 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x02"
+                channel.channel_id_bytes(), b"\x02"
             )
             session_2.set(KEY, b"world")
             self.assertEqual(session_2.get(KEY), b"world")
@@ -271,14 +218,14 @@ class TestStorageCache(TestCaseWithContext):
             channel = thp_common.get_new_channel(self.interface)
 
             session_1 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x01"
+                channel.channel_id_bytes(), b"\x01"
             )
             session_1.set_int(KEY, 1234)
 
             self.assertEqual(session_1.get_int(KEY), 1234)
 
             session_2 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x02"
+                channel.channel_id_bytes(), b"\x02"
             )
             session_2.set_int(KEY, 5678)
             self.assertEqual(session_2.get_int(KEY), 5678)
@@ -293,7 +240,7 @@ class TestStorageCache(TestCaseWithContext):
             channel = thp_common.get_new_channel(self.interface)
 
             session_1 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x01"
+                channel.channel_id_bytes(), b"\x01"
             )
             with self.assertRaises(AssertionError):
                 session_1.set_bool(KEY_BOOL, True)
@@ -306,7 +253,7 @@ class TestStorageCache(TestCaseWithContext):
             self.assertEqual(session_1.get_bool(KEY_BOOL), True)
 
             session_2 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x02"
+                channel.channel_id_bytes(), b"\x02"
             )
             session_2.fields = session_2.fields = (0,) + session_2.fields[1:]
             session_2.set_bool(KEY_BOOL, False)
@@ -323,7 +270,7 @@ class TestStorageCache(TestCaseWithContext):
         def test_delete(self):
             channel = thp_common.get_new_channel(self.interface)
             session_1 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x01"
+                channel.channel_id_bytes(), b"\x01"
             )
 
             self.assertIsNone(session_1.get(KEY))
@@ -334,7 +281,7 @@ class TestStorageCache(TestCaseWithContext):
 
             session_1.set(KEY, b"hello")
             session_2 = cache_thp.create_or_replace_session(
-                channel.channel_cache, b"\x02"
+                channel.channel_id_bytes(), b"\x02"
             )
 
             self.assertIsNone(session_2.get(KEY))
@@ -528,7 +475,7 @@ class TestStorageCache(TestCaseWithContext):
             self.assertEqual(get_active_session().get(KEY), b"hello")
 
             # supplying a different session ID starts a new session
-            call_Initialize(session_id=b"A" * _PROTOCOL_CACHE.SESSION_ID_LENGTH)
+            call_Initialize(session_id=b"A" * _PROTOCOL_CACHE._SESSION_ID_LENGTH)
             self.assertIsNone(get_active_session().get(KEY))
 
             # but resuming a session loads the previous one
diff --git a/core/tests/test_trezor.wire.thp.channel.py b/core/tests/test_trezor.wire.thp.channel.py
deleted file mode 100644
index 567c9df7..00000000
--- a/core/tests/test_trezor.wire.thp.channel.py
+++ /dev/null
@@ -1,109 +0,0 @@
-# flake8: noqa: F403,F405
-from common import *  # isort:skip
-from typing import Callable
-
-from mock import patch
-
-if utils.USE_THP:
-    import fixtures
-    import thp_common
-    from trezor import protobuf, wire
-    from trezor.messages import Ping
-    from trezor.wire.thp import channel as channel_module
-    from trezor.wire.thp.channel import Reassembler
-    from trezor.wire.thp.memory_manager import _PROTOBUF_BUFFER_SIZE, ThpBuffer
-    from trezor.wire.thp.writer import MAX_PAYLOAD_LEN
-
-    def _encoded_len_patch(first_len: int) -> patch:
-        """
-        Patches `protobuf.encoded_length`:
-
-        - the first call of `protobuf.encoded_length(msg)` returns `first_len`,
-        - subsequent calls of `protobuf.encoded_length(msg)` return `trezorproto.encoded_length(msg)` (correct value).
-        """
-
-        def _patch_first_encoded_len(
-            first_len: int,
-        ) -> Callable[[protobuf.MessageType], int]:
-            import trezorproto
-
-            called = False
-
-            def wrapper(msg: protobuf.MessageType) -> int:
-                nonlocal called
-                if not called:
-                    called = True
-                    return first_len
-                return trezorproto.encoded_length(msg)
-
-            return wrapper
-
-        return patch(protobuf, "encoded_length", _patch_first_encoded_len(first_len))
-
-
-@unittest.skipUnless(utils.USE_THP, "only needed for THP")
-class TestTrezorHostProtocolChannel(TestCaseWithContext):
-    def test_reassembler_get_buffer(self):
-        """
-        Test request of a reassembly buffer (various sizes).
-        """
-        reassembler = Reassembler(ThpBuffer())
-        read_buffer = reassembler.thp_read_buf
-
-        # Should pass
-        for buffer_len in (0, 5, 100, 4096, _PROTOBUF_BUFFER_SIZE):
-            buffer = read_buffer.get(buffer_len)
-            assert buffer is not None  # to make typechecker happy
-            self.assertEqual(len(buffer), buffer_len)
-
-        # Should fail
-        for buffer_len in (-1, -5, -100):
-            with self.assertRaises(AssertionError):
-                buffer = read_buffer.get(buffer_len)
-
-        # Should return None
-        for buffer_len in (
-            _PROTOBUF_BUFFER_SIZE + 1,
-            2 * _PROTOBUF_BUFFER_SIZE,
-            2 * _PROTOBUF_BUFFER_SIZE + 1,
-            MAX_PAYLOAD_LEN,  # Currently holds that: _PROTOBUF_BUFFER_SIZE < MAX_PAYLOAD_LEN
-        ):
-            buffer = read_buffer.get(buffer_len)
-            self.assertIsNone(buffer)
-
-    def test_write_too_big_message_mocked_size(self):
-        """
-        Action: Try to send a message with size greater than `_PROTOBUF_BUFFER_SIZE`. The size of the message is mocked.
-
-        Expected: FirmwareError is raised.
-        """
-        channel = thp_common.PatchedChannel()
-        gen = channel.write(Ping(message="Test"), 0)
-        with _encoded_len_patch(first_len=_PROTOBUF_BUFFER_SIZE + 1):
-            with self.assertRaises(wire.FirmwareError) as e:
-                gen.send(None)
-            self.assertEqual(
-                e.value.message,
-                "Failed to get a sufficiently large write buffer.",
-            )
-
-    def test_write_too_big_message(self):
-        """
-        Action: Try to send a message with size greater than `_PROTOBUF_BUFFER_SIZE`,
-                but smaller than `MAX_PAYLOAD_LEN`. The size of the message is real.
-
-        Expected: Message is sent successfully.
-        """
-        channel = thp_common.PatchedChannel()
-        ping_message = fixtures.LONG_STRING_50000
-        gen = channel.write(Ping(message=ping_message), 0)
-        with self.assertRaises(wire.FirmwareError) as e:
-            gen.send(None)
-        self.assertEqual(
-            e.value.message,
-            "Failed to get a sufficiently large write buffer.",
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()
diff --git a/core/tests/test_trezor.wire.thp.checksum.py b/core/tests/test_trezor.wire.thp.checksum.py
deleted file mode 100644
index 307a8c5c..00000000
--- a/core/tests/test_trezor.wire.thp.checksum.py
+++ /dev/null
@@ -1,89 +0,0 @@
-# flake8: noqa: F403,F405
-from common import *  # isort:skip
-
-if utils.USE_THP:
-    from trezor.wire.thp import checksum
-
-
-@unittest.skipUnless(utils.USE_THP, "only needed for THP")
-class TestTrezorHostProtocolChecksum(unittest.TestCase):
-    vectors_correct = [
-        (
-            b"",
-            b"\x00\x00\x00\x00",
-        ),
-        (
-            b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
-            b"\x19\x0a\x55\xad",
-        ),
-        (
-            b"a",
-            b"\xe8\xb7\xbe\x43",
-        ),
-        (
-            b"abc",
-            b"\x35\x24\x41\xc2",
-        ),
-        (
-            b"123456789",
-            b"\xcb\xf4\x39\x26",
-        ),
-        (
-            b"12345678901234567890123456789012345678901234567890123456789012345678901234567890",
-            b"\x7c\xa9\x4a\x72",
-        ),
-        (
-            b"\x76\x61\x72\x69\x6f\x75\x73\x20\x43\x52\x43\x20\x61\x6c\x67\x6f\x72\x69\x74\x68\x6d\x73\x20\x69\x6e\x70\x75\x74\x20\x64\x61\x74\x61",
-            b"\x9b\xd3\x66\xae",
-        ),
-        (
-            b"\x67\x3a\x5f\x0e\x39\xc0\x3c\x79\x58\x22\x74\x76\x64\x9e\x36\xe9\x0b\x04\x8c\xd2\xc0\x4d\x76\x63\x1a\xa2\x17\x85\xe8\x50\xa7\x14\x18\xfb\x86\xed\xa3\x59\x2d\x62\x62\x49\x64\x62\x26\x12\xdb\x95\x3d\xd6\xb5\xca\x4b\x22\x0d\xc5\x78\xb2\x12\x97\x8e\x54\x4e\x06\xb7\x9c\x90\xf5\xa0\x21\xa6\xc7\xd8\x39\xfd\xea\x3a\xf1\x7b\xa2\xe8\x71\x41\xd6\xcb\x1e\x5b\x0e\x29\xf7\x0c\xc7\x57\x8b\x53\x20\x1d\x2b\x41\x1c\x25\xf9\x07\xbb\xb4\x37\x79\x6a\x13\x1f\x6c\x43\x71\xc1\x1e\x70\xe6\x74\xd3\x9c\xbf\x32\x15\xee\xf2\xa7\x86\xbe\x59\x99\xc4\x10\x09\x8a\x6a\xaa\xd4\xd1\xd0\x71\xd2\x06\x1a\xdd\x2a\xa0\x08\xeb\x08\x6c\xfb\xd2\x2d\xfb\xaa\x72\x56\xeb\xd1\x92\x92\xe5\x0e\x95\x67\xf8\x38\xc3\xab\x59\x37\xe6\xfd\x42\xb0\xd0\x31\xd0\xcb\x8a\x66\xce\x2d\x53\x72\x1e\x72\xd3\x84\x25\xb0\xb8\x93\xd2\x61\x5b\x32\xd5\xe7\xe4\x0e\x31\x11\xaf\xdc\xb4\xb8\xee\xa4\x55\x16\x5f\x78\x86\x8b\x50\x4d\xc5\x6d\x6e\xfc\xe1\x6b\x06\x5b\x37\x84\x2a\x67\x95\x28\x00\xa4\xd1\x32\x9f\xbf\xe1\x64\xf8\x17\x47\xe1\xad\x8b\x72\xd2\xd9\x45\x5b\x73\x43\x3c\xe6\x21\xf7\x53\xa3\x73\xf9\x2a\xb0\xe9\x75\x5e\xa6\xbe\x9a\xad\xfc\xed\xb5\x46\x5b\x9f\xa9\x5a\x4f\xcb\xb6\x60\x96\x31\x91\x42\xca\xaf\xee\xa5\x0c\xe0\xab\x3e\x83\xb8\xac\x88\x10\x2c\x63\xd3\xc9\xd2\xf2\x44\xef\xea\x3d\x19\x24\x3c\x5b\xe7\x0c\x52\xfd\xfe\x47\x41\x14\xd5\x4c\x67\x8d\xdb\xe5\xd9\xfa\x67\x9c\x06\x31\x01\x92\xba\x96\xc4\x0d\xef\xf7\xc1\xe9\x23\x28\x0f\xae\x27\x9b\xff\x28\x0b\x3e\x85\x0c\xae\x02\xda\x27\xb6\x04\x51\x04\x43\x04\x99\x8c\xa3\x97\x1d\x84\xec\x55\x59\xfb\xf3\x84\xe5\xf8\x40\xf8\x5f\x81\x65\x92\x4c\x92\x7a\x07\x51\x8d\x6f\xff\x8d\x15\x36\x5c\x57\x7a\x5b\x3a\x63\x1c\x87\x65\xee\x54\xd5\x96\x50\x73\x1a\x9c\xff\x59\xe5\xea\x6f\x89\xd2\xbb\xa9\x6a\x12\x21\xf5\x08\x8e\x8a\xc0\xd8\xf5\x14\xe9\x9d\x7e\x99\x13\x88\x29\xa8\xb4\x22\x2a\x41\x7c\xc5\x10\xdf\x11\x5e\xf8\x8d\x0e\xd9\x98\xd5\xaf\xa8\xf9\x55\x1e\xe3\x29\xcd\x2c\x51\x7b\x8a\x8d\x52\xaa\x8b\x87\xae\x8e\xb2\xfa\x31\x27\x60\x90\xcb\x01\x6f\x7a\x79\x38\x04\x05\x7c\x11\x79\x10\x40\x33\x70\x75\xfd\x0b\x88\xa5\xcd\x35\xd8\xa6\x3b\xb0\x45\x82\x64\xd1\xb5\xdc\x06\xc9\x89\xf4\x16\x3e\xc7\xb3\xf1\x9d\xd3\xc5\xe3\xaf\xe8\x25\x86\x7a\x4a\xfd\x10\x5d\x20\xe5\x76\x5a\x22\x5f\x8f\xbc\xaa\x97\xee\xf2\xc2\x4c\x0e\xdc\x7b\xc4\xee\x53\xa3\xe0\xfa\xcd\x1e\x4e\x54\x1d\x5e\xe1\x51\x17\x1f\x1a\x75\x7f\xed\x12\xd7\xf7\xe3\x18\x56\x24\xcf\xc6\x96\x30\x77\x0d\x73\x98\x9c\x09\x69\xa3\xbc\x96\x5e\xaf\xde\x76\xa4\x66\x04\x6b\x36\x2a\xac\x6d\x37\xf8\x1e\xe1\x2a\x3e\x42\x2d\x1d\xe6\x46\xdd\x28\xb9\x08\x44\xa1\x9e\xb2\x22\x7a\x45\x8a\x37\x39\x74\xb4\xae\xc8\x3b\x40\xf7\xec\xbf\xfd\xe5\xde\xb2\x83\x5e\xa4\x46\x19\xa6\x9d\xb0\xe8\x76\x80\xbd\xc1\x80\x7a\xd9\xeb\xe7\x90\x5b\x81\x25\x21\xd9\x5b\x4a\x80\x48\x92\x71\x77\x04\xb2\xac\x05\xc9\xdf\x5e\x44\x5a\xae\x6e\xb3\xd8\x30\x5e\xdc\x77\x2f\x79\xc2\x8e\x8b\x28\x24\x06\x1b\x6f\x8d\x88\x53\x80\x55\x0c\x3a\x7b\x85\xb8\x96\x85\xe9\xf0\x57\x63\xfe\x32\x80\xff\x57\xc9\x3c\xdb\xf6\xcd\x67\x14\x47\x6c\x43\x3d\x6d\x48\x3f\x9c\x00\x60\x0e\xf5\x94\xe4\x52\x97\x86\xcd\xac\xbc\xe4\xe3\xe7\xee\xa2\x91\x6e\x92\xbb\xd1\x55\x0c\x5c\x0d\x63\xdb\x6b\xb8\x6e\x45\x48\x0f\xdf\x44\x48\xd2\xf5\xf7\x4d\x7b\xd4\x4d\xd3\xcd\xcd\x5b\x40\x60\xb1\xb2\x8e\xc9\x9a\x65\xc5\x06\x24\xcf\xe9\xcc\x5e\x2c\x49\x47\x38\x45\x5d\xc5\xc0\x0d\x8a\x07\x1c\xb3\xbb\xb1\x69\xf5\x6d\x0e\x9c\x96\x14\x93\x58\x0c\xc9\x48\x74\xfc\x35\xda\x7d\x4e\x32\x73\xa3\x77\x4a\x9e\xc5\xd1\x08\xfe\xa6\xa0\xf1\x66\x72\xea\xc7\xae\x21\x81\x0e\x8a\xba\x99\x06\x97\xfc\xc6\x2b\x69\x53\xc6\x67\xec\x5d\xa1\xfc\xa1\x3b\xdd\x2a\xd6\x8f\x31\xa7\x8d\xec\xfe\x0a\x3b\x6b\x39\x70\x70\x09\x72\x12\xbc\x84\x67\xca\xd2\x4a\x17\x33\x94\x45\x25\xc7\xfd\x1e\xa2\x4a\x9e\x27\x9d\xfb\x87\xea\xe4\xfd\xb0\x11\x06\x9d\x72\xb9\x1d\xea\x9b\x81\x2e\x6a\x36\x76\x62\xfa\xbe\x96\x67\x7d\x35\xdd\x5e\x5c\x4f\x41\x0d\xce\xdb\x13\xb0\x46\x89\x92\x45\x02\x39\x0f\xe6\xd1\x20\x96\x1c\x34\x00\x8c\xc9\xdf\xe3\xf0\xb6\x92\x3a\xda\x5c\x96\xd9\x0b\x7d\x57\xf5\x78\x11\xc0\xcf\xbf\xb0\x92\x3d\xe5\x6a\x67\x34\xce\xd9\x16\x08\xa0\x09\x42\x0b\x07\x13\x7c\x73\x0c\xc6\x50\x17\x42\xcf\xd9\x85\xd9\x23\x3c\xb1\x40\x40\x0f\x94\x20\xed\x2d\xbf\x10\x44\x6e\x64\x65\xe5\x1d\x5f\xec\x24\xd8\x4b\xe8\xc2\xfb\x06\x11\x24\x3f\xdf\x54\x2d\xe8\x4d\xc2\x1c\x27\x11\xb8\xb3\xd4",
-            b"\x6b\xa4\xec\x92",
-        ),
-    ]
-    vectors_incorrect = [
-        (
-            b"",
-            b"\x00\x00\x00\x00\x00",
-        ),
-        (
-            b"",
-            b"",
-        ),
-        (
-            b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
-            b"\x19\x0a\x55\xae",
-        ),
-        (
-            b"A",
-            b"\xe8\xb7\xbe\x43",
-        ),
-        (
-            b"abc ",
-            b"\x35\x24\x41\xc2",
-        ),
-        (
-            b"1234567890",
-            b"\xcb\xf4\x39\x26",
-        ),
-        (
-            b"1234567890123456789012345678901234567890123456789012345678901234567890123456789",
-            b"\x7c\xa9\x4a\x72",
-        ),
-    ]
-
-    def test_computation(self):
-        for data, chksum in self.vectors_correct:
-            self.assertEqual(checksum.compute(data), chksum)
-
-    def test_validation_correct(self):
-        for data, chksum in self.vectors_correct:
-            self.assertTrue(checksum.is_valid(chksum, data))
-
-    def test_validation_incorrect(self):
-        for data, chksum in self.vectors_incorrect:
-            self.assertFalse(checksum.is_valid(chksum, data))
-
-
-if __name__ == "__main__":
-    unittest.main()
diff --git a/core/tests/test_trezor.wire.thp.crypto.py b/core/tests/test_trezor.wire.thp.crypto.py
deleted file mode 100644
index fc87fc81..00000000
--- a/core/tests/test_trezor.wire.thp.crypto.py
+++ /dev/null
@@ -1,151 +0,0 @@
-# flake8: noqa: F403,F405
-from common import *  # isort:skip
-from trezorcrypto import aesgcm_encrypt, curve25519
-
-import storage
-
-if utils.USE_THP:
-    import thp_common
-    from trezor.wire.thp import crypto
-    from trezor.wire.thp.crypto import IV_1, IV_2, Handshake
-
-    def get_dummy_device_secret():
-        return b"\x01\x02\x03\x04\x05\x06\x07\x08\x01\x02\x03\x04\x05\x06\x07\x08"
-
-
-@unittest.skipUnless(utils.USE_THP, "only needed for THP")
-class TestTrezorHostProtocolCrypto(unittest.TestCase):
-    if utils.USE_THP:
-        handshake = Handshake()
-        key_1 = b"\x00\x01\x02\x03\x04\x05\x06\x07\x00\x01\x02\x03\x04\x05\x06\x07\x00\x01\x02\x03\x04\x05\x06\x07\x00\x01\x02\x03\x04\x05\x06\x07"
-        # 0:key, 1:nonce, 2:auth_data, 3:plaintext, 4:expected_ciphertext, 5:expected_tag
-        vectors_enc = [
-            (
-                key_1,
-                0,
-                b"\x55\x64",
-                b"\x00\x01\x02\x03\x04\05\x06\x07\x08\x09",
-                b"e2c9dd152fbee5821ea7",
-                b"10625812de81b14a46b9f1e5100a6d0c",
-            ),
-            (
-                key_1,
-                1,
-                b"\x55\x64",
-                b"\x00\x01\x02\x03\x04\05\x06\x07\x08\x09",
-                b"79811619ddb07c2b99f8",
-                b"71c6b872cdc499a7e9a3c7441f053214",
-            ),
-            (
-                key_1,
-                369,
-                b"\x55\x64",
-                b"\x00\x01\x02\x03\x04\05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f",
-                b"03bd030390f2dfe815a61c2b157a064f",
-                b"c1200f8a7ae9a6d32cef0fff878d55c2",
-            ),
-            (
-                key_1,
-                369,
-                b"\x55\x64\x73\x82\x91",
-                b"\x00\x01\x02\x03\x04\05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f",
-                b"03bd030390f2dfe815a61c2b157a064f",
-                b"693ac160cd93a20f7fc255f049d808d0",
-            ),
-        ]
-        # 0:chaining key, 1:input, 2:output_1, 3:output:2
-        vectors_hkdf = [
-            (
-                crypto.PROTOCOL_NAME,
-                b"\x01\x02",
-                b"c784373a217d6be057cddc6068e6748f255fc8beb6f99b7b90cbc64aad947514",
-                b"12695451e29bf08ffe5e4e6ab734b0c3d7cdd99b16cd409f57bd4eaa874944ba",
-            ),
-            (
-                b"\xc7\x84\x37\x3a\x21\x7d\x6b\xe0\x57\xcd\xdc\x60\x68\xe6\x74\x8f\x25\x5f\xc8\xbe\xb6\xf9\x9b\x7b\x90\xcb\xc6\x4a\xad\x94\x75\x14",
-                b"\x31\x41\x59\x26\x52\x12\x34\x56\x78\x89\x04\xaa",
-                b"f88c1e08d5c3bae8f6e4a3d3324c8cbc60a805603e399e69c4bf4eacb27c2f48",
-                b"5f0216bdb7110ee05372286974da8c9c8b96e2efa15b4af430755f462bd79a76",
-            ),
-        ]
-        vectors_iv = [
-            (0, b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"),
-            (1, b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01"),
-            (7, b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07"),
-            (1025, b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x01"),
-            (4294967295, b"\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff"),
-            (0xFFFFFFFFFFFFFFFF, b"\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff"),
-        ]
-
-    def test_encryption(self):
-        for v in self.vectors_enc:
-            buffer = bytearray(v[3])
-            tag = crypto.enc(buffer, v[0], v[1], v[2])
-            self.assertEqual(hexlify(buffer), v[4])
-            self.assertEqual(hexlify(tag), v[5])
-            self.assertTrue(crypto.dec(buffer, tag, v[0], v[1], v[2]))
-            self.assertEqual(buffer, v[3])
-
-    def test_hkdf(self):
-        for v in self.vectors_hkdf:
-            ck, k = crypto._hkdf(v[0], v[1])
-            self.assertEqual(hexlify(ck), v[2])
-            self.assertEqual(hexlify(k), v[3])
-
-    def test_iv_from_nonce(self):
-        for v in self.vectors_iv:
-            # x = v[0]
-            # y = x.to_bytes(8, "big")
-            iv = crypto._get_iv_from_nonce(v[0])
-            self.assertEqual(iv, v[1])
-        with self.assertRaises(AssertionError) as e:
-            iv = crypto._get_iv_from_nonce(0xFFFFFFFFFFFFFFFF + 1)
-        self.assertEqual(e.value.value, "Nonce overflow, terminate the channel")
-
-    def test_incorrect_vectors(self):
-        pass
-
-    def test_th1_crypto(self):
-        storage.device.get_device_secret = get_dummy_device_secret
-        handshake = self.handshake
-
-        host_ephemeral_private_key = curve25519.generate_secret()
-        host_ephemeral_public_key = curve25519.publickey(host_ephemeral_private_key)
-        handshake.handle_th1_crypto(b"", host_ephemeral_public_key, payload=b"\x00")
-
-    def test_th2_crypto(self):
-        handshake = self.handshake
-
-        host_static_private_key = curve25519.generate_secret()
-        host_static_public_key = curve25519.publickey(host_static_private_key)
-        aes_ctx = aesgcm_encrypt(handshake.k, IV_2)
-        aes_ctx.auth(handshake.h)
-        encrypted_host_static_public_key = bytearray(
-            aes_ctx.encrypt(host_static_public_key) + aes_ctx.finish()
-        )
-
-        # Code to encrypt Host's noise encrypted payload correctly:
-        protomsg = bytearray(b"\x10\x02\x10\x03")
-        temp_k = handshake.k
-        temp_h = handshake.h
-
-        temp_h = crypto._hash_of_two(temp_h, encrypted_host_static_public_key)
-        _, temp_k = crypto._hkdf(
-            handshake.ck,
-            curve25519.multiply(
-                handshake.trezor_ephemeral_private_key, host_static_public_key
-            ),
-        )
-        aes_ctx = aesgcm_encrypt(temp_k, IV_1)
-        aes_ctx.encrypt_in_place(protomsg)
-        aes_ctx.auth(temp_h)
-        tag = aes_ctx.finish()
-        encrypted_payload = bytearray(protomsg + tag)
-        # end of encrypted payload generation
-
-        handshake.handle_th2_crypto(encrypted_host_static_public_key, encrypted_payload)
-        self.assertEqual(encrypted_payload[:4], b"\x10\x02\x10\x03")
-
-
-if __name__ == "__main__":
-    unittest.main()
diff --git a/core/tests/test_trezor.wire.thp.writer.py b/core/tests/test_trezor.wire.thp.writer.py
deleted file mode 100644
index 36b2d11b..00000000
--- a/core/tests/test_trezor.wire.thp.writer.py
+++ /dev/null
@@ -1,183 +0,0 @@
-# flake8: noqa: F403,F405
-from common import *  # isort:skip
-
-from typing import Any, Awaitable
-
-if utils.USE_THP:
-    import thp_common
-    from mock_wire_interface import MockHID
-    from trezor.loop import Timeout, race
-    from trezor.wire.thp import ENCRYPTED, PacketHeader
-    from trezor.wire.thp import alternating_bit_protocol as ABP
-    from trezor.wire.thp.channel import _MAX_RETRANSMISSION_COUNT
-    from trezor.wire.thp.interface_context import ThpContext
-
-
-@unittest.skipUnless(utils.USE_THP, "only needed for THP")
-class TestTrezorHostProtocolWriter(unittest.TestCase):
-    short_payload_expected = b"04123400050700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
-    longer_payload_expected = [
-        b"0412340100000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a",
-        b"8012343b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f7071727374757677",
-        b"80123478797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4",
-        b"801234b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1",
-        b"801234f2f3f4f5f6f7f8f9fafbfcfdfeff0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
-    ]
-    eight_longer_payloads_expected = [
-        b"0412340800000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a",
-        b"8012343b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f7071727374757677",
-        b"80123478797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4",
-        b"801234b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1",
-        b"801234f2f3f4f5f6f7f8f9fafbfcfdfeff000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e",
-        b"8012342f303132333435363738393a3b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b",
-        b"8012346c6d6e6f707172737475767778797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8",
-        b"801234a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5",
-        b"801234e6e7e8e9eaebecedeeeff0f1f2f3f4f5f6f7f8f9fafbfcfdfeff000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122",
-        b"801234232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f",
-        b"801234606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c",
-        b"8012349d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9",
-        b"801234dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1f2f3f4f5f6f7f8f9fafbfcfdfeff000102030405060708090a0b0c0d0e0f10111213141516",
-        b"8012341718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f404142434445464748494a4b4c4d4e4f50515253",
-        b"8012345455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f90",
-        b"8012349192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccd",
-        b"801234cecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1f2f3f4f5f6f7f8f9fafbfcfdfeff000102030405060708090a",
-        b"8012340b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f4041424344454647",
-        b"80123448494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f8081828384",
-        b"80123485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1",
-        b"801234c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1f2f3f4f5f6f7f8f9fafbfcfdfe",
-        b"801234ff000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a3b",
-        b"8012343c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f707172737475767778",
-        b"801234797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5",
-        b"801234b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1f2",
-        b"801234f3f4f5f6f7f8f9fafbfcfdfeff000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f",
-        b"801234303132333435363738393a3b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c",
-        b"8012346d6e6f707172737475767778797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9",
-        b"801234aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6",
-        b"801234e7e8e9eaebecedeeeff0f1f2f3f4f5f6f7f8f9fafbfcfdfeff000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20212223",
-        b"8012342425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f60",
-        b"8012346162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d",
-        b"8012349e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9da",
-        b"801234dbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1f2f3f4f5f6f7f8f9fafbfcfdfeff000000000000000000000000000000000000000000000000",
-    ]
-    empty_payload_with_checksum_expected = b"0412340004edbd479c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
-    longer_payload_with_checksum_expected = [
-        b"0412340100000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a",
-        b"8012343b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f7071727374757677",
-        b"80123478797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4",
-        b"801234b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1",
-        b"801234f2f3f4f5f6f7f8f9fafbfcfdfefff40c65ee00000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
-    ]
-
-    def await_until_result(self, task: Awaitable) -> Any:
-        with self.assertRaises(StopIteration):
-            while True:
-                task.send(None)
-
-    def setUp(self):
-        self.interface = MockHID()
-        thp_ctx = ThpContext(self.interface)
-        (self.ctx,) = thp_ctx._iface_ctxs
-
-    def test_write_empty_payload(self):
-        header = PacketHeader(ENCRYPTED, 4660, 4)
-        await_result(self.ctx._write_payload_chunks(header, b""))
-        self.assertEqual(len(self.interface.data), 0)
-
-    def test_write_short_payload(self):
-        header = PacketHeader(ENCRYPTED, 4660, 5)
-        data = b"\x07"
-        self.await_until_result(self.ctx._write_payload_chunks(header, data))
-        self.assertEqual(hexlify(self.interface.data[0]), self.short_payload_expected)
-
-    def test_write_longer_payload(self):
-        data = bytearray(range(256))
-        header = PacketHeader(ENCRYPTED, 4660, 256)
-        self.await_until_result(self.ctx._write_payload_chunks(header, data))
-
-        for i in range(len(self.longer_payload_expected)):
-            self.assertEqual(
-                hexlify(self.interface.data[i]), self.longer_payload_expected[i]
-            )
-
-    def test_write_eight_longer_payloads(self):
-        data = bytearray(range(256))
-        header = PacketHeader(ENCRYPTED, 4660, 2048)
-        chunks = [data] * 8
-        self.await_until_result(self.ctx._write_payload_chunks(header, *chunks))
-
-        for i in range(len(self.eight_longer_payloads_expected)):
-            self.assertEqual(
-                hexlify(self.interface.data[i]), self.eight_longer_payloads_expected[i]
-            )
-
-    def test_write_empty_payload_with_checksum(self):
-        header = PacketHeader(ENCRYPTED, 4660, 4)
-        self.await_until_result(self.ctx.write_payload(header, b""))
-
-        self.assertEqual(
-            hexlify(self.interface.data[0]), self.empty_payload_with_checksum_expected
-        )
-
-    def test_write_longer_payload_with_checksum(self):
-        data = bytearray(range(256))
-        header = PacketHeader(ENCRYPTED, 4660, 256)
-        self.await_until_result(self.ctx.write_payload(header, data))
-
-        for i in range(len(self.longer_payload_with_checksum_expected)):
-            self.assertEqual(
-                hexlify(self.interface.data[i]),
-                self.longer_payload_with_checksum_expected[i],
-            )
-
-    def test_write_timeout(self):
-        channel = thp_common.get_new_channel(self.interface)
-        seq_bit = ABP.get_send_seq_bit(channel.channel_cache)
-
-        task = channel.write_encrypted_payload(ENCRYPTED, b"PAYLOAD")
-        race_obj = task.send(None)  # start the generator
-        assert isinstance(race_obj, race)
-        _wait_for_ack, write_loop = race_obj.children
-        write_loop.send(None)  # start the generator
-
-        for _ in range(_MAX_RETRANSMISSION_COUNT - 1):
-            write_loop.send(None)  # complete write
-            write_loop.send(None)  # complete sleep
-
-        write_loop.send(None)  # complete write last time
-        with self.assertRaises(Timeout) as ctx:
-            write_loop.send(None)  # complete sleep & raise Timeout
-
-        with self.assertRaises(Timeout):
-            task.throw(ctx.value)  # re-raise timeout in `write_encrypted_payload`
-
-        # next write should use the next `seq_bit` (see #6138)
-        self.assertNotEqual(ABP.get_send_seq_bit(channel.channel_cache), seq_bit)
-
-    def test_write_blocked(self):
-        channel = thp_common.get_new_channel(self.interface)
-        seq_bit = ABP.get_send_seq_bit(channel.channel_cache)
-
-        task = channel.write_encrypted_payload(ENCRYPTED, b"PAYLOAD")
-        race_obj = task.send(None)  # start the generator
-        assert isinstance(race_obj, race)
-        _wait_for_ack, write_loop = race_obj.children
-        write_loop.send(None)  # start the generator
-
-        # Re-transmit a few times
-        for _ in range(3):
-            write_loop.send(None)  # complete write
-            write_loop.send(None)  # complete sleep
-
-        with self.assertRaises(Timeout) as ctx:
-            # timeout `_write_payload_once()` (as if `loop.sleep` has completed) using dummy "ticks" integer value
-            write_loop.send(12345)
-
-        with self.assertRaises(Timeout):
-            task.throw(ctx.value)  # re-raise timeout in `write_encrypted_payload`
-
-        # next write should use the next `seq_bit` (see #6138)
-        self.assertNotEqual(ABP.get_send_seq_bit(channel.channel_cache), seq_bit)
-
-
-if __name__ == "__main__":
-    unittest.main()
diff --git a/core/tests/thp_common.py b/core/tests/thp_common.py
index fb244791..a69e7ce1 100644
--- a/core/tests/thp_common.py
+++ b/core/tests/thp_common.py
@@ -9,8 +9,7 @@ if utils.USE_THP:
     from storage import cache_thp
     from trezor.wire import context
     from trezor.wire.thp.channel import Channel
-    from trezor.wire.thp.channel_manager import create_new_channel
-    from trezor.wire.thp.interface_context import ThpContext
+    from trezor.wire.thp.interface_context import InterfaceContext, ThpContext
     from trezor.wire.thp.memory_manager import ThpBuffer
     from trezor.wire.thp.session_context import SessionContext
 
@@ -20,19 +19,36 @@ if utils.USE_THP:
         from trezor import protobuf
         from trezor.wire import WireInterface
 
+    class MockChannel:
+        def __init__(
+            self,
+            channel_id: int,
+            iface_ctx: InterfaceContext,
+            buffers: tuple[ThpBuffer, ThpBuffer],
+        ) -> None:
+            self.iface = iface_ctx
+            self.channel_id = channel_id
+
+        def channel_id_bytes(self):
+            return self.channel_id.to_bytes(2, "big")
+
+    NEXT_CHANNEL_ID = 0x0FFF
+
     def create_context() -> SessionContext:
         mock_iface = MockHID()
         channel = get_new_channel(mock_iface)
         session_cache = cache_thp.create_or_replace_session(
-            channel.channel_cache, session_id=b"\x01"
+            channel.channel_id_bytes(), session_id=b"\x01"
         )
         return SessionContext(channel, session_cache)
 
-    def get_new_channel(iface: WireInterface) -> Channel:
-        channel_cache = create_new_channel(iface)
+    def get_new_channel(iface: WireInterface) -> MockChannel:
+        global NEXT_CHANNEL_ID
+        # channel_cache = create_new_channel(iface)
         thp_ctx = ThpContext(iface)
         (iface_ctx,) = thp_ctx._iface_ctxs
-        return Channel(channel_cache, iface_ctx, (ThpBuffer(), ThpBuffer()))
+        NEXT_CHANNEL_ID += 1
+        return MockChannel(NEXT_CHANNEL_ID, iface_ctx, (ThpBuffer(), ThpBuffer()))
 
     def _encrypt_patch() -> patch:
         return patch(Channel, "_encrypt", lambda self, buffer, noise_payload_len: None)
diff --git a/python/src/trezorlib/debuglink.py b/python/src/trezorlib/debuglink.py
index 230fde68..1b1271af 100644
--- a/python/src/trezorlib/debuglink.py
+++ b/python/src/trezorlib/debuglink.py
@@ -1485,6 +1485,9 @@ class TrezorTestContext:
         if self.is_thp():
             assert isinstance(self.client, TrezorClientThp)
             self.client.channel = channel
+            # destroy the old interactive context associated with the previous channel
+            # tests in device_tests/thp/test_handshake.py don't work without it
+            self.client._interact_ctx = self.client._interact()
             return
         raise AttributeError("Channel is not available for this protocol")
 
diff --git a/tests/device_tests/thp/test_basic.py b/tests/device_tests/thp/test_basic.py
index f735b00f..080e64ed 100644
--- a/tests/device_tests/thp/test_basic.py
+++ b/tests/device_tests/thp/test_basic.py
@@ -1,11 +1,12 @@
 import pytest
 
-from trezorlib import messages, protocol_v1
+from trezorlib import btc, messages, protocol_v1
 from trezorlib.debuglink import TrezorTestContext as Client
 from trezorlib.mapping import DEFAULT_MAPPING
 from trezorlib.thp import control_byte, thp_io
 from trezorlib.thp.exceptions import ThpErrorCode
 from trezorlib.thp.message import Message
+from trezorlib.tools import parse_path
 from trezorlib.transport import Timeout, Transport
 
 pytestmark = [
@@ -48,3 +49,25 @@ def test_v2_unallocated(client: Client):
     assert response.cid == 0x789A
     assert response.ctrl_byte == control_byte.ERROR
     assert response.data == bytes([ThpErrorCode.UNALLOCATED_CHANNEL])
+
+
+@pytest.mark.setup_client(uninitialized=False)
+def test_message_length(test_ctx: Client):
+    # slightly under _PROTOBUF_BUFFER_SIZE, should pass
+    test_ctx.channel.BUSY_RETRIES = 1
+    session = test_ctx.get_session()
+    btc.sign_message(
+        session,
+        coin_name="Bitcoin",
+        n=parse_path("m/44h/0h/0h/0/0"),
+        message=("u" * 8_600),
+    )
+
+    # slightly over _PROTOBUF_BUFFER_SIZE, should time out
+    with pytest.raises(Timeout):
+        btc.sign_message(
+            session,
+            coin_name="Bitcoin",
+            n=parse_path("m/44h/0h/0h/0/0"),
+            message=("u" * 8_700),
+        )
diff --git a/tests/device_tests/thp/test_multiple_hosts.py b/tests/device_tests/thp/test_multiple_hosts.py
index 629eb184..50790556 100644
--- a/tests/device_tests/thp/test_multiple_hosts.py
+++ b/tests/device_tests/thp/test_multiple_hosts.py
@@ -3,10 +3,13 @@ import time
 
 import pytest
 
-from trezorlib.debuglink import TrezorTestContext as Client
+from trezorlib.debuglink import TrezorTestContext
 from trezorlib.thp.channel import Channel
 from trezorlib.thp.exceptions import ThpError, ThpErrorCode
 
+from .connect import prepare_channel_for_pairing
+
+Client = TrezorTestContext
 pytestmark = [pytest.mark.protocol("thp")]
 
 
@@ -17,18 +20,30 @@ def _new_channel(client) -> Channel:
 
 
 def test_concurrent_handshakes(client: Client) -> None:
-    channel_1 = _new_channel(client)
-    channel_2 = _new_channel(client)
+    MAX = 4  # See `MAX_CHANNELS_OPENING` in core/embed/rust/src/thp/mod.rs
+    channels = []
+
+    # Start the handshake for MAX+1 channels
+    for _ in range(MAX + 1):
+        channel = _new_channel(client)
+        channel.BUSY_RETRIES = 0
+        channel._send_handshake_init_request(unlock=False)
+        channel._read_handshake_init_response()
+        channels.append(channel)
+
+    # Oldest handshake is forgotten
+    with pytest.raises(ThpError) as err:
+        channels[0]._send_handshake_completion_request([])
+        channels[0]._read_handshake_completion_response()
+    assert err.value.code == ThpErrorCode.UNALLOCATED_CHANNEL
 
-    # The first host starts handshake
-    channel_1._send_handshake_init_request(unlock=False)
-    channel_1._read_handshake_init_response()
+    # Others finish successfully
+    for channel in channels[1:]:
+        channel._send_handshake_completion_request([])
+        channel._read_handshake_completion_response()
+        channel._flush_ack()
 
-    channel_2.BUSY_RETRIES = 0
-    with pytest.raises(ThpError) as err:
-        # The second host should not be able to interrupt the first host's handshake immediately
-        channel_2.open([])
-    assert err.value.code == ThpErrorCode.TRANSPORT_BUSY
+    assert all(channel.is_open() for channel in channels[1:])
 
 
 def test_concurrent_handshakes_busy_retries(client: Client) -> None:
@@ -60,3 +75,30 @@ def test_concurrent_handshakes_busy_retries(client: Client) -> None:
     # both channels should be open
     assert channel_1.is_open()
     assert channel_2.is_open()
+
+
+def test_concurrent_channels(test_ctx: TrezorTestContext) -> None:
+    MAX = 10  # See `MAX_CHANNELS_APPDATA` in core/embed/rust/src/thp/mod.rs
+    channels = []
+
+    # Open MAX+1 channels
+    for _ in range(MAX + 1):
+        pairing = prepare_channel_for_pairing(test_ctx)
+        pairing.skip()
+        pairing.finish()
+        channels.append(pairing.client.channel)
+
+    # Oldest channel gets evicted
+    with pytest.raises(ThpError) as err:
+        test_ctx.channel = channels[0]
+        test_ctx.ping("will raise")
+    assert err.value.code == ThpErrorCode.UNALLOCATED_CHANNEL
+
+    # Others keep working
+    for channel in channels[1:]:
+        test_ctx.channel = channel
+        test_ctx.ping("hi")
+
+    for channel in channels[1:]:
+        test_ctx.channel = channel
+        test_ctx.ping("hi2")