refactor(python): rework session-based API

diff --git a/core/emu.py b/core/emu.py
index 445f0106..b6ec27e9 100755
--- a/core/emu.py
+++ b/core/emu.py
@@ -65,14 +65,30 @@ def watch_emulator(emulator: CoreEmulator) -> int:
     return 0
 
 
-def run_debugger(emulator: CoreEmulator, gdb_script_file: str | Path | None, valgrind: bool = False, run_command: list[str] = []) -> None:
+def run_debugger(
+    emulator: CoreEmulator,
+    gdb_script_file: str | Path | None,
+    valgrind: bool = False,
+    run_command: list[str] = [],
+) -> None:
     os.chdir(emulator.workdir)
     env = emulator.make_env()
     if valgrind:
-        dbg_command = ["valgrind", "-v", "--tool=callgrind", "--read-inline-info=yes", str(emulator.executable)] + emulator.make_args()
+        dbg_command = [
+            "valgrind",
+            "-v",
+            "--tool=callgrind",
+            "--read-inline-info=yes",
+            str(emulator.executable),
+        ] + emulator.make_args()
     elif platform.system() == "Darwin":
         env["PATH"] = "/usr/bin"
-        dbg_command = ["lldb", "-f", str(emulator.executable), "--"] + emulator.make_args()
+        dbg_command = [
+            "lldb",
+            "-f",
+            str(emulator.executable),
+            "--",
+        ] + emulator.make_args()
     else:
         # Optionally run a gdb script from a file
         if gdb_script_file is None:
@@ -286,10 +302,9 @@ def cli(
             label = "Emulator"
 
         assert emulator.client is not None
-        trezorlib.device.wipe(emulator.client.get_seedless_session())
-
+        emulator.client.wipe_device()
         trezorlib.debuglink.load_device(
-            emulator.client.get_seedless_session(),
+            emulator.client.get_session(passphrase=None),
             mnemonics,
             pin=None,
             passphrase_protection=False,
diff --git a/python/pyproject.toml b/python/pyproject.toml
index 24bfa61a..10558802 100644
--- a/python/pyproject.toml
+++ b/python/pyproject.toml
@@ -31,6 +31,8 @@ dependencies = [
     "construct-classes>=0.1.2",
     "cryptography>=41",
     "noiseprotocol>=0.3.1,<0.4.0",
+    "platformdirs>=4.4.0",
+    "keyring>=25.7.0",
 ]
 
 [project.optional-dependencies]
diff --git a/python/pyrightconfig.json b/python/pyrightconfig.json
index 397dc629..b84084b5 100644
--- a/python/pyrightconfig.json
+++ b/python/pyrightconfig.json
@@ -5,7 +5,7 @@
         "helper-scripts"
     ],
     "stubPath": "./stubs",
-    "pythonVersion": "3.8",
+    "pythonVersion": "3.9",
     "typeCheckingMode": "basic",
     "reportMissingImports": false,
     "reportUntypedFunctionDecorator": true,
diff --git a/python/src/trezorlib/_internal/emulator.py b/python/src/trezorlib/_internal/emulator.py
index b06fdc96..0ab7cf32 100644
--- a/python/src/trezorlib/_internal/emulator.py
+++ b/python/src/trezorlib/_internal/emulator.py
@@ -14,6 +14,8 @@
 # You should have received a copy of the License along with this library.
 # If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
 
+from __future__ import annotations
+
 import atexit
 import logging
 import os
@@ -23,7 +25,7 @@ import time
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional, Sequence, TextIO, Union, cast
 
-from ..debuglink import DebugLinkNotFound, TrezorClientDebugLink
+from ..debuglink import DebugLinkNotFound, TrezorTestContext
 from ..transport import Transport
 from ..transport.udp import UdpTransport
 
@@ -181,8 +183,8 @@ class Emulator:
             self.logfile = self.profile_dir / "trezor.log"
 
         # Using `client` property instead to assert `not None`
-        self._client: Optional[TrezorClientDebugLink] = None
-        self.process: Optional[subprocess.Popen] = None
+        self._client: TrezorTestContext | None = None
+        self.process: subprocess.Popen | None = None
 
         self.port = 21324
         self.headless = headless
@@ -200,15 +202,13 @@ class Emulator:
         pass
 
     @property
-    def client(self) -> TrezorClientDebugLink:
+    def client(self) -> TrezorTestContext:
         """So that type-checkers do not see `client` as `Optional`.
 
         (it is not None between `start()` and `stop()` calls)
         """
         if self._client is None:
             raise RuntimeError
-        if self._client.is_invalidated:
-            self._client = self._client.get_new_client()
         return self._client
 
     def make_args(self) -> List[str]:
@@ -227,7 +227,7 @@ class Emulator:
         start = time.monotonic()
         try:
             while True:
-                if self.transport.ping():
+                if self.transport.is_ready():
                     break
                 if self.process.poll() is not None:
                     raise RuntimeError("Emulator process died")
@@ -300,10 +300,9 @@ class Emulator:
         (self.profile_dir / "trezor.port").write_text(str(self.port) + "\n")
 
         try:
-            self._client = TrezorClientDebugLink(
-                self.transport,
+            self._client = TrezorTestContext(
+                transport=self.transport,
                 auto_interact=self.auto_interact,
-                open_transport=True,
                 debug_transport=debug_transport,
             )
         except DebugLinkNotFound as e:
@@ -312,7 +311,7 @@ class Emulator:
 
     def stop(self) -> None:
         if self._client:
-            self._client.close_transport()
+            self._client.transport.close()
         self._client = None
 
         if self.process:
diff --git a/python/src/trezorlib/authentication.py b/python/src/trezorlib/authentication.py
index b0d91594..91059b75 100644
--- a/python/src/trezorlib/authentication.py
+++ b/python/src/trezorlib/authentication.py
@@ -27,7 +27,8 @@ from cryptography.hazmat.primitives.asymmetric import ec, ed25519, utils
 from cryptography.x509.oid import NameOID, ObjectIdentifier, SignatureAlgorithmOID
 
 from . import device
-from .transport.session import Session
+from .client import Session
+from .tools import workflow
 
 LOG = logging.getLogger(__name__)
 
@@ -476,6 +477,7 @@ def verify_authentication_response(
     return root
 
 
+@workflow()
 def authenticate_device(
     session: Session,
     challenge: bytes | None = None,
diff --git a/python/src/trezorlib/benchmark.py b/python/src/trezorlib/benchmark.py
index a6c58ff0..1f2f73d0 100644
--- a/python/src/trezorlib/benchmark.py
+++ b/python/src/trezorlib/benchmark.py
@@ -17,17 +17,20 @@
 from typing import TYPE_CHECKING
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 
+@workflow()
 def list_names(
     session: "Session",
 ) -> messages.BenchmarkNames:
     return session.call(messages.BenchmarkListNames(), expect=messages.BenchmarkNames)
 
 
+@workflow()
 def run(session: "Session", name: str) -> messages.BenchmarkResult:
     return session.call(
         messages.BenchmarkRun(name=name), expect=messages.BenchmarkResult
diff --git a/python/src/trezorlib/ble.py b/python/src/trezorlib/ble.py
index b469d567..df39fa12 100644
--- a/python/src/trezorlib/ble.py
+++ b/python/src/trezorlib/ble.py
@@ -17,10 +17,12 @@
 import typing as t
 
 from . import messages
+from .tools import workflow
 
 if t.TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 
+@workflow(capability=messages.Capability.BLE)
 def unpair(session: "Session", all: bool) -> None:
     session.call(messages.BleUnpair(all=all), expect=messages.Success)
diff --git a/python/src/trezorlib/btc.py b/python/src/trezorlib/btc.py
index 3d5a72b7..b15f120e 100644
--- a/python/src/trezorlib/btc.py
+++ b/python/src/trezorlib/btc.py
@@ -25,11 +25,11 @@ from typing import TYPE_CHECKING, Any, AnyStr, List, Optional, Sequence, Tuple
 from typing_extensions import Protocol, TypedDict
 
 from . import exceptions, messages
-from .tools import _return_success, prepare_message_bytes
+from .tools import prepare_message_bytes, workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
     class ScriptSig(TypedDict):
         asm: str
@@ -104,6 +104,7 @@ def from_json(json_dict: "Transaction") -> messages.TransactionType:
     )
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def get_public_node(
     session: "Session",
     n: "Address",
@@ -138,6 +139,7 @@ def get_address(*args: Any, **kwargs: Any) -> str:
     return get_authenticated_address(*args, **kwargs).address
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def get_authenticated_address(
     session: "Session",
     coin_name: str,
@@ -170,6 +172,7 @@ def get_authenticated_address(
     )
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def get_ownership_id(
     session: "Session",
     coin_name: str,
@@ -188,6 +191,7 @@ def get_ownership_id(
     ).ownership_id
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def get_ownership_proof(
     session: "Session",
     coin_name: str,
@@ -218,6 +222,7 @@ def get_ownership_proof(
     return res.ownership_proof, res.signature
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def sign_message(
     session: "Session",
     coin_name: str,
@@ -264,6 +269,7 @@ def verify_message(
         return False
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def sign_tx(
     session: "Session",
     coin_name: str,
@@ -418,6 +424,7 @@ def sign_tx(
     return signatures, serialized_tx
 
 
+@workflow(capability=messages.Capability.Bitcoin)
 def authorize_coinjoin(
     session: "Session",
     coordinator: str,
@@ -427,8 +434,8 @@ def authorize_coinjoin(
     n: "Address",
     coin_name: str,
     script_type: messages.InputScriptType = messages.InputScriptType.SPENDADDRESS,
-) -> str | None:
-    resp = session.call(
+) -> None:
+    session.call(
         messages.AuthorizeCoinJoin(
             coordinator=coordinator,
             max_rounds=max_rounds,
@@ -440,4 +447,3 @@ def authorize_coinjoin(
         ),
         expect=messages.Success,
     )
-    return _return_success(resp)
diff --git a/python/src/trezorlib/cardano.py b/python/src/trezorlib/cardano.py
index 467e76c0..e789ac39 100644
--- a/python/src/trezorlib/cardano.py
+++ b/python/src/trezorlib/cardano.py
@@ -35,7 +35,7 @@ from . import messages as m
 from . import tools
 
 if TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 PROTOCOL_MAGICS = {
     "mainnet": 764824073,
@@ -840,6 +840,7 @@ def get_address(*args: Any, **kwargs: Any) -> str:
     return get_authenticated_address(*args, **kwargs).address
 
 
+@tools.workflow(capability=m.Capability.Cardano)
 def get_authenticated_address(
     session: "Session",
     address_parameters: m.CardanoAddressParametersType,
@@ -862,6 +863,7 @@ def get_authenticated_address(
     )
 
 
+@tools.workflow(capability=m.Capability.Cardano)
 def get_public_key(
     session: "Session",
     address_n: List[int],
@@ -878,6 +880,7 @@ def get_public_key(
     )
 
 
+@tools.workflow(capability=m.Capability.Cardano)
 def get_native_script_hash(
     session: "Session",
     native_script: m.CardanoNativeScript,
@@ -894,6 +897,7 @@ def get_native_script_hash(
     )
 
 
+@tools.workflow(capability=m.Capability.Cardano)
 def sign_tx(
     session: "Session",
     signing_mode: m.CardanoTxSigningMode,
@@ -1019,6 +1023,7 @@ def sign_tx(
     return sign_tx_response
 
 
+@tools.workflow(capability=m.Capability.Cardano)
 def sign_message(
     session: "Session",
     signing_path: Path,
diff --git a/python/src/trezorlib/cli/__init__.py b/python/src/trezorlib/cli/__init__.py
index 25efa9e6..37d906dd 100644
--- a/python/src/trezorlib/cli/__init__.py
+++ b/python/src/trezorlib/cli/__init__.py
@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import atexit
 import functools
 import logging
 import os
@@ -24,19 +23,25 @@ import re
 import sys
 import typing as t
 from contextlib import contextmanager
+from enum import Enum
 
 import click
 
-from .. import exceptions, transport, ui
-from ..client import PASSPHRASE_ON_DEVICE, ProtocolVersion, TrezorClient
-from ..messages import Capability
+from .. import exceptions, protocol_v1, transport, ui
+from ..client import (
+    AppManifest,
+    PassphraseSetting,
+    Session,
+    TrezorClient,
+    get_client,
+    get_default_session,
+)
+from ..thp import client as thp_client
 from ..transport import Transport
-from ..transport.session import Session, SessionV1, SessionV2
+from . import credentials
 
 LOG = logging.getLogger(__name__)
 
-_TRANSPORT: Transport | None = None
-
 if t.TYPE_CHECKING:
     # Needed to enforce a return value from decorators
     # More details: https://www.python.org/dev/peps/pep-0612/
@@ -60,7 +65,7 @@ class ChoiceType(click.Choice):
         else:
             self.typemap = {k.lower(): v for k, v in typemap.items()}
 
-    def convert(self, value: t.Any, param: t.Any, ctx: click.Context) -> t.Any:
+    def convert(self, value: t.Any, param: t.Any, ctx: click.Context | None) -> t.Any:
         if value in self.typemap.values():
             return value
         value = super().convert(value, param, ctx)
@@ -69,12 +74,25 @@ class ChoiceType(click.Choice):
         return self.typemap[value]
 
 
-def get_passphrase(
-    available_on_device: bool, passphrase_on_host: bool
-) -> t.Union[str, object]:
-    if available_on_device and not passphrase_on_host:
-        return PASSPHRASE_ON_DEVICE
+class PassphraseSource(Enum):
+    """Passphrase source configured by the user."""
+
+    AUTO = "auto"
+    """If passphrase is enabled and the device supports it, request passphrase
+    entry on the device. Otherwise, open the default wallet with no
+    passphrase."""
+    PROMPT = "prompt"
+    """Request passphrase entry on the host."""
+    EMPTY = "empty"
+    """Open the default wallet with no passphrase."""
+    DEVICE = "device"
+    """Request passphrase entry on the device."""
+
+    def ok_if_disabled(self) -> bool:
+        return self in (self.AUTO, self.EMPTY)
 
+
+def get_passphrase() -> str:
     env_passphrase = os.getenv("PASSPHRASE")
     if env_passphrase is not None:
         ui.echo("Passphrase required. Using PASSPHRASE environment variable.")
@@ -105,7 +123,7 @@ def get_passphrase(
             raise exceptions.Cancelled from None
 
 
-def get_code_entry_code() -> int:
+def get_code_entry_code() -> str:
     while True:
         try:
             code_input = ui.prompt(
@@ -121,137 +139,128 @@ def get_code_entry_code() -> int:
             if len(code_str) != 6:
                 ui.echo("Code must be 6-digits long.")
                 continue
-            code = int(code_str)
-            return code
+            return code_str
         except click.Abort:
             raise exceptions.Cancelled from None
 
 
-def get_client(transport: Transport) -> TrezorClient:
-    return TrezorClient(transport)
-
-
 class TrezorConnection:
-
     def __init__(
         self,
         path: str,
-        session_id: bytes | None,
-        passphrase_on_host: bool,
+        session_id: str | None,
+        passphrase_source: PassphraseSource,
         script: bool,
+        *,
+        app_name: str = "trezorctl",
     ) -> None:
         self.path = path
         self.session_id = session_id
-        self.passphrase_on_host = passphrase_on_host
+        self.passphrase_source = passphrase_source
         self.script = script
+        self.credentials = credentials.CredentialStore(app_name)
+        self.app = AppManifest(app_name=app_name, credentials=self.credentials.list)
+        if self.script:
+            self.app.button_callback = ui.ScriptUI.button_request
+            self.app.pin_callback = ui.ScriptUI.get_pin
+        else:
+            click_ui = ui.ClickUI()
+            self.app.button_callback = click_ui.button_request
+            self.app.pin_callback = click_ui.get_pin
 
     def get_session(
         self,
+        use_passphrase: bool = True,
+        seedless: bool = False,
         derive_cardano: bool = False,
-        empty_passphrase: bool = False,
-        must_resume: bool = False,
     ) -> Session:
         client = self.get_client()
-        if must_resume and self.session_id is None:
-            click.echo("Failed to resume session - no session id provided")
-            raise RuntimeError("Failed to resume session - no session id provided")
+        client.ensure_unlocked()
+        if (
+            not client.features.passphrase_protection
+            and not self.passphrase_source.ok_if_disabled()
+        ):
+            raise click.ClickException("Passphrase protection is not enabled")
+
+        # if empty passphrase is requested, do not try to resume and instead
+        # create a new session
+        if not use_passphrase or self.passphrase_source == PassphraseSource.EMPTY:
+            return client.get_session(passphrase=PassphraseSetting.STANDARD_WALLET)
+        if seedless:
+            return client.get_session(passphrase=None)
 
         # Try resume session from id
         if self.session_id is not None:
-            if client.protocol_version is ProtocolVersion.V1:
-                session = SessionV1.resume_from_id(
-                    client=client, session_id=self.session_id
-                )
-            elif client.protocol_version is ProtocolVersion.V2:
-                session = SessionV2(client, self.session_id)
-                # TODO fix resumption on THP
-            else:
-                raise Exception("Unsupported client protocol", client.protocol_version)
-            if must_resume:
-                if session.id != self.session_id or session.id is None:
-                    click.echo("Failed to resume session")
-                    env_var = os.environ.get("TREZOR_SESSION_ID")
-                    if env_var and bytes.fromhex(env_var) == self.session_id:
-                        click.echo(
-                            "Session-id stored in TREZOR_SESSION_ID is no longer valid. Call 'unset TREZOR_SESSION_ID' to clear it."
-                        )
-                    raise exceptions.FailedSessionResumption(
-                        received_session_id=session.id
+            try:
+                if isinstance(client, protocol_v1.TrezorClientV1):
+                    session = protocol_v1.SessionV1(
+                        client, id=bytes.fromhex(self.session_id)
                     )
-            return session
-
-        features = client.features
-
-        passphrase_protection = features.passphrase_protection
-        if passphrase_protection is None:
-            raise RuntimeError("Device is locked")
-
-        if not passphrase_protection:
-            return client.get_session(derive_cardano=derive_cardano)
-
-        if empty_passphrase:
-            passphrase = ""
-        elif self.script:
-            passphrase = None
-        else:
-            available_on_device = Capability.PassphraseEntry in features.capabilities
-            passphrase = get_passphrase(available_on_device, self.passphrase_on_host)
-        session = client.get_session(
-            passphrase=passphrase, derive_cardano=derive_cardano
+                    session.initialize()
+                elif isinstance(client, thp_client.TrezorClientThp):
+                    session = thp_client.ThpSession(client, id=int(self.session_id))
+                    # TODO what here?
+                else:
+                    raise click.ClickException(
+                        f"Unsupported client type: {type(client).__name__}"
+                    )
+            except exceptions.InvalidSessionError:
+                LOG.error("Failed to resume session", exc_info=True)
+                env_var = os.environ.get("TREZOR_SESSION_ID")
+                if env_var != self.session_id:
+                    click.echo(
+                        "Session-id stored in TREZOR_SESSION_ID is no longer valid. Call\n"
+                        "  unset TREZOR_SESSION_ID\n"
+                        "to clear it."
+                    )
+                raise
+            else:
+                return session
+
+        if self.passphrase_source == PassphraseSource.PROMPT:
+            passphrase = get_passphrase()
+            return client.get_session(passphrase=passphrase)
+        if self.passphrase_source == PassphraseSource.DEVICE:
+            return client.get_session(passphrase=PassphraseSetting.ON_DEVICE)
+        if self.passphrase_source == PassphraseSource.AUTO:
+            return get_default_session(client, derive_cardano=derive_cardano)
+        raise NotImplementedError(
+            f"Passphrase source {self.passphrase_source} not implemented"
         )
-        return session
 
-    def get_transport(self, _clear_cache: bool = False) -> "Transport":
-        global _TRANSPORT
-        if _TRANSPORT is not None:
-            if not _clear_cache:
-                return _TRANSPORT
-
-            # remove previously cached transport
-            try:
-                atexit.unregister(_TRANSPORT.close)
-                _TRANSPORT.close()
-            except Exception as e:
-                self._print_exception(e, "Failed to close transport")
-            finally:
-                _TRANSPORT = None
+    def get_transport(self) -> Transport:
         try:
             # look for transport without prefix search
-            _TRANSPORT = transport.get_transport(self.path, prefix_search=False)
+            return transport.get_transport(self.path, prefix_search=False)
         except Exception:
             # most likely not found. try again below.
             pass
 
         # look for transport with prefix search
         # if this fails, we want the exception to bubble up to the caller
-        if not _TRANSPORT:
-            _TRANSPORT = transport.get_transport(self.path, prefix_search=True)
-
-        _TRANSPORT.open()
-        atexit.register(_TRANSPORT.close)
-        return _TRANSPORT
+        return transport.get_transport(self.path, prefix_search=True)
 
     def get_client(self) -> TrezorClient:
-        client = get_client(self.get_transport())
-        if self.script:
-            client.button_callback = ui.ScriptUI.button_request
-            client.pin_callback = ui.ScriptUI.get_pin
-        else:
-            click_ui = ui.ClickUI()
-            client.button_callback = click_ui.button_request
-            client.pin_callback = click_ui.get_pin
-        return client
+        client = get_client(self.app, self.get_transport())
+        if not client.pairing.is_paired():
+            from ..thp import pairing
 
-    def get_seedless_session(self) -> Session:
-        client = self.get_client()
-        seedless_session = client.get_seedless_session()
-        return seedless_session
+            credential = pairing.default_pairing_flow(
+                client.pairing, code_entry_callback=get_code_entry_code
+            )
+            if credential is not None:
+                self.credentials.add(credential)
+
+        return client
 
     def _connection_context(
-        self, connect_fn: t.Callable[[], R]
+        self,
+        connect_fn: t.Callable[P, R],
+        *args: P.args,
+        **kwargs: P.kwargs,
     ) -> t.Generator[R, None, None]:
         try:
-            conn = connect_fn()
+            conn = connect_fn(*args, **kwargs)
         except Exception as e:
             self._print_exception(e, "Failed to connect")
             sys.exit(1)
@@ -281,19 +290,16 @@ class TrezorConnection:
     @contextmanager
     def session_context(
         self,
-        empty_passphrase: bool = False,
+        *,
+        use_passphrase: bool = True,
         derive_cardano: bool = False,
         seedless: bool = False,
-        must_resume: bool = False,
     ) -> t.Generator[Session, None, None]:
         yield from self._connection_context(
-            self.get_seedless_session
-            if seedless
-            else lambda: self.get_session(
-                derive_cardano=derive_cardano,
-                empty_passphrase=empty_passphrase,
-                must_resume=must_resume,
-            )
+            self.get_session,
+            use_passphrase=use_passphrase,
+            derive_cardano=derive_cardano,
+            seedless=seedless,
         )
 
     def _print_exception(self, exc: Exception, message: str) -> None:
@@ -307,16 +313,32 @@ class TrezorConnection:
             click.echo(f"Using path: {self.path}")
 
 
+@t.overload
+def with_session(func: t.Callable[Concatenate[Session, P], R]) -> t.Callable[P, R]: ...
+
+
+@t.overload
+def with_session(
+    *,
+    passphrase: bool = True,
+    cardano: bool = False,
+    seedless: bool = False,
+) -> t.Callable[[FuncWithSession[P, R]], t.Callable[P, R]]: ...
+
+
 def with_session(
-    func: "t.Callable[Concatenate[Session, P], R]|None" = None,
+    func: t.Callable[Concatenate[Session, P], R] | None = None,
     *,
-    empty_passphrase: bool = False,
-    derive_cardano: bool = False,
+    passphrase: bool = True,
+    cardano: bool = False,
     seedless: bool = False,
-    must_resume: bool = False,
-) -> t.Callable[[FuncWithSession], t.Callable[P, R]]:
+) -> t.Callable[[FuncWithSession[P, R]], t.Callable[P, R]] | t.Callable[P, R]:
     """Provides a Click command with parameter `session=obj.get_session(...)`
-    based on the parameters provided.
+    based on the parameters provided:
+
+    * if `passphrase` is set to False, a standard wallet is always used for this session
+    * if `cardano` is set to True, Cardano-specific operations are enabled for this session
+    * if `seedless` is set to True, a seedless session is used for this session
 
     If default parameters are ok, this decorator can be used without parentheses.
     """
@@ -330,34 +352,39 @@ def with_session(
         def function_with_session(
             obj: TrezorConnection, *args: "P.args", **kwargs: "P.kwargs"
         ) -> "R":
-            is_resume_mandatory = must_resume or obj.session_id is not None
-
             with obj.session_context(
-                empty_passphrase=empty_passphrase,
-                derive_cardano=derive_cardano,
+                use_passphrase=passphrase,
+                derive_cardano=cardano,
                 seedless=seedless,
-                must_resume=is_resume_mandatory,
             ) as session:
                 try:
                     return func(session, *args, **kwargs)
 
                 finally:
-                    if (
-                        not is_resume_mandatory
-                        and not session.features.bootloader_mode
-                        and not session.client.is_invalidated
-                    ):
-                        session.end()
+                    if obj.session_id is None and not session.features.bootloader_mode:
+                        session.close()
 
         return function_with_session
 
     # If the decorator @get_session is used without parentheses
     if func and callable(func):
-        return decorator(func)  # type: ignore [Function return type]
+        return decorator(func)
 
     return decorator
 
 
+def with_client(func: t.Callable[Concatenate[TrezorClient, P], R]) -> t.Callable[P, R]:
+    @click.pass_obj
+    @functools.wraps(func)
+    def function_with_client(
+        obj: TrezorConnection, *args: P.args, **kwargs: P.kwargs
+    ) -> R:
+        with obj.client_context() as client:
+            return func(client, *args, **kwargs)
+
+    return function_with_client
+
+
 class AliasedGroup(click.Group):
     """Command group that handles aliases and Click 6.x compatibility.
 
diff --git a/python/src/trezorlib/cli/benchmark.py b/python/src/trezorlib/cli/benchmark.py
index 0ca3794a..03d8bc4d 100644
--- a/python/src/trezorlib/cli/benchmark.py
+++ b/python/src/trezorlib/cli/benchmark.py
@@ -24,7 +24,7 @@ from . import with_session
 
 if TYPE_CHECKING:
 
-    from ..transport.session import Session
+    from ..client import Session
 
 
 def list_names_patern(session: "Session", pattern: Optional[str] = None) -> List[str]:
@@ -41,7 +41,7 @@ def cli() -> None:
 
 @cli.command()
 @click.argument("pattern", required=False)
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def list_names(session: "Session", pattern: Optional[str] = None) -> None:
     """List names of all supported benchmarks"""
     names = list_names_patern(session, pattern)
@@ -54,7 +54,7 @@ def list_names(session: "Session", pattern: Optional[str] = None) -> None:
 
 @cli.command()
 @click.argument("pattern", required=False)
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def run(session: "Session", pattern: Optional[str]) -> None:
     """Run benchmark"""
     names = list_names_patern(session, pattern)
diff --git a/python/src/trezorlib/cli/ble.py b/python/src/trezorlib/cli/ble.py
index 8bc35d63..7c86421e 100644
--- a/python/src/trezorlib/cli/ble.py
+++ b/python/src/trezorlib/cli/ble.py
@@ -24,7 +24,7 @@ from ..transport.ble import BleProxy
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 
 @click.group(name="ble")
diff --git a/python/src/trezorlib/cli/btc.py b/python/src/trezorlib/cli/btc.py
index ac629e62..48e0f35a 100644
--- a/python/src/trezorlib/cli/btc.py
+++ b/python/src/trezorlib/cli/btc.py
@@ -27,7 +27,7 @@ from .. import btc, messages, protobuf, tools
 from . import ChoiceType, with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PURPOSE_BIP44 = 44
 PURPOSE_BIP48 = 48
diff --git a/python/src/trezorlib/cli/cardano.py b/python/src/trezorlib/cli/cardano.py
index b103f858..1bb4c43e 100644
--- a/python/src/trezorlib/cli/cardano.py
+++ b/python/src/trezorlib/cli/cardano.py
@@ -23,7 +23,7 @@ from .. import cardano, messages, tools
 from . import ChoiceType, with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path to key, e.g. m/44h/1815h/0h/0/0"
 
@@ -62,7 +62,7 @@ def cli() -> None:
 @click.option("-i", "--include-network-id", is_flag=True)
 @click.option("-C", "--chunkify", is_flag=True)
 @click.option("-T", "--tag-cbor-sets", is_flag=True)
-@with_session(derive_cardano=True)
+@with_session(cardano=True)
 def sign_tx(
     session: "Session",
     file: TextIO,
@@ -208,7 +208,7 @@ def sign_tx(
     default=messages.CardanoDerivationType.ICARUS,
 )
 @click.option("-C", "--chunkify", is_flag=True)
-@with_session(derive_cardano=True)
+@with_session(cardano=True)
 def get_address(
     session: "Session",
     address: str,
@@ -281,7 +281,7 @@ def get_address(
     default=messages.CardanoDerivationType.ICARUS,
 )
 @click.option("-d", "--show-display", is_flag=True)
-@with_session(derive_cardano=True)
+@with_session(cardano=True)
 def get_public_key(
     session: "Session",
     address: str,
@@ -309,7 +309,7 @@ def get_public_key(
     type=ChoiceType({m.name: m for m in messages.CardanoDerivationType}),
     default=messages.CardanoDerivationType.ICARUS,
 )
-@with_session(derive_cardano=True)
+@with_session(cardano=True)
 def get_native_script_hash(
     session: "Session",
     file: TextIO,
@@ -333,7 +333,7 @@ def get_native_script_hash(
     type=ChoiceType({m.name: m for m in messages.CardanoDerivationType}),
     default=messages.CardanoDerivationType.ICARUS,
 )
-@with_session(derive_cardano=True)
+@with_session(cardano=True)
 def sign_message(
     session: "Session",
     file: TextIO,
diff --git a/python/src/trezorlib/cli/credentials.py b/python/src/trezorlib/cli/credentials.py
new file mode 100644
index 00000000..a956bbba
--- /dev/null
+++ b/python/src/trezorlib/cli/credentials.py
@@ -0,0 +1,155 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+import typing as t
+from functools import cached_property
+from pathlib import Path
+
+import keyring
+import platformdirs
+from typing_extensions import Self
+
+from ..thp.credentials import Credential
+
+LOG = logging.getLogger(__name__)
+
+
+class KeyringCredential:
+    def __init__(self, app_name: str, trezor_pubkey: bytes) -> None:
+        self.app_name = app_name
+        self.trezor_pubkey = trezor_pubkey
+
+    @property
+    def _system(self) -> str:
+        return f"{self.app_name}/thp-credentials"
+
+    @property
+    def _system_privkey(self) -> str:
+        return self._system + "/privkey"
+
+    @property
+    def _system_credential(self) -> str:
+        return self._system + "/credential"
+
+    @cached_property
+    def _username(self) -> str:
+        return base64.b64encode(self.trezor_pubkey).decode()
+
+    @cached_property
+    def host_privkey(self) -> bytes:
+        privkey_b64 = keyring.get_password(self._system_privkey, self._username)
+        if privkey_b64 is None:
+            raise ValueError("Private key not found")
+        return base64.b64decode(privkey_b64)
+
+    @cached_property
+    def credential(self) -> bytes:
+        credential_b64 = keyring.get_password(self._system_credential, self._username)
+        if credential_b64 is None:
+            raise ValueError("Credential not found")
+        return base64.b64decode(credential_b64)
+
+    @classmethod
+    def save(cls, app_name: str, credential: Credential) -> Self:
+        new = cls(app_name, credential.trezor_pubkey)
+        new.host_privkey = credential.host_privkey
+        new.credential = credential.credential
+        keyring.set_password(
+            new._system_privkey,
+            new._username,
+            base64.b64encode(new.host_privkey).decode(),
+        )
+        keyring.set_password(
+            new._system_credential,
+            new._username,
+            base64.b64encode(new.credential).decode(),
+        )
+        return new
+
+    def delete(self) -> None:
+        try:
+            keyring.delete_password(self._system_privkey, self._username)
+        except Exception:
+            pass
+        try:
+            keyring.delete_password(self._system_credential, self._username)
+        except Exception:
+            pass
+
+    def as_credential(self) -> Credential:
+        # limitation of pyright:
+        # https://github.com/microsoft/pyright/issues/10252
+        # at runtime, KeyringCredential does conform to Credential, but this
+        # can't be cleanly expressed via the type system.
+        return self  # type: ignore ["cached_property[bytes]" is not assignable to "bytes"]
+
+
+class CredentialStore:
+    def __init__(
+        self,
+        app_name: str,
+        config_file: Path | None = None,
+        config_appname: str = "trezorctl",
+    ) -> None:
+        self.app_name = app_name
+        if config_file is not None:
+            self.config_path = config_file
+        else:
+            config_dir = Path(
+                platformdirs.user_config_dir(config_appname, ensure_exists=True)
+            )
+            self.config_path = config_dir / "thp-credentials.json"
+
+    def _load(self) -> dict[str, t.Any]:
+        if not self.config_path.exists():
+            return {}
+        return json.loads(self.config_path.read_text())
+
+    def _save(self, data: dict[str, t.Any]) -> None:
+        self.config_path.write_text(json.dumps(data, indent=2) + "\n")
+
+    def list(self) -> t.Collection[Credential]:
+        data = self._load()
+        app_data = data.get(self.app_name, ())
+        return [
+            KeyringCredential(
+                self.app_name, base64.b64decode(credential)
+            ).as_credential()
+            for credential in app_data
+        ]
+
+    def add(self, credential: Credential) -> None:
+        data = self._load()
+        app_data = data.setdefault(self.app_name, [])
+        saved_credential = KeyringCredential.save(self.app_name, credential)
+        app_data.append(saved_credential._username)
+        self._save(data)
+        LOG.info(
+            "Added credential for %s: %s", self.app_name, credential.trezor_pubkey.hex()
+        )
+
+    def delete(self, trezor_pubkey: bytes) -> None:
+        data = self._load()
+        app_data = data.setdefault(self.app_name, [])
+        credential = KeyringCredential(self.app_name, trezor_pubkey)
+        app_data.remove(credential._username)
+        credential.delete()
+        self._save(data)
diff --git a/python/src/trezorlib/cli/crypto.py b/python/src/trezorlib/cli/crypto.py
index e7ac4ddf..5dca37c6 100644
--- a/python/src/trezorlib/cli/crypto.py
+++ b/python/src/trezorlib/cli/crypto.py
@@ -22,7 +22,7 @@ from .. import misc, tools
 from . import ChoiceType, with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 
 PROMPT_TYPE = ChoiceType(
@@ -42,7 +42,7 @@ def cli() -> None:
 
 @cli.command()
 @click.argument("size", type=int)
-@with_session(empty_passphrase=True)
+@with_session
 def get_entropy(session: "Session", size: int) -> str:
     """Get random bytes from device."""
     return misc.get_entropy(session, size).hex()
@@ -55,7 +55,7 @@ def get_entropy(session: "Session", size: int) -> str:
 )
 @click.argument("key")
 @click.argument("value")
-@with_session(empty_passphrase=True)
+@with_session
 def encrypt_keyvalue(
     session: "Session",
     address: str,
@@ -91,7 +91,7 @@ def encrypt_keyvalue(
 )
 @click.argument("key")
 @click.argument("value")
-@with_session(empty_passphrase=True)
+@with_session
 def decrypt_keyvalue(
     session: "Session",
     address: str,
diff --git a/python/src/trezorlib/cli/debug.py b/python/src/trezorlib/cli/debug.py
index 22e459ba..9eeb5fe6 100644
--- a/python/src/trezorlib/cli/debug.py
+++ b/python/src/trezorlib/cli/debug.py
@@ -18,12 +18,12 @@ from typing import TYPE_CHECKING, Union
 
 import click
 
-from ..debuglink import DebugLink, TrezorClientDebugLink
+from ..client import Session
+from ..debuglink import DebugLink, TrezorTestContext
 from ..debuglink import optiga_set_sec_max as debuglink_optiga_set_sec_max
 from ..debuglink import prodtest_t1 as debuglink_prodtest_t1
 from ..debuglink import record_screen
 from ..debuglink import set_log_filter as debuglink_set_log_filter
-from ..transport.session import Session
 from . import with_session
 
 if TYPE_CHECKING:
@@ -52,9 +52,8 @@ def record_screen_from_connection(
 ) -> None:
     """Record screen helper to transform TrezorConnection into TrezorClientDebugLink."""
     transport = obj.get_transport()
-    debug_client = TrezorClientDebugLink(transport, auto_interact=False)
+    debug_client = TrezorTestContext(transport=transport, auto_interact=False)
     record_screen(debug_client, directory, report_func=click.echo)
-    debug_client.close_transport()
 
 
 @cli.command()
diff --git a/python/src/trezorlib/cli/device.py b/python/src/trezorlib/cli/device.py
index 6571542f..8e9d2017 100644
--- a/python/src/trezorlib/cli/device.py
+++ b/python/src/trezorlib/cli/device.py
@@ -29,7 +29,7 @@ from ..tools import format_path
 from . import ChoiceType, with_session
 
 if t.TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
     from . import TrezorConnection
 
 RECOVERY_DEVICE_INPUT_METHOD = {
@@ -319,7 +319,7 @@ def reboot_to_bootloader(obj: "TrezorConnection") -> None:
     # avoid using @with_session because it closes the session afterwards,
     # which triggers double prompt on device
     with obj.client_context() as client:
-        device.reboot_to_bootloader(client.get_seedless_session())
+        device.reboot_to_bootloader(client.get_session(passphrase=None))
 
 
 @cli.command()
diff --git a/python/src/trezorlib/cli/eos.py b/python/src/trezorlib/cli/eos.py
index 353cba94..0ef5a898 100644
--- a/python/src/trezorlib/cli/eos.py
+++ b/python/src/trezorlib/cli/eos.py
@@ -24,7 +24,7 @@ from . import with_session
 
 if TYPE_CHECKING:
     from .. import messages
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path, e.g. m/44h/194h/0h/0/0"
 
diff --git a/python/src/trezorlib/cli/ethereum.py b/python/src/trezorlib/cli/ethereum.py
index 50b1b18e..478d3a87 100644
--- a/python/src/trezorlib/cli/ethereum.py
+++ b/python/src/trezorlib/cli/ethereum.py
@@ -33,7 +33,7 @@ if TYPE_CHECKING:
     from eth_typing import ChecksumAddress  # noqa: I900
     from web3.types import Wei
 
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path, e.g. m/44h/60h/0h/0/0"
 
diff --git a/python/src/trezorlib/cli/evolu.py b/python/src/trezorlib/cli/evolu.py
index cfb1774f..36980bc7 100644
--- a/python/src/trezorlib/cli/evolu.py
+++ b/python/src/trezorlib/cli/evolu.py
@@ -24,7 +24,7 @@ from .. import evolu
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 
 @click.group(name="evolu")
diff --git a/python/src/trezorlib/cli/fido.py b/python/src/trezorlib/cli/fido.py
index 9fd27a9d..ccda9f01 100644
--- a/python/src/trezorlib/cli/fido.py
+++ b/python/src/trezorlib/cli/fido.py
@@ -22,7 +22,7 @@ from .. import fido
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 ALGORITHM_NAME = {-7: "ES256 (ECDSA w/ SHA-256)", -8: "EdDSA"}
 
@@ -40,7 +40,7 @@ def credentials() -> None:
 
 
 @credentials.command(name="list")
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def credentials_list(session: "Session") -> None:
     """List all resident credentials on the device."""
     creds = fido.list_credentials(session)
@@ -79,7 +79,7 @@ def credentials_list(session: "Session") -> None:
 
 @credentials.command(name="add")
 @click.argument("hex_credential_id")
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def credentials_add(session: "Session", hex_credential_id: str) -> None:
     """Add the credential with the given ID as a resident credential.
 
@@ -92,7 +92,7 @@ def credentials_add(session: "Session", hex_credential_id: str) -> None:
 @click.option(
     "-i", "--index", required=True, type=click.IntRange(0, 99), help="Credential index."
 )
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def credentials_remove(session: "Session", index: int) -> None:
     """Remove the resident credential at the given index."""
     fido.remove_credential(session, index)
@@ -110,14 +110,14 @@ def counter() -> None:
 
 @counter.command(name="set")
 @click.argument("counter", type=int)
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def counter_set(session: "Session", counter: int) -> None:
     """Set FIDO/U2F counter value."""
     fido.set_counter(session, counter)
 
 
 @counter.command(name="get-next")
-@with_session(empty_passphrase=True)
+@with_session(passphrase=False)
 def counter_get_next(session: "Session") -> int:
     """Get-and-increase value of FIDO/U2F counter.
 
diff --git a/python/src/trezorlib/cli/firmware.py b/python/src/trezorlib/cli/firmware.py
index a29649e6..4585086e 100644
--- a/python/src/trezorlib/cli/firmware.py
+++ b/python/src/trezorlib/cli/firmware.py
@@ -39,8 +39,7 @@ from ..models import TrezorModel
 from . import ChoiceType, with_session
 
 if TYPE_CHECKING:
-    from ..client import TrezorClient
-    from ..transport.session import Session
+    from ..client import Session, TrezorClient
     from . import TrezorConnection
 
 MODEL_CHOICE = ChoiceType(
@@ -641,7 +640,7 @@ def update(
     against data.trezor.io information, if available.
     """
     with obj.client_context() as client:
-        seedless_session = client.get_seedless_session()
+        seedless_session = client.get_session(passphrase=None)
         if sum(bool(x) for x in (filename, url, version)) > 1:
             click.echo("You can use only one of: filename, url, version.")
             sys.exit(1)
@@ -691,8 +690,7 @@ def update(
             while True:
                 time.sleep(0.5)
                 try:
-                    # uncache previous transport to force re-connection attempt
-                    obj.get_transport(_clear_cache=True)
+                    obj.get_transport()
                     break
                 except Exception:
                     pass
@@ -703,7 +701,7 @@ def update(
             sys.exit(1)
 
         upload_firmware_into_device(
-            session=client.get_seedless_session(), firmware_data=firmware_data
+            session=client.get_session(passphrase=None), firmware_data=firmware_data
         )
 
 
diff --git a/python/src/trezorlib/cli/monero.py b/python/src/trezorlib/cli/monero.py
index b5d3bdc4..7289f39f 100644
--- a/python/src/trezorlib/cli/monero.py
+++ b/python/src/trezorlib/cli/monero.py
@@ -22,7 +22,7 @@ from .. import messages, monero, tools
 from . import ChoiceType, with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path, e.g. m/44h/128h/0h"
 
diff --git a/python/src/trezorlib/cli/nem.py b/python/src/trezorlib/cli/nem.py
index 43d2075f..2c3cdb00 100644
--- a/python/src/trezorlib/cli/nem.py
+++ b/python/src/trezorlib/cli/nem.py
@@ -24,7 +24,7 @@ from .. import nem, tools
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path, e.g. m/44h/134h/0h/0h"
 
diff --git a/python/src/trezorlib/cli/nostr.py b/python/src/trezorlib/cli/nostr.py
index 25b9807a..996d8304 100644
--- a/python/src/trezorlib/cli/nostr.py
+++ b/python/src/trezorlib/cli/nostr.py
@@ -25,7 +25,7 @@ from .. import messages, nostr, tools
 from . import with_session
 
 if t.TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 
 PATH_TEMPLATE = "m/44h/1237h/{}h/0/0"
diff --git a/python/src/trezorlib/cli/ripple.py b/python/src/trezorlib/cli/ripple.py
index 6e1043c7..8848a8f3 100644
--- a/python/src/trezorlib/cli/ripple.py
+++ b/python/src/trezorlib/cli/ripple.py
@@ -23,7 +23,7 @@ from .. import ripple, tools
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path to key, e.g. m/44h/144h/0h/0/0"
 
diff --git a/python/src/trezorlib/cli/settings.py b/python/src/trezorlib/cli/settings.py
index 4c3cfcc0..1af8f70f 100644
--- a/python/src/trezorlib/cli/settings.py
+++ b/python/src/trezorlib/cli/settings.py
@@ -27,7 +27,7 @@ from .. import device, messages, toif
 from . import AliasedGroup, ChoiceType, with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 try:
     from PIL import Image
diff --git a/python/src/trezorlib/cli/solana.py b/python/src/trezorlib/cli/solana.py
index 24811907..3769af10 100644
--- a/python/src/trezorlib/cli/solana.py
+++ b/python/src/trezorlib/cli/solana.py
@@ -26,7 +26,7 @@ from .. import definitions, messages, solana, tools
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path to key, e.g. m/44h/501h/0h/0h"
 DEFAULT_PATH = "m/44h/501h/0h/0h"
diff --git a/python/src/trezorlib/cli/stellar.py b/python/src/trezorlib/cli/stellar.py
index edf185d4..08452009 100644
--- a/python/src/trezorlib/cli/stellar.py
+++ b/python/src/trezorlib/cli/stellar.py
@@ -24,7 +24,7 @@ from .. import stellar, tools
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 try:
     from stellar_sdk import (
diff --git a/python/src/trezorlib/cli/tezos.py b/python/src/trezorlib/cli/tezos.py
index d7f6084d..8940bf8d 100644
--- a/python/src/trezorlib/cli/tezos.py
+++ b/python/src/trezorlib/cli/tezos.py
@@ -23,7 +23,7 @@ from .. import messages, protobuf, tezos, tools
 from . import with_session
 
 if TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
 PATH_HELP = "BIP-32 path, e.g. m/44h/1729h/0h"
 
diff --git a/python/src/trezorlib/cli/trezorctl.py b/python/src/trezorlib/cli/trezorctl.py
index 0c0b220b..59475609 100755
--- a/python/src/trezorlib/cli/trezorctl.py
+++ b/python/src/trezorlib/cli/trezorctl.py
@@ -16,22 +16,25 @@
 # You should have received a copy of the License along with this library.
 # If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
 
+from __future__ import annotations
+
 import importlib.metadata
 import json
 import logging
 import os
 import time
-from typing import TYPE_CHECKING, Any, Callable, Iterable, Optional, TypeVar, cast
+from typing import Any, Callable, Optional, TypeVar, cast
 
 import click
 
 from .. import log, messages, protobuf
+from ..client import TrezorClient
 from ..transport import DeviceIsBusy, enumerate_devices
-from ..transport.session import Session
 from ..transport.ble import BleTransport
 from ..transport.udp import UdpTransport
 from . import (
     AliasedGroup,
+    PassphraseSource,
     TrezorConnection,
     benchmark,
     ble,
@@ -55,14 +58,11 @@ from . import (
     telemetry,
     tezos,
     tron,
-    with_session,
+    with_client,
 )
 
 F = TypeVar("F", bound=Callable)
 
-if TYPE_CHECKING:
-    from ..transport import Transport
-
 LOG = logging.getLogger(__name__)
 
 COMMAND_ALIASES = {
@@ -226,14 +226,12 @@ def cli_main(
 
     BleTransport.ENABLED = ble or (os.environ.get("TREZOR_BLE") == "1")
 
-    bytes_session_id: Optional[bytes] = None
-    if session_id is not None:
-        try:
-            bytes_session_id = bytes.fromhex(session_id)
-        except ValueError:
-            raise click.ClickException(f"Not a valid session id: {session_id}")
+    if passphrase_on_host:
+        passphrase_source = PassphraseSource.PROMPT
+    else:
+        passphrase_source = PassphraseSource.AUTO
 
-    ctx.obj = TrezorConnection(path, bytes_session_id, passphrase_on_host, script)
+    ctx.obj = TrezorConnection(path, session_id, passphrase_source, script)
 
     # Optionally record the screen into a specified directory.
     if record:
@@ -302,21 +300,21 @@ def format_device_name(features: messages.Features) -> str:
 @cli.command(name="list")
 @click.option("-n", "no_resolve", is_flag=True, help="Do not resolve Trezor names")
 @click.pass_obj
-def list_devices(
-    obj: TrezorConnection, no_resolve: bool
-) -> Optional[Iterable["Transport"]]:
+def list_devices(obj: TrezorConnection, no_resolve: bool) -> None:
     """List connected Trezor devices."""
     if no_resolve:
         for d in enumerate_devices():
             click.echo(d.get_path())
         return
 
-    from . import get_client
+    from ..client import AppManifest, get_client
+
+    app = AppManifest(app_name="trezorctl")
 
     for transport in enumerate_devices():
         try:
             transport.open()
-            client = get_client(transport)
+            client = get_client(app, transport)
             description = format_device_name(client.features)
         except DeviceIsBusy:
             description = "Device is in use by another process"
@@ -325,7 +323,6 @@ def list_devices(
         finally:
             transport.close()
         click.echo(f"{transport.get_path()} - {description}")
-    return None
 
 
 @cli.command()
@@ -342,10 +339,10 @@ def version() -> str:
 @cli.command()
 @click.argument("message")
 @click.option("-b", "--button-protection", is_flag=True)
-@with_session(seedless=True)
-def ping(session: "Session", message: str, button_protection: bool) -> str:
+@with_client
+def ping(client: TrezorClient, message: str, button_protection: bool) -> str:
     """Send ping message."""
-    return session.ping(message, button_protection)
+    return client.ping(message, button_protection)
 
 
 @cli.command()
@@ -371,22 +368,34 @@ def get_session(obj: TrezorConnection, derive_cardano: bool = False) -> str:
         if session.id is None:
             raise click.ClickException("Passphrase not enabled or firmware too old.")
         else:
+            # TODO
             return session.id.hex()
 
 
 @cli.command()
-@with_session(must_resume=True, empty_passphrase=True)
-def clear_session(session: "Session") -> None:
-    """Clear session (remove cached PIN, passphrase, etc.)."""
-    session.call(messages.LockDevice())
-    session.end()
+@click.pass_obj
+def clear_session(obj: TrezorConnection) -> None:
+    """Clear current session and lock the device.
+
+    Clears cached passphrase from the current session previously obtained
+    with `trezorctl get-session`.
+
+    Additionally, locks the device with PIN, if configured.
+    """
+    if obj.session_id is not None:
+        try:
+            session = obj.get_session()
+            session.close()
+        except Exception:
+            LOG.debug("Failed to clear session.", exc_info=True)
+    obj.get_client().lock()
 
 
 @cli.command()
-@with_session(seedless=True)
-def get_features(session: "Session") -> messages.Features:
+@with_client
+def get_features(client: TrezorClient) -> messages.Features:
     """Retrieve device features and settings."""
-    return session.features
+    return client.features
 
 
 @cli.command()
diff --git a/python/src/trezorlib/client.py b/python/src/trezorlib/client.py
index 7d2ca6c0..d6b25ed8 100644
--- a/python/src/trezorlib/client.py
+++ b/python/src/trezorlib/client.py
@@ -16,279 +16,409 @@
 
 from __future__ import annotations
 
+import enum
 import logging
 import os
-import socket
+import platform
 import typing as t
+import unicodedata
 import warnings
-from enum import IntEnum
-from hashlib import sha256
+from abc import ABCMeta, abstractmethod
+from dataclasses import dataclass
 
-from . import exceptions, mapping, messages, models
-from .tools import parse_path
+import typing_extensions as tx
+
+from . import exceptions, messages, models
+from .mapping import DEFAULT_MAPPING
+from .protobuf import MessageType
+from .tools import enter_context, parse_path
 from .transport import Transport, get_transport
-from .transport.thp.channel import Channel
-from .transport.thp.cpace import Cpace
-from .transport.thp.protocol_v1 import ProtocolV1Channel, UnexpectedMagicError
-from .transport.thp.protocol_v2 import ProtocolV2Channel
 
 if t.TYPE_CHECKING:
-    from .transport.session import Session, SessionV1, SessionV2
+    from .mapping import ProtobufMapping
+    from .thp import pairing
+    from .thp.credentials import Credential
+
+MT = t.TypeVar("MT", bound=MessageType)
+ClientType = t.TypeVar("ClientType", bound="TrezorClient")
+SessionType = t.TypeVar("SessionType", bound="Session")
+SessionIdType = t.TypeVar("SessionIdType", contravariant=True)
 
 LOG = logging.getLogger(__name__)
 
 MAX_PASSPHRASE_LENGTH = 50
 MAX_PIN_LENGTH = 50
 
-PASSPHRASE_ON_DEVICE = object()
-SEEDLESS = object()
-PASSPHRASE_TEST_PATH = parse_path("44h/1h/0h/0/0")
+_DEFAULT_READ_TIMEOUT: int | None = None
+
+
+class PassphraseSetting(enum.Enum):
+    """Passphrase setting for a session."""
+
+    STANDARD_WALLET = ""
+    """Open the default wallet with no passphrase."""
+    ON_DEVICE = object()
+    """Request passphrase entry on the device."""
+    AUTO = object()
+    """If passphrase is enabled and the device supports it, request passphrase
+    entry on the device. Otherwise, open the default wallet with no
+    passphrase."""
+    NONE = None
+    """Create a management session where wallet operations are disabled."""
+
+
+GET_ROOT_FINGERPRINT_MESSAGE = messages.GetPublicKey(
+    address_n=parse_path("m/0h"),
+    show_display=False,
+    ignore_xpub_magic=True,
+    ecdsa_curve_name="secp256k1",
+)
+
+
+class Session(t.Generic[ClientType, SessionIdType]):
+    def __init__(
+        self,
+        client: ClientType,
+        id: SessionIdType,
+        *,
+        root_fingerprint: bytes | None = None,
+    ) -> None:
+        self.client = client
+        self.id = id
+        self.is_invalid = False
+        self._root_fingerprint = root_fingerprint
+
+    def _log_short_id(self) -> str:
+        if self.id is None:
+            return f"(none:{id(self)})"
+        return repr(self.id)[:8]
+
+    @enter_context
+    def get_root_fingerprint(self) -> bytes:
+        if self._root_fingerprint is None:
+            self.ensure_unlocked()
+            assert self._root_fingerprint is not None
+        return self._root_fingerprint
+
+    def call(
+        self,
+        msg: MessageType,
+        *,
+        expect: type[MT] = MessageType,
+        timeout: float | None = None,
+    ) -> MT:
+        """Call a method on this session, process and return the response."""
+        if self.is_invalid:
+            raise exceptions.InvalidSessionError(self.id)
+        with self:
+            return self.client._call(self, msg, expect=expect, timeout=timeout)
+
+    def call_raw(self, msg: MessageType, timeout: float | None = None) -> MessageType:
+        """Invoke a single call-response round-trip to the device.
+
+        No processing is done on the response: errors are not converted to exceptions,
+        internal workflow callbacks are not triggered.
+        """
+        return self.client._call_raw(self, msg, timeout)
+
+    def read(self, timeout: float | None = None) -> MessageType:
+        """Read a single message from the device."""
+        return self.client._read(self, timeout)
 
-OUTDATED_FIRMWARE_ERROR = """
-Your Trezor firmware is out of date. Update it with the following command:
-  trezorctl firmware update
-Or visit https://suite.trezor.io/
-""".strip()
+    def write(self, msg: MessageType) -> None:
+        """Write a single message to the device."""
+        return self.client._write(self, msg)
 
+    def close(self) -> None:
+        """End and invalidate this session."""
+        LOG.info("Closing session %s", self)
+        try:
+            self.call(messages.EndSession())
+            self.is_invalid = True
+        except Exception as e:
+            LOG.warning("Failed to end session: %s", e)
 
-class ProtocolVersion(IntEnum):
-    V1 = 0x01  # Codec
-    V2 = 0x02  # THP
+    def cancel(self) -> None:
+        """Send a Cancel signal to the device, interrupting the current workflow."""
+        self.write(messages.Cancel())
+
+    @property
+    def features(self) -> messages.Features:
+        return self.client.features
+
+    def refresh_features(self) -> messages.Features:
+        return self.client.refresh_features()
+
+    @property
+    def model(self) -> models.TrezorModel:
+        return self.client.model
+
+    @property
+    def version(self) -> tuple[int, int, int]:
+        return self.client.version
+
+    def __enter__(self) -> tx.Self:
+        self.client.__enter__()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: t.Any,
+    ) -> None:
+        self.client.__exit__(exc_type, exc_value, traceback)
 
+    @enter_context
+    def ensure_unlocked(self) -> None:
+        resp = self.call(GET_ROOT_FINGERPRINT_MESSAGE, expect=messages.PublicKey)
+        assert resp.root_fingerprint is not None
+        root_fingerprint = resp.root_fingerprint.to_bytes(4, "big")
+        if self._root_fingerprint is None:
+            self._root_fingerprint = root_fingerprint
+        assert self._root_fingerprint == root_fingerprint
+        self.refresh_features()
+
+    @enter_context
+    def lock(self) -> None:
+        self.client.lock(_use_session=self)
+
+
+@dataclass
+class AppManifest:
+    app_name: str
+    host_name: str = platform.node()
 
-class TrezorClient:
     button_callback: t.Callable[[messages.ButtonRequest], None] | None = None
     pin_callback: t.Callable[[messages.PinMatrixRequest], str] | None = None
 
-    _model: models.TrezorModel
-    _features: messages.Features | None = None
-    _protocol_version: int
-    _setup_pin: str | None = None  # Should be used only by conftest
-    _last_active_session: SessionV1 | None = None
+    credentials: (
+        t.Collection[Credential] | t.Callable[[], t.Collection[Credential]]
+    ) = ()
 
-    _session_id_counter: int = 0
+    def _callback_pin(self, msg: messages.PinMatrixRequest) -> str:
+        if self.pin_callback is None:
+            raise RuntimeError("PIN callback was not specified")
+        return self.pin_callback(msg)
+
+    def _callback_button(self, msg: messages.ButtonRequest) -> None:
+        if self.button_callback is not None:
+            self.button_callback(msg)
+
+    def get_credentials(self) -> t.Collection[Credential]:
+        if callable(self.credentials):
+            return self.credentials()
+        return self.credentials
+
+
+class TrezorClient(t.Generic[SessionType], metaclass=ABCMeta):
+    _features: messages.Features | None = None
 
     def __init__(
         self,
+        app: AppManifest,
         transport: Transport,
-        protocol: Channel | None = None,
-        model: models.TrezorModel | None = None,
-        app_name: str | None = None,
-        host_name: str | None = None,
+        *,
+        model: models.TrezorModel | None,
+        mapping: ProtobufMapping | None,
+        pairing: pairing.PairingController,
     ) -> None:
         """
-        Transport needs to be opened before calling a method (or accessing
-        an attribute) for the first time. It should be closed after you're
-        done using the client.
-
-        The parameters app_name and host_name are only used for protocol
-        version 2 (THP). Because they are displayed on Trezor during the
-        pairing phase, it is recommended to provide app_name matching the
-        name of your application.
+        TODO
         """
+        LOG.info(
+            f"creating client instance {type(self).__name__} for device: {transport}"
+        )
+        self.app = app
+        self.transport = transport
+        self._model = model
+        self._mapping = mapping
+        self._features = None
+        self.pairing = pairing
 
-        LOG.info(f"creating client instance for device: {transport.get_path()}")
-        # Here, self.model could be set to None. Unless _init_device is False, it will
-        # get correctly reconfigured as part of the init_device flow.
-        self._model = model  # type: ignore ["None" is not assignable to "TrezorModel"]
-        if self._model:
-            self.mapping = self.model.default_mapping
-        else:
-            self.mapping = mapping.DEFAULT_MAPPING
+    # ===== Internal methods for overriding in subclasses =====
 
-        self._is_invalidated: bool = False
-        self.transport = transport
-        self.app_name = app_name
-        self.host_name = host_name
+    @abstractmethod
+    def _write(self, session: SessionType, msg: MessageType) -> None:
+        """Convert a message to the appropriate bytes representation for the given session
+        and write it to the transport.
+        """
+        raise NotImplementedError
 
-        if protocol is None:
-            self.protocol = self._get_protocol()
-        else:
-            self.protocol = protocol
-        self.protocol.mapping = self.mapping
+    @abstractmethod
+    def _read(self, session: SessionType, timeout: float | None = None) -> MessageType:
+        """Read the next message from the transport that is intended for the given session."""
+        raise NotImplementedError
 
-        if isinstance(self.protocol, ProtocolV1Channel):
-            self._protocol_version = ProtocolVersion.V1
-        elif isinstance(self.protocol, ProtocolV2Channel):
-            self._protocol_version = ProtocolVersion.V2
-        else:
-            raise Exception("Unknown protocol version")
+    @abstractmethod
+    def _get_any_session(self) -> SessionType:
+        """Get an arbitrary but valid session.
+
+        Used for internal calls that do not want to activate a specific session.
+        Users of the library SHOULD NOT use this method; use `get_session()`
+        with the appropriate parameters instead.
+        """
+        raise NotImplementedError
 
-    def do_pairing(
-        self, pairing_method: messages.ThpPairingMethod | None = None
+    @abstractmethod
+    def _get_session(
+        self,
+        *,
+        passphrase: str | t.Literal[PassphraseSetting.ON_DEVICE] | None,
+        derive_cardano: bool,
+    ) -> SessionType:
+        """Get a new session with the given passphrase and `derive_cardano` flag.
+
+        This internal method is used by `get_session()`, so that TrezorClient
+        subclasses do not have to check for Cardano in capabilities.
+        """
+        raise NotImplementedError
+
+    # ===== Common implementations =====
+
+    def __enter__(self) -> tx.Self:
+        """(Re)Open a connection to the device."""
+        self.transport.__enter__()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: t.Any,
     ) -> None:
-        from .transport.session import SessionV2
-
-        assert self.protocol_version == ProtocolVersion.V2
-        if pairing_method is None:
-            supported_methods = self.device_properties.pairing_methods
-            if messages.ThpPairingMethod.SkipPairing in supported_methods:
-                pairing_method = messages.ThpPairingMethod.SkipPairing
-            elif messages.ThpPairingMethod.CodeEntry in supported_methods:
-                pairing_method = messages.ThpPairingMethod.CodeEntry
-            else:
-                raise RuntimeError(
-                    "Connected Trezor does not support any trezorlib-compatible pairing method."
-                )
-        LOG.debug("Starting pairing: %r", pairing_method)
-        session = SessionV2.seedless(self)
-        app_name = self.app_name or "trezorlib"
-        host_name = self.host_name or socket.gethostname()
-        session.call(
-            messages.ThpPairingRequest(host_name=host_name, app_name=app_name),
-            expect=messages.ThpPairingRequestApproved,
-            skip_firmware_version_check=True,
-        )
-        if pairing_method is messages.ThpPairingMethod.SkipPairing:
-            return self._handle_skip_pairing(session)
-        if pairing_method is messages.ThpPairingMethod.CodeEntry:
-            return self._handle_code_entry(session)
-
-        raise RuntimeError("Unexpected pairing method")
-
-    def _handle_skip_pairing(self, session: SessionV2) -> None:
-        session.call(
-            messages.ThpSelectMethod(
-                selected_pairing_method=messages.ThpPairingMethod.SkipPairing
-            ),
-            expect=messages.ThpEndResponse,
-            skip_firmware_version_check=True,
-        )
-        assert isinstance(self.protocol, ProtocolV2Channel)
-        self.protocol._is_paired = True
-
-    def _handle_code_entry(self, session: SessionV2) -> None:
-        from .cli import get_code_entry_code
-
-        commitment_msg = session.call(
-            messages.ThpSelectMethod(
-                selected_pairing_method=messages.ThpPairingMethod.CodeEntry
-            ),
-            expect=messages.ThpCodeEntryCommitment,
-            skip_firmware_version_check=True,
-        )
-        challenge = os.urandom(16)
-        cpace_trezor_msg = session.call(
-            messages.ThpCodeEntryChallenge(challenge=challenge),
-            expect=messages.ThpCodeEntryCpaceTrezor,
-            skip_firmware_version_check=True,
-        )
+        self.transport.__exit__(exc_type, exc_value, traceback)
 
-        code = get_code_entry_code()
-        assert isinstance(session.client.protocol, ProtocolV2Channel)
-        cpace = Cpace(handshake_hash=session.client.protocol.handshake_hash)
-        cpace.random_bytes = os.urandom
-        assert cpace_trezor_msg.cpace_trezor_public_key is not None
-        cpace.generate_keys_and_secret(
-            f"{code:06}".encode("ascii"), cpace_trezor_msg.cpace_trezor_public_key
-        )
-        sha_ctx = sha256(cpace.shared_secret)
-        tag = sha_ctx.digest()
-
-        secret_msg = session.call(
-            messages.ThpCodeEntryCpaceHostTag(
-                cpace_host_public_key=cpace.host_public_key,
-                tag=tag,
-            ),
-            expect=messages.ThpCodeEntrySecret,
-            skip_firmware_version_check=True,
-        )
+    def connect(self) -> None:
+        """Establish a connection to the device.
 
-        # Check `commitment` and `code`
-        assert secret_msg.secret is not None
-        sha_ctx = sha256(secret_msg.secret)
-        computed_commitment = sha_ctx.digest()
-
-        assert commitment_msg.commitment == computed_commitment
-
-        sha_ctx = sha256(messages.ThpPairingMethod.CodeEntry.to_bytes(1, "big"))
-        sha_ctx.update(session.client.protocol.handshake_hash)
-        sha_ctx.update(secret_msg.secret)
-        sha_ctx.update(challenge)
-        code_hash = sha_ctx.digest()
-        computed_code = int.from_bytes(code_hash, "big") % 1000000
-        assert code == computed_code
-
-        session.call(
-            messages.ThpEndRequest(),
-            expect=messages.ThpEndResponse,
-            skip_firmware_version_check=True,
-        )
+        When connecting to a THP device with active screen lock, the channel
+        will fail to establish. The user must unlock their device first, even if
+        a valid credential is available.
+
+        Normally, calling `get_session()` will trigger the PIN prompt if
+        required to open the channel. However, other operations that are
+        normally "silent" from the user's perspective can fail with a
+        `DeviceLockedError`. (One notable example is reading `client.features`.)
 
-        assert isinstance(self.protocol, ProtocolV2Channel)
-        self.protocol._is_paired = True
+        `connect()` ensures an open channel, triggering a PIN unlock if
+        required. Subsequent silent operations will be able to proceed.
+
+        `connect()` does nothing if a connection is already established.
+        Notably, if the device was locked _after_ the channel was established,
+        `connect()` will not cause it to unlock. If that is your requirement,
+        use `client.ensure_unlocked()` instead.
+        """
+
+    def is_connected(self) -> bool:
+        return True
 
     def get_session(
         self,
-        passphrase: str | object = "",
+        passphrase: str | PassphraseSetting | None = PassphraseSetting.STANDARD_WALLET,
+        *,
         derive_cardano: bool = False,
-    ) -> Session:
-        """
-        Returns a new session.
+    ) -> SessionType:
+        """Get a new session with the given passphrase.
+
+        Passphrase can be provided as a string or a [`PassphraseSetting`] enum
+        value. Set `passphrase=PassphraseSetting.ON_DEVICE` to request the
+        passphrase on the device. Set `passphrase=PassphraseSetting.AUTO` to
+        automatically determine whether to request the passphrase on the device
+        based on the device's capabilities.
+
+        If passphrase is None or `PassphraseSetting.NONE`, the returned session
+        will be "seedless", that is, it will not be possible to call any methods
+        that require the user's seed (such as wallet addresses or signature
+        operations).
+
+        Use `derive_cardano=True` to request activation of Cardano-specific
+        operations in this session. If Cardano is not available, an exception
+        will be raised. (Note that Cardano operations may still be available
+        even if `derive_cardano` is set to False.)
+
+        The value of `derive_cardano` is ignored if `passphrase` is set to None.
         """
-        if isinstance(self.protocol, ProtocolV1Channel):
-            from .transport.session import SessionV1, derive_seed
-
-            if passphrase is SEEDLESS:
-                return SessionV1.new(client=self, derive_cardano=False)
-            session = SessionV1.new(
-                self,
-                derive_cardano=derive_cardano,
+        self.connect()
+        self.check_firmware_version()
+
+        if (
+            derive_cardano
+            and messages.Capability.Cardano not in self.features.capabilities
+        ):
+            raise exceptions.TrezorException("Cardano is not available on this device.")
+        if (
+            passphrase is PassphraseSetting.ON_DEVICE
+            and messages.Capability.PassphraseEntry not in self.features.capabilities
+        ):
+            raise exceptions.PassphraseError(
+                "Passphrase entry is not available on this device."
             )
-            derive_seed(session, passphrase)
-            return session
-        if isinstance(self.protocol, ProtocolV2Channel):
-            from .transport.session import SessionV2
-
-            if not self.protocol._is_paired:
-                self.do_pairing()
 
-            if passphrase is SEEDLESS:
-                return SessionV2.seedless(self)
+        if isinstance(passphrase, str):
+            passphrase = unicodedata.normalize("NFKD", passphrase)
 
-            if self._session_id_counter >= 255:
-                self._session_id_counter = 0
-
-            self._session_id_counter += 1
-
-            return SessionV2.new(
-                self, passphrase, derive_cardano, self._session_id_counter
+        passphrase_is_nonempty = isinstance(passphrase, str) and passphrase != ""
+        must_request_passphrase = (
+            passphrase_is_nonempty or passphrase is PassphraseSetting.ON_DEVICE
+        )
+        if must_request_passphrase and not self.features.passphrase_protection:
+            raise exceptions.PassphraseError(
+                "Passphrase protection is disabled on this device."
+            )
+        if passphrase_is_nonempty and self.features.passphrase_always_on_device is True:
+            raise exceptions.PassphraseError(
+                "Only on-device entry allowed for passphrase."
             )
 
-        raise NotImplementedError
-
-    def get_seedless_session(self) -> Session:
-        return self.get_session(passphrase=SEEDLESS)
+        # coerce PassphraseSetting to str, None, or ON_DEVICE
+        if passphrase is PassphraseSetting.STANDARD_WALLET:
+            passphrase = ""
+        elif passphrase is PassphraseSetting.AUTO:
+            if (
+                self.features.passphrase_protection
+                and messages.Capability.PassphraseEntry in self.features.capabilities
+            ):
+                passphrase = PassphraseSetting.ON_DEVICE
+            else:
+                passphrase = ""
+        elif passphrase is PassphraseSetting.NONE:
+            passphrase = None
 
-    def invalidate(self) -> None:
-        self._is_invalidated = True
+        return self._get_session(passphrase=passphrase, derive_cardano=derive_cardano)
 
     @property
     def features(self) -> messages.Features:
-        if self._features is None:
-            self._features = self._get_features()
-            self.check_firmware_version(warn_only=True)
-        assert self._features is not None
+        if self._features is not None:
+            return self._features
+        self._features = self._get_features()
+        self.check_firmware_version(warn_only=True)
         return self._features
 
     def _get_features(self) -> messages.Features:
-        if isinstance(self.protocol, ProtocolV2Channel):
-            if not self.protocol._is_paired:
-                self.do_pairing()
-        return self.protocol.get_features()
+        with self._get_any_session() as session:
+            resp = session.call_raw(messages.GetFeatures())
+        return messages.Features.ensure_isinstance(resp)
 
     @property
-    def protocol_version(self) -> int:
-        return self._protocol_version
+    def model(self) -> models.TrezorModel:
+        if self._model is None:
+            self._model = models.detect(self.features)
+            if self.features.vendor not in self._model.vendors:
+                raise exceptions.TrezorException(
+                    f"Unrecognized vendor: {self.features.vendor}"
+                )
+        return self._model
 
     @property
-    def model(self) -> models.TrezorModel:
-        model = models.detect(self.features)
-        if self.features.vendor not in model.vendors:
-            raise exceptions.TrezorException(
-                f"Unrecognized vendor: {self.features.vendor}"
-            )
-        return model
+    def mapping(self) -> ProtobufMapping:
+        if self._mapping is None:
+            if self._model is None:
+                # short-circuit the case where we need some mapping in order
+                # to run model detection via GetFeatures
+                return DEFAULT_MAPPING
+            self._mapping = self.model.default_mapping
+        return self._mapping
 
     @property
     def version(self) -> tuple[int, int, int]:
@@ -300,53 +430,11 @@ class TrezorClient:
         )
         return ver
 
-    @property
-    def is_invalidated(self) -> bool:
-        return self._is_invalidated
-
-    @property
-    def device_properties(self) -> messages.ThpDeviceProperties:
-        if self.protocol_version == ProtocolVersion.V1:
-            raise RuntimeError("Device properties are not avaialble with ProtocolV1.")
-        assert isinstance(self.protocol, ProtocolV2Channel)
-        if self.protocol.device_properties is None:
-            raise RuntimeError("Device properties are not avaialble.")
-        dp = self.mapping.decode_without_wire_type(
-            messages.ThpDeviceProperties, self.protocol.device_properties
-        )
-        assert isinstance(dp, messages.ThpDeviceProperties)
-        return dp
-
     def refresh_features(self) -> messages.Features:
-        self.protocol.update_features()
-        self._features = self.protocol.get_features()
-        self.check_firmware_version(warn_only=True)
-        return self._features
-
-    def _get_protocol(self) -> Channel:
-        protocol = ProtocolV1Channel(self.transport, mapping.DEFAULT_MAPPING)
-        protocol.write(messages.Initialize())
-        while True:
-            try:
-                response = protocol.read()
-            except UnexpectedMagicError:
-                continue
-            break
-
-        if isinstance(response, messages.Failure):
-            if response.code == messages.FailureType.InvalidProtocol:
-                LOG.debug("Protocol V2 detected")
-                protocol = ProtocolV2Channel(self.transport, self.mapping)
-        return protocol
-
-    def reset_protocol(self) -> None:
-        if self._protocol_version == ProtocolVersion.V1:
-            self.protocol = ProtocolV1Channel(self.transport, self.mapping)
-        elif self._protocol_version == ProtocolVersion.V2:
-            self.protocol = ProtocolV2Channel(self.transport, self.mapping)
-        else:
-            assert False
+        # clear cached features
         self._features = None
+        # trigger a refresh
+        return self.features
 
     def is_outdated(self) -> bool:
         if self.features.bootloader_mode:
@@ -358,55 +446,210 @@ class TrezorClient:
             if warn_only:
                 warnings.warn("Firmware is out of date", stacklevel=2)
             else:
-                raise exceptions.OutdatedFirmwareError(OUTDATED_FIRMWARE_ERROR)
-
-    def _write(self, msg: t.Any, session_id: int | None = None) -> None:
-        if isinstance(self.protocol, ProtocolV1Channel):
-            self.protocol.write(msg)
-        elif isinstance(self.protocol, ProtocolV2Channel):
-            assert session_id is not None
-            self.protocol.write(session_id=session_id, msg=msg)
-        else:
-            raise Exception("Unknown client protocol")
-
-    def _read(self, session_id: int | None = None) -> t.Any:
-        if isinstance(self.protocol, ProtocolV1Channel):
-            return self.protocol.read()
-        elif isinstance(self.protocol, ProtocolV2Channel):
-            assert session_id is not None
-            return self.protocol.read(session_id=session_id)
+                raise exceptions.OutdatedFirmwareError
+
+    def _call_raw(
+        self,
+        session: SessionType,
+        msg: MessageType,
+        timeout: float | None = None,
+    ) -> MessageType:
+        """Send a message to the transport and return the raw response.
+
+        Does not perform any sort of handling on the response: errors are not
+        converted to exceptions, internal workflow callbacks are not triggered.
+        """
+        self._write(session, msg)
+        return self._read(session, timeout)
+
+    def ping(
+        self,
+        message: str,
+        button_protection: bool | None = None,
+        timeout: float | None = None,
+    ) -> str:
+        with self._get_any_session() as session:
+            resp = session.call(
+                messages.Ping(message=message, button_protection=button_protection),
+                expect=messages.Success,
+                timeout=timeout,
+            )
+            assert resp.message is not None
+            return resp.message
+
+    def _call(
+        self,
+        session: SessionType,
+        msg: MessageType,
+        *,
+        expect: type[MT] = MessageType,
+        timeout: float | None = None,
+    ) -> MT:
+        resp = session.call_raw(msg, timeout=timeout)
+        while True:
+            if isinstance(resp, messages.PinMatrixRequest):
+                resp = self._callback_pin(session, resp)
+            elif isinstance(resp, messages.ButtonRequest):
+                resp = self._callback_button(session, resp)
+            elif isinstance(resp, messages.Failure):
+                if resp.code in (
+                    messages.FailureType.ActionCancelled,
+                    messages.FailureType.PinCancelled,
+                ):
+                    raise exceptions.Cancelled
+                elif resp.code == messages.FailureType.InvalidSession:
+                    raise exceptions.InvalidSessionError(session.id)
+                raise exceptions.TrezorFailure(resp)
+            elif isinstance(resp, messages.PassphraseRequest):
+                raise exceptions.InvalidSessionError(session.id, from_message=resp)
+            elif not isinstance(resp, expect):
+                raise exceptions.UnexpectedMessageError(expect, resp)
+            else:
+                return resp
+
+    def _callback_pin(
+        self, session: SessionType, msg: messages.PinMatrixRequest
+    ) -> MessageType:
+        try:
+            pin = self.app._callback_pin(msg)
+        except exceptions.Cancelled:
+            session.call_raw(messages.Cancel())
+            raise
+
+        if any(d not in "123456789" for d in pin) or not (
+            1 <= len(pin) <= MAX_PIN_LENGTH
+        ):
+            session.call_raw(messages.Cancel())
+            raise ValueError("Invalid PIN provided")
+
+        resp = session.call_raw(messages.PinMatrixAck(pin=pin))
+        if isinstance(resp, messages.Failure) and resp.code in (
+            messages.FailureType.PinInvalid,
+            messages.FailureType.PinCancelled,
+            messages.FailureType.PinExpected,
+        ):
+            raise exceptions.PinException(resp.code, resp.message)
         else:
-            raise Exception("Unknown client protocol")
+            return resp
+
+    def _callback_button(
+        self, session: SessionType, msg: messages.ButtonRequest
+    ) -> MessageType:
+        __tracebackhide__ = True  # for pytest # pylint: disable=W0612
+        # do this raw - send ButtonAck first, notify UI later
+        session.write(messages.ButtonAck())
+        self.app._callback_button(msg)
+        return session.read()
+
+    def cancel(self) -> None:
+        """Send a Cancel signal to the device, interrupting the current workflow."""
+        try:
+            with self._get_any_session() as session:
+                session.call_raw(messages.Cancel())
+        except exceptions.Cancelled:
+            pass
+
+    def lock(self, *, _use_session: SessionType | None = None) -> None:
+        """Lock the device with a PIN prompt, if enabled."""
+        session = _use_session or self._get_any_session()
+        with session:
+            session.call_raw(messages.LockDevice())
+            self.refresh_features()
+
+    def ensure_unlocked(self) -> None:
+        """Ensure the device is unlocked."""
+        session = self.get_session(passphrase=PassphraseSetting.STANDARD_WALLET)
+        with session:
+            session.ensure_unlocked()
+
+    def _invalidate(self) -> None:
+        """Invalidate the client after a device wipe.
+
+        All state that is no longer valid after a wipe should be cleared here.
+        """
+        self._features = None
 
 
 def get_default_client(
-    path: t.Optional[str] = None,
+    app_name: str,
+    path_or_transport: str | Transport | None = None,
+    *,
+    credentials: t.Collection[Credential] = (),
+    button_callback: t.Callable[[messages.ButtonRequest], None] | None = None,
+    pin_callback: t.Callable[[messages.PinMatrixRequest], str] | None = None,
+    code_entry_callback: t.Callable[[], str] | None = None,
     **kwargs: t.Any,
 ) -> "TrezorClient":
     """Get a client for a connected Trezor device.
 
     Returns a TrezorClient instance with minimum fuss.
 
-    Transport is opened and should be closed after you're done with the client.
-
     If path is specified, does a prefix-search for the specified device. Otherwise, uses
     the value of TREZOR_PATH env variable, or finds first connected Trezor.
     """
-
-    if path is None:
-        path = os.getenv("TREZOR_PATH")
-
-    transport = get_transport(path, prefix_search=True)
-    transport.open()
-
-    return TrezorClient(transport, **kwargs)
-
-
-def get_callback_passphrase_v1(
-    passphrase: str = "",
-) -> t.Callable[[Session, t.Any], t.Any] | None:
-
-    def _callback_passphrase_v1(session: Session, msg: t.Any) -> t.Any:
-        return session.call(messages.PassphraseAck(passphrase=passphrase))
-
-    return _callback_passphrase_v1
+    if path_or_transport is None:
+        path_or_transport = os.getenv("TREZOR_PATH")
+    if isinstance(path_or_transport, Transport):
+        transport = path_or_transport
+    else:
+        transport = get_transport(path_or_transport, prefix_search=True)
+
+    app = AppManifest(
+        app_name=app_name,
+        credentials=credentials,
+        button_callback=button_callback,
+        pin_callback=pin_callback,
+    )
+    client = get_client(app, transport, **kwargs)
+
+    if not client.pairing.is_paired():
+        from .thp.pairing import default_pairing_flow
+
+        default_pairing_flow(client.pairing, code_entry_callback=code_entry_callback)
+    return client
+
+
+def get_default_session(
+    client: TrezorClient,
+    passphrase_callback: t.Callable[[], str] | None = None,
+    *,
+    derive_cardano: bool = False,
+) -> Session:
+    """Get a default session for a connected Trezor device.
+
+    The first argument must be a previously created and paired client instance,
+    e.g., via `get_client` or `get_default_client`.
+
+    The logic for determining what passphrase to use is as follows:
+
+    1. If the device has passphrase disabled, the default wallet is used
+    2. If the device allows on-device entry, passphrase is requested on the
+       device
+    3. If `passphrase_callback` is provided, it is used to get the passphrase
+    4. Otherwise, the default wallet is used
+    """
+    passphrase = PassphraseSetting.STANDARD_WALLET
+    client.ensure_unlocked()
+    if client.features.passphrase_protection:
+        if messages.Capability.PassphraseEntry in client.features.capabilities:
+            passphrase = PassphraseSetting.ON_DEVICE
+        elif passphrase_callback is not None:
+            passphrase = passphrase_callback()
+    return client.get_session(passphrase=passphrase, derive_cardano=derive_cardano)
+
+
+def get_client(
+    app: AppManifest,
+    transport: Transport,
+    *,
+    mapping: ProtobufMapping | None = None,
+    model: models.TrezorModel | None = None,
+) -> TrezorClient:
+    from .protocol_v1 import TrezorClientV1, probe
+    from .thp.client import TrezorClientThp
+
+    if probe(transport):
+        cls = TrezorClientV1
+    else:
+        cls = TrezorClientThp
+    return cls(app=app, transport=transport, mapping=mapping, model=model)
diff --git a/python/src/trezorlib/debuglink.py b/python/src/trezorlib/debuglink.py
index 0781e877..914ef258 100644
--- a/python/src/trezorlib/debuglink.py
+++ b/python/src/trezorlib/debuglink.py
@@ -32,15 +32,13 @@ from pathlib import Path
 
 from mnemonic import Mnemonic
 
-from . import btc, mapping, messages, models, protobuf
-from .client import ProtocolVersion, TrezorClient
-from .exceptions import Cancelled, TrezorFailure, UnexpectedMessageError
+from . import client, mapping, messages, models, protobuf, protocol_v1
+from .exceptions import DeviceLockedError, TrezorFailure
 from .log import DUMP_BYTES
 from .messages import DebugTouchEventType, DebugWaitType
-from .tools import parse_path
+from .thp.channel import Channel
+from .thp.client import TrezorClientThp
 from .transport import Timeout
-from .transport.session import ProtocolV2Channel, Session
-from .transport.thp.protocol_v1 import ProtocolV1Channel
 
 if t.TYPE_CHECKING:
     from typing_extensions import Protocol
@@ -48,9 +46,12 @@ if t.TYPE_CHECKING:
     from .transport import Transport
 
     ExpectedMessage = t.Union[
-        protobuf.MessageType, t.Type[protobuf.MessageType], "MessageFilter"
+        protobuf.MessageType, type[protobuf.MessageType], "MessageFilter"
     ]
 
+    ExpectedResponse = t.Union[ExpectedMessage, tuple[bool, ExpectedMessage]]
+    ExpectedResponses = t.Sequence[ExpectedResponse]
+
     AnyDict = t.Dict[str, t.Any]
     Coords = t.Tuple[int, int]
 
@@ -63,9 +64,11 @@ if t.TYPE_CHECKING:
 
     InputFlowType = t.Generator[None, messages.ButtonRequest, None]
 
+T = t.TypeVar("T")
+S = t.TypeVar("S", bound=client.Session)
+MT = t.TypeVar("MT", bound=protobuf.MessageType)
 
 EXPECTED_RESPONSES_CONTEXT_LINES = 3
-PASSPHRASE_TEST_PATH = parse_path("44h/1h/0h/0/0")
 
 LOG = logging.getLogger(__name__)
 
@@ -98,6 +101,11 @@ class LayoutType(Enum):
         return f"LayoutType.{self.name}"
 
 
+class ProtocolVersion(Enum):
+    V1 = "v1"
+    THP = "thp"
+
+
 class UnstructuredJSONReader:
     """Contains data-parsing helpers for JSON data that have unknown structure."""
 
@@ -509,11 +517,16 @@ def _make_input_func(
 
 class DebugLink:
     def __init__(self, transport: "Transport", auto_interact: bool = True) -> None:
+        try:
+            self.transport.close()
+        except Exception:
+            pass
+        transport.open()
+
         self.transport = transport
         self.allow_interactions = auto_interact
         self.mapping = mapping.DEFAULT_MAPPING
 
-        self.protocol = ProtocolV1Channel(self.transport, self.mapping)
         # To be set by TrezorClientDebugLink (is not known during creation time)
         self.model: models.TrezorModel | None = None
         self.version: tuple[int, int, int] = (0, 0, 0)
@@ -589,25 +602,29 @@ class DebugLink:
             DUMP_BYTES,
             f"encoded as type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
         )
-        self.protocol.write(msg)
+        protocol_v1.write(self.transport, msg_type, msg_bytes)
 
     def _read(self, timeout: float | None = None) -> protobuf.MessageType:
-        msg = self.protocol.read(timeout=timeout)
+        msg_type, msg_bytes = protocol_v1.read(self.transport, timeout=timeout)
+        msg = self.mapping.decode(msg_type, msg_bytes)
 
         # Collapse tokens to make log use less lines.
-        msg_for_log = msg
         if isinstance(msg, (messages.DebugLinkState, messages.DebugLinkLayout)):
-            msg_for_log = deepcopy(msg)
-            msg_for_log.tokens = ["".join(msg_for_log.tokens)]
+            msg.tokens = ["".join(msg.tokens)]
 
         return msg
 
-    def _call(self, msg: protobuf.MessageType, timeout: float | None = None) -> t.Any:
+    def _call(
+        self,
+        msg: protobuf.MessageType,
+        timeout: float | None = None,
+        expect: type[MT] = protobuf.MessageType,
+    ) -> MT:
         self._write(msg)
         result = self._read(timeout=timeout)
         if isinstance(result, messages.Failure):
             raise TrezorFailure(result)
-        return result
+        return expect.ensure_isinstance(result)
 
     def state(self, wait_type: DebugWaitType | None = None) -> messages.DebugLinkState:
         if wait_type is None:
@@ -617,10 +634,8 @@ class DebugLink:
                 else DebugWaitType.IMMEDIATE
             )
         result = self._call(messages.DebugLinkGetState(wait_layout=wait_type))
-        while not isinstance(result, (messages.Failure, messages.DebugLinkState)):
+        while not isinstance(result, messages.DebugLinkState):
             result = self._read()
-        if isinstance(result, messages.Failure):
-            raise TrezorFailure(result)
         return result
 
     def pairing_info(
@@ -636,10 +651,8 @@ class DebugLink:
                 nfc_secret_host=nfc_secret_host,
             )
         )
-        while not isinstance(result, (messages.Failure, messages.DebugLinkPairingInfo)):
+        while not isinstance(result, messages.DebugLinkPairingInfo):
             result = self._read()
-        if isinstance(result, messages.Failure):
-            raise TrezorFailure(result)
         return result
 
     def read_layout(self, wait: bool | None = None) -> LayoutContent:
@@ -666,7 +679,8 @@ class DebugLink:
             self.reset_debug_events()
 
         obj = self._call(
-            messages.DebugLinkGetState(wait_layout=DebugWaitType.NEXT_LAYOUT)
+            messages.DebugLinkGetState(wait_layout=DebugWaitType.NEXT_LAYOUT),
+            expect=messages.DebugLinkState,
         )
         return LayoutContent(obj.tokens)
 
@@ -686,8 +700,7 @@ class DebugLink:
             self.waiting_for_layout_change = False
 
         # wait for the reply
-        resp = self._read()
-        assert isinstance(resp, messages.DebugLinkState)
+        messages.DebugLinkState.ensure_isinstance(self._read())
 
     @contextmanager
     def hold_touch(self, pos: tuple[int, int]) -> t.Iterator[None]:
@@ -715,8 +728,7 @@ class DebugLink:
     def reset_debug_events(self) -> None:
         # Only supported on TT and above certain version
         if (self.model is not models.T1B1) and not self.legacy_debug:
-            return self._call(messages.DebugLinkResetDebugEvents())
-        return None
+            self._call(messages.DebugLinkResetDebugEvents(), expect=messages.Success)
 
     def synchronize_at(
         self, layout_text: str | list[str], timeout: float = 5
@@ -739,7 +751,7 @@ class DebugLink:
         The message is missing on T1. Use `TrezorClientDebugLink.watch_layout` for
         cross-version compatibility.
         """
-        self._call(messages.DebugLinkWatchLayout(watch=watch))
+        self._call(messages.DebugLinkWatchLayout(watch=watch), expect=messages.Success)
 
     def encode_pin(self, pin: str, matrix: str | None = None) -> str:
         """Transform correct PIN according to the displayed matrix."""
@@ -756,7 +768,8 @@ class DebugLink:
         return (state.recovery_fake_word, state.recovery_word_pos)
 
     def read_reset_word(self) -> str:
-        state = self._call(messages.DebugLinkGetState(wait_word_list=True))
+        state = self.state()
+        assert state.reset_word is not None
         return state.reset_word
 
     def _decision(
@@ -855,8 +868,8 @@ class DebugLink:
     def stop(self) -> None:
         self._write(messages.DebugLinkStop())
 
-    def reseed(self, value: int) -> protobuf.MessageType:
-        return self._call(messages.DebugLinkReseedRandom(value=value))
+    def reseed(self, value: int) -> None:
+        self._call(messages.DebugLinkReseedRandom(value=value), expect=messages.Success)
 
     def start_recording(self, directory: str, refresh_index: int | None = None) -> None:
         self.screenshot_recording_dir = directory
@@ -865,7 +878,8 @@ class DebugLink:
             self._call(
                 messages.DebugLinkRecordScreen(
                     target_directory=directory, refresh_index=refresh_index
-                )
+                ),
+                expect=messages.Success,
             )
         else:
             self.t1_screenshot_directory = Path(directory)
@@ -893,11 +907,10 @@ class DebugLink:
     def flash_erase(self, sector: int) -> None:
         self._write(messages.DebugLinkFlashErase(sector=sector))
 
-    def erase_sd_card(self, format: bool = True) -> messages.Success:
-        res = self._call(messages.DebugLinkEraseSdCard(format=format))
-        if not isinstance(res, messages.Success):
-            raise UnexpectedMessageError(messages.Success, res)
-        return res
+    def erase_sd_card(self, format: bool = True) -> None:
+        self._call(
+            messages.DebugLinkEraseSdCard(format=format), expect=messages.Success
+        )
 
     def snapshot_legacy(self) -> None:
         """Snapshot the current state of the device."""
@@ -940,10 +953,9 @@ class DebugLink:
         if not self.has_gc_info:
             return
 
-        resp = self._call(messages.DebugLinkGetGcInfo())
-        while not isinstance(resp, messages.DebugLinkGcInfo):
-            resp = self._read()
-
+        resp = self._call(
+            messages.DebugLinkGetGcInfo(), expect=messages.DebugLinkGcInfo
+        )
         info = dict(sorted((item.name, item.value) for item in resp.items))
         if info["total"]:
             LOG.debug(
@@ -980,16 +992,19 @@ class NullDebugLink(DebugLink):
     def close(self) -> None:
         pass
 
-    def _call(
-        self, msg: protobuf.MessageType, nowait: bool = False
-    ) -> messages.DebugLinkState | None:
-        if not nowait:
-            if isinstance(msg, messages.DebugLinkGetState):
-                return messages.DebugLinkState()
-            else:
-                raise RuntimeError("unexpected call to a fake debuglink")
+    def _write(self, msg: protobuf.MessageType) -> None:
+        pass
 
-        return None
+    def _call(
+        self,
+        msg: protobuf.MessageType,
+        timeout: float | None = None,
+        expect: type[MT] = protobuf.MessageType,
+    ) -> protobuf.MessageType:
+        if isinstance(msg, messages.DebugLinkGetState):
+            return messages.DebugLinkState()
+        else:
+            raise RuntimeError("unexpected call to a fake debuglink")
 
 
 class UnexpectedMenuError(Exception):
@@ -1155,7 +1170,7 @@ class DebugUI:
             except StopIteration:
                 self.input_flow = self.INPUT_FLOW_DONE
 
-    def get_pin(self) -> str:
+    def get_pin(self, _request: messages.PinMatrixRequest | None = None) -> str:
         self.debuglink.snapshot_legacy()
 
         if self.pins is None:
@@ -1166,16 +1181,8 @@ class DebugUI:
         except StopIteration:
             raise AssertionError("PIN sequence ended prematurely")
 
-    def get_passphrase(self, available_on_device: bool) -> str | None | object:
-        self.debuglink.snapshot_legacy()
-        return self.passphrase
-
-    def confirm_screen(self) -> None:
-        self.debuglink.press_yes()
-
 
 class MessageFilter:
-
     def __init__(
         self, message_type: t.Type[protobuf.MessageType], **fields: t.Any
     ) -> None:
@@ -1269,105 +1276,41 @@ class MessageFilterGenerator:
 message_filters = MessageFilterGenerator()
 
 
-class SessionDebugWrapper(Session):
-    def __init__(self, session: Session) -> None:
-        if isinstance(session, SessionDebugWrapper):
-            raise Exception("Cannot wrap already wrapped session!")
-        self.__dict__["_session"] = session
-
-    def __getattr__(self, name: str) -> t.Any:
-        return getattr(self._session, name)
-
-    def __setattr__(self, name: str, value: t.Any) -> None:
-        if hasattr(self._session, name):
-            setattr(self._session, name, value)
-        else:
-            self.__dict__[name] = value
-
-    @property
-    def protocol_version(self) -> int:
-        return self.client.protocol_version
-
-    @property
-    def debug_client(self) -> TrezorClientDebugLink:
-        if not isinstance(self.client, TrezorClientDebugLink):
-            raise Exception("Debug client not available")
-        return self.client
-
-    def _write(self, msg: t.Any) -> None:
-        if isinstance(self.client, TrezorClientDebugLink):
-            msg = self.client._filter_message(msg)
-        self._session._write(msg)
-
-    def _read(self, timeout: float | None = None) -> t.Any:
-        msg = self._session._read(timeout)
-        if isinstance(self.client, TrezorClientDebugLink):
-            msg = self.client._filter_message(msg)
-        return msg
-
-    def resume(self) -> None:
-        self._session.resume()
-
-    def lock(self) -> None:
-        """Lock the device.
-
-        If the device does not have a PIN configured, this will do nothing.
-        Otherwise, a lock screen will be shown and the device will prompt for PIN
-        before further actions.
-
-        This call does _not_ invalidate passphrase cache. If passphrase is in use,
-        the device will not prompt for it after unlocking.
+if t.TYPE_CHECKING:
 
-        To invalidate passphrase cache, use `session.end()`. To lock _and_ invalidate
-        passphrase cache, use `session.lock()` followed by `session.end()`.
-        """
-        self.call(messages.LockDevice())
-        self.refresh_features()
+    class DebugSession(client.Session[client.TrezorClient, t.Any]):
+        test_ctx: "TrezorTestContext"
+        debug: DebugLink
+        layout_type: LayoutType
 
-    def ensure_unlocked(self) -> None:
-        btc.get_address(self, "Testnet", PASSPHRASE_TEST_PATH)
-        self.refresh_features()
+else:
+    DebugSession = "do not use this type at runtime"
 
 
 class DebugLinkNotFound(Exception):
     pass
 
 
-class TrezorClientDebugLink(TrezorClient):
-    # This class implements automatic responses
-    # and other functionality for unit tests
-    # for various callbacks, created in order
-    # to automatically pass unit tests.
-    #
-    # This mixing should be used only for purposes
-    # of unit testing, because it will fail to work
-    # without special DebugLink interface provided
-    # by the device.
-
-    protocol: ProtocolV1Channel | ProtocolV2Channel
-    actual_responses: list[protobuf.MessageType] | None = None
-    filters: t.Dict[
-        t.Type[protobuf.MessageType],
-        t.Callable[[protobuf.MessageType], protobuf.MessageType] | None,
-    ] = {}
+class TrezorTestContext:
+    # This class implements automatic responses and other functionality for unit
+    # tests for various callbacks, created in order to automatically pass unit
+    # tests.
 
     def __init__(
         self,
         transport: Transport,
+        *,
         auto_interact: bool = True,
-        open_transport: bool = True,
         debug_transport: Transport | None = None,
-        app_name: str = "trezorlib-debug",
-        host_name: str = "testhost",
+        force_wipe: bool = False,
+        host_name: str = "debughost",
     ) -> None:
         try:
             debug_transport = debug_transport or transport.find_debug()
+            with debug_transport:
+                if not debug_transport.is_ready():
+                    raise DebugLinkNotFound(debug_transport.get_path())
             self.debug = DebugLink(debug_transport, auto_interact)
-            if open_transport:
-                self.debug.open()
-                # try to open debuglink, see if it works
-                if not self.debug.transport.ping():
-                    raise DebugLinkNotFound(self.debug.transport.get_path())
 
         except Exception:
             if not auto_interact:
@@ -1375,85 +1318,170 @@ class TrezorClientDebugLink(TrezorClient):
             else:
                 raise
 
-        if open_transport:
-            transport.open()
-
-        # set transport explicitly so that sync_responses can work
         self.transport = transport
-        self.ui: DebugUI = DebugUI(self.debug)
+        self.app = client.AppManifest(app_name="debuglink", host_name=host_name)
 
-        def get_pin(_msg: messages.PinMatrixRequest) -> str:
-            try:
-                pin = self.ui.get_pin()
-            except Cancelled:
+        if protocol_v1.probe(self.transport):
+            self.protocol_version = ProtocolVersion.V1
+        else:
+            self.protocol_version = ProtocolVersion.THP
+
+        try:
+            self.reset_instance()
+            # peek at features to maybe trigger the DeviceLockedError
+            _ = self.client.features
+        except DeviceLockedError:
+            if force_wipe:
+                self.debug._call(messages.WipeDevice(), expect=messages.Success)
+                self.reset_instance()
+            else:
                 raise
-            return pin
 
-        self.pin_callback = get_pin
-        self.button_callback = self.ui.button_request
+        self.capabilities = self.client.features.capabilities
+        self.sd_card_present = self.client.features.sd_card_present
+        self.debug.version = self.version = self.client.version
+        self.debug.model = self.model = self.client.model
+        self.layout_type = self.debug.layout_type
+
+    def _get_client(self) -> client.TrezorClient:
+        if self.protocol_version is ProtocolVersion.V1:
+            cls = protocol_v1.TrezorClientV1
+        else:
+            cls = TrezorClientThp
+        client = cls(self.app, self.transport, model=None, mapping=None)
+        client._write = self._wrap_write(client._write)  # type: ignore [Missing keyword parameter;;"SessionV1" is not assignable to "ThpSession"]
+        client._read = self._wrap_read(client._read)  # type: ignore [Missing keyword parameter;;"SessionV1" is not assignable to "ThpSession"]
+        client._invalidate = self._wrap_invalidate(client._invalidate)
+        return client
+
+    def reset_instance(self) -> None:
+        try:
+            height = self.transport._opened
+            self.transport.close()
+        except Exception:
+            pass
+        self.transport.open()
+        self.transport._opened = height
+
+        # debug-specific initialization
+        self.ui: DebugUI = DebugUI(self.debug)
+        self.app.pin_callback = self.ui.get_pin
+        self.app.button_callback = self.ui.button_request
+        self.reset_debug_features()
+
+        self.client = self._get_client()
 
-        super().__init__(transport, app_name=app_name, host_name=host_name)
+        self.pairing = self.client.pairing
+        if self.client.is_connected():
+            self.pairing.skip()
         self.sync_responses()
 
-        # So that we can choose right screenshotting logic (T1 vs TT)
-        # and know the supported debug capabilities
-        if self.protocol_version is ProtocolVersion.V2:
-            assert isinstance(self.protocol, ProtocolV2Channel)
-            self.do_pairing(pairing_method=messages.ThpPairingMethod.SkipPairing)
-        self.debug.model = self.model
-        self.debug.version = self.version
+    def reset_debug_features(self) -> None:
+        """Prepare the debugging session for a new testcase.
 
-        self.reset_debug_features()
+        Clears all debugging state that might have been modified by a testcase.
+        """
+        self.debug.input_wait_type = DebugWaitType.IMMEDIATE
+        self.ui.clear()
+        self.in_with_statement = False
+        self.expected_responses: list[MessageFilter] | None = None
+        self.actual_responses: list[protobuf.MessageType] = []
+        self.filters: dict[
+            type[protobuf.MessageType],
+            t.Callable[[protobuf.MessageType], protobuf.MessageType],
+        ] = {}
+
+    # === required overrides for the base class ===
+
+    def _wrap_write(
+        self, write_fn: t.Callable[[S, protobuf.MessageType], None]
+    ) -> t.Callable[[S, protobuf.MessageType], None]:
+        def wrapped_write(session: S, msg: protobuf.MessageType) -> None:
+            __tracebackhide__ = True  # for pytest # pylint: disable=W0612
+            filtered_msg = self._filter_message(msg)
+            write_fn(session, filtered_msg)
+
+        return wrapped_write
+
+    def _wrap_read(
+        self, read_fn: t.Callable[[S, float | None], protobuf.MessageType]
+    ) -> t.Callable[[S, float | None], protobuf.MessageType]:
+        def wrapped_read(
+            session: S, timeout: float | None = None
+        ) -> protobuf.MessageType:
+            __tracebackhide__ = True  # for pytest # pylint: disable=W0612
+            inner_resp = read_fn(session, timeout)
+            resp = self._filter_message(inner_resp)
+            self.actual_responses.append(resp)
+            return resp
+
+        return wrapped_read
+
+    def _wrap_invalidate(
+        self, invalidate_fn: t.Callable[[], None]
+    ) -> t.Callable[[], None]:
+        def wrapped_invalidate() -> None:
+            __tracebackhide__ = True  # for pytest # pylint: disable=W0612
+            invalidate_fn()
+            self.reset_instance()
+
+        return wrapped_invalidate
 
     @property
-    def layout_type(self) -> LayoutType:
-        return self.debug.layout_type
-
-    def get_new_client(self) -> TrezorClientDebugLink:
-        new_client = TrezorClientDebugLink(
-            self.transport,
-            self.debug.allow_interactions,
-            open_transport=False,
-            debug_transport=self.debug.transport,
-        )
-        new_client.debug.screenshot_recording_dir = self.debug.screenshot_recording_dir
-        new_client.debug.t1_screenshot_directory = self.debug.t1_screenshot_directory
-        new_client.debug.t1_screenshot_counter = self.debug.t1_screenshot_counter
-        new_client.debug.t1_take_screenshots = self.debug.t1_take_screenshots
-        new_client.debug.prev_gc_info = self.debug.prev_gc_info
-        return new_client
-
-    def close_transport(self) -> None:
-        self.transport.close()
-        self.debug.close()
+    def features(self) -> messages.Features:
+        return self.client.features
 
-    def lock(self) -> None:
-        s = self.get_seedless_session()
-        s.lock()
+    def refresh_features(self) -> messages.Features:
+        return self.client.refresh_features()
+
+    @property
+    def channel(self) -> Channel:
+        if self.is_thp():
+            assert isinstance(self.client, TrezorClientThp)
+            return self.client.channel
+        raise AttributeError("Channel is not available for this protocol")
+
+    @channel.setter
+    def channel(self, channel: Channel) -> None:
+        if self.is_thp():
+            assert isinstance(self.client, TrezorClientThp)
+            self.client.channel = channel
+            return
+        raise AttributeError("Channel is not available for this protocol")
+
+    def is_protocol_v1(self) -> bool:
+        return self.protocol_version is ProtocolVersion.V1
+
+    def is_thp(self) -> bool:
+        return self.protocol_version is ProtocolVersion.THP
+
+    def _wrap_session(self, session: client.Session) -> DebugSession:
+        dbg_session = t.cast(DebugSession, session)
+        dbg_session.debug = self.debug
+        dbg_session.test_ctx = self
+        dbg_session.layout_type = self.layout_type
+        return dbg_session
 
     def get_session(
         self,
-        passphrase: str | object = "",
+        passphrase: (
+            str | client.PassphraseSetting | None
+        ) = client.PassphraseSetting.STANDARD_WALLET,
         derive_cardano: bool = False,
-    ) -> SessionDebugWrapper:
-        if isinstance(passphrase, str):
-            passphrase = Mnemonic.normalize_string(passphrase)
-        session = SessionDebugWrapper(
-            super().get_session(
-                passphrase,
-                derive_cardano,
-            )
+    ) -> DebugSession:
+        session = self.client.get_session(
+            passphrase=passphrase, derive_cardano=derive_cardano
         )
-        return session
+        return self._wrap_session(session)
 
-    # FIXME: can be deleted
-    def get_seedless_session(
-        self, *args: t.Any, **kwargs: t.Any
-    ) -> SessionDebugWrapper:
-        session = super().get_seedless_session(*args, **kwargs)
-        if not isinstance(session, SessionDebugWrapper):
-            session = SessionDebugWrapper(session)
-        return session
+    def get_seedless_session(self) -> DebugSession:
+        return self.get_session(passphrase=None)
+
+    def lock(self) -> None:
+        self.client.lock()
+
+    def ping(self, message: str) -> str:
+        return self.client.ping(message)
 
     def watch_layout(self, watch: bool = True) -> None:
         """Enable or disable watching layout changes.
@@ -1489,28 +1517,10 @@ class TrezorClientDebugLink(TrezorClient):
         This function will call `Ping` and read responses until it locates a `Success`
         with the expected text. This means that we are reading up-to-date responses.
         """
-        import secrets
-
-        if self.protocol_version is ProtocolVersion.V1:
-            assert isinstance(self.protocol, ProtocolV1Channel)
-            if self.model is models.T1B1:
-                # Start by canceling whatever is on screen. This will work to cancel T1 PIN
-                # prompt, which is in TINY mode and does not respond to `Ping`.
-                self.protocol.write(messages.Cancel())
-
-            message = "SYNC" + secrets.token_hex(8)
-            self.protocol.write(messages.Ping(message=message))
-            success = messages.Success(message=message)
-            while True:
-                try:
-                    if self.protocol.read() == success:
-                        return
-                except Exception:
-                    pass
-
-        if self.protocol_version is ProtocolVersion.V2:
-            assert isinstance(self.protocol, ProtocolV2Channel)
-            self.protocol.sync_responses()
+        if self.is_protocol_v1():
+            protocol_v1.sync_responses(self.transport)
+        else:
+            self.channel.sync_responses()
 
     def mnemonic_callback(self, _: t.Any) -> str:
         word, pos = self.debug.read_recovery_word()
@@ -1521,10 +1531,7 @@ class TrezorClientDebugLink(TrezorClient):
 
         raise RuntimeError("Unexpected call")
 
-    def set_expected_responses(
-        self,
-        expected: list["ExpectedMessage" | t.Tuple[bool, "ExpectedMessage"]],
-    ) -> None:
+    def set_expected_responses(self, expected: ExpectedResponses) -> None:
         """Set a sequence of expected responses to session calls.
 
         Within a given with-block, the list of received responses from device must
@@ -1579,7 +1586,10 @@ class TrezorClientDebugLink(TrezorClient):
         if not self.in_with_statement:
             raise RuntimeError("Must be called inside 'with' statement")
 
-        self.filters[message_type] = callback
+        if callback is None:
+            del self.filters[message_type]
+        else:
+            self.filters[message_type] = callback
 
     def _filter_message(self, msg: protobuf.MessageType) -> protobuf.MessageType:
         message_type = msg.__class__
@@ -1589,21 +1599,7 @@ class TrezorClientDebugLink(TrezorClient):
         else:
             return msg
 
-    def reset_debug_features(self) -> None:
-        """Prepare the debugging session for a new testcase.
-
-        Clears all debugging state that might have been modified by a testcase.
-        """
-        self.ui.clear()
-        self.in_with_statement = False
-        self.expected_responses: list[MessageFilter] | None = None
-        self.actual_responses: list[protobuf.MessageType] | None = None
-        self.filters: t.Dict[
-            t.Type[protobuf.MessageType],
-            t.Callable[[protobuf.MessageType], protobuf.MessageType] | None,
-        ] = {}
-
-    def __enter__(self) -> "TrezorClientDebugLink":
+    def __enter__(self) -> "TrezorTestContext":
         # For usage in with/expected_responses
         if self.in_with_statement:
             raise RuntimeError("Do not nest!")
@@ -1633,22 +1629,19 @@ class TrezorClientDebugLink(TrezorClient):
             # Propagate the exception through the input flow, so that we see in
             # traceback where it is stuck.
             input_flow.throw(value)
-        self.actual_responses = None
+        self.actual_responses = []
 
     @classmethod
     def _verify_responses(
         cls,
         expected: list[MessageFilter] | None,
-        actual: list[protobuf.MessageType] | None,
+        actual: list[protobuf.MessageType],
     ) -> None:
         __tracebackhide__ = True  # for pytest # pylint: disable=W0612
 
-        if expected is None and actual is None:
+        if expected is None:
             return
 
-        assert expected is not None
-        assert actual is not None
-
         for i, (exp, act) in enumerate(zip_longest(expected, actual)):
             if exp is None:
                 output = cls._expectation_lines(expected, i)
@@ -1728,16 +1721,27 @@ class TrezorClientDebugLink(TrezorClient):
 
         next(input_flow)  # start the generator
 
-    def notify_read(self, msg: protobuf.MessageType) -> None:
-        try:
-            if self.actual_responses is not None:
-                self.actual_responses.append(msg)
-        except Exception as e:
-            print(e)
+    def wipe_device(self, reseed: bool = True) -> None:
+        """Wipe device storage.
+
+        Unless disabled via `reseed=False`, reseed the device before wiping,
+        so that we get consistent storage contents (specifically device id)
+        after the wipe.
+        """
+        if reseed:
+            self.debug.reseed(0)
+        if self.model is models.T1B1:
+            self.client._get_any_session().call(
+                messages.WipeDevice(),
+                expect=messages.Success,
+            )
+        else:
+            self.debug._call(messages.WipeDevice(), expect=messages.Success)
+        self.reset_instance()
 
 
 def load_device(
-    session: "Session",
+    session: client.Session,
     mnemonic: str | t.Iterable[str],
     pin: str | None,
     passphrase_protection: bool,
@@ -1777,7 +1781,7 @@ def load_device(
 load_device_by_mnemonic = load_device
 
 
-def prodtest_t1(session: "Session") -> None:
+def prodtest_t1(session: client.Session) -> None:
     if session.features.bootloader_mode is not True:
         raise RuntimeError("Device must be in bootloader mode")
 
@@ -1790,7 +1794,7 @@ def prodtest_t1(session: "Session") -> None:
 
 
 def record_screen(
-    debug_client: "TrezorClientDebugLink",
+    debug_client: "TrezorTestContext",
     directory: str | None,
     report_func: t.Callable[[str], None] | None = None,
 ) -> None:
@@ -1815,7 +1819,8 @@ def record_screen(
         raise RuntimeError("Recording is only supported on emulator.")
 
     if directory is None:
-        debug_client.debug.stop_recording()
+        with debug_client.debug.transport:
+            debug_client.debug.stop_recording()
         if report_func is not None:
             report_func("Recording stopped.")
     else:
@@ -1827,12 +1832,13 @@ def record_screen(
             abs_directory.mkdir(parents=True, exist_ok=True)
         # Getting a new screenshot dir for the current session
         current_session_dir = get_session_screenshot_dir(abs_directory)
-        debug_client.debug.start_recording(str(current_session_dir))
+        with debug_client.debug.transport:
+            debug_client.debug.start_recording(str(current_session_dir))
         if report_func is not None:
             report_func(f"Recording started into {current_session_dir}.")
 
 
-def _is_emulator(debug_client: "TrezorClientDebugLink") -> bool:
+def _is_emulator(debug_client: "TrezorTestContext") -> bool:
     """Check if we are connected to emulator, in contrast to hardware device."""
     return debug_client.features.fw_vendor == "EMULATOR"
 
diff --git a/python/src/trezorlib/device.py b/python/src/trezorlib/device.py
index ff733661..86f5de50 100644
--- a/python/src/trezorlib/device.py
+++ b/python/src/trezorlib/device.py
@@ -28,10 +28,10 @@ from slip10 import SLIP10
 
 from . import messages
 from .exceptions import Cancelled, TrezorException
-from .tools import Address, _deprecation_retval_helper, _return_success, parse_path
+from .tools import Address, parse_path, workflow
 
 if TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 
 RECOVERY_BACK = "\x08"  # backspace character, sent literally
@@ -41,6 +41,7 @@ ENTROPY_CHECK_MIN_VERSION = (2, 8, 7)
 HOMESCREEN_STREAMING_MIN_VERSION = (2, 8, 11)
 
 
+@workflow()
 def apply_settings(
     session: "Session",
     label: Optional[str] = None,
@@ -55,7 +56,7 @@ def apply_settings(
     experimental_features: Optional[bool] = None,
     hide_passphrase_from_host: Optional[bool] = None,
     haptic_feedback: Optional[bool] = None,
-) -> str | None:
+) -> None:
     if language is not None:
         warnings.warn(
             "language ignored. Use change_language() to set device language.",
@@ -78,12 +79,10 @@ def apply_settings(
         settings.homescreen_length = len(homescreen)
         response = session.call(settings, expect=messages.DataChunkRequest)
         _send_chunked_data(session, response, homescreen)
-        out = messages.Success()
     else:
         settings.homescreen = homescreen
-        out = session.call(settings, expect=messages.Success)
+        session.call(settings, expect=messages.Success)
     session.refresh_features()
-    return _return_success(out)
 
 
 def _send_chunked_data(
@@ -100,11 +99,12 @@ def _send_chunked_data(
         response = session.call(messages.DataChunkAck(data_chunk=chunk))
 
 
+@workflow()
 def change_language(
     session: "Session",
     language_data: bytes,
     show_display: bool | None = None,
-) -> str | None:
+) -> None:
     data_length = len(language_data)
     msg = messages.ChangeLanguage(data_length=data_length, show_display=show_display)
 
@@ -115,41 +115,39 @@ def change_language(
     else:
         messages.Success.ensure_isinstance(response)
     session.refresh_features()  # changing the language in features
-    return _return_success(messages.Success(message="Language changed."))
 
 
-def apply_flags(session: "Session", flags: int) -> str | None:
-    out = session.call(messages.ApplyFlags(flags=flags), expect=messages.Success)
+@workflow()
+def apply_flags(session: "Session", flags: int) -> None:
+    session.call(messages.ApplyFlags(flags=flags), expect=messages.Success)
     session.refresh_features()
-    return _return_success(out)
 
 
-def change_pin(session: "Session", remove: bool = False) -> str | None:
-    ret = session.call(messages.ChangePin(remove=remove), expect=messages.Success)
+@workflow()
+def change_pin(session: "Session", remove: bool = False) -> None:
+    session.call(messages.ChangePin(remove=remove), expect=messages.Success)
     session.refresh_features()
-    return _return_success(ret)
 
 
-def change_wipe_code(session: "Session", remove: bool = False) -> str | None:
-    ret = session.call(messages.ChangeWipeCode(remove=remove), expect=messages.Success)
+@workflow()
+def change_wipe_code(session: "Session", remove: bool = False) -> None:
+    session.call(messages.ChangeWipeCode(remove=remove), expect=messages.Success)
     session.refresh_features()
-    return _return_success(ret)
 
 
-def sd_protect(
-    session: "Session", operation: messages.SdProtectOperationType
-) -> str | None:
-    ret = session.call(messages.SdProtect(operation=operation), expect=messages.Success)
+@workflow()
+def sd_protect(session: "Session", operation: messages.SdProtectOperationType) -> None:
+    session.call(messages.SdProtect(operation=operation), expect=messages.Success)
     session.refresh_features()
-    return _return_success(ret)
 
 
-def wipe(session: "Session") -> str | None:
-    ret = session.call(messages.WipeDevice(), expect=messages.Success)
-    session.invalidate()
-    return _return_success(ret)
+@workflow()
+def wipe(session: "Session") -> None:
+    session.call(messages.WipeDevice(), expect=messages.Success)
+    session.client._invalidate()
 
 
+@workflow()
 def recover(
     session: "Session",
     word_count: int = 24,
@@ -163,7 +161,7 @@ def recover(
     u2f_counter: Optional[int] = None,
     *,
     type: Optional[messages.RecoveryType] = None,
-) -> messages.Success | None:
+) -> None:
     if language is not None:
         warnings.warn(
             "language ignored. Use change_language() to set device language.",
@@ -230,8 +228,6 @@ def recover(
     # reinitialize the device
     session.refresh_features()
 
-    return _deprecation_retval_helper(res)
-
 
 def is_slip39_backup_type(backup_type: messages.BackupType) -> bool:
     return backup_type in (
@@ -318,13 +314,12 @@ def reset(
         backup_type=backup_type,
     )
 
-    return _return_success(messages.Success(message="Initialized"))
-
 
 def _get_external_entropy() -> bytes:
     return secrets.token_bytes(32)
 
 
+@workflow()
 def setup(
     session: "Session",
     *,
@@ -563,12 +558,13 @@ def _reset_with_entropycheck(
     return xpubs
 
 
+@workflow()
 def backup(
     session: "Session",
     group_threshold: Optional[int] = None,
     groups: Iterable[tuple[int, int]] = (),
-) -> str | None:
-    ret = session.call(
+) -> None:
+    session.call(
         messages.BackupDevice(
             group_threshold=group_threshold,
             groups=[
@@ -579,14 +575,14 @@ def backup(
         expect=messages.Success,
     )
     session.refresh_features()
-    return _return_success(ret)
 
 
-def cancel_authorization(session: "Session") -> str | None:
-    ret = session.call(messages.CancelAuthorization(), expect=messages.Success)
-    return _return_success(ret)
+@workflow()
+def cancel_authorization(session: "Session") -> None:
+    session.call(messages.CancelAuthorization(), expect=messages.Success)
 
 
+@workflow()
 def unlock_path(session: "Session", n: "Address") -> bytes:
     resp = session.call(
         messages.UnlockPath(address_n=n), expect=messages.UnlockedPathRequest
@@ -601,42 +597,43 @@ def unlock_path(session: "Session", n: "Address") -> bytes:
         raise TrezorException("Unexpected response in UnlockPath flow")
 
 
+@workflow()
 def reboot_to_bootloader(
     session: "Session",
     boot_command: messages.BootCommand = messages.BootCommand.STOP_AND_WAIT,
     firmware_header: Optional[bytes] = None,
-) -> str | None:
-    ret = session.call(
+) -> None:
+    session.call(
         messages.RebootToBootloader(
             boot_command=boot_command,
             firmware_header=firmware_header,
         ),
         expect=messages.Success,
     )
-    return _return_success(ret)
 
 
-def show_device_tutorial(session: "Session") -> str | None:
-    ret = session.call(messages.ShowDeviceTutorial(), expect=messages.Success)
-    return _return_success(ret)
+@workflow()
+def show_device_tutorial(session: "Session") -> None:
+    session.call(messages.ShowDeviceTutorial(), expect=messages.Success)
 
 
-def unlock_bootloader(session: "Session") -> str | None:
-    ret = session.call(messages.UnlockBootloader(), expect=messages.Success)
-    return _return_success(ret)
+@workflow()
+def unlock_bootloader(session: "Session") -> None:
+    session.call(messages.UnlockBootloader(), expect=messages.Success)
 
 
-def set_busy(session: "Session", expiry_ms: Optional[int]) -> str | None:
+@workflow()
+def set_busy(session: "Session", expiry_ms: Optional[int]) -> None:
     """Sets or clears the busy state of the device.
 
     In the busy state the device shows a "Do not disconnect" message instead of the homescreen.
     Setting `expiry_ms=None` clears the busy state.
     """
-    ret = session.call(messages.SetBusy(expiry_ms=expiry_ms), expect=messages.Success)
+    session.call(messages.SetBusy(expiry_ms=expiry_ms), expect=messages.Success)
     session.refresh_features()
-    return _return_success(ret)
 
 
+@workflow()
 def authenticate(session: "Session", challenge: bytes) -> messages.AuthenticityProof:
     return session.call(
         messages.AuthenticateDevice(challenge=challenge),
@@ -644,11 +641,12 @@ def authenticate(session: "Session", challenge: bytes) -> messages.AuthenticityP
     )
 
 
-def set_brightness(session: "Session", value: Optional[int] = None) -> str | None:
-    ret = session.call(messages.SetBrightness(value=value), expect=messages.Success)
-    return _return_success(ret)
+@workflow()
+def set_brightness(session: "Session", value: Optional[int] = None) -> None:
+    session.call(messages.SetBrightness(value=value), expect=messages.Success)
 
 
+@workflow()
 def get_serial_number(session: "Session") -> str:
     ret = session.call(messages.GetSerialNumber(), expect=messages.SerialNumber)
     return ret.serial_number
diff --git a/python/src/trezorlib/eos.py b/python/src/trezorlib/eos.py
index 819b829c..da4e40b1 100644
--- a/python/src/trezorlib/eos.py
+++ b/python/src/trezorlib/eos.py
@@ -18,11 +18,11 @@ from datetime import datetime
 from typing import TYPE_CHECKING, List, Tuple
 
 from . import exceptions, messages
-from .tools import b58decode
+from .tools import b58decode, workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 
 def name_to_number(name: str) -> int:
@@ -318,6 +318,7 @@ def parse_transaction_json(
 # ====== Client functions ====== #
 
 
+@workflow(capability=messages.Capability.EOS)
 def get_public_key(
     session: "Session", n: "Address", show_display: bool = False
 ) -> messages.EosPublicKey:
@@ -327,6 +328,7 @@ def get_public_key(
     )
 
 
+@workflow(capability=messages.Capability.EOS)
 def sign_tx(
     session: "Session",
     address: "Address",
diff --git a/python/src/trezorlib/ethereum.py b/python/src/trezorlib/ethereum.py
index 92ba41f5..49ec7d87 100644
--- a/python/src/trezorlib/ethereum.py
+++ b/python/src/trezorlib/ethereum.py
@@ -18,11 +18,11 @@ import re
 from typing import TYPE_CHECKING, Any, AnyStr, Dict, List, Optional, Tuple
 
 from . import exceptions, messages
-from .tools import prepare_message_bytes
+from .tools import prepare_message_bytes, workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 
 def int_to_big_endian(value: int) -> bytes:
@@ -149,6 +149,7 @@ def get_address(*args: Any, **kwargs: Any) -> str:
     return resp.address
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def get_authenticated_address(
     session: "Session",
     n: "Address",
@@ -168,6 +169,7 @@ def get_authenticated_address(
     return resp
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def get_public_node(
     session: "Session", n: "Address", show_display: bool = False
 ) -> messages.EthereumPublicKey:
@@ -177,6 +179,7 @@ def get_public_node(
     )
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def sign_tx(
     session: "Session",
     n: "Address",
@@ -237,6 +240,7 @@ def sign_tx(
     return response.signature_v, response.signature_r, response.signature_s
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def sign_tx_eip1559(
     session: "Session",
     n: "Address",
@@ -288,6 +292,7 @@ def sign_tx_eip1559(
     return response.signature_v, response.signature_r, response.signature_s
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def sign_message(
     session: "Session",
     n: "Address",
@@ -306,6 +311,7 @@ def sign_message(
     )
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def sign_typed_data(
     session: "Session",
     n: "Address",
@@ -383,6 +389,7 @@ def sign_typed_data(
     return messages.EthereumTypedDataSignature.ensure_isinstance(response)
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def verify_message(
     session: "Session",
     address: str,
@@ -405,6 +412,7 @@ def verify_message(
         return False
 
 
+@workflow(capability=messages.Capability.Ethereum)
 def sign_typed_data_hash(
     session: "Session",
     n: "Address",
diff --git a/python/src/trezorlib/evolu.py b/python/src/trezorlib/evolu.py
index 921d5f1b..8063f2b7 100644
--- a/python/src/trezorlib/evolu.py
+++ b/python/src/trezorlib/evolu.py
@@ -19,11 +19,13 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Optional
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 
+@workflow()
 def get_node(session: Session, proof: bytes) -> bytes:
     return session.call(
         messages.EvoluGetNode(proof_of_delegated_identity=proof),
diff --git a/python/src/trezorlib/exceptions.py b/python/src/trezorlib/exceptions.py
index 4713d7a0..a5ff4ceb 100644
--- a/python/src/trezorlib/exceptions.py
+++ b/python/src/trezorlib/exceptions.py
@@ -16,13 +16,20 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import typing as t
 
-if TYPE_CHECKING:
+if t.TYPE_CHECKING:
     from .messages import Failure
     from .protobuf import MessageType
 
 
+OUTDATED_FIRMWARE_ERROR = """
+Your Trezor firmware is out of date. Update it with the following command:
+  trezorctl firmware update
+Or visit https://suite.trezor.io/
+""".strip()
+
+
 class TrezorException(Exception):
     """General Trezor exception."""
 
@@ -65,21 +72,48 @@ class Cancelled(TrezorException):
     """Action was cancelled.
 
     Cancellation can be either received from Trezor or caused by the library, typically
-    in response to user action."""
+    in response to user action.
+    """
+
+    def __init__(self, message: str = "Action was cancelled") -> None:
+        self.message = message
+        super().__init__(self.message)
+
+
+class DeviceLockedError(TrezorException):
+    """Device is locked.
+
+    Raised when an action cannot proceed because the device is locked.
+
+    Typically, an action will trigger an unlock prompt on device. In specific
+    cases, that is not the appropriate action (e.g., when establishing a THP channel).
+    In such cases, this exception will be raised for the caller to handle,
+    by, e.g., explicitly triggering the unlock prompt.
+    """
+
+    def __init__(self, message: str = "Device is locked") -> None:
+        self.message = message
+        super().__init__(self.message)
 
 
 class OutdatedFirmwareError(TrezorException):
     """Trezor firmware is too old.
 
     Raised when interfacing with a Trezor whose firmware version is no longer supported
-    by current library version."""
+    by current library version.
+    """
+
+    def __init__(self, message: str = OUTDATED_FIRMWARE_ERROR) -> None:
+        self.message = message
+        super().__init__(self.message)
 
 
 class UnexpectedMessageError(TrezorException):
     """Unexpected message received from Trezor.
 
     Raised when the library receives a response from Trezor that does not match the
-    previous request."""
+    previous request.
+    """
 
     def __init__(self, expected: type[MessageType], actual: MessageType) -> None:
         self.expected = expected
@@ -87,43 +121,41 @@ class UnexpectedMessageError(TrezorException):
         super().__init__(f"Expected {expected.__name__} but Trezor sent {actual}")
 
 
-class FailedSessionResumption(TrezorException):
-    """Provided session_id is not valid / session cannot be resumed.
-
-    Raised when `trezorctl -s <sesssion_id>` is used or `TREZOR_SESSION_ID = <session_id>`
-    is set and resumption of session with the `session_id` fails."""
-
-    def __init__(self, received_session_id: bytes | None = None) -> None:
-        # We keep the session id that was received from Trezor for test purposes
-        self.received_session_id = received_session_id
-        super().__init__("Failed to resume session")
-
-
 class InvalidSessionError(TrezorException):
-    """Session expired and is no longer valid.
-
-    Raised when Trezor returns unexpected PassphraseRequest"""
+    """Session is invalid or expired."""
 
+    def __init__(
+        self, session_id: t.Any, *, from_message: MessageType | None = None
+    ) -> None:
+        self.session_id = session_id
+        self.from_message = from_message
+        super().__init__(session_id)
 
-class ThpError(TrezorException):
-    pass
 
+class ProtocolError(TrezorException):
+    """Response from Trezor could not be understood.
 
-class TransportBusy(ThpError):
-    pass
+    This could indicate invalid magic bytes or another kind of error in the
+    low-level message encoding.
+    """
 
 
-class UnallocatedChannel(ThpError):
-    pass
+class PassphraseError(TrezorException):
+    """Unable to create a passphrase session because passphrase is disabled on device."""
 
 
-class DecryptionFailed(ThpError):
-    pass
+class NotPairedError(TrezorException):
+    """Pairing is required before this client can be used."""
 
+    def __init__(self, message: str | None = None) -> None:
+        if message is None:
+            message = self.__doc__
+        super().__init__(message)
 
-class DeviceLocked(ThpError):
-    pass
 
+class StateMismatchError(TrezorException):
+    """Expected state mismatch.
 
-class ThpUnknownError(ThpError):
-    pass
+    Raised when the caller invokes a function that does not match the current
+    state of a flow.
+    """
diff --git a/python/src/trezorlib/fido.py b/python/src/trezorlib/fido.py
index a1b4bbc0..1f3509ec 100644
--- a/python/src/trezorlib/fido.py
+++ b/python/src/trezorlib/fido.py
@@ -19,40 +19,43 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Sequence
 
 from . import messages
-from .tools import _return_success
+from .tools import workflow
 
 if TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 
+@workflow()
 def list_credentials(session: "Session") -> Sequence[messages.WebAuthnCredential]:
     return session.call(
         messages.WebAuthnListResidentCredentials(), expect=messages.WebAuthnCredentials
     ).credentials
 
 
-def add_credential(session: "Session", credential_id: bytes) -> str | None:
-    ret = session.call(
+@workflow()
+def add_credential(session: "Session", credential_id: bytes) -> None:
+    session.call(
         messages.WebAuthnAddResidentCredential(credential_id=credential_id),
         expect=messages.Success,
     )
-    return _return_success(ret)
 
 
-def remove_credential(session: "Session", index: int) -> str | None:
-    ret = session.call(
+@workflow()
+def remove_credential(session: "Session", index: int) -> None:
+    session.call(
         messages.WebAuthnRemoveResidentCredential(index=index), expect=messages.Success
     )
-    return _return_success(ret)
 
 
-def set_counter(session: "Session", u2f_counter: int) -> str | None:
-    ret = session.call(
+@workflow()
+def set_counter(session: "Session", u2f_counter: int) -> None:
+    session.call(
         messages.SetU2FCounter(u2f_counter=u2f_counter), expect=messages.Success
     )
-    return _return_success(ret)
 
 
+@workflow()
 def get_next_counter(session: "Session") -> int:
-    ret = session.call(messages.GetNextU2FCounter(), expect=messages.NextU2FCounter)
-    return ret.u2f_counter
+    return session.call(
+        messages.GetNextU2FCounter(), expect=messages.NextU2FCounter
+    ).u2f_counter
diff --git a/python/src/trezorlib/firmware/__init__.py b/python/src/trezorlib/firmware/__init__.py
index af8cb369..81f476c3 100644
--- a/python/src/trezorlib/firmware/__init__.py
+++ b/python/src/trezorlib/firmware/__init__.py
@@ -41,7 +41,7 @@ if True:
     from .vendor import *  # noqa: F401, F403
 
 if t.TYPE_CHECKING:
-    from ..transport.session import Session
+    from ..client import Session
 
     T = t.TypeVar("T", bound="FirmwareType")
 
diff --git a/python/src/trezorlib/log.py b/python/src/trezorlib/log.py
index 67bd9dd6..ca283b62 100644
--- a/python/src/trezorlib/log.py
+++ b/python/src/trezorlib/log.py
@@ -14,20 +14,15 @@
 # You should have received a copy of the License along with this library.
 # If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
 
-import logging
-from typing import Optional, Set, Type
-
-from typing_extensions import Protocol, runtime_checkable
-
-from . import protobuf
+from __future__ import annotations
 
+import logging
+import typing as t
 
-@runtime_checkable
-class HasProtobuf(Protocol):
-    protobuf: protobuf.MessageType
-
+if t.TYPE_CHECKING:
+    from . import protobuf
 
-OMITTED_MESSAGES: Set[Type[protobuf.MessageType]] = set()
+OMITTED_MESSAGES: set[type[protobuf.MessageType]] = set()
 
 DUMP_BYTES = 5
 DUMP_PACKETS = 4
@@ -38,23 +33,35 @@ logging.addLevelName(DUMP_PACKETS, "PACKETS")
 
 class PrettyProtobufFormatter(logging.Formatter):
     def format(self, record: logging.LogRecord) -> str:
+        from . import client, protobuf
+
+        session = getattr(record, "session", None)
+        if isinstance(session, client.Session):
+            session = f" [s:{session._log_short_id()}]"
+        else:
+            session = ""
+
         time = self.formatTime(record)
-        message = "[{time}] {source} {level}: {msg}".format(
+        message = "[{time}] {source} {level}{session}: {msg}".format(
             time=time,
             level=record.levelname.upper(),
             source=record.name,
             msg=super().format(record),
+            session=session,
         )
-        if isinstance(record, HasProtobuf):
-            if type(record.protobuf) in OMITTED_MESSAGES:
-                message += f" ({record.protobuf.ByteSize()} bytes)"
+
+        proto_msg = getattr(record, "protobuf", None)
+        if isinstance(proto_msg, protobuf.MessageType):
+            if type(proto_msg) in OMITTED_MESSAGES:
+                message += f" ({proto_msg.ByteSize()} bytes)"
             else:
-                message += "\n" + protobuf.format_message(record.protobuf)
+                message += "\n" + protobuf.format_message(proto_msg)
+
         return message
 
 
 def enable_debug_output(
-    verbosity: int = 1, handler: Optional[logging.Handler] = None
+    verbosity: int = 1, handler: logging.Handler | None = None
 ) -> None:
     if handler is None:
         handler = logging.StreamHandler()
diff --git a/python/src/trezorlib/mapping.py b/python/src/trezorlib/mapping.py
index 525874ee..d2ddf919 100644
--- a/python/src/trezorlib/mapping.py
+++ b/python/src/trezorlib/mapping.py
@@ -71,30 +71,12 @@ class ProtobufMapping:
         protobuf.dump_message(buf, msg)
         return wire_type, buf.getvalue()
 
-    def encode_without_wire_type(self, msg: protobuf.MessageType) -> bytes:
-        """Serialize a Python protobuf class.
-
-        Returns the byte representation of the protobuf message.
-        """
-
-        buf = io.BytesIO()
-        protobuf.dump_message(buf, msg)
-        return buf.getvalue()
-
     def decode(self, msg_wire_type: int, msg_bytes: bytes) -> protobuf.MessageType:
         """Deserialize a protobuf message into a Python class."""
         cls = self.type_to_class[msg_wire_type]
         buf = io.BytesIO(msg_bytes)
         return protobuf.load_message(buf, cls)
 
-    def decode_without_wire_type(
-        self, message_type: type[MT], msg_bytes: bytes
-    ) -> protobuf.MessageType:
-        """Deserialize a protobuf message into a Python class."""
-        cls = message_type
-        buf = io.BytesIO(msg_bytes)
-        return protobuf.load_message(buf, cls)
-
     @classmethod
     def from_module(cls, module: ModuleType) -> Self:
         """Generate a mapping from a module.
diff --git a/python/src/trezorlib/misc.py b/python/src/trezorlib/misc.py
index 3a580312..dfa156e5 100644
--- a/python/src/trezorlib/misc.py
+++ b/python/src/trezorlib/misc.py
@@ -17,16 +17,19 @@
 from typing import TYPE_CHECKING, Optional
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 
+@workflow(capability=messages.Capability.Crypto)
 def get_entropy(session: "Session", size: int) -> bytes:
     return session.call(messages.GetEntropy(size=size), expect=messages.Entropy).entropy
 
 
+@workflow(capability=messages.Capability.Crypto)
 def sign_identity(
     session: "Session",
     identity: messages.IdentityType,
@@ -45,6 +48,7 @@ def sign_identity(
     )
 
 
+@workflow(capability=messages.Capability.Crypto)
 def get_ecdh_session_key(
     session: "Session",
     identity: messages.IdentityType,
@@ -61,6 +65,7 @@ def get_ecdh_session_key(
     )
 
 
+@workflow(capability=messages.Capability.Crypto)
 def encrypt_keyvalue(
     session: "Session",
     n: "Address",
@@ -84,6 +89,7 @@ def encrypt_keyvalue(
     ).value
 
 
+@workflow(capability=messages.Capability.Crypto)
 def decrypt_keyvalue(
     session: "Session",
     n: "Address",
@@ -107,10 +113,12 @@ def decrypt_keyvalue(
     ).value
 
 
+@workflow(capability=messages.Capability.Crypto)
 def get_nonce(session: "Session") -> bytes:
     return session.call(messages.GetNonce(), expect=messages.Nonce).nonce
 
 
+@workflow()
 def payment_notification(
     session: "Session", payment_req: messages.PaymentRequest
 ) -> None:
diff --git a/python/src/trezorlib/models.py b/python/src/trezorlib/models.py
index 48dfb05e..a6022437 100644
--- a/python/src/trezorlib/models.py
+++ b/python/src/trezorlib/models.py
@@ -160,6 +160,18 @@ def by_internal_name(name: str | None) -> TrezorModel | None:
     return None
 
 
+def unknown_model(name: str | None, internal_name: str | None) -> TrezorModel:
+    return TrezorModel(
+        name=name or "Unknown",
+        internal_name=internal_name or "????",
+        minimum_version=(0, 0, 0),
+        vendors=VENDORS,
+        usb_ids=(),
+        default_mapping=mapping.DEFAULT_MAPPING,
+        is_unknown=True,
+    )
+
+
 def detect(features: messages.Features) -> TrezorModel:
     """Detect Trezor model from its Features response.
 
@@ -178,14 +190,4 @@ def detect(features: messages.Features) -> TrezorModel:
     if model is not None:
         return model
 
-    return TrezorModel(
-        name=features.model or "Unknown",
-        internal_name=features.internal_model or "????",
-        minimum_version=(0, 0, 0),
-        # Allowed vendors are the internal VENDORS list instead of trusting features.vendor.
-        # That way, an unrecognized non-Trezor device will fail the check in TrezorClient.
-        vendors=VENDORS,
-        usb_ids=(),
-        default_mapping=mapping.DEFAULT_MAPPING,
-        is_unknown=True,
-    )
+    return unknown_model(features.model, features.internal_model)
diff --git a/python/src/trezorlib/monero.py b/python/src/trezorlib/monero.py
index 5f754830..e0de4f2e 100644
--- a/python/src/trezorlib/monero.py
+++ b/python/src/trezorlib/monero.py
@@ -17,10 +17,11 @@
 from typing import TYPE_CHECKING
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 
 # MAINNET = 0
@@ -29,6 +30,7 @@ if TYPE_CHECKING:
 # FAKECHAIN = 3
 
 
+@workflow(capability=messages.Capability.Monero)
 def get_address(
     session: "Session",
     n: "Address",
@@ -47,6 +49,7 @@ def get_address(
     ).address
 
 
+@workflow(capability=messages.Capability.Monero)
 def get_watch_key(
     session: "Session",
     n: "Address",
diff --git a/python/src/trezorlib/nem.py b/python/src/trezorlib/nem.py
index c0834ace..dbedea90 100644
--- a/python/src/trezorlib/nem.py
+++ b/python/src/trezorlib/nem.py
@@ -18,10 +18,11 @@ import json
 from typing import TYPE_CHECKING
 
 from . import exceptions, messages
+from .tools import workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 TYPE_TRANSACTION_TRANSFER = 0x0101
 TYPE_IMPORTANCE_TRANSFER = 0x0801
@@ -194,6 +195,7 @@ def create_sign_tx(transaction: dict, chunkify: bool = False) -> messages.NEMSig
 # ====== Client functions ====== #
 
 
+@workflow(capability=messages.Capability.NEM)
 def get_address(
     session: "Session",
     n: "Address",
@@ -209,6 +211,7 @@ def get_address(
     ).address
 
 
+@workflow(capability=messages.Capability.NEM)
 def sign_tx(
     session: "Session", n: "Address", transaction: dict, chunkify: bool = False
 ) -> messages.NEMSignedTx:
diff --git a/python/src/trezorlib/nostr.py b/python/src/trezorlib/nostr.py
index 70408f2e..f46ff0b6 100644
--- a/python/src/trezorlib/nostr.py
+++ b/python/src/trezorlib/nostr.py
@@ -17,12 +17,14 @@
 from typing import TYPE_CHECKING
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 
+@workflow()
 def get_pubkey(session: "Session", n: "Address") -> bytes:
     return session.call(
         messages.NostrGetPubkey(
@@ -32,6 +34,7 @@ def get_pubkey(session: "Session", n: "Address") -> bytes:
     ).pubkey
 
 
+@workflow()
 def sign_event(
     session: "Session",
     sign_event: messages.NostrSignEvent,
diff --git a/python/src/trezorlib/protocol_v1.py b/python/src/trezorlib/protocol_v1.py
index afab415e..d7ac5941 100644
--- a/python/src/trezorlib/protocol_v1.py
+++ b/python/src/trezorlib/protocol_v1.py
@@ -18,16 +18,23 @@ from __future__ import annotations
 
 import io
 import logging
+import secrets
 import struct
 import typing as t
+import warnings
 
-from . import client, exceptions, messages
+import typing_extensions as tx
+
+from . import client, exceptions, mapping, messages, models
 from .log import DUMP_BYTES
-from .transport import Transport
+from .thp import pairing
+from .tools import enter_context
 
 if t.TYPE_CHECKING:
     from .mapping import ProtobufMapping
     from .models import TrezorModel
+    from .protobuf import MessageType
+    from .transport import Transport
 
 LOG = logging.getLogger(__name__)
 
@@ -45,7 +52,7 @@ def write(transport: Transport, message_type: int, message_data: bytes) -> None:
         return
 
     buffer = io.BytesIO(b"##" + header + message_data)
-    while chunk_payload := buffer.read(chunk_size):
+    while chunk_payload := buffer.read(chunk_size - 1):
         chunk = b"?" + chunk_payload
         # pad to chunk size
         chunk = chunk.ljust(chunk_size, b"\x00")
@@ -88,161 +95,297 @@ def read(transport: Transport, timeout: float | None = None) -> tuple[int, bytes
     return msg_type, bytes(buffer[:datalen])
 
 
-class SessionV1(client.Session["TrezorClientV1"]):
+class SessionV1(client.Session["TrezorClientV1", t.Optional[bytes]]):
     def __init__(
         self,
         client: TrezorClientV1,
         *,
-        session_id: bytes | None = None,
+        id: bytes | None = None,
         seedless: bool = False,
     ) -> None:
-        super().__init__(client)
-        self.session_id = session_id
+        super().__init__(client, id=id)
         self.seedless = seedless
-        self.is_invalid = False
 
-    def resume(self) -> None:
-        if self.session_id is None:
-            raise RuntimeError("resuming session without id")
-        self.initialize()
+    def __str__(self) -> str:
+        return f"SessionV1(id={self.id.hex() if self.id else '(none)'})"
 
-    def _activate_self(self) -> None:
-        if self.is_invalid:
-            raise exceptions.InvalidSessionError(self.session_id)
-        if self.client._last_active_session is not self:
-            self.client._last_active_session = self
-            self.resume()
+    def _log_short_id(self) -> str:
+        if self.id is None:
+            return super()._log_short_id()
+        return self.id.hex()[:8]
 
-    def _write(self, msg: t.Any) -> None:
-        self._activate_self()
-        LOG.debug(
-            f"sending message: {msg.__class__.__name__}",
-            extra={"protobuf": msg},
-        )
-        msg_type, msg_bytes = self.client.mapping.encode(msg)
-        LOG.log(
-            DUMP_BYTES,
-            f"encoded as type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
-        )
-        write(self.client.transport, msg_type, msg_bytes)
+    @enter_context
+    def initialize(self, *, derive_cardano: bool | None = None) -> messages.Features:
+        """Initialize the session.
 
-    def _read(self, timeout: float | None = None) -> t.Any:
+        This can:
+        - create a new session if this instance has not been initialized yet
+        - activate an existing session, and/or trigger an InvalidSessionError
+          if the session has expired.
+        """
         if self.is_invalid:
-            raise exceptions.InvalidSessionError(self.session_id)
-        assert self.client._last_active_session is self
-        msg_type, msg_bytes = self._read(timeout=timeout)
-        LOG.log(
-            DUMP_BYTES,
-            f"received type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
-        )
-        msg = self.client.mapping.decode(msg_type, msg_bytes)
-        LOG.debug(
-            f"received message: {msg.__class__.__name__}",
-            extra={"protobuf": msg},
-        )
+            raise exceptions.InvalidSessionError(self.id)
 
-        from .debuglink import TrezorClientDebugLink
-
-        if isinstance(self.client, TrezorClientDebugLink):
-            self.client.notify_read(msg)
-
-        return msg
-
-    def initialize(self, *, derive_cardano: bool | None = None) -> None:
-        # avoid triggering a resume() in _activate_self()
+        # notify the client that this is now the active session
         self.client._last_active_session = self
+        LOG.info("Activating session %s", self, extra={"session": self})
         resp = self.call_raw(
-            messages.Initialize(
-                session_id=self.session_id, derive_cardano=derive_cardano
-            )
+            messages.Initialize(session_id=self.id, derive_cardano=derive_cardano)
         )
         features = messages.Features.ensure_isinstance(resp)
         session_id = features.session_id
-        if self.session_id is None or self.seedless:
-            self.session_id = session_id
-        elif self.session_id != session_id:
+        if session_id is None:
+            LOG.error(
+                "Trezor did not return a session ID. Session management is now broken."
+            )
+            warnings.warn("Your Trezor firmware does not support sessions.")
+
+        if self.id is None or self.seedless:
+            LOG.info("New session id %s", session_id.hex() if session_id else "(none)")
+            self.id = session_id
+        elif self.id != session_id:
             self.is_invalid = True
-            raise exceptions.InvalidSessionError(session_id)
+            self.client._close_session(self)
+            LOG.error("Failed to resume session id %s", self.id.hex())
+            raise exceptions.InvalidSessionError(session_id, from_message=resp)
+        else:
+            LOG.info("Resumed session id %s", self.id.hex())
+        return features
+
+    @classmethod
+    def derive(
+        cls,
+        client_: TrezorClientV1,
+        passphrase: str | t.Literal[client.PassphraseSetting.ON_DEVICE],
+        derive_cardano: bool,
+    ) -> tx.Self:
+        new = cls(client_)
+        new._derive(passphrase, derive_cardano)
+        return new
 
-    def derive_seed(
+    @enter_context
+    def _derive(
         self,
-        passphrase: str | type[client.PassphraseOnDevice],
+        passphrase: str | t.Literal[client.PassphraseSetting.ON_DEVICE],
         derive_cardano: bool,
     ) -> None:
-        if self.session_id is not None:
-            raise exceptions.TrezorException("Session already initialized")
+        """Create a new session with pre-derived seed for the given passphrase."""
         self.initialize(derive_cardano=derive_cardano)
-        resp = self.call(
-            messages.GetAddress(
-                address_n=client.PASSPHRASE_TEST_PATH, coin_name="Testnet"
+
+        try:
+            resp = self.call(client.GET_ROOT_FINGERPRINT_MESSAGE)
+        except exceptions.InvalidSessionError as e:
+            # raised by call() when an unexpected PassphraseRequest is received
+            resp = e.from_message
+
+        if isinstance(resp, messages.PassphraseRequest):
+            # process PassphraseRequest / PassphraseAck
+            if passphrase is client.PassphraseSetting.ON_DEVICE:
+                ack = messages.PassphraseAck(on_device=True)
+            else:
+                ack = messages.PassphraseAck(passphrase=passphrase)
+            resp = self.call(ack)
+
+        elif (
+            self.features.passphrase_always_on_device is True
+            and passphrase is client.PassphraseSetting.ON_DEVICE
+        ):
+            # Passphrase was processed on device without asking the host. This is OK.
+            pass
+
+        elif passphrase:
+            # We didn't get a PassphraseRequest, but passphrase_protection is enabled.
+            # Looks like the session is already initialized. Bail out.
+            raise exceptions.TrezorException(
+                f"Failed to activate passphrase session {resp}"
             )
-        )
-        # no passphrase was requested
-        if isinstance(resp, messages.Address):
-            if self.features.passphrase_protection is True:
-                raise exceptions.TrezorException(
-                    "Failed to activate passphrase session"
-                )
-            if passphrase not in (None, client.PassphraseOnDevice):
-                raise exceptions.PassphraseDisabledError
 
-            return
+        # after processing any PassphraseRequest, we should have an Address response
+        resp = messages.PublicKey.ensure_isinstance(resp)
 
-        resp = messages.PassphraseRequest.ensure_isinstance(resp)
-        if passphrase is client.PassphraseOnDevice:
-            ack = messages.PassphraseAck(on_device=True)
+        # `root_fingerprint` is not available on older models.
+        min_version = (1, 9, 4) if self.model is models.T1B1 else (2, 3, 5)
+        if self.version >= min_version:
+            assert resp.root_fingerprint is not None
+            self._root_fingerprint = resp.root_fingerprint.to_bytes(4, "big")
         else:
-            assert isinstance(passphrase, str)
-            ack = messages.PassphraseAck(passphrase=passphrase)
-        resp = self.call(ack)
-        if isinstance(resp, messages.Deprecated_PassphraseStateRequest):
-            self.session_id = resp.state
-            resp = self.call(messages.Deprecated_PassphraseStateAck())
-        messages.Address.ensure_isinstance(resp)
-        self.refresh_features()
+            warnings.warn("Your Trezor firmware does not support root fingerprint.")
+
+        self.client.refresh_features()
+
+    def close(self) -> None:
+        super().close()
+        self.client._close_session(self)
+        if self.seedless:
+            self.is_invalid = False
+            self.id = None
+
+
+class NullPairing(pairing.PairingController):
+    def __init__(self) -> None:
+        # not calling super because we're not initializing the
+        # parent class, which assumes it's getting a TrezorClientThp instance.
+        # This should not be a problem in practice because we're pretending
+        # to be paired already.
+        pass
+
+    @property
+    def state(self) -> pairing.ControllerLifecycle:
+        return pairing.ControllerLifecycle.FINISHED
+
+    @property
+    def methods(self) -> t.Collection[type[pairing.PairingMethod]]:
+        return (pairing.SkipPairing,)
+
+    def is_paired(self) -> bool:
+        return True
+
+    def finish(self, _no_call: bool = False) -> None:
+        pass
+
+    def skip(self) -> None:
+        pass
+
+    def request_credential(self, autoconnect: bool = False) -> pairing.Credential:
+        raise ValueError("Pairing not available in protocol-v1")
 
 
 class TrezorClientV1(client.TrezorClient[SessionV1]):
-    _last_active_session: SessionV1 | None = None
+    _seedless_session: SessionV1
+    """Shared session instance for seedless calls. Will regenerate if it fails to resume."""
+
+    _last_active_session: SessionV1 | None
+    """The currently active session on the connected Trezor."""
 
     def __init__(
         self,
+        app: client.AppManifest,
         transport: Transport,
         *,
         model: TrezorModel | None,
         mapping: ProtobufMapping | None,
-        app_name: str,
-        host_name: str | None,
     ) -> None:
-        """
-        TODO
-        """
         super().__init__(
+            app=app,
+            transport=transport,
             model=model,
             mapping=mapping,
-            app_name=app_name,
-            host_name=host_name,
+            pairing=NullPairing(),
         )
-        LOG.info(f"creating client instance for device: {transport.get_path()}")
-        self.transport = transport
         self._seedless_session = SessionV1(client=self, seedless=True)
+        self._last_active_session = None
 
-    def get_session(
+    def _close_session(self, session: SessionV1) -> None:
+        if self._last_active_session is session:
+            self._last_active_session = None
+
+    def _invalidate(self) -> None:
+        super()._invalidate()
+        self._last_active_session = None
+
+    def _activate(self, session: SessionV1) -> None:
+        if self._last_active_session is not session:
+            self._last_active_session = session
+            session.initialize()
+
+    def _get_any_session(self) -> SessionV1:
+        if self._last_active_session is None:
+            # create a new session instance
+            return SessionV1(client=self, seedless=True)
+        assert not self._last_active_session.is_invalid
+        return self._last_active_session
+
+    def _get_features(self) -> messages.Features:
+        if self._last_active_session is None:
+            # return the features that we got from Initialize()
+            session = self._get_any_session()
+            return session.initialize()
+        assert not self._last_active_session.is_invalid
+        return super()._get_features()
+
+    def _get_session(
         self,
-        passphrase: str | type[client.PassphraseOnDevice] | None = "",
         *,
-        derive_cardano: bool = False,
+        passphrase: str | t.Literal[client.PassphraseSetting.ON_DEVICE] | None,
+        derive_cardano: bool,
     ) -> SessionV1:
         """
         Returns a new session.
         """
         if passphrase is None:
             return self._seedless_session
-        session = SessionV1(client=self)
-        session.derive_seed(passphrase, derive_cardano)
-        return session
+        return SessionV1.derive(self, passphrase, derive_cardano)
 
-    def _get_features(self) -> messages.Features:
-        return self._seedless_session.call(
-            messages.GetFeatures(), expect=messages.Features
+    def _write(self, session: SessionV1, msg: MessageType) -> None:
+        self._activate(session)
+        LOG.debug(
+            f"sending message: {msg.__class__.__name__}",
+            extra={"protobuf": msg, "session": session},
+        )
+        msg_type, msg_bytes = self.mapping.encode(msg)
+        LOG.log(
+            DUMP_BYTES,
+            f"encoded as type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
+            extra={"session": session},
+        )
+        write(self.transport, msg_type, msg_bytes)
+
+    def _read(self, session: SessionV1, timeout: float | None = None) -> MessageType:
+        if session.is_invalid:
+            raise exceptions.InvalidSessionError(session.id)
+        if self._last_active_session is not session:
+            raise exceptions.TrezorException("Reading from the wrong session")
+        msg_type, msg_bytes = read(self.transport, timeout=timeout)
+        LOG.log(
+            DUMP_BYTES,
+            f"received type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
+            extra={"session": session},
+        )
+        msg = self.mapping.decode(msg_type, msg_bytes)
+        LOG.debug(
+            f"received message: {msg.__class__.__name__}",
+            extra={"protobuf": msg, "session": session},
         )
+        return msg
+
+
+@enter_context
+def probe(
+    transport: Transport, *, mapping: ProtobufMapping = mapping.DEFAULT_MAPPING
+) -> bool:
+    """Probe the transport to see if it supports protocol v1."""
+    cancel_msg = messages.Cancel()
+    cancel_msg_type, cancel_msg_bytes = mapping.encode(cancel_msg)
+    write(transport, cancel_msg_type, cancel_msg_bytes)
+    resp_type, resp_bytes = read(transport)
+    resp = mapping.decode(resp_type, resp_bytes)
+    if isinstance(resp, messages.Failure):
+        if resp.code == messages.FailureType.InvalidProtocol:
+            return False
+    return True
+
+
+@enter_context
+def sync_responses(
+    transport: Transport,
+    *,
+    mapping: ProtobufMapping = mapping.DEFAULT_MAPPING,
+    retries: int = 10,
+) -> None:
+    """Sync responses from the transport."""
+    # cancel anything on screen -- on T1B1 this is the only way to exit e.g. a PIN prompt.
+    cancel_msg = mapping.encode(messages.Cancel())
+    write(transport, *cancel_msg)
+
+    # prepare an unique message to wait for
+    sync_string = "SYNC" + secrets.token_hex(8)
+    ping_msg = mapping.encode(messages.Ping(message=sync_string))
+    # prepare
+    write(transport, *ping_msg)
+
+    for _ in range(retries):
+        resp_type, resp_bytes = read(transport)
+        resp = mapping.decode(resp_type, resp_bytes)
+        if isinstance(resp, messages.Success) and resp.message == sync_string:
+            return
+    raise exceptions.ProtocolError("Failed to sync responses")
diff --git a/python/src/trezorlib/ripple.py b/python/src/trezorlib/ripple.py
index 1faf78ec..c6b20a77 100644
--- a/python/src/trezorlib/ripple.py
+++ b/python/src/trezorlib/ripple.py
@@ -18,11 +18,11 @@ from typing import TYPE_CHECKING, Any, Optional
 
 from . import messages
 from .protobuf import dict_to_proto
-from .tools import dict_from_camelcase
+from .tools import dict_from_camelcase, workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 REQUIRED_FIELDS = ("Fee", "Sequence", "TransactionType", "Payment")
 REQUIRED_PAYMENT_FIELDS = ("Amount", "Destination")
@@ -32,6 +32,7 @@ def get_address(*args: Any, **kwargs: Any) -> str:
     return get_authenticated_address(*args, **kwargs).address
 
 
+@workflow(capability=messages.Capability.Ripple)
 def get_authenticated_address(
     session: "Session",
     address_n: "Address",
@@ -46,6 +47,7 @@ def get_authenticated_address(
     )
 
 
+@workflow(capability=messages.Capability.Ripple)
 def sign_tx(
     session: "Session",
     address_n: "Address",
diff --git a/python/src/trezorlib/solana.py b/python/src/trezorlib/solana.py
index ad576fb6..45624fe7 100644
--- a/python/src/trezorlib/solana.py
+++ b/python/src/trezorlib/solana.py
@@ -17,11 +17,13 @@
 from typing import TYPE_CHECKING, Any, List, Optional
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
-    from .transport.session import Session
+    from .client import Session
 
 
+@workflow(capability=messages.Capability.Solana)
 def get_public_key(
     session: "Session",
     address_n: List[int],
@@ -37,6 +39,7 @@ def get_address(*args: Any, **kwargs: Any) -> str:
     return get_authenticated_address(*args, **kwargs).address
 
 
+@workflow(capability=messages.Capability.Solana)
 def get_authenticated_address(
     session: "Session",
     address_n: List[int],
@@ -53,6 +56,7 @@ def get_authenticated_address(
     )
 
 
+@workflow(capability=messages.Capability.Solana)
 def sign_tx(
     session: "Session",
     address_n: List[int],
diff --git a/python/src/trezorlib/stellar.py b/python/src/trezorlib/stellar.py
index 552eb4ab..6f5bf1ae 100644
--- a/python/src/trezorlib/stellar.py
+++ b/python/src/trezorlib/stellar.py
@@ -18,10 +18,11 @@ from decimal import Decimal
 from typing import TYPE_CHECKING, Any, List, Tuple, Union
 
 from . import exceptions, messages
+from .tools import workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
     StellarMessageType = Union[
         messages.StellarAccountMergeOp,
@@ -325,6 +326,7 @@ def get_address(*args: Any, **kwargs: Any) -> str:
     return get_authenticated_address(*args, **kwargs).address
 
 
+@workflow(capability=messages.Capability.Stellar)
 def get_authenticated_address(
     session: "Session",
     address_n: "Address",
@@ -339,6 +341,7 @@ def get_authenticated_address(
     )
 
 
+@workflow(capability=messages.Capability.Stellar)
 def sign_tx(
     session: "Session",
     tx: messages.StellarSignTx,
diff --git a/python/src/trezorlib/tezos.py b/python/src/trezorlib/tezos.py
index 988f40ca..bd38ab4f 100644
--- a/python/src/trezorlib/tezos.py
+++ b/python/src/trezorlib/tezos.py
@@ -17,16 +17,18 @@
 from typing import TYPE_CHECKING, Any
 
 from . import messages
+from .tools import workflow
 
 if TYPE_CHECKING:
+    from .client import Session
     from .tools import Address
-    from .transport.session import Session
 
 
 def get_address(*args: Any, **kwargs: Any) -> str:
     return get_authenticated_address(*args, **kwargs).address
 
 
+@workflow(capability=messages.Capability.Tezos)
 def get_authenticated_address(
     session: "Session",
     address_n: "Address",
@@ -41,6 +43,7 @@ def get_authenticated_address(
     )
 
 
+@workflow(capability=messages.Capability.Tezos)
 def get_public_key(
     session: "Session",
     address_n: "Address",
@@ -55,6 +58,7 @@ def get_public_key(
     ).public_key
 
 
+@workflow(capability=messages.Capability.Tezos)
 def sign_tx(
     session: "Session",
     address_n: "Address",
diff --git a/python/src/trezorlib/thp/__init__.py b/python/src/trezorlib/thp/__init__.py
new file mode 100644
index 00000000..469ccd77
--- /dev/null
+++ b/python/src/trezorlib/thp/__init__.py
@@ -0,0 +1,15 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
diff --git a/python/src/trezorlib/thp/channel.py b/python/src/trezorlib/thp/channel.py
index 44747331..ae52187f 100644
--- a/python/src/trezorlib/thp/channel.py
+++ b/python/src/trezorlib/thp/channel.py
@@ -16,26 +16,451 @@
 
 from __future__ import annotations
 
+import io
+import logging
+import secrets
+import time
 import typing as t
+from enum import Enum, IntEnum, auto
 
-from ... import messages
-from ...mapping import ProtobufMapping
-from .. import Transport
+import typing_extensions as tx
+from noise.connection import Keypair, NoiseConnection
+
+from .. import client, messages, protobuf, transport
+from ..exceptions import (
+    DeviceLockedError,
+    ProtocolError,
+    StateMismatchError,
+    TrezorException,
+)
+from . import control_byte, curve25519, exceptions, thp_io
+from .credentials import TrezorPublicKeys, find_credential
+from .message import Message
+
+if t.TYPE_CHECKING:
+    from noise.functions.keypair import KeyPair
+    from noise.noise_protocol import NoiseProtocol
+
+    from .credentials import Credential
+
+LOG = logging.getLogger(__name__)
+
+DEFAULT_SESSION_ID: int = 0
+
+MAX_RETRANSMISSION_COUNT = 20
+ACK_TIMEOUT = 0.5
+
+
+class ChannelState(IntEnum):
+    """Lifecycle state of the channel.
+
+    Values are ordered and you can use numeric comparison operators to check
+    the current phase.
+    """
+
+    UNALLOCATED = auto()
+    """Channel has not been allocated."""
+    ALLOCATED = auto()
+    """Channel has been allocated."""
+    HANDSHAKE_PHASE = auto()
+    """Handshake in progress."""
+    PAIRING_PHASE = auto()
+    """Pairing in progress."""
+    CREDENTIAL_PHASE = auto()
+    """Pairing complete, credentials can be requested."""
+    ENCRYPTED_TRANSPORT = auto()
+    """Encrypted transport in progress."""
+
+    def is_handshake_done(self) -> bool:
+        return self > ChannelState.HANDSHAKE_PHASE
+
+
+class PairingState(Enum):
+    UNPAIRED = b"\x00"
+    PAIRED = b"\x01"
+    PAIRED_AUTOCONNECT = b"\x02"
+
+    def is_paired(self) -> bool:
+        return self in (self.PAIRED, self.PAIRED_AUTOCONNECT)
+
+
+MT = t.TypeVar("MT", bound=protobuf.MessageType)
+
+
+def encode_proto(message: protobuf.MessageType) -> bytes:
+    buf = io.BytesIO()
+    protobuf.dump_message(buf, message)
+    return buf.getvalue()
+
+
+def decode_proto(message_type: type[MT], message_data: bytes) -> MT:
+    buf = io.BytesIO(message_data)
+    return protobuf.load_message(buf, message_type)
+
+
+def _keypair_from_private_bytes(noise: NoiseProtocol, private_bytes: bytes) -> KeyPair:
+    return noise.dh_fn.klass.from_private_bytes(private_bytes)
+
+
+class ChannelClosedError(TrezorException):
+    """Channel is closed."""
+
+    def __init__(self) -> None:
+        super().__init__(self.__doc__)
 
 
 class Channel:
-    _DEFAULT_READ_TIMEOUT: t.ClassVar[float | None] = None
+    CHUNK_SIZE: t.ClassVar[int | None] = None
+
+    pairing_state: PairingState = PairingState.UNPAIRED
+    sync_bit_send: bool = False
+    sync_bit_receive: bool = False
+
+    BUSY_RETRIES: int = MAX_RETRANSMISSION_COUNT
+    BUSY_BACKOFF_TIME: float = 0.1
 
     def __init__(
         self,
-        transport: Transport,
-        mapping: ProtobufMapping,
+        *,
+        transport: transport.Transport,
+        channel_id: int,
+        device_properties: messages.ThpDeviceProperties,
+        prologue: bytes,
+        channel_state: ChannelState = ChannelState.UNALLOCATED,
     ) -> None:
+        LOG.info("Initializing channel %04x", channel_id)
         self.transport = transport
-        self.mapping = mapping
+        self.channel_id = channel_id
+        self.device_properties = device_properties
+        self.sync_bit_send = False
+        self.sync_bit_receive = False
+        self.prologue = prologue
+        self.host_static_privkey: bytes = secrets.token_bytes(32)
+        self._noise: NoiseConnection | None = None
+        self.state = channel_state
+
+    @property
+    def noise(self) -> NoiseConnection:
+        if self._noise is None:
+            raise ChannelClosedError
+        return self._noise
+
+    @property
+    def handshake_hash(self) -> bytes:
+        self._assert_handshake_done()
+        return self.noise.get_handshake_hash()
+
+    def _assert_handshake_done(self) -> None:
+        if not self.state.is_handshake_done():
+            raise StateMismatchError("Handshake is not finished")
+
+    def get_host_static_pubkey(self) -> bytes:
+        return curve25519.get_public_key(self.host_static_privkey)
+
+    def sync_responses(
+        self, retries: int = MAX_RETRANSMISSION_COUNT, timeout: float = 10.0
+    ) -> None:
+        """Make sure the event loop is running and ready."""
+        with self.transport:
+            nonce = secrets.token_bytes(8)
+            message = Message.broadcast(control_byte.PING, nonce)
+            thp_io.write_payload_to_wire(self.transport, message)
+            for _ in range(1 + retries):
+                message = self._read(timeout=timeout)
+                if not message.is_pong():
+                    LOG.debug(
+                        "Discarding non-pong message: %s", message.to_bytes().hex()
+                    )
+                    continue
+                if message.data != nonce:
+                    LOG.warning(
+                        "Read pong with unexpected nonce: %s",
+                        message.to_bytes().hex(),
+                    )
+                    continue
+                return
+
+        raise transport.Timeout(f"Failed to sync in {retries} retries")
+
+    @classmethod
+    def allocate(
+        cls, transport: transport.Transport, retries: int = thp_io.DEFAULT_MAX_RETRIES
+    ) -> tx.Self:
+        # send channel allocation request
+        LOG.info("Allocating new channel")
+        nonce = secrets.token_bytes(8)
+        message = Message.broadcast(control_byte.CHANNEL_ALLOCATION_REQ, nonce)
+        with transport:
+            thp_io.write_payload_to_wire(transport, message)
+            # read channel allocation response
+            for r in range(1 + retries):
+                message = thp_io.read(transport, max_retries=retries - r)
+                if not message.is_channel_allocation_response():
+                    LOG.info("Not a channel allocation response, ignoring: %s", message)
+                    continue
+                if len(message.data) < 10 or message.data[:8] != nonce:
+                    LOG.warning(
+                        "Unexpected channel allocation nonce. Expected: %s, got: %s",
+                        nonce.hex(),
+                        message.data[:8].hex() or "(empty)",
+                    )
+                    continue
+                channel_id = int.from_bytes(message.data[8:10], "big")
+                LOG.info("Allocated channel %04x", channel_id)
+                prologue = message.data[10:]
+                device_properties = decode_proto(messages.ThpDeviceProperties, prologue)
+                LOG.debug("Device properties: %s", device_properties)
+                return cls(
+                    transport=transport,
+                    channel_id=channel_id,
+                    device_properties=device_properties,
+                    prologue=prologue,
+                    channel_state=ChannelState.ALLOCATED,
+                )
+        raise ProtocolError("Retries exceeded while allocating channel")
+
+    def _init_noise(
+        self,
+        *,
+        static_privkey: bytes | None = None,
+        ephemeral_privkey: bytes | None = None,
+    ) -> None:
+        if static_privkey is not None:
+            self.host_static_privkey = static_privkey
+
+        noise = NoiseConnection.from_name(b"Noise_XX_25519_AESGCM_SHA256")
+        noise.set_as_initiator()
+        noise.set_keypair_from_private_bytes(Keypair.STATIC, self.host_static_privkey)
+        if ephemeral_privkey is not None:
+            noise.set_keypair_from_private_bytes(Keypair.EPHEMERAL, ephemeral_privkey)
+        noise.set_prologue(self.prologue)
+        noise.start_handshake()
+
+        self._noise = noise
+
+    def is_open(self) -> bool:
+        return self.state.is_handshake_done()
+
+    def open(
+        self,
+        credentials: t.Iterable[Credential],
+        *,
+        force_unlock: bool = False,
+    ) -> None:
+        if self.state is ChannelState.UNALLOCATED:
+            raise StateMismatchError("Channel is not allocated")
+        if self.state > ChannelState.ALLOCATED:
+            # channel is already open
+            return
+
+        if self._noise is None:
+            self._init_noise()
+
+        self.state = ChannelState.HANDSHAKE_PHASE
+        with self.transport:
+            try:
+                LOG.info("Performing handshake for channel %04x", self.channel_id)
+                self._send_handshake_init_request(force_unlock)
+                self._read_handshake_init_response()
+                self._send_handshake_completion_request(credentials)
+                self._read_handshake_completion_response()
+                LOG.info("Handshake completed for channel %04x", self.channel_id)
+            except Exception:
+                # any other failure during handshake will close the channel
+                self.close()
+                raise
+
+    def close(self) -> None:
+        self._noise = None
+        self.state = ChannelState.UNALLOCATED
+        self.pairing_state = PairingState.UNPAIRED
+
+    def _send_handshake_init_request(self, unlock: bool) -> None:
+        payload = self.noise.write_message(bytes([unlock]))
+        ha_init_req_message = Message(
+            control_byte.HANDSHAKE_INIT_REQ, self.channel_id, bytes(payload)
+        )
+        self._send_message(ha_init_req_message)
+
+    def _read_handshake_init_response(self) -> None:
+        try:
+            message = self._read()
+        except exceptions.ThpError as e:
+            if e.code == exceptions.ThpErrorCode.DEVICE_LOCKED:
+                raise DeviceLockedError from e
+            raise
+        self._send_ack(message)
+        if not message.is_handshake_init_response():
+            raise ProtocolError(f"Not a valid handshake init response: {message}")
+
+        empty_string = self.noise.read_message(message.data)
+        if empty_string != b"":
+            raise ProtocolError(
+                f"Unexpected data in handshake init response: {empty_string.hex()}"
+            )
+
+    def _send_handshake_completion_request(
+        self, credentials: t.Iterable[Credential]
+    ) -> None:
+        trezor_public_keys = TrezorPublicKeys.from_noise(
+            self.noise.noise_protocol.handshake_state
+        )
+        cred = find_credential(credentials, trezor_public_keys)
+        if cred is not None:
+            LOG.info(
+                "Found credential for channel %04x: %s",
+                self.channel_id,
+                cred.trezor_pubkey.hex(),
+            )
+            # load the credential
+            credential = cred.credential
+            # set the appropriate host static privkey
+            self.host_static_privkey = cred.host_privkey
+            keypair = _keypair_from_private_bytes(
+                self.noise.noise_protocol, cred.host_privkey
+            )
+            self.noise.noise_protocol.handshake_state.s = keypair
+        else:
+            # credential was not found
+            credential = None
+
+        msg_data = encode_proto(
+            messages.ThpHandshakeCompletionReqNoisePayload(
+                host_pairing_credential=credential
+            )
+        )
+        message = self.noise.write_message(payload=msg_data)
+
+        ha_completion_req_message = Message(
+            control_byte.HANDSHAKE_COMP_REQ, self.channel_id, bytes(message)
+        )
+        self._send_message(ha_completion_req_message)
+
+    def _read_handshake_completion_response(self) -> None:
+        # Read handshake completion response
+        message = self._read()
+        self._send_ack(message)
+        if not message.is_handshake_comp_response():
+            LOG.error(
+                "Received message is not a valid handshake completion response: %s",
+                message,
+            )
+            raise ProtocolError(f"Not a valid handshake completion response: {message}")
+
+        trezor_state = self.noise.decrypt(bytes(message.data))
+        try:
+            self.pairing_state = PairingState(trezor_state)
+        except ValueError:
+            raise ProtocolError(f"Invalid trezor state: {trezor_state.hex()}")
+
+        LOG.info("Channel %04x is %s", self.channel_id, self.pairing_state.name)
+        if not self.pairing_state.is_paired():
+            self.state = ChannelState.PAIRING_PHASE
+        else:
+            self.state = ChannelState.CREDENTIAL_PHASE
+
+    def _send_message(self, message: Message) -> None:
+        msg_with_seq_bit = message.with_seq_bit(self.sync_bit_send)
+        self.sync_bit_send = not self.sync_bit_send
+
+        retries_left = self.BUSY_RETRIES
+        retry_backoff_time = self.BUSY_BACKOFF_TIME
+
+        def should_back_off() -> bool:
+            nonlocal retries_left, retry_backoff_time
+            if retries_left <= 0:
+                return False
+            retries_left -= 1
+            time.sleep(retry_backoff_time)
+            retry_backoff_time *= 2
+            return True
+
+        while True:
+            try:
+                thp_io.write_payload_to_wire(self.transport, msg_with_seq_bit)
+                try:
+                    self._read_ack(msg_with_seq_bit)
+                except transport.Timeout:
+                    if should_back_off():
+                        continue
+                    raise
+                break
+            except exceptions.ThpError as e:
+                if (
+                    e.code == exceptions.ThpErrorCode.TRANSPORT_BUSY
+                    and should_back_off()
+                ):
+                    continue
+                raise
+
+    def _send_ack(self, acked_message: Message) -> None:
+        ack = control_byte.make_ack_for(acked_message.ctrl_byte)
+        ack_message = Message(ack, acked_message.cid, b"")
+        thp_io.write_payload_to_wire(self.transport, ack_message)
+
+    def _read_ack(self, message: Message) -> None:
+        expected_seq_bit = message.seq_bit
+        retries = MAX_RETRANSMISSION_COUNT
+        time_start = time.monotonic()
+        for _ in range(1 + retries):
+            time_elapsed = time.monotonic() - time_start
+            message = self._read(timeout=ACK_TIMEOUT - time_elapsed)
+            if not message.is_ack() or len(message.data) > 0:
+                LOG.error("Received message is not a valid ACK: %s", message)
+                # data messages and their acks should have been handled by _read()
+                continue
+            if message.ack_bit != expected_seq_bit:
+                LOG.warning("Received ACK with unexpected sequence bit: %s", message)
+                continue
+            return
+        raise transport.Timeout(
+            f"Failed to read ACK in {retries} retries for message: {message}"
+        )
+
+    def write_chunk(self, data: bytes, /) -> None:
+        self._assert_handshake_done()
+        encrypted_data = self.noise.encrypt(data)
+        message = Message(
+            control_byte.ENCRYPTED_TRANSPORT, self.channel_id, encrypted_data
+        )
+        self._send_message(message)
+
+    def read_chunk(self, *, timeout: float | None = None) -> bytes:
+        self._assert_handshake_done()
+        while True:
+            message = self._read(timeout)
+            if message.cid != self.channel_id:
+                LOG.info("Discarding message from different channel: %s", message)
+                continue
+            if control_byte.is_ack(message.ctrl_byte):
+                LOG.warning("Unexpected ACK: %s", message)
+                continue
+            if not message.is_encrypted_transport():
+                LOG.error("Trying to decrypt not encrypted message! (%s)", message)
+
+            self._send_ack(message)
+            return self.noise.decrypt(bytes(message.data))
+
+    def _read(self, timeout: float | None = None) -> Message:
+        if timeout is None:
+            timeout = client._DEFAULT_READ_TIMEOUT
+
+        while True:
+            message = thp_io.read(self.transport, timeout)
+            if message.seq_bit is not None:
+                if message.seq_bit != self.sync_bit_receive:
+                    LOG.warning(
+                        "Received unexpected message: sync bit=%d, expected=%d",
+                        message.seq_bit,
+                        self.sync_bit_receive,
+                    )
+                    self._send_ack(message)
+                    continue
+
+                self.sync_bit_receive = not self.sync_bit_receive
 
-    def get_features(self) -> messages.Features:
-        raise NotImplementedError()
+            if control_byte.is_error(message.ctrl_byte):
+                code = message.data[0]
+                raise exceptions.ThpError(code)
 
-    def update_features(self) -> None:
-        raise NotImplementedError
+            return message
diff --git a/python/src/trezorlib/thp/checksum.py b/python/src/trezorlib/thp/checksum.py
deleted file mode 100644
index 04de5168..00000000
--- a/python/src/trezorlib/thp/checksum.py
+++ /dev/null
@@ -1,35 +0,0 @@
-# This file is part of the Trezor project.
-#
-# Copyright (C) SatoshiLabs and contributors
-#
-# This library is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Lesser General Public License version 3
-# as published by the Free Software Foundation.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Lesser General Public License for more details.
-#
-# You should have received a copy of the License along with this library.
-# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
-
-import zlib
-
-CHECKSUM_LENGTH = 4
-
-
-def compute(data: bytes) -> bytes:
-    """
-    Returns a CRC-32 checksum of the provided `data`.
-    """
-    return zlib.crc32(data).to_bytes(CHECKSUM_LENGTH, "big")
-
-
-def is_valid(checksum: bytes, data: bytes) -> 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/python/src/trezorlib/thp/client.py b/python/src/trezorlib/thp/client.py
new file mode 100644
index 00000000..da08a5fa
--- /dev/null
+++ b/python/src/trezorlib/thp/client.py
@@ -0,0 +1,201 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+
+from __future__ import annotations
+
+import logging
+import struct
+import typing as t
+from collections import defaultdict
+
+from .. import client, exceptions, messages, models, protobuf
+from ..log import DUMP_BYTES
+from .channel import Channel
+from .pairing import PairingController
+
+if t.TYPE_CHECKING:
+    from ..mapping import ProtobufMapping
+    from ..transport import Transport
+
+LOG = logging.getLogger(__name__)
+
+HEADER_FMT = ">BH"
+HEADER_LEN = struct.calcsize(HEADER_FMT)
+
+
+class ThpSession(client.Session["TrezorClientThp", int]):
+    def derive(
+        self,
+        passphrase: str | client.PassphraseSetting,
+        derive_cardano: bool,
+    ) -> None:
+        msg = messages.ThpCreateNewSession(derive_cardano=derive_cardano)
+        if passphrase is client.PassphraseSetting.ON_DEVICE:
+            msg.on_device = True
+        else:
+            assert isinstance(passphrase, str)
+            msg.passphrase = passphrase
+        self.call(msg, expect=messages.Success)
+
+    @property
+    def channel(self) -> Channel:
+        return self.client.channel
+
+
+class TrezorClientThp(client.TrezorClient[ThpSession]):
+    _channel: Channel | None = None
+    _device_properties: messages.ThpDeviceProperties | None = None
+
+    def __init__(
+        self,
+        app: client.AppManifest,
+        transport: Transport,
+        *,
+        mapping: ProtobufMapping | None,
+        model: models.TrezorModel | None,
+    ) -> None:
+        channel = Channel.allocate(transport)
+        try:
+            # try to open the channel
+            channel.open(app.get_credentials())
+        except exceptions.DeviceLockedError:
+            # If opening failed, the channel is now invalid.
+            # Allocate a new channel for someone else to open.
+            channel = Channel.allocate(transport)
+        self.channel = channel
+
+        if model is None:
+            model = self.detect_model(self.device_properties)
+        if mapping is None:
+            mapping = model.default_mapping
+
+        super().__init__(
+            app=app,
+            transport=transport,
+            mapping=mapping,
+            model=model,
+            pairing=PairingController(self),
+        )
+        self._session_id_counter = 0
+        self._session_message_queue: dict[int, list[protobuf.MessageType]] = (
+            defaultdict(list)
+        )
+
+    def connect(self) -> None:
+        if self.channel.is_open():
+            return
+        self.channel.open(self.app.get_credentials(), force_unlock=True)
+
+    def is_connected(self) -> bool:
+        return self.channel.is_open()
+
+    def _get_any_session(self) -> ThpSession:
+        if not self.channel.is_open():
+            raise exceptions.DeviceLockedError
+        if not self.pairing.is_paired():
+            raise exceptions.NotPairedError
+        else:
+            self.pairing.finish()
+        return ThpSession(self, self._session_id_counter)
+
+    def _get_pairing_session(self) -> ThpSession:
+        return ThpSession(self, 0)
+
+    def _get_session(
+        self,
+        *,
+        passphrase: str | client.PassphraseSetting | None,
+        derive_cardano: bool,
+    ) -> ThpSession:
+        if not self.pairing.is_paired():
+            raise exceptions.NotPairedError
+        else:
+            self.pairing.finish()
+
+        if passphrase is None:
+            return ThpSession(self, 0)
+
+        self._session_id_counter += 1
+        if self._session_id_counter >= 0xFF:
+            self._session_id_counter = 1
+        session = ThpSession(self, self._session_id_counter)
+        session.derive(passphrase, derive_cardano)
+        return session
+
+    def _invalidate(self) -> None:
+        super()._invalidate()
+        # Close the channel. The client cannot be used until a channel is
+        # re-established.
+        self.channel.close()
+        self._session_id_counter = 0
+        self._session_message_queue.clear()
+
+    def _write(self, session: ThpSession, msg: protobuf.MessageType) -> None:
+        if not self.channel.is_open():
+            raise exceptions.DeviceLockedError
+
+        LOG.debug(
+            f"sending message: {msg.__class__.__name__}",
+            extra={"protobuf": msg, "session": session},
+        )
+        msg_type, msg_bytes = self.mapping.encode(msg)
+        LOG.log(
+            DUMP_BYTES,
+            f"encoded as type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
+        )
+        header = struct.pack(HEADER_FMT, session.id, msg_type)
+        self.channel.write_chunk(header + msg_bytes)
+
+    def _read(
+        self, session: ThpSession, timeout: float | None = None
+    ) -> protobuf.MessageType:
+        if not self.channel.is_open():
+            raise exceptions.DeviceLockedError
+
+        if self._session_message_queue[session.id]:
+            return self._session_message_queue[session.id].pop(0)
+        if session.is_invalid:
+            raise exceptions.InvalidSessionError(session.id)
+        while True:
+            msg = self.channel.read_chunk(timeout=timeout)
+            session_id, msg_type = struct.unpack(HEADER_FMT, msg[:HEADER_LEN])
+            msg_bytes = msg[HEADER_LEN:]
+            LOG.log(
+                DUMP_BYTES,
+                f"received type {msg_type} ({len(msg_bytes)} bytes): {msg_bytes.hex()}",
+                extra={"session": session_id},
+            )
+            msg = self.mapping.decode(msg_type, msg_bytes)
+            LOG.debug(
+                f"received message: {msg.__class__.__name__}",
+                extra={"protobuf": msg, "session": session_id},
+            )
+            if session_id == session.id:
+                return msg
+            else:
+                self._session_message_queue[session_id].append(msg)
+
+    @staticmethod
+    def detect_model(props: messages.ThpDeviceProperties) -> models.TrezorModel:
+        internal_model = props.internal_model
+        model = models.by_internal_name(internal_model)
+        if model is None:
+            model = models.unknown_model(None, internal_model)
+        return model
+
+    @property
+    def device_properties(self) -> messages.ThpDeviceProperties:
+        return self.channel.device_properties
diff --git a/python/src/trezorlib/thp/control_byte.py b/python/src/trezorlib/thp/control_byte.py
index 59bc0ea8..9fdab9e0 100644
--- a/python/src/trezorlib/thp/control_byte.py
+++ b/python/src/trezorlib/thp/control_byte.py
@@ -14,72 +14,128 @@
 # You should have received a copy of the License along with this library.
 # If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
 
-from typing import Optional
+from __future__ import annotations
 
-CODEC_V1 = 0x3F
-CONTINUATION_PACKET = 0x80
+CONTINUATION_BIT = 0b1000_0000
+
+# data packets
+DATA_MASK = 0xE7
 HANDSHAKE_INIT_REQ = 0x00
 HANDSHAKE_INIT_RES = 0x01
 HANDSHAKE_COMP_REQ = 0x02
 HANDSHAKE_COMP_RES = 0x03
 ENCRYPTED_TRANSPORT = 0x04
+DATA_SEQ_BIT = 0b0001_0000
+DATA_ACK_SEQ_BIT = 0b0000_1000
 
-CONTINUATION_PACKET_MASK = 0x80
-ACK_MASK = 0xF7
-DATA_MASK = 0xE7
+DATA_DETECT_BASE = 0b0000_0000
+DATA_DETECT_MASK = 0b1110_0000
 
-ACK_MESSAGE = 0x20
-_ERROR = 0x42
-CHANNEL_ALLOCATION_REQ = 0x40
-_CHANNEL_ALLOCATION_RES = 0x41
+# ack packets
+ACK_MASK = 0b1111_0111
+ACK_BASE = 0b0010_0000
+ACK_SEQ_BIT = 0b0000_1000
 
-TREZOR_STATE_UNPAIRED = b"\x00"
-TREZOR_STATE_PAIRED = b"\x01"
-
-
-def add_seq_bit_to_ctrl_byte(ctrl_byte: int, seq_bit: int) -> int:
-    if seq_bit == 0:
-        return ctrl_byte & 0xEF
-    if seq_bit == 1:
-        return ctrl_byte | 0x10
-    raise Exception("Unexpected sequence bit")
+# special values
+CODEC_V1 = 0x3F
+CHANNEL_ALLOCATION_REQ = 0x40
+CHANNEL_ALLOCATION_RES = 0x41
+ERROR = 0x42
+PING = 0x43
+PONG = 0x44
+
+
+HANDSHAKE_SEQ_BITS = {
+    HANDSHAKE_INIT_REQ: False,
+    HANDSHAKE_INIT_RES: False,
+    HANDSHAKE_COMP_REQ: True,
+    HANDSHAKE_COMP_RES: True,
+}
+
+
+FIXED_NAMES = {
+    CODEC_V1: "CODEC_V1",
+    CHANNEL_ALLOCATION_REQ: "CHANNEL_ALLOCATION_REQ",
+    CHANNEL_ALLOCATION_RES: "CHANNEL_ALLOCATION_RES",
+    ERROR: "ERROR",
+    PING: "PING",
+    PONG: "PONG",
+}
+
+DATA_NAMES = {
+    HANDSHAKE_INIT_REQ: "HANDSHAKE_INIT_REQ",
+    HANDSHAKE_INIT_RES: "HANDSHAKE_INIT_RES",
+    HANDSHAKE_COMP_REQ: "HANDSHAKE_COMP_REQ",
+    HANDSHAKE_COMP_RES: "HANDSHAKE_COMP_RES",
+    ENCRYPTED_TRANSPORT: "ENCRYPTED_TRANSPORT",
+}
+
+
+def to_string(ctrl_byte: int) -> str:
+    hex = f"0x{ctrl_byte:02x}"
+    if is_continuation(ctrl_byte):
+        return f"{hex} (CONTINUATION)"
+    if ctrl_byte in FIXED_NAMES:
+        return f"{hex} ({FIXED_NAMES[ctrl_byte]})"
+    if is_ack(ctrl_byte):
+        ack_bit = bool(ctrl_byte & ACK_SEQ_BIT)
+        return f"{hex} (ACK{int(ack_bit)})"
+    if ctrl_byte & DATA_MASK in DATA_NAMES:
+        seq_bit = int(get_seq_bit(ctrl_byte) or 0)
+        ack_bit = int(bool(ctrl_byte & DATA_ACK_SEQ_BIT) or 0)
+        return f"{hex} ({DATA_NAMES[ctrl_byte & DATA_MASK]} seq{seq_bit} ack{ack_bit})"
+    return f"{hex} (reserved)"
+
+
+def set_seq_bit(ctrl_byte: int, seq_bit: bool) -> int:
+    if not is_data(ctrl_byte):
+        return ctrl_byte
+    return ctrl_byte | (DATA_SEQ_BIT * seq_bit)
 
 
 def add_ack_bit_to_ctrl_byte(ctrl_byte: int, ack_bit: int) -> int:
-    if ack_bit == 0:
-        return ctrl_byte & 0xF7
-    if ack_bit == 1:
-        return ctrl_byte | 0x08
-    raise Exception("Unexpected acknowledgement bit")
+    return ctrl_byte | (DATA_ACK_SEQ_BIT * ack_bit)
 
 
-def get_seq_bit(ctrl_byte: int) -> Optional[int]:
-    if ctrl_byte & 0xE0:
+def get_seq_bit(ctrl_byte: int) -> bool | None:
+    if ctrl_byte in HANDSHAKE_SEQ_BITS:
+        return HANDSHAKE_SEQ_BITS[ctrl_byte]
+    if not is_data(ctrl_byte):
         # not all message types contain SEQ bit
         return None
+    return bool(ctrl_byte & DATA_SEQ_BIT)
 
-    return (ctrl_byte & 0x10) >> 4
 
+def get_ack_bit(ctrl_byte: int) -> bool | None:
+    if not is_ack(ctrl_byte):
+        return None
+    return bool(ctrl_byte & ACK_SEQ_BIT)
 
-def is_ack(ctrl_byte: int) -> bool:
-    return ctrl_byte & ACK_MASK == ACK_MESSAGE
 
+def make_ack(ack_bit: bool) -> int:
+    return ACK_BASE | (ACK_SEQ_BIT * ack_bit)
 
-def is_error(ctrl_byte: int) -> bool:
-    return ctrl_byte == _ERROR
 
+def make_ack_for(ctrl_byte: int) -> int:
+    bit = get_seq_bit(ctrl_byte)
+    if bit is None:
+        raise ValueError(
+            f"Cannot make ack for non-data control byte: {to_string(ctrl_byte)}"
+        )
+    return make_ack(bit)
 
-def is_continuation(ctrl_byte: int) -> bool:
-    return ctrl_byte & CONTINUATION_PACKET_MASK == CONTINUATION_PACKET
+
+def is_ack(ctrl_byte: int) -> bool:
+    return ctrl_byte & ACK_MASK == ACK_BASE
 
 
-def is_encrypted_transport(ctrl_byte: int) -> bool:
-    return ctrl_byte & DATA_MASK == ENCRYPTED_TRANSPORT
+def is_data(ctrl_byte: int) -> bool:
+    return ctrl_byte & DATA_DETECT_MASK == DATA_DETECT_BASE
 
 
-def is_handshake_init_req(ctrl_byte: int) -> bool:
-    return ctrl_byte & DATA_MASK == HANDSHAKE_INIT_REQ
+def is_error(ctrl_byte: int) -> bool:
+    return ctrl_byte == ERROR
 
 
-def is_handshake_comp_req(ctrl_byte: int) -> bool:
-    return ctrl_byte & DATA_MASK == HANDSHAKE_COMP_REQ
+def is_continuation(ctrl_byte: int) -> bool:
+    return bool(ctrl_byte & CONTINUATION_BIT)
diff --git a/python/src/trezorlib/thp/cpace.py b/python/src/trezorlib/thp/cpace.py
index 6df9678c..b739b19e 100644
--- a/python/src/trezorlib/thp/cpace.py
+++ b/python/src/trezorlib/thp/cpace.py
@@ -13,44 +13,92 @@
 #
 # You should have received a copy of the License along with this library.
 # If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+"""
+CPace, a balanced composable PAKE: https://datatracker.ietf.org/doc/draft-irtf-cfrg-cpace/
+"""
 
+from __future__ import annotations
+
+import secrets
 import typing as t
 from hashlib import sha512
 
 from . import curve25519
 
-_PREFIX = b"\x08\x43\x50\x61\x63\x65\x32\x35\x35\x06"
-_PADDING = b"\x6f\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\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\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\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x20"
+DSI = b"CPace255"
+HASH_BLOCK_SIZE = sha512().block_size
+FIELD_SIZE_BYTES = 32
 
 
-class Cpace:
-    """
-    CPace, a balanced composable PAKE: https://datatracker.ietf.org/doc/draft-irtf-cfrg-cpace/
-    """
+class CpaceResult(t.NamedTuple):
+    """Result of a CPace exchange."""
+
+    a_pubkey: bytes
+    """Public key of the initiator, to be sent to the counterparty."""
+    shared_secret: bytes
+    """Shared secret, to be used for further communication."""
+
+
+def _leb128(value: int) -> bytes:
+    if value > 0x7F:
+        raise NotImplementedError
+    return value.to_bytes(1, "little")
+
+
+def _prepend_len(data: bytes) -> bytes:
+    return _leb128(len(data)) + data
+
+
+def _lv_cat(*args: bytes) -> bytes:
+    return b"".join(_prepend_len(arg) for arg in args)
 
-    random_bytes: t.Callable[[int], bytes]
-
-    def __init__(self, handshake_hash: bytes) -> None:
-        self.handshake_hash: bytes = handshake_hash
-        self.shared_secret: bytes
-        self.host_private_key: bytes
-        self.host_public_key: bytes
-
-    def generate_keys_and_secret(
-        self, code_code_entry: bytes, trezor_public_key: bytes
-    ) -> None:
-        """
-        Generate ephemeral key pair and a shared secret using Elligator2 with X25519.
-        """
-        sha_ctx = sha512(_PREFIX)
-        sha_ctx.update(code_code_entry)
-        sha_ctx.update(_PADDING)
-        sha_ctx.update(self.handshake_hash)
-        sha_ctx.update(b"\x00")
-        pregenerator = sha_ctx.digest()[:32]
-        generator = curve25519.elligator2(pregenerator)
-        self.host_private_key = self.random_bytes(32)
-        self.host_public_key = curve25519.multiply(self.host_private_key, generator)
-        self.shared_secret = curve25519.multiply(
-            self.host_private_key, trezor_public_key
-        )
+
+def _generator_string(
+    *,
+    prs: bytes,
+    ci: bytes,
+    sid: bytes = b"",
+) -> bytes:
+    dsi_bytes = _prepend_len(DSI)
+    prs_bytes = _prepend_len(prs)
+    len_zpad = max(0, HASH_BLOCK_SIZE - (len(dsi_bytes) + len(prs_bytes) + 1))
+    return _lv_cat(DSI, prs, b"\x00" * len_zpad, ci, sid)
+
+
+def _generator(prs: bytes, ci: bytes, sid: bytes = b"") -> bytes:
+    gen_str = _generator_string(prs=prs, ci=ci, sid=sid)
+    gen_str_hashed = sha512(gen_str).digest()[:FIELD_SIZE_BYTES]
+    return curve25519.elligator2(gen_str_hashed)
+
+
+def cpace(
+    *,
+    prs: bytes,
+    ci: bytes,
+    sid: bytes = b"",
+    b_pubkey: bytes,
+    _a_privkey: bytes | None = None,
+) -> CpaceResult:
+    """Perform the CPace255 protocol.
+
+    That is, an instance of CPace for group object G_X25519.
+
+    Detailed specification is available at https://datatracker.ietf.org/doc/draft-irtf-cfrg-cpace/,
+    argument names match the specification.
+
+    Arguments:
+        prs: Possibly low-entropy shared passphrase.
+        ci: Channel identifier that binds both participantsto the current communication channel.
+        sid: Optional session identifier.
+        b_pubkey: Public key of the counterparty.
+
+    Returns: the result of the CPace protocol. See `CpaceResult` for details.
+    """
+    generator = _generator(prs=prs, ci=ci, sid=sid)
+    if _a_privkey is not None:
+        a_privkey = _a_privkey
+    else:
+        a_privkey = secrets.token_bytes(FIELD_SIZE_BYTES)
+    a_pubkey = curve25519.multiply(a_privkey, generator)
+    shared_secret = curve25519.multiply(a_privkey, b_pubkey)
+    return CpaceResult(a_pubkey=a_pubkey, shared_secret=shared_secret)
diff --git a/python/src/trezorlib/thp/credentials.py b/python/src/trezorlib/thp/credentials.py
new file mode 100644
index 00000000..c3a3f6c7
--- /dev/null
+++ b/python/src/trezorlib/thp/credentials.py
@@ -0,0 +1,73 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+
+from __future__ import annotations
+
+import typing as t
+from dataclasses import dataclass
+from hashlib import sha256
+
+from cryptography.hazmat.primitives.asymmetric import x25519
+from typing_extensions import Self
+
+if t.TYPE_CHECKING:
+    from noise.state import HandshakeState
+
+
+class Credential(t.Protocol):
+    @property
+    def trezor_pubkey(self) -> bytes: ...
+    @property
+    def host_privkey(self) -> bytes: ...
+    @property
+    def credential(self) -> bytes: ...
+
+
+class TrezorPublicKeys(t.NamedTuple):
+    ephemeral: bytes
+    static_masked: bytes
+
+    @classmethod
+    def from_noise(cls, handshake_state: HandshakeState) -> Self:
+        return cls(
+            ephemeral=handshake_state.re.public_bytes,
+            static_masked=handshake_state.rs.public_bytes,
+        )
+
+
+@dataclass(frozen=True)
+class StaticCredential:
+    trezor_pubkey: bytes
+    host_privkey: bytes
+    credential: bytes
+
+
+def matches(credential: Credential, trezor_public_keys: TrezorPublicKeys) -> bool:
+    mask = sha256(credential.trezor_pubkey + trezor_public_keys.ephemeral).digest()
+    mask_as_privkey = x25519.X25519PrivateKey.from_private_bytes(mask)
+    pubkey = x25519.X25519PublicKey.from_public_bytes(credential.trezor_pubkey)
+    shared_secret = mask_as_privkey.exchange(pubkey)
+    return shared_secret == trezor_public_keys.static_masked
+
+
+def find_credential(
+    credentials: t.Iterable[Credential],
+    trezor_public_keys: TrezorPublicKeys,
+) -> Credential | None:
+    for credential in credentials:
+        if matches(credential, trezor_public_keys):
+            return credential
+    return None
diff --git a/python/src/trezorlib/thp/exceptions.py b/python/src/trezorlib/thp/exceptions.py
new file mode 100644
index 00000000..35922b0d
--- /dev/null
+++ b/python/src/trezorlib/thp/exceptions.py
@@ -0,0 +1,46 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+
+from __future__ import annotations
+
+from enum import IntEnum
+
+from .. import exceptions
+
+
+class ThpErrorCode(IntEnum):
+    TRANSPORT_BUSY = 1
+    UNALLOCATED_CHANNEL = 2
+    DECRYPTION_FAILED = 3
+    DEVICE_LOCKED = 5
+
+    @classmethod
+    def to_exception(cls, code: int) -> ThpError:
+        try:
+            valid_code = cls(code)
+            return ThpError(valid_code)
+        except ValueError:
+            return ThpError(code)
+
+
+class ThpError(exceptions.TrezorException):
+    def __init__(self, code: ThpErrorCode | int) -> None:
+        self.code = code
+        if isinstance(code, ThpErrorCode):
+            self.name = code.name
+        else:
+            self.name = "unknown"
+        super().__init__(code, self.name)
diff --git a/python/src/trezorlib/thp/message.py b/python/src/trezorlib/thp/message.py
new file mode 100644
index 00000000..f3b7c44a
--- /dev/null
+++ b/python/src/trezorlib/thp/message.py
@@ -0,0 +1,158 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+
+from __future__ import annotations
+
+import io
+import struct
+import typing as t
+import zlib
+from dataclasses import dataclass
+from functools import cached_property
+
+from typing_extensions import Self
+
+from .. import exceptions
+from . import control_byte
+
+BROADCAST_CHANNEL_ID = 0xFFFF
+
+FORMAT_STR_INIT = ">BHH"
+FORMAT_STR_CONT = ">BH"
+
+CHECKSUM_LENGTH = 4
+
+
+def packet_header(ctrl_byte: int, cid: int) -> bytes:
+    return struct.pack(">BH", ctrl_byte, cid)
+
+
+def packet_length(data: bytes) -> bytes:
+    return struct.pack(">H", len(data) + CHECKSUM_LENGTH)
+
+
+def _crc32(data: bytes) -> bytes:
+    return zlib.crc32(data).to_bytes(CHECKSUM_LENGTH, "big")
+
+
+class ChecksumError(exceptions.ProtocolError):
+    """Invalid checksum in message."""
+
+    def __init__(self, message: Message, received_checksum: bytes) -> None:
+        self.message = message
+        self.received_checksum = received_checksum
+        super().__init__(message, received_checksum)
+
+
+@dataclass(frozen=True)
+class Message:
+    ctrl_byte: int
+    cid: int
+    data: bytes
+
+    @staticmethod
+    def checked_bytes(ctrl_byte: int, cid: int, data: bytes) -> bytes:
+        return packet_header(ctrl_byte, cid) + packet_length(data) + data
+
+    def checksum(self) -> bytes:
+        return _crc32(self.checked_bytes(self.ctrl_byte, self.cid, self.data))
+
+    def __str__(self) -> str:
+        props = {
+            "ctrl": control_byte.to_string(self.ctrl_byte),
+            "cid": f"0x{self.cid:04x}",
+            "data": self.data.hex(),
+        }
+        props_str = ", ".join(f"{k}={v}" for k, v in props.items())
+        return f"Message({props_str})"
+
+    def to_bytes(self) -> bytes:
+        """Return the message as a single byte string with the appropriate header."""
+        return self.checked_bytes(self.ctrl_byte, self.cid, self.data) + self.checksum()
+
+    def chunks(self, chunk_size: int) -> t.Iterator[bytes]:
+        """Yield chunks of the message, properly delineated by the right
+        control bytes and padded to the chunk size.
+        """
+        payload_reader = io.BytesIO(self.to_bytes())
+        first_chunk = payload_reader.read(chunk_size)
+        yield first_chunk.ljust(chunk_size, b"\x00")
+
+        cont_header = packet_header(control_byte.CONTINUATION_BIT, self.cid)
+        cont_chunk_size = chunk_size - len(cont_header)
+        while buffer := payload_reader.read(cont_chunk_size):
+            chunk = cont_header + buffer
+            yield chunk.ljust(chunk_size, b"\x00")
+
+    @classmethod
+    def parse(cls, ctrl_byte: int, cid: int, payload: bytes) -> Self:
+        if len(payload) < CHECKSUM_LENGTH:
+            raise exceptions.ProtocolError("Payload too short")
+        data, checksum = payload[:-CHECKSUM_LENGTH], payload[-CHECKSUM_LENGTH:]
+        new = cls(ctrl_byte, cid, data)
+        if new.checksum() != checksum:
+            raise ChecksumError(new, checksum)
+        return new
+
+    @classmethod
+    def ack(cls, cid: int, ack_bit: bool) -> Self:
+        return cls(control_byte.make_ack(ack_bit), cid, b"")
+
+    @classmethod
+    def broadcast(cls, ctrl_byte: int, data: bytes) -> Self:
+        return cls(ctrl_byte, BROADCAST_CHANNEL_ID, data)
+
+    def with_seq_bit(self, seq_bit: bool) -> Self:
+        return self.__class__(
+            control_byte.set_seq_bit(self.ctrl_byte, seq_bit),
+            self.cid,
+            self.data,
+        )
+
+    @cached_property
+    def seq_bit(self) -> bool | None:
+        return control_byte.get_seq_bit(self.ctrl_byte)
+
+    @cached_property
+    def ack_bit(self) -> bool | None:
+        return control_byte.get_ack_bit(self.ctrl_byte)
+
+    def is_ack(self) -> bool:
+        return control_byte.is_ack(self.ctrl_byte)
+
+    def is_channel_allocation_response(self) -> bool:
+        return (
+            self.cid == BROADCAST_CHANNEL_ID
+            and self.ctrl_byte == control_byte.CHANNEL_ALLOCATION_RES
+        )
+
+    def is_pong(self) -> bool:
+        return self.cid == BROADCAST_CHANNEL_ID and self.ctrl_byte == control_byte.PONG
+
+    def is_handshake_init_response(self) -> bool:
+        return (
+            self.ctrl_byte & control_byte.DATA_MASK == control_byte.HANDSHAKE_INIT_RES
+        )
+
+    def is_handshake_comp_response(self) -> bool:
+        return (
+            self.ctrl_byte & control_byte.DATA_MASK == control_byte.HANDSHAKE_COMP_RES
+        )
+
+    def is_encrypted_transport(self) -> bool:
+        return (
+            self.ctrl_byte & control_byte.DATA_MASK == control_byte.ENCRYPTED_TRANSPORT
+        )
diff --git a/python/src/trezorlib/thp/message_header.py b/python/src/trezorlib/thp/message_header.py
deleted file mode 100644
index 525522f9..00000000
--- a/python/src/trezorlib/thp/message_header.py
+++ /dev/null
@@ -1,112 +0,0 @@
-# This file is part of the Trezor project.
-#
-# Copyright (C) SatoshiLabs and contributors
-#
-# This library is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Lesser General Public License version 3
-# as published by the Free Software Foundation.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Lesser General Public License for more details.
-#
-# You should have received a copy of the License along with this library.
-# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
-
-from __future__ import annotations
-
-import struct
-
-from typing_extensions import Self
-
-CODEC_V1 = 0x3F
-CONTINUATION_PACKET = 0x80
-HANDSHAKE_INIT_REQ = 0x00
-HANDSHAKE_INIT_RES = 0x01
-HANDSHAKE_COMP_REQ = 0x02
-HANDSHAKE_COMP_RES = 0x03
-ENCRYPTED_TRANSPORT = 0x04
-
-CONTINUATION_PACKET_MASK = 0x80
-ACK_MASK = 0xF7
-DATA_MASK = 0xE7
-
-ACK_MESSAGE = 0x20
-_ERROR = 0x42
-CHANNEL_ALLOCATION_REQ = 0x40
-_CHANNEL_ALLOCATION_RES = 0x41
-
-PING = 0x43
-PONG = 0x44
-
-TREZOR_STATE_UNPAIRED = b"\x00"
-TREZOR_STATE_PAIRED = b"\x01"
-
-BROADCAST_CHANNEL_ID = 0xFFFF
-
-
-class MessageHeader:
-    format_str_init = ">BHH"
-    format_str_cont = ">BH"
-
-    def __init__(self, ctrl_byte: int, cid: int, length: int) -> None:
-        self.ctrl_byte = ctrl_byte
-        self.cid = cid
-        self.data_length = length
-
-    def to_bytes_init(self) -> bytes:
-        return struct.pack(
-            self.format_str_init, self.ctrl_byte, self.cid, self.data_length
-        )
-
-    def to_bytes_cont(self) -> bytes:
-        return struct.pack(self.format_str_cont, CONTINUATION_PACKET, self.cid)
-
-    def pack_to_init_buffer(self, buffer: bytearray, buffer_offset: int = 0) -> None:
-        struct.pack_into(
-            self.format_str_init,
-            buffer,
-            buffer_offset,
-            self.ctrl_byte,
-            self.cid,
-            self.data_length,
-        )
-
-    def pack_to_cont_buffer(self, buffer: bytearray, buffer_offset: int = 0) -> None:
-        struct.pack_into(
-            self.format_str_cont, buffer, buffer_offset, CONTINUATION_PACKET, self.cid
-        )
-
-    def is_ack(self) -> bool:
-        return self.ctrl_byte & ACK_MASK == ACK_MESSAGE
-
-    def is_channel_allocation_response(self) -> bool:
-        return (
-            self.cid == BROADCAST_CHANNEL_ID
-            and self.ctrl_byte == _CHANNEL_ALLOCATION_RES
-        )
-
-    def is_pong(self) -> bool:
-        return self.cid == BROADCAST_CHANNEL_ID and self.ctrl_byte == PONG
-
-    def is_handshake_init_response(self) -> bool:
-        return self.ctrl_byte & DATA_MASK == HANDSHAKE_INIT_RES
-
-    def is_handshake_comp_response(self) -> bool:
-        return self.ctrl_byte & DATA_MASK == HANDSHAKE_COMP_RES
-
-    def is_encrypted_transport(self) -> bool:
-        return self.ctrl_byte & DATA_MASK == ENCRYPTED_TRANSPORT
-
-    @classmethod
-    def get_error_header(cls, cid: int, length: int) -> Self:
-        return cls(_ERROR, cid, length)
-
-    @classmethod
-    def get_channel_allocation_request_header(cls, length: int) -> Self:
-        return cls(CHANNEL_ALLOCATION_REQ, BROADCAST_CHANNEL_ID, length)
-
-    @classmethod
-    def get_ping_header(cls, length: int) -> Self:
-        return cls(PING, BROADCAST_CHANNEL_ID, length)
diff --git a/python/src/trezorlib/thp/pairing.py b/python/src/trezorlib/thp/pairing.py
new file mode 100644
index 00000000..b7d49d9f
--- /dev/null
+++ b/python/src/trezorlib/thp/pairing.py
@@ -0,0 +1,399 @@
+# This file is part of the Trezor project.
+#
+# Copyright (C) SatoshiLabs and contributors
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the License along with this library.
+# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
+
+from __future__ import annotations
+
+import secrets
+import typing as t
+from abc import ABCMeta, abstractmethod
+from dataclasses import dataclass
+from enum import Enum, auto
+from hashlib import sha256
+
+import typing_extensions as tx
+
+from .. import messages
+from ..exceptions import ProtocolError, StateMismatchError, TrezorException
+from ..protobuf import MessageType
+from .channel import Channel, ChannelState, PairingState
+from .cpace import cpace
+from .credentials import Credential, StaticCredential
+
+if t.TYPE_CHECKING:
+    from ..client import MT
+    from .client import TrezorClientThp
+
+
+class ControllerLifecycle(Enum):
+    INITIAL = auto()
+    PAIRING_REQUESTED = auto()
+    PAIRING_COMPLETED = auto()
+    FINISHED = auto()
+    FAILED = auto()
+
+    def abort_if_failed(self) -> None:
+        if self is ControllerLifecycle.FAILED:
+            raise ValueError("Pairing failed")
+
+
+@dataclass
+class CodeEntryState:
+    challenge: bytes
+    commitment: bytes
+    cpace_trezor_public_key: bytes
+
+
+class PairingController:
+    def __init__(self, client: TrezorClientThp) -> None:
+        self.opened = False
+        self.client = client
+        self.session = client._get_pairing_session()
+        self._pairing_requested = False
+        self._failed = False
+
+    @property
+    def state(self) -> ControllerLifecycle:
+        if self._failed:
+            return ControllerLifecycle.FAILED
+        if self.channel.state is ChannelState.CREDENTIAL_PHASE:
+            return ControllerLifecycle.PAIRING_COMPLETED
+        elif self.channel.state is ChannelState.ENCRYPTED_TRANSPORT:
+            return ControllerLifecycle.FINISHED
+        elif (
+            self.channel.state is ChannelState.PAIRING_PHASE and self._pairing_requested
+        ):
+            return ControllerLifecycle.PAIRING_REQUESTED
+        else:
+            return ControllerLifecycle.INITIAL
+
+    @state.setter
+    def state(self, state: ControllerLifecycle) -> None:
+        self._failed = state is ControllerLifecycle.FAILED
+        if state is ControllerLifecycle.INITIAL:
+            self._pairing_requested = False
+        elif state is ControllerLifecycle.PAIRING_REQUESTED:
+            if self.channel.state > ChannelState.PAIRING_PHASE:
+                raise StateMismatchError(
+                    "Tried to revert to pairing phase from a later state"
+                )
+            self.channel.state = ChannelState.PAIRING_PHASE
+            self._pairing_requested = True
+        elif state is ControllerLifecycle.PAIRING_COMPLETED:
+            self.channel.state = ChannelState.CREDENTIAL_PHASE
+            self._pairing_requested = False
+        elif state is ControllerLifecycle.FINISHED:
+            self.channel.state = ChannelState.ENCRYPTED_TRANSPORT
+            self._pairing_requested = False
+        else:
+            raise ValueError(f"Invalid state: {state}")
+
+    def _maybe_open(self) -> None:
+        if self.opened:
+            return
+        self.opened = True
+        self.client.connect()
+        self.session.__enter__()
+
+    def _maybe_close(self) -> None:
+        if not self.opened:
+            return
+        self.opened = False
+        self.session.__exit__(None, None, None)
+
+    def start(self) -> None:
+        self.state.abort_if_failed()
+        self._maybe_open()
+        if self.state is not ControllerLifecycle.INITIAL:
+            return
+        self.session.call(
+            messages.ThpPairingRequest(
+                host_name=self.client.app.host_name,
+                app_name=self.client.app.app_name,
+            ),
+            expect=messages.ThpPairingRequestApproved,
+        )
+        self.state = ControllerLifecycle.PAIRING_REQUESTED
+
+    def _call(self, message: MessageType, *, expect: type[MT]) -> MT:
+        self.start()
+        return self.session.call(message, expect=expect)
+
+    @property
+    def channel(self) -> Channel:
+        return self.client.channel
+
+    @property
+    def methods(self) -> t.Collection[type["PairingMethod"]]:
+        return {
+            m
+            for m in PairingMethod.METHODS_AVAILABLE
+            if m.PAIRING_METHOD in self.channel.device_properties.pairing_methods
+        }
+
+    def is_paired(self) -> bool:
+        return self.channel.pairing_state.is_paired()
+
+    def set_paired(self) -> None:
+        self.state.abort_if_failed()
+        if not self.channel.pairing_state.is_paired():
+            self.channel.pairing_state = PairingState.PAIRED
+        self.state = ControllerLifecycle.PAIRING_COMPLETED
+
+    def finish(self, _no_call: bool = False) -> None:
+        if self.state is ControllerLifecycle.FINISHED:
+            return
+        self.state.abort_if_failed()
+        if not _no_call:
+            self._call(messages.ThpEndRequest(), expect=messages.ThpEndResponse)
+        self.state = ControllerLifecycle.FINISHED
+        self._maybe_close()
+
+    def abort(self) -> None:
+        self.state = ControllerLifecycle.FAILED
+        self.channel.close()
+        self._maybe_close()
+
+    def _check_state(self, required_state: ControllerLifecycle) -> None:
+        if self.state != required_state:
+            raise StateMismatchError(
+                f"Tried to execute a {required_state.name} operation in the {self.state.name} state"
+            )
+
+    def request_credential(self, autoconnect: bool = False) -> Credential:
+        self._check_state(ControllerLifecycle.PAIRING_COMPLETED)
+        pubkey = self.channel.get_host_static_pubkey()
+        credential_response = self._call(
+            messages.ThpCredentialRequest(
+                host_static_public_key=pubkey,
+                autoconnect=autoconnect,
+            ),
+            expect=messages.ThpCredentialResponse,
+        )
+        return StaticCredential(
+            host_privkey=self.channel.host_static_privkey,
+            credential=credential_response.credential,
+            trezor_pubkey=credential_response.trezor_static_public_key,
+        )
+
+    # ==== Available pairing flows ====
+    def skip(self) -> None:
+        if not self.is_paired():
+            SkipPairing(self)
+
+
+class PairingMethod(metaclass=ABCMeta):
+    METHODS_AVAILABLE: t.ClassVar[set[type[tx.Self]]] = set()
+
+    PAIRING_METHOD: t.ClassVar[messages.ThpPairingMethod]
+
+    def __init__(self, controller: PairingController) -> None:
+        controller.start()
+        controller._check_state(ControllerLifecycle.PAIRING_REQUESTED)
+        self.controller = controller
+        self.setup()
+
+    def __init_subclass__(cls, **kwargs: t.Any) -> None:
+        super().__init_subclass__(**kwargs)
+        cls.METHODS_AVAILABLE.add(cls)
+
+    @abstractmethod
+    def setup(self) -> None:
+        raise NotImplementedError
+
+    @property
+    def handshake_hash(self) -> bytes:
+        return self.controller.channel.handshake_hash
+
+    def _abort_if_not_equal(self, expected: t.Any, actual: t.Any) -> None:
+        if actual != expected:
+            self.controller.abort()
+            raise ProtocolError("Code or commitment mismatch")
+
+    def _select_method(
+        self,
+        *,
+        expect: type[MT] = messages.ThpPairingPreparationsFinished,
+    ) -> MT:
+        if not any(
+            self.PAIRING_METHOD == m.PAIRING_METHOD for m in self.controller.methods
+        ):
+            raise ValueError(
+                f"Pairing method {self.PAIRING_METHOD.name} not supported by the device."
+            )
+
+        return self.controller._call(
+            messages.ThpSelectMethod(selected_pairing_method=self.PAIRING_METHOD),
+            expect=expect,
+        )
+
+
+class SkipPairing(PairingMethod):
+    PAIRING_METHOD = messages.ThpPairingMethod.SkipPairing
+
+    def setup(self) -> None:
+        self._select_method(expect=messages.ThpEndResponse)
+        self.controller.set_paired()
+        self.controller.finish(_no_call=True)
+
+
+class CodeEntry(PairingMethod):
+    PAIRING_METHOD = messages.ThpPairingMethod.CodeEntry
+
+    code_entry_state: CodeEntryState | None = None
+
+    def setup(self) -> None:
+        commitment_msg = self._select_method(expect=messages.ThpCodeEntryCommitment)
+        # create a challenge
+        challenge = secrets.token_bytes(16)
+        cpace_trezor_msg = self.controller._call(
+            messages.ThpCodeEntryChallenge(challenge=challenge),
+            expect=messages.ThpCodeEntryCpaceTrezor,
+        )
+        self.code_entry_state = CodeEntryState(
+            challenge=challenge,
+            commitment=commitment_msg.commitment,
+            cpace_trezor_public_key=cpace_trezor_msg.cpace_trezor_public_key,
+        )
+
+    def _perform_cpace(self, code: str) -> messages.ThpCodeEntryCpaceHostTag:
+        # perform the CPace protocol
+        assert self.code_entry_state is not None
+        cpace_result = cpace(
+            prs=code.encode("ascii"),
+            ci=self.handshake_hash,
+            b_pubkey=self.code_entry_state.cpace_trezor_public_key,
+        )
+        tag = sha256(cpace_result.shared_secret).digest()
+        return messages.ThpCodeEntryCpaceHostTag(
+            cpace_host_public_key=cpace_result.a_pubkey,
+            tag=tag,
+        )
+
+    def send_code(self, code: str) -> None:
+        assert self.code_entry_state is not None
+        if len(code) != 6 or not code.isdigit():
+            raise ValueError("Code must be a 6-digit number")
+
+        msg = self._perform_cpace(code)
+        secret_msg = self.controller._call(msg, expect=messages.ThpCodeEntrySecret)
+
+        # check the commitment
+        computed_commitment = sha256(secret_msg.secret).digest()
+        self._abort_if_not_equal(self.code_entry_state.commitment, computed_commitment)
+
+        # check the code
+        sha_ctx = sha256(messages.ThpPairingMethod.CodeEntry.to_bytes(1, "big"))
+        sha_ctx.update(self.handshake_hash)
+        sha_ctx.update(secret_msg.secret)
+        sha_ctx.update(self.code_entry_state.challenge)
+        code_hash = sha_ctx.digest()
+        computed_code = int.from_bytes(code_hash, "big") % 1_000_000
+        self._abort_if_not_equal(code, f"{computed_code:06}")
+
+        self.controller.set_paired()
+
+
+class QrCode(PairingMethod):
+    PAIRING_METHOD = messages.ThpPairingMethod.QrCode
+
+    def setup(self) -> None:
+        self._select_method()
+
+    def send_qr_code(self, code: bytes) -> None:
+        tag = sha256(self.handshake_hash + code).digest()
+        secret_msg = self.controller._call(
+            messages.ThpQrCodeTag(tag=tag),
+            expect=messages.ThpQrCodeSecret,
+        )
+
+        sha_ctx = sha256()
+        sha_ctx.update(messages.ThpPairingMethod.QrCode.to_bytes(1, "big"))
+        sha_ctx.update(self.handshake_hash)
+        sha_ctx.update(secret_msg.secret)
+        computed_code = sha_ctx.digest()[:16]
+        self._abort_if_not_equal(code, computed_code)
+
+        self.controller.set_paired()
+
+
+class Nfc(PairingMethod):
+    PAIRING_METHOD = messages.ThpPairingMethod.NFC
+
+    nfc_host_secret: bytes
+
+    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.nfc_host_secret = secrets.token_bytes(16)
+
+    def setup(self) -> None:
+        self._select_method()
+
+    def send_nfc_tag(self, tag_trezor: bytes) -> None:
+        sha_ctx = sha256(messages.ThpPairingMethod.NFC.to_bytes(1, "big"))
+        sha_ctx.update(self.handshake_hash)
+        sha_ctx.update(tag_trezor)
+        tag_host = sha_ctx.digest()
+
+        tag_trezor_msg = self.controller._call(
+            messages.ThpNfcTagHost(tag=tag_host),
+            expect=messages.ThpNfcTagTrezor,
+        )
+
+        sha_ctx = sha256(messages.ThpPairingMethod.NFC.to_bytes(1, "big"))
+        sha_ctx.update(self.handshake_hash)
+        sha_ctx.update(self.nfc_host_secret)
+        computed_tag = sha_ctx.digest()
+        self._abort_if_not_equal(tag_trezor_msg.tag, computed_tag)
+
+        self.controller.set_paired()
+
+
+def default_pairing_flow(
+    pairing: PairingController,
+    *,
+    code_entry_callback: t.Callable[[], str] | None = None,
+    request_credential: bool = True,
+) -> Credential | None:
+    if pairing.is_paired():
+        return
+
+    if SkipPairing in pairing.methods:
+        pairing.skip()
+        return
+
+    if CodeEntry not in pairing.methods:
+        raise NotImplementedError(
+            "CodeEntry pairing method not supported by the device."
+        )
+
+    if code_entry_callback is None:
+        raise TrezorException(
+            "code_entry_callback is required when the device is not paired"
+        )
+
+    method = CodeEntry(pairing)
+    code = code_entry_callback()
+    method.send_code(code)
+
+    assert pairing.state is ControllerLifecycle.PAIRING_COMPLETED
+
+    if request_credential:
+        credential = pairing.request_credential()
+    else:
+        credential = None
+
+    pairing.finish()
+    return credential
diff --git a/python/src/trezorlib/thp/protocol_v2.py b/python/src/trezorlib/thp/protocol_v2.py
deleted file mode 100644
index 0ab6e2ea..00000000
--- a/python/src/trezorlib/thp/protocol_v2.py
+++ /dev/null
@@ -1,427 +0,0 @@
-# This file is part of the Trezor project.
-#
-# Copyright (C) SatoshiLabs and contributors
-#
-# This library is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Lesser General Public License version 3
-# as published by the Free Software Foundation.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Lesser General Public License for more details.
-#
-# You should have received a copy of the License along with this library.
-# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
-
-from __future__ import annotations
-
-import logging
-import os
-import typing as t
-from binascii import hexlify
-
-from noise.connection import Keypair, NoiseConnection
-
-from ... import exceptions, messages, protobuf
-from ...mapping import ProtobufMapping
-from .. import Transport
-from ..thp import checksum, thp_io
-from ..thp.checksum import CHECKSUM_LENGTH
-from ..thp.message_header import MessageHeader
-from . import control_byte
-from .channel import Channel
-
-LOG = logging.getLogger(__name__)
-
-DEFAULT_SESSION_ID: int = 0
-
-MAX_RETRANSMISSION_COUNT = 50
-
-TREZOR_STATE_UNPAIRED = b"\x00"
-TREZOR_STATE_PAIRED = b"\x01"
-TREZOR_STATE_PAIRED_AUTOCONNECT = b"\x02"
-TREZOR_STATES = [
-    TREZOR_STATE_UNPAIRED,
-    TREZOR_STATE_PAIRED,
-    TREZOR_STATE_PAIRED_AUTOCONNECT,
-]
-
-if t.TYPE_CHECKING:
-    pass
-MT = t.TypeVar("MT", bound=protobuf.MessageType)
-
-
-class ProtocolV2Channel(Channel):
-    channel_id: int
-    sync_bit_send: int
-    sync_bit_receive: int
-    handshake_hash: bytes
-    device_properties: bytes
-
-    _features: messages.Features | None = None
-    _is_paired: bool = False
-
-    def __init__(
-        self,
-        transport: Transport,
-        mapping: ProtobufMapping,
-        credential: bytes | None = None,
-        prepare_channel_without_pairing: bool = True,
-    ) -> None:
-        super().__init__(transport, mapping)
-        self._reset_sync_bits()
-        if prepare_channel_without_pairing:
-            # allow skipping unrelated response packets (e.g. in case of retransmissions)
-            self._do_channel_allocation(retries=MAX_RETRANSMISSION_COUNT)
-            LOG.debug("THP channel allocated: %04x", self.channel_id)
-            self._do_handshake(credential=credential)
-            LOG.debug("THP handshake done: is_paired=%s", self._is_paired)
-
-    def get_channel(self) -> ProtocolV2Channel:
-        if not self._is_paired:
-            raise RuntimeError("Channel is not paired")
-        return self
-
-    def read(self, session_id: int, timeout: float | None = None) -> t.Any:
-        sid, msg_type, msg_data = self.read_and_decrypt(timeout)
-        if sid != session_id:
-            raise Exception(
-                f"Received messsage on a different session (expected/received): ({session_id}/{sid}) "
-            )
-        return self.mapping.decode(msg_type, msg_data)
-
-    def write(self, session_id: int, msg: t.Any) -> None:
-        msg_type, msg_data = self.mapping.encode(msg)
-        self._encrypt_and_write(session_id, msg_type, msg_data)
-
-    def get_features(self) -> messages.Features:
-        if not self._is_paired:
-            raise RuntimeError("Channel is not paired")
-        if self._features is None:
-            self.update_features()
-        assert self._features is not None
-        return self._features
-
-    def update_features(self, timeout: float | None = None) -> None:
-        message = messages.GetFeatures()
-        message_type, message_data = self.mapping.encode(message)
-        self.session_id: int = DEFAULT_SESSION_ID
-        self._encrypt_and_write(DEFAULT_SESSION_ID, message_type, message_data)
-        header, _payload = self._read_until_valid_crc_check()
-        if not header.is_ack():
-            raise exceptions.TrezorException("ACK expected")
-        _, msg_type, msg_data = self.read_and_decrypt(timeout)
-        features = self.mapping.decode(msg_type, msg_data)
-        if not isinstance(features, messages.Features):
-            raise exceptions.TrezorException("Unexpected response to GetFeatures")
-        self._features = features
-
-    def _send_message(
-        self,
-        message: protobuf.MessageType,
-        session_id: int = DEFAULT_SESSION_ID,
-    ) -> None:
-        message_type, message_data = self.mapping.encode(message)
-        self._encrypt_and_write(session_id, message_type, message_data)
-        self._read_ack()
-
-    def _read_message(self, message_type: type[MT], timeout: float | None = None) -> MT:
-        _, msg_type, msg_data = self.read_and_decrypt(timeout)
-        msg = self.mapping.decode(msg_type, msg_data)
-        assert isinstance(msg, message_type)
-        return msg
-
-    def _reset_sync_bits(self) -> None:
-        self.sync_bit_send = 0
-        self.sync_bit_receive = 0
-
-    def sync_responses(
-        self, retries: int = MAX_RETRANSMISSION_COUNT, timeout: float = 10.0
-    ) -> None:
-        """Make sure the event loop is running and ready."""
-        nonce = os.urandom(8)
-        thp_io.write_payload_to_wire_and_add_checksum(
-            self.transport,
-            MessageHeader.get_ping_header(len(nonce) + CHECKSUM_LENGTH),
-            nonce,
-        )
-        for _ in range(1 + retries):
-            header, payload = self._read_until_valid_crc_check(timeout=timeout)
-            if self._is_valid_pong(header, payload, nonce):
-                break
-        else:
-            raise RuntimeError("Invalid ping response")
-
-    def _do_channel_allocation(self, retries: int = 0) -> None:
-        channel_allocation_nonce = os.urandom(8)
-        self._send_channel_allocation_request(channel_allocation_nonce)
-        cid, dp = self._read_channel_allocation_response(
-            channel_allocation_nonce, retries=retries
-        )
-        self.channel_id = cid
-        self.device_properties = dp
-
-    def _send_channel_allocation_request(self, nonce: bytes) -> None:
-        thp_io.write_payload_to_wire_and_add_checksum(
-            self.transport,
-            MessageHeader.get_channel_allocation_request_header(
-                len(nonce) + CHECKSUM_LENGTH
-            ),
-            nonce,
-        )
-
-    def _read_channel_allocation_response(
-        self, expected_nonce: bytes, retries: int = 0
-    ) -> tuple[int, bytes]:
-        for _ in range(1 + retries):
-            header, payload = self._read_until_valid_crc_check()
-            if self._is_valid_channel_allocation_response(
-                header, payload, expected_nonce
-            ):
-                break
-        else:
-            raise Exception("Invalid channel allocation response.")
-
-        channel_id = int.from_bytes(payload[8:10], "big")
-        device_properties = payload[10:]
-        return (channel_id, device_properties)
-
-    def _init_noise(
-        self,
-        randomness_static: bytes | None = None,
-        randomness_ephemeral: bytes | None = None,
-    ) -> None:
-        randomness_static = randomness_static or os.urandom(32)
-        self._noise = NoiseConnection.from_name(b"Noise_XX_25519_AESGCM_SHA256")
-        self._noise.set_as_initiator()
-        self._noise.set_keypair_from_private_bytes(Keypair.STATIC, randomness_static)
-        if randomness_ephemeral is not None:
-            self._noise.set_keypair_from_private_bytes(
-                Keypair.EPHEMERAL, randomness_ephemeral
-            )
-        prologue = bytes(self.device_properties)
-        self._noise.set_prologue(prologue)
-        self._noise.start_handshake()
-
-    def _do_handshake(
-        self,
-        credential: bytes | None = None,
-        host_static_randomness: bytes | None = None,
-        host_ephemeral_randomness: bytes | None = None,
-    ) -> None:
-
-        randomness_static = host_static_randomness or os.urandom(32)
-        if host_ephemeral_randomness is not None:
-            self._init_noise(randomness_static, host_ephemeral_randomness)
-        else:
-            self._init_noise(randomness_static)
-        self._send_handshake_init_request()
-        self._read_ack()
-        self._read_handshake_init_response()
-        self._send_handshake_completion_request(
-            credential,
-        )
-        self._read_ack()
-        return self._read_handshake_completion_response()
-
-    def _send_handshake_init_request(self, try_to_unlock: bool = True) -> None:
-        payload = self._noise.write_message(bytes([try_to_unlock]))
-        ha_init_req_header = MessageHeader(
-            0, self.channel_id, len(payload) + CHECKSUM_LENGTH
-        )
-
-        thp_io.write_payload_to_wire_and_add_checksum(
-            self.transport, ha_init_req_header, payload
-        )
-
-    def _read_handshake_init_response(self) -> bytes:
-        header, payload = self._read_until_valid_crc_check()
-
-        if not header.is_handshake_init_response():
-            LOG.error("Received message is not a valid handshake init response message")
-
-        self._send_ack_bit(bit=0)
-        self._noise.read_message(payload)
-        return payload
-
-    def _send_handshake_completion_request(
-        self,
-        credential: bytes | None = None,
-    ) -> None:
-        # TODO implement key recognition
-        # print(
-        #     "TREZOR's static pubkey:\n",
-        #     self.noise.noise_protocol.handshake_state.rs.public.public_bytes_raw(),
-        # )
-
-        msg_data = self.mapping.encode_without_wire_type(
-            messages.ThpHandshakeCompletionReqNoisePayload(
-                host_pairing_credential=credential,
-            )
-        )
-        message2 = self._noise.write_message(payload=msg_data)
-
-        ha_completion_req_header = MessageHeader(
-            0x12,
-            self.channel_id,
-            len(message2) + CHECKSUM_LENGTH,
-        )
-        thp_io.write_payload_to_wire_and_add_checksum(
-            self.transport,
-            ha_completion_req_header,
-            message2,  # encrypted_host_static_pubkey + encrypted_payload,
-        )
-        self.handshake_hash = self._noise.get_handshake_hash()
-
-    def _read_handshake_completion_response(self) -> None:
-        # Read handshake completion response
-        header, data = self._read_until_valid_crc_check()
-        if not header.is_handshake_comp_response():
-            LOG.error("Received message is not a valid handshake completion response")
-        trezor_state = self._noise.decrypt(bytes(data))
-        assert trezor_state in TREZOR_STATES
-        self._send_ack_bit(bit=1)
-        self._is_paired = trezor_state != TREZOR_STATE_UNPAIRED
-
-    def _read_ack(self) -> None:
-        header, payload = self._read_until_valid_crc_check()
-        if not header.is_ack() or len(payload) > 0:
-            LOG.error("Received message is not a valid ACK")
-
-    def _send_ack_bit(self, bit: int) -> None:
-        if bit not in (0, 1):
-            raise ValueError("Invalid ACK bit")
-        LOG.debug(f"sending ack {bit}")
-        ctrl_byte = 0x20 if bit == 0 else 0x28
-        header = MessageHeader(ctrl_byte, self.channel_id, 4)
-        thp_io.write_payload_to_wire_and_add_checksum(self.transport, header, b"")
-
-    def _encrypt_and_write(
-        self,
-        session_id: int,
-        message_type: int,
-        message_data: bytes,
-        ctrl_byte: int | None = None,
-    ) -> None:
-
-        if ctrl_byte is None:
-            ctrl_byte = control_byte.add_seq_bit_to_ctrl_byte(0x04, self.sync_bit_send)
-            self.sync_bit_send = 1 - self.sync_bit_send
-
-        sid = session_id.to_bytes(1, "big")
-        msg_type = message_type.to_bytes(2, "big")
-        data = sid + msg_type + message_data
-
-        encrypted_message = self._noise.encrypt(data)
-
-        header = MessageHeader(
-            ctrl_byte, self.channel_id, len(encrypted_message) + CHECKSUM_LENGTH
-        )
-
-        thp_io.write_payload_to_wire_and_add_checksum(
-            self.transport, header, encrypted_message
-        )
-
-    def read_and_decrypt(
-        self, timeout: float | None = None
-    ) -> t.Tuple[int, int, bytes]:
-        while True:
-            header, raw_payload = self._read_until_valid_crc_check(timeout)
-            if header.cid != self.channel_id:
-                # Received message from different channel - discard
-                continue
-            if control_byte.is_ack(header.ctrl_byte):
-                continue
-            if not header.is_encrypted_transport():
-                LOG.error(
-                    "Trying to decrypt not encrypted message! ("
-                    + hexlify(header.to_bytes_init() + raw_payload).decode()
-                    + ")"
-                )
-
-            seq_bit = control_byte.get_seq_bit(header.ctrl_byte)
-            assert seq_bit is not None
-            LOG.debug(
-                "--> Get sequence bit %d %s %s",
-                seq_bit,
-                "from control byte",
-                hexlify(header.ctrl_byte.to_bytes(1, "big")).decode(),
-            )
-            self._send_ack_bit(bit=seq_bit)
-
-            message = self._noise.decrypt(bytes(raw_payload))
-            session_id = message[0]
-            message_type = message[1:3]
-            message_data = message[3:]
-            return (
-                session_id,
-                int.from_bytes(message_type, "big"),
-                message_data,
-            )
-
-    def _read_until_valid_crc_check(
-        self, timeout: float | None = None
-    ) -> t.Tuple[MessageHeader, bytes]:
-        if timeout is None:
-            timeout = self._DEFAULT_READ_TIMEOUT
-
-        while True:
-            header, payload, chksum = thp_io.read(self.transport, timeout)
-            if not checksum.is_valid(chksum, header.to_bytes_init() + payload):
-                LOG.error(
-                    "Received a message with an invalid checksum:"
-                    + hexlify(header.to_bytes_init() + payload + chksum).decode()
-                )
-                continue
-
-            seq_bit = control_byte.get_seq_bit(header.ctrl_byte)
-            if seq_bit is not None:
-                if