feat(core): allow preempting stale THP channels

diff --git a/core/src/apps/thp/pairing.py b/core/src/apps/thp/pairing.py
index 77acb123b..61f01b608 100644
--- a/core/src/apps/thp/pairing.py
+++ b/core/src/apps/thp/pairing.py
@@ -143,7 +143,8 @@ async def handle_pairing_request(
             # Should raise UnexpectedMessageException
             result = await ctx.show_pairing_method_screen()
         except UnexpectedMessageException as e:
-            raw_response = e.msg
+            if (raw_response := e.msg) is None:
+                raise  # propagate stale channel preemption
             req_type = protobuf.type_for_wire(
                 ctx.message_type_enum_name, raw_response.type
             )
diff --git a/core/src/trezor/wire/__init__.py b/core/src/trezor/wire/__init__.py
index 99cb8a4d3..a2565d25b 100644
--- a/core/src/trezor/wire/__init__.py
+++ b/core/src/trezor/wire/__init__.py
@@ -104,10 +104,16 @@ if utils.USE_THP:
         if __debug__:
             _THP_CHANNELS.append(ctx._channels)
         try:
-            channel = await ctx.get_next_message()
+            while (channel := await ctx.get_next_message()) is None:
+                if __debug__:
+                    # happens if another interface is active and using THP buffers.
+                    log.error(__name__, "Another interface is active", iface=iface)
+
             while await received_message_handler.handle_received_message(channel):
                 pass
         finally:
+            if __debug__:
+                log.debug(__name__, "Finished THP session", iface=iface)
             # Wait for all active workflows to finish.
             await workflow.join_all()
             if __debug__:
diff --git a/core/src/trezor/wire/context.py b/core/src/trezor/wire/context.py
index 96598f0c0..8fc01e14f 100644
--- a/core/src/trezor/wire/context.py
+++ b/core/src/trezor/wire/context.py
@@ -40,9 +40,11 @@ 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:
+    def __init__(self, msg: Message | None) -> None:
         super().__init__()
         self.msg = msg
 
diff --git a/core/src/trezor/wire/thp/channel.py b/core/src/trezor/wire/thp/channel.py
index 01676bb20..9d90fcf3c 100644
--- a/core/src/trezor/wire/thp/channel.py
+++ b/core/src/trezor/wire/thp/channel.py
@@ -1,4 +1,5 @@
 import ustruct
+import utime
 from micropython import const
 from typing import TYPE_CHECKING
 
@@ -22,6 +23,7 @@ from storage.cache_thp import (
 )
 from trezor import protobuf, utils, workflow
 from trezor.loop import Timeout, race, sleep
+from trezor.wire.context import UnexpectedMessageException
 
 from ..protocol_common import Message
 from . import (
@@ -63,6 +65,10 @@ _MIN_RETRANSMISSION_COUNT = const(2)
 _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)
+
 
 class Reassembler:
     def __init__(self, cid: int, read_buf: ThpBuffer) -> None:
@@ -134,6 +140,13 @@ def verify_checksum(buffer: memoryview) -> memoryview | None:
     return None
 
 
+class ChannelPreemptedException(UnexpectedMessageException):
+    """Raising this exception should restart the event loop."""
+
+    def __init__(self) -> None:
+        super().__init__(msg=None)
+
+
 class Channel:
     """
     THP protocol encrypted communication channel.
@@ -158,6 +171,7 @@ class Channel:
         # Shared variables
         self.sessions: dict[int, GenericSessionContext] = {}
         self.reassembler = Reassembler(self.get_channel_id_int(), self.read_buf)
+        self.last_write_ms: int = utime.ticks_ms()
 
         # Temporary objects
         self.credential: ThpPairingCredential | None = None
@@ -275,6 +289,10 @@ class Channel:
         while self.reassembler.message is None:
             # receive and reassemble a new message from this channel
             channel = await self.ctx.get_next_message(timeout_ms=timeout_ms)
+            if channel is None:
+                self.preempt_if_stale()
+                continue
+
             if channel is self:
                 break
 
@@ -378,6 +396,15 @@ class Channel:
     ) -> Awaitable[None]:
         return self.write_encrypted_payload(ctrl_byte, payload)
 
+    def preempt_if_stale(self) -> None:
+        elapsed_ms = utime.ticks_diff(utime.ticks_ms(), self.last_write_ms)
+        preempt = elapsed_ms > _PREEMPT_TIMEOUT_MS
+        if __debug__:
+            logger = log.error if preempt else log.warning
+            self._log(f"Interrupted channel after {elapsed_ms} ms", logger=logger)
+        if preempt:
+            raise ChannelPreemptedException
+
     async def write_encrypted_payload(self, ctrl_byte: int, payload: bytes) -> None:
         if __debug__:
             self._log("write_encrypted_payload_loop")
@@ -392,6 +419,9 @@ class Channel:
         # 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()
+
         for i in range(_MAX_RETRANSMISSION_COUNT):
             result = await race(self.ctx.write_payload(header, payload), _WRITE_TIMEOUT)
             if isinstance(result, int):
diff --git a/core/src/trezor/wire/thp/interface_context.py b/core/src/trezor/wire/thp/interface_context.py
index 819cd0934..22bbc7f26 100644
--- a/core/src/trezor/wire/thp/interface_context.py
+++ b/core/src/trezor/wire/thp/interface_context.py
@@ -47,7 +47,7 @@ class ThpContext:
         self._write = loop.wait(iface.iface_num() | io.POLL_WRITE)
         self._channels: dict[int, Channel] = {}
 
-    async def get_next_message(self, timeout_ms: int | None = None) -> Channel:
+    async def get_next_message(self, timeout_ms: int | None = None) -> Channel | None:
         """
         Reassemble a valid THP payload and return its channel.
 
@@ -82,7 +82,8 @@ class ThpContext:
                 if (buffers := THP_BUFFERS_PROVIDER.take()) is None:
                     # concurrent payload reassembly is not supported
                     await self.write_error(cid, ThpErrorType.TRANSPORT_BUSY)
-                    continue
+                    return None  # try to preempt the caller (if stale)
+
                 channel = self._channels[cid] = Channel(cache, self, buffers)
 
             if channel.reassemble(packet):
diff --git a/core/src/trezor/wire/thp/session_context.py b/core/src/trezor/wire/thp/session_context.py
index 1dcef8a73..7714feaf7 100644
--- a/core/src/trezor/wire/thp/session_context.py
+++ b/core/src/trezor/wire/thp/session_context.py
@@ -55,10 +55,11 @@ class GenericSessionContext(Context):
                     log.exception(__name__, e, iface=self.iface)
                 await self.write(failure(e))
             except UnexpectedMessageException as unexpected:
-                # The workflow was interrupted by an unexpected message. We need to
-                # process it as if it was a new message...
-                message = unexpected.msg
-                continue
+                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
             except Exception as exc:
                 if __debug__:
                     log.exception(__name__, exc, iface=self.iface)