Merge remote-tracking branch 'refs/remotes/origin/master'

ChaCha20
chore(deps): update chacha20 from 0.10.1 to 0.10.2
2026-08-27 22:43:40 +02:00 · 2026-08-27 22:42:32 +02:00 · 2026-08-27 22:33:20 +02:00 · 2026-08-27 22:16:16 +02:00 · 2026-08-27 21:55:36 +02:00 · 2026-08-27 21:10:37 +02:00
32 changed files with 1317 additions and 472 deletions
--- a/.envrc
+++ b/.envrc
@ -0,0 +1 @@
+use flake
--- a/.forgejo/workflows/ci.yml
+++ b/.forgejo/workflows/ci.yml
@ -29,8 +29,6 @@ jobs:
            cargo machete

            pnpm install --frozen-lockfile
-            pnpm add --save-dev --save-exact --workspace-root jscpd-linux-x64-gnu@5.0.14
-            pnpm run dup

            RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm
            RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack build wasm --target web
--- a/.gitignore
+++ b/.gitignore
@ -6,3 +6,4 @@ dist/
 *.tgz
 wasm/pkg/
 web_client/
+.direnv
--- a/Cargo.lock
+++ b/Cargo.lock
@ -256,9 +256,9 @@ dependencies = [

 [[package]]
 name = "chacha20"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
 dependencies = [
 "cfg-if",
 "cpufeatures 0.3.0",
@ -1326,9 +1326,6 @@ dependencies = [
 "mtp-transport",
 "mtp-type-map",
 "mtp-webserver",
- "rand",
- "rcgen",
- "tokio",
 ]

 [[package]]
@ -1363,7 +1360,6 @@ name = "mtp-common"
 version = "0.3.0"
 dependencies = [
 "quinn",
- "rustls",
 "thiserror 2.0.20",
 "wtransport",
 ]
@ -1804,7 +1800,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "chacha20 0.10.1",
+ "chacha20 0.10.2",
 "getrandom 0.4.3",
 "rand_core 0.10.1",
 ]
--- a/Cargo.toml
+++ b/Cargo.toml
@ -113,10 +113,5 @@ tls = ["crypto", "mtp-crypto?/tls"]
 # Requires MTP_INSECURE_TLS=1 at runtime.
 insecure-tls = ["dep:mtp-transport", "mtp-transport?/insecure-tls"]

-[dev-dependencies]
-tokio = { version = "1", features = ["full"] }
-rcgen = "0.14"
-rand = "0.10.1"
-
 [package.metadata.cargo-machete]
 ignored = ["mtp-transport"]
--- a/client/src/pipe.rs
+++ b/client/src/pipe.rs
@ -78,9 +78,41 @@ pub struct PipeRequest {
    pub(crate) pipe_id: u32,
    pub(crate) description: String,
    pub(crate) sender: Sender,
+    pub(crate) receiver: Receiver,
    pub(crate) dispatcher: Arc<PipeDispatcher>,
 }

+#[cfg(feature = "pipes")]
+struct ExpectedPipeGuard {
+    receiver: Receiver,
+    pipe_id: u32,
+    armed: bool,
+}
+
+#[cfg(feature = "pipes")]
+impl ExpectedPipeGuard {
+    fn new(receiver: Receiver, pipe_id: u32) -> Self {
+        Self {
+            receiver,
+            pipe_id,
+            armed: true,
+        }
+    }
+
+    fn disarm(&mut self) {
+        self.armed = false;
+    }
+}
+
+#[cfg(feature = "pipes")]
+impl Drop for ExpectedPipeGuard {
+    fn drop(&mut self) {
+        if self.armed {
+            self.receiver.cancel_expected_pipe(self.pipe_id);
+        }
+    }
+}
+
 #[cfg(feature = "pipes")]
 impl PipeRequest {
    pub fn id(&self) -> u32 {
@ -92,6 +124,10 @@ impl PipeRequest {
    }

    pub async fn accept(self) -> Result<mtp_transport::PipeReader, PipeError> {
+        self.receiver
+            .expect_pipe(self.pipe_id)
+            .map_err(PipeError::from)?;
+        let mut expected_pipe = ExpectedPipeGuard::new(self.receiver.clone(), self.pipe_id);
        let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel();
        {
            let mut pending = self.dispatcher.pending_pipes.lock().await;
@ -115,7 +151,10 @@ impl PipeRequest {

        let timeout = self.dispatcher.policy.read_timeout;
        match tokio::time::timeout(timeout, pipe_rx).await {
-            Ok(Ok(reader)) => Ok(reader),
+            Ok(Ok(reader)) => {
+                expected_pipe.disarm();
+                Ok(reader)
+            }
            Ok(Err(_)) => {
                self.dispatcher
                    .pending_pipes
@ -413,6 +452,7 @@ pub(crate) async fn run_dispatcher(
                        pipe_id,
                        description,
                        sender: sender.clone(),
+                        receiver: receiver.clone(),
                        dispatcher: dispatcher.clone(),
                    };
                    let _ = pipe_req_tx.send(req).await;
--- a/codec/Cargo.lock
+++ b/codec/Cargo.lock
@ -187,9 +187,9 @@ dependencies = [

 [[package]]
 name = "chacha20"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
 dependencies = [
 "cfg-if",
 "cpufeatures 0.3.0",
@ -1151,7 +1151,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "chacha20 0.10.1",
+ "chacha20 0.10.2",
 "getrandom 0.4.3",
 "rand_core 0.10.1",
 ]
--- a/common/Cargo.lock
+++ b/common/Cargo.lock
@ -139,9 +139,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"

 [[package]]
 name = "chacha20"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
 dependencies = [
 "cfg-if",
 "cpufeatures",
--- a/common/Cargo.toml
+++ b/common/Cargo.toml
@ -15,7 +15,6 @@ wtransport = { version = "0.7.1", default-features = false, features = [
    "quinn",
    "self-signed",
 ] }
-rustls = { version = "0.23.41" }
 quinn = { version = "0.11.11", default-features = false, features = [
    "rustls-aws-lc-rs",
    "rustls",
--- a/common/src/lib.rs
+++ b/common/src/lib.rs
@ -164,6 +164,9 @@ pub enum CommunicationError {
    #[error("Stream Error")]
    StreamError,

+    #[error("Stream failed after delivery may have started")]
+    DeliveryUnknown,
+
    #[error("Stream Error: {0}")]
    #[cfg(not(target_arch = "wasm32"))]
    StreamWriteError(#[from] wtransport::error::StreamWriteError),
@ -182,6 +185,38 @@ pub enum CommunicationError {
    Other(String),
 }

+/// How the protocol layer should handle the first frame on a receive stream.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum FirstFrameDisposition {
+    Message,
+    Pipe(u32),
+}
+
+/// Classify a first frame without tying the decision to a WebTransport backend.
+///
+/// `PipeRequest` is used both as a control message and as the header of the raw
+/// stream opened after that request is accepted. Only the protocol layer knows
+/// which raw stream IDs are currently expected.
+pub fn classify_first_frame(
+    is_pipe_request: bool,
+    pipe_id: Option<u32>,
+    pipe_is_expected: bool,
+) -> Result<FirstFrameDisposition, CommunicationError> {
+    if !is_pipe_request {
+        return Ok(FirstFrameDisposition::Message);
+    }
+
+    let pipe_id = pipe_id.filter(|id| *id != 0).ok_or_else(|| {
+        CommunicationError::Other("PipeRequest frame must contain a non-zero id".into())
+    })?;
+
+    if pipe_is_expected {
+        Ok(FirstFrameDisposition::Pipe(pipe_id))
+    } else {
+        Ok(FirstFrameDisposition::Message)
+    }
+}
+
 // ---- manual PartialEq (quinn / wtransport types don't impl PartialEq) ----

 impl PartialEq for CommunicationError {
@ -212,6 +247,7 @@ impl PartialEq for CommunicationError {
            (Self::ReadExactError(_), Self::ReadExactError(_)) => true,
            (Self::StreamClosed, Self::StreamClosed) => true,
            (Self::StreamError, Self::StreamError) => true,
+            (Self::DeliveryUnknown, Self::DeliveryUnknown) => true,
            #[cfg(not(target_arch = "wasm32"))]
            (Self::StreamWriteError(_), Self::StreamWriteError(_)) => true,
            #[cfg(not(target_arch = "wasm32"))]
--- a/crypto/Cargo.lock
+++ b/crypto/Cargo.lock
@ -181,9 +181,9 @@ dependencies = [

 [[package]]
 name = "chacha20"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
 dependencies = [
 "cfg-if",
 "cpufeatures 0.3.0",
@ -853,7 +853,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "chacha20 0.10.1",
+ "chacha20 0.10.2",
 "getrandom 0.4.3",
 "rand_core 0.10.1",
 ]
--- a/docs/WASM-CLIENT.md
+++ b/docs/WASM-CLIENT.md
@ -104,6 +104,9 @@ if (!MTPClient.isSupported()) {
 | `requestTimeoutMs` | 30 seconds | Default `request()` timeout. |
 | `pings` | `false` | Protocol pings, or an object with `intervalMs`. |
 | `logger` | No-op | Receives SDK state and error events. |
+| `schemas` | None | Client-wide request and response schema registry. |
+| `throwProtocolErrors` | `false` | Reject requests whose correlated response is an `Error*` frame. |
+| `onValidationError` | No-op | Receives subscription validation failures. |
 | `sessionStorage` | In-memory | E2EE session state storage. |
 | `encryptedSecretProvider` | In-memory | Independent caller-managed encrypted secret storage. |
 | `defaultSignatureVerificationPolicy` | `"ed25519"` | Receiver policy for protected signatures. |
@ -471,6 +474,60 @@ const unsubscribe = client.subscribe("SomeType", (message) => {
 unsubscribe();
 ```

+### Zod request and response schemas
+
+Applications can provide their request and response schemas once when creating
+the client. MTP uses `parseAsync`, so synchronous schemas, async refinements,
+defaults, coercions, and transforms all work. MTP has no runtime dependency on
+Zod; the application supplies its preferred Zod version.
+
+```typescript
+import { z } from "zod";
+import { MTPClient, MTPValidationError } from "mtp";
+
+const schemas = {
+  GetUser: {
+    request: z.object({ UserId: z.number().int().positive() }),
+    response: z.object({
+      UserId: z.number().int().positive(),
+      Display: z.string(),
+    }),
+  },
+};
+
+const client = await MTPClient.create({
+  url,
+  schemas,
+  throwProtocolErrors: true,
+  onValidationError(error) {
+    console.error(error.messageType, error.cause);
+  },
+});
+
+const response = await client.request("GetUser", { UserId: 42 });
+console.log(response.data.Display);
+```
+
+Request schemas run before frame encoding and transmission. Their transformed
+output is sent. Response schemas run after request correlation, and their
+transformed output replaces `frame.data`; `frame.raw`, when present, remains the
+original wire frame. Invalid requests and responses reject with
+`MTPValidationError`. Invalid subscription messages do not reach the handler
+and are reported through `onValidationError`.
+
+`throwProtocolErrors: true` converts correlated `Error*` frames into
+`MTPProtocolError`. It defaults to `false` for compatibility.
+
+`MTPProxyConnection` applies the same schema registry to another TypeScript
+request/subscription transport, such as a Tauri command and event proxy:
+
+```typescript
+const connection = new MTPProxyConnection(adapter, {
+  schemas,
+  throwProtocolErrors: true,
+});
+```
+
 Protocol ping behavior is defined in [Protocol Reference](PROTOCOL-REFERENCE.md#protocol-keepalive). The SDK configuration is:

 ```typescript
--- a/example/Cargo.lock
+++ b/example/Cargo.lock
@ -225,9 +225,9 @@ dependencies = [

 [[package]]
 name = "chacha20"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
 dependencies = [
 "cfg-if",
 "cpufeatures 0.3.0",
@ -1305,7 +1305,6 @@ name = "mtp-common"
 version = "0.3.0"
 dependencies = [
 "quinn",
- "rustls",
 "thiserror 2.0.20",
 "wtransport",
 ]
@ -1702,7 +1701,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "chacha20 0.10.1",
+ "chacha20 0.10.2",
 "getrandom 0.4.3",
 "rand_core 0.10.1",
 ]
--- a/example/server/src/main.rs
+++ b/example/server/src/main.rs
@ -38,6 +38,7 @@ async fn handle_pipe_loopback(
    conn: &mtp::webserver::WebMTPConnection,
    request: mtp::host::PipeRequest<
        mtp::webserver::WebMtpSender,
+        mtp::webserver::WebMtpReceiver,
        mtp::webserver::H3TransportReceiver,
    >,
 ) -> Result<u64, Box<dyn std::error::Error>> {
--- a/flake.nix
+++ b/flake.nix
@ -5,26 +5,30 @@
    rust-overlay.url = "github:oxalica/rust-overlay";
  };

-  outputs = {
+  outputs =
+    {
      self,
      nixpkgs,
      rust-overlay,
-  }: let
+    }:
+    let
      systems = [
        "aarch64-darwin"
        "aarch64-linux"
        "x86_64-darwin"
        "x86_64-linux"
      ];
-    eachSystem = f:
-      nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate {} (
-        map (system: nixpkgs.lib.mapAttrs (_: value: {${system} = value;}) (f system)) systems
+      eachSystem =
+        f:
+        nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate { } (
+          map (system: nixpkgs.lib.mapAttrs (_: value: { ${system} = value; }) (f system)) systems
        );
    in
    eachSystem (
-      system: let
-        overlays = [rust-overlay.overlays.default];
-        pkgs = import nixpkgs {inherit system overlays;};
+      system:
+      let
+        overlays = [ rust-overlay.overlays.default ];
+        pkgs = import nixpkgs { inherit system overlays; };

        rustToolchain = pkgs.rust-bin.stable.latest.default.override {
          extensions = [
@ -32,12 +36,12 @@
            "clippy"
            "rustfmt"
          ];
-          targets = ["wasm32-unknown-unknown"];
+          targets = [ "wasm32-unknown-unknown" ];
        };

        clippyCheck = pkgs.writeShellApplication {
          name = "mtp-clippy";
-          runtimeInputs = [rustToolchain];
+          runtimeInputs = [ rustToolchain ];
          text = ''
            export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}"
            cargo clippy --workspace --exclude mtp-wasm --all-targets --all-features -- -D warnings -W unreachable-pub
@ -46,7 +50,7 @@

        macheteCheck = pkgs.writeShellApplication {
          name = "mtp-machete";
-          runtimeInputs = [pkgs.cargo-machete];
+          runtimeInputs = [ pkgs.cargo-machete ];
          text = ''
            cargo machete "$@"
          '';
@ -54,7 +58,15 @@

        buildAll = pkgs.writeShellApplication {
          name = "mtp-build-all";
-          runtimeInputs = [rustToolchain pkgs.cargo-deny pkgs.wasm-pack pkgs.pnpm pkgs.coreutils clippyCheck macheteCheck];
+          runtimeInputs = [
+            rustToolchain
+            pkgs.cargo-deny
+            pkgs.wasm-pack
+            pkgs.pnpm
+            pkgs.coreutils
+            clippyCheck
+            macheteCheck
+          ];
          text = ''
            export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}"

@ -66,7 +78,6 @@
              cargo check --manifest-path example/Cargo.toml --workspace --all-targets --all-features
            mtp-clippy
            mtp-machete
-            pnpm run dup
            pnpm run build
            RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm
            pnpm run test:e2e
@ -79,13 +90,17 @@

        healthCheck = pkgs.writeShellApplication {
          name = "mtp-health";
-          runtimeInputs = [clippyCheck macheteCheck];
+          runtimeInputs = [
+            clippyCheck
+            macheteCheck
+          ];
          text = ''
            mtp-clippy
            mtp-machete
          '';
        };
-      in {
+      in
+      {
        devShells = {
          default = pkgs.mkShell {
            name = "mtp-dev";
--- a/host/src/config.rs
+++ b/host/src/config.rs
@ -251,6 +251,8 @@ pub struct HostConfig {
    #[cfg(feature = "crypto")]
    pub authentication_policy: AuthenticationPolicy,
    #[cfg(feature = "crypto")]
+    authentication_policy_explicit: bool,
+    #[cfg(feature = "crypto")]
    pub auth_timeout: Duration,
    #[cfg(feature = "crypto")]
    pub require_pq: bool,
@ -288,6 +290,8 @@ impl HostConfig {
            #[cfg(feature = "crypto")]
            authentication_policy: AuthenticationPolicy::Unauthenticated,
            #[cfg(feature = "crypto")]
+            authentication_policy_explicit: false,
+            #[cfg(feature = "crypto")]
            auth_timeout: Duration::from_secs(30),
            #[cfg(feature = "crypto")]
            require_pq: true,
@ -341,7 +345,9 @@ impl HostConfig {
        get_existing_client: GetExistingClient,
        complete_register: CompleteRegister,
    ) -> Self {
+        if !self.authentication_policy_explicit {
            self.authentication_policy = AuthenticationPolicy::ForceAuthentication;
+        }
        self.host_keyring = host_keyring;
        self.get_existing_client = Box::new(get_existing_client);
        self.complete_register = Box::new(complete_register);
@ -351,6 +357,7 @@ impl HostConfig {
    #[cfg(feature = "crypto")]
    pub fn with_authentication_policy(mut self, policy: AuthenticationPolicy) -> Self {
        self.authentication_policy = policy;
+        self.authentication_policy_explicit = true;
        self
    }

@ -441,4 +448,78 @@ mod tests {
                .expect("repeated registration decision")
        );
    }
+
+    fn test_keyring() -> mtp_crypto::Keyring {
+        mtp_crypto::Keyring::new(
+            mtp_crypto::KemPublicKey::new(Vec::new()),
+            mtp_crypto::KemPrivateKey::new(Vec::new()),
+            mtp_crypto::SignaturePqPublicKey::new(Vec::new()),
+            mtp_crypto::SignaturePqPrivateKey::new(Vec::new()),
+            mtp_crypto::SignaturePublicKey::new(Vec::new()),
+            mtp_crypto::SignaturePrivateKey::new(Vec::new()),
+        )
+    }
+
+    fn test_get_existing_client() -> GetExistingClient {
+        Box::new(|_, _| Box::pin(async { None }))
+    }
+
+    fn test_complete_register() -> CompleteRegister {
+        Box::new(|_, _| Box::pin(async { 1 }))
+    }
+
+    fn test_config() -> HostConfig {
+        HostConfig::new(
+            IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
+            4433,
+            Vec::new(),
+            Vec::new(),
+        )
+    }
+
+    #[test]
+    fn with_authentication_defaults_to_force_authentication() {
+        let config = test_config().with_authentication(
+            test_keyring(),
+            test_get_existing_client(),
+            test_complete_register(),
+        );
+
+        assert_eq!(
+            config.authentication_policy,
+            AuthenticationPolicy::ForceAuthentication
+        );
+    }
+
+    #[test]
+    fn explicit_authentication_policy_before_with_authentication_is_preserved() {
+        let config = test_config()
+            .with_authentication_policy(AuthenticationPolicy::AllowAuthentication)
+            .with_authentication(
+                test_keyring(),
+                test_get_existing_client(),
+                test_complete_register(),
+            );
+
+        assert_eq!(
+            config.authentication_policy,
+            AuthenticationPolicy::AllowAuthentication
+        );
+    }
+
+    #[test]
+    fn explicit_authentication_policy_after_with_authentication_is_preserved() {
+        let config = test_config()
+            .with_authentication(
+                test_keyring(),
+                test_get_existing_client(),
+                test_complete_register(),
+            )
+            .with_authentication_policy(AuthenticationPolicy::AllowAuthentication);
+
+        assert_eq!(
+            config.authentication_policy,
+            AuthenticationPolicy::AllowAuthentication
+        );
+    }
 }
--- a/host/src/connection.rs
+++ b/host/src/connection.rs
@ -73,7 +73,7 @@ pub struct MTPConnection<
    #[cfg(feature = "pipes")]
    pub(crate) app_rx: Mutex<mpsc::Receiver<Result<CommunicationValue, CommunicationError>>>,
    #[cfg(feature = "pipes")]
-    pub(crate) pipe_req_rx: Mutex<mpsc::Receiver<PipeRequest<S, P>>>,
+    pub(crate) pipe_req_rx: Mutex<mpsc::Receiver<PipeRequest<S, R, P>>>,
    #[cfg(feature = "pipes")]
    pub(crate) pipe_dispatcher: Arc<PipeDispatcher<P>>,
    #[cfg(not(feature = "pipes"))]
@ -381,7 +381,7 @@ where
        })
    }

-    pub async fn receive_pipe(&self) -> Result<PipeRequest<S, P>, CommunicationError> {
+    pub async fn receive_pipe(&self) -> Result<PipeRequest<S, R, P>, CommunicationError> {
        self.pipe_req_rx
            .lock()
            .await
--- a/host/src/pipe.rs
+++ b/host/src/pipe.rs
@ -27,6 +27,10 @@ pub trait PipeReceiver<P>: Clone + Send + Sync + 'static
 where
    P: tokio::io::AsyncRead + Send + Unpin + 'static,
 {
+    fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError>;
+
+    fn cancel_expected_pipe(&self, pipe_id: u32);
+
    fn receive_pipe_event(
        &self,
    ) -> impl std::future::Future<Output = Result<TransportEvent<P>, CommunicationError>> + Send;
@ -52,6 +56,14 @@ impl PipeSender for mtp_transport::Sender {
 }

 impl PipeReceiver<wtransport::RecvStream> for mtp_transport::Receiver {
+    fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
+        self.expect_pipe(pipe_id)
+    }
+
+    fn cancel_expected_pipe(&self, pipe_id: u32) {
+        self.cancel_expected_pipe(pipe_id);
+    }
+
    async fn receive_pipe_event(
        &self,
    ) -> Result<TransportEvent<wtransport::RecvStream>, CommunicationError> {
@ -87,6 +99,14 @@ where
    C: mtp_transport::TransportConnection,
    C::RecvStream: tokio::io::AsyncRead + Send + Unpin + 'static,
 {
+    fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
+        self.expect_pipe(pipe_id)
+    }
+
+    fn cancel_expected_pipe(&self, pipe_id: u32) {
+        self.cancel_expected_pipe(pipe_id);
+    }
+
    async fn receive_pipe_event(
        &self,
    ) -> Result<TransportEvent<C::RecvStream>, CommunicationError> {
@ -152,16 +172,60 @@ where
    }
 }

-pub struct PipeRequest<S, P> {
+pub struct PipeRequest<S, R, P> {
    pub(crate) pipe_id: u32,
    pub(crate) description: String,
    pub(crate) sender: S,
+    pub(crate) receiver: R,
    pub(crate) dispatcher: Arc<PipeDispatcher<P>>,
 }

-impl<S, P> PipeRequest<S, P>
+struct ExpectedPipeGuard<R, P>
+where
+    R: PipeReceiver<P>,
+    P: tokio::io::AsyncRead + Send + Unpin + 'static,
+{
+    receiver: R,
+    pipe_id: u32,
+    armed: bool,
+    _stream: std::marker::PhantomData<P>,
+}
+
+impl<R, P> ExpectedPipeGuard<R, P>
+where
+    R: PipeReceiver<P>,
+    P: tokio::io::AsyncRead + Send + Unpin + 'static,
+{
+    fn new(receiver: R, pipe_id: u32) -> Self {
+        Self {
+            receiver,
+            pipe_id,
+            armed: true,
+            _stream: std::marker::PhantomData,
+        }
+    }
+
+    fn disarm(&mut self) {
+        self.armed = false;
+    }
+}
+
+impl<R, P> Drop for ExpectedPipeGuard<R, P>
+where
+    R: PipeReceiver<P>,
+    P: tokio::io::AsyncRead + Send + Unpin + 'static,
+{
+    fn drop(&mut self) {
+        if self.armed {
+            self.receiver.cancel_expected_pipe(self.pipe_id);
+        }
+    }
+}
+
+impl<S, R, P> PipeRequest<S, R, P>
 where
    S: PipeSender,
+    R: PipeReceiver<P>,
    P: tokio::io::AsyncRead + Send + Unpin + 'static,
 {
    pub fn id(&self) -> u32 {
@ -173,6 +237,10 @@ where
    }

    pub async fn accept(self) -> Result<PipeReader<P>, PipeError> {
+        self.receiver
+            .expect_pipe(self.pipe_id)
+            .map_err(PipeError::from)?;
+        let mut expected_pipe = ExpectedPipeGuard::<R, P>::new(self.receiver.clone(), self.pipe_id);
        let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel();
        self.dispatcher
            .pending_pipes
@ -196,7 +264,10 @@ where
        }

        match tokio::time::timeout(self.dispatcher.policy.read_timeout, pipe_rx).await {
-            Ok(Ok(reader)) => Ok(reader),
+            Ok(Ok(reader)) => {
+                expected_pipe.disarm();
+                Ok(reader)
+            }
            Ok(Err(_)) => {
                self.dispatcher
                    .pending_pipes
@ -361,7 +432,7 @@ pub(crate) async fn run_dispatcher<S, R, P>(
    receiver: R,
    sender: S,
    app_tx: mpsc::Sender<Result<CommunicationValue, CommunicationError>>,
-    pipe_req_tx: mpsc::Sender<PipeRequest<S, P>>,
+    pipe_req_tx: mpsc::Sender<PipeRequest<S, R, P>>,
    dispatcher: Arc<PipeDispatcher<P>>,
 ) where
    S: PipeSender,
@ -388,6 +459,7 @@ pub(crate) async fn run_dispatcher<S, R, P>(
                            .unwrap_or("")
                            .to_owned(),
                        sender: sender.clone(),
+                        receiver: receiver.clone(),
                        dispatcher: dispatcher.clone(),
                    };
                    let _ = pipe_req_tx.send(request).await;
@ -446,6 +518,7 @@ pub(crate) async fn run_dispatcher<S, R, P>(
                    pipe_id,
                    description: reader.description().to_owned(),
                    sender: sender.clone(),
+                    receiver: receiver.clone(),
                    dispatcher: dispatcher.clone(),
                };
                let _ = pipe_req_tx.send(request).await;
--- a/mtp-webserver/src/transport.rs
+++ b/mtp-webserver/src/transport.rs
@ -57,14 +57,14 @@ impl TransportSendStream for H3TransportSender {
        self.stream
            .write_all(buf)
            .await
-            .map_err(|_| CommunicationError::StreamError)?;
+            .map_err(|_| CommunicationError::DeliveryUnknown)?;
        // Control/authentication frames use a persistent stream. h3 keeps
        // those writes buffered until flushed; without this the peer can wait
        // for the challenge while the server waits for its proof.
        self.stream
            .flush()
            .await
-            .map_err(|_| CommunicationError::StreamError)
+            .map_err(|_| CommunicationError::DeliveryUnknown)
    }

    async fn finish(&mut self) -> Result<(), CommunicationError> {
@ -73,6 +73,11 @@ impl TransportSendStream for H3TransportSender {
            .await
            .map_err(|_| CommunicationError::StreamError)
    }
+
+    fn reset(&mut self, code: u32) -> Result<(), CommunicationError> {
+        h3::quic::SendStream::reset(&mut self.stream, code as u64);
+        Ok(())
+    }
 }

 #[async_trait::async_trait]
@ -140,6 +145,11 @@ impl TransportRecvStream for H3TransportReceiver {
            }
        }
    }
+
+    fn stop(mut self, code: u32) -> Result<(), CommunicationError> {
+        h3::quic::RecvStream::stop_sending(&mut self.stream, code as u64);
+        Ok(())
+    }
 }

 impl tokio::io::AsyncWrite for H3TransportSender {
--- a/package.json
+++ b/package.json
@ -61,7 +61,6 @@
    "pack": "pnpm run release:web",
    "release:web": "node create-web-release.mjs",
    "build:all": "nix run .#build-all",
-    "dup": "jscpd --pattern '**/*.{rs,ts}' --ignore 'target/**' --ignore 'wasm/pkg/**' --min-lines 8 --min-tokens 80 --threshold 4 --reporters console --no-tips .",
    "test:e2e": "tsc && node test/e2ee.mjs",
    "test:secrets": "tsc && node --test --test-isolation=none test/encrypted-secret.mjs",
    "test:wasm-init": "tsc && node --test test/wasm-init.mjs",
--- a/src/sdk/client.ts
+++ b/src/sdk/client.ts
@ -10,6 +10,15 @@ import * as bindings from "mtp/raw";
 import { unixTimeMillis, utf8Encode } from "./utils.js";
 import type * as RawBindings from "../raw/index";
 import type { MTPCommunicationType } from "../type-map/index";
+import { MTPProtocol } from "./schema.js";
+import type {
+  MTPMessageType,
+  MTPFrame,
+  MTPNoSchemas,
+  MTPRequestData,
+  MTPResponseFrame,
+  MTPSchemaRegistry,
+} from "./schema.js";
 import type { MTPSessionStorage, MTPSessionState } from "./session";
 import { MTPSessionManager } from "./session.js";
 import {
@ -249,7 +258,9 @@ export interface MTPPublicKeyBundleKeys {
  sigClPublicKey: Uint8Array;
 }

-export interface MTPClientOptions {
+export interface MTPClientOptions<
+  Registry extends MTPSchemaRegistry = MTPNoSchemas,
+> {
  url: string;
  descriptor?: string;
  hostPublicKey?: MTPKeyMaterialInput;
@ -282,6 +293,12 @@ export interface MTPClientOptions {
  securityProfile?: MTPSecurityProfile;
  /** One receive resource policy shared by frame and protected-value opening. */
  receiveLimits?: MTPReceiveLimits;
+  /** Application request and response schemas, keyed by communication type. */
+  schemas?: Registry;
+  /** Reject `request()` when the correlated response is an `Error*` frame. */
+  throwProtocolErrors?: boolean;
+  /** Receives subscription validation failures. Request failures reject normally. */
+  onValidationError?: (error: import("./schema.js").MTPValidationError) => void;
 }

 export interface MTPSecurityProfile {
@ -562,8 +579,8 @@ export interface MTPAcceptEncryptedPipeOptions {
  signaturePolicy?: MTPSignatureVerificationPolicy;
 }

-type NormalizedMTPClientOptions = Omit<
-  MTPClientOptions,
+type NormalizedMTPClientOptions<Registry extends MTPSchemaRegistry> = Omit<
+  MTPClientOptions<Registry>,
  "hostPublicKey" | "receiveLimits"
 > & {
  hostPublicKey?: Uint8Array;
@ -914,14 +931,34 @@ function validateOptions(options) {
  ) {
    throw new TypeError("requestTimeoutMs must be a positive safe integer");
  }
+  if (options.schemas != null) {
+    if (typeof options.schemas !== "object" || Array.isArray(options.schemas)) {
+      throw new TypeError("schemas must be an object");
+    }
+    for (const [type, pair] of Object.entries(options.schemas)) {
+      if (
+        !pair ||
+        typeof pair !== "object" ||
+        typeof (pair as { request?: { parseAsync?: unknown } }).request
+          ?.parseAsync !== "function" ||
+        typeof (pair as { response?: { parseAsync?: unknown } }).response
+          ?.parseAsync !== "function"
+      ) {
+        throw new TypeError(
+          `schemas.${type} must contain request and response schemas with parseAsync()`,
+        );
+      }
+    }
+  }
 }

-export class MTPClient {
+export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
  static readonly crypto = crypto;
  static readonly codec = codec;

  #credentials: InternalCredentials | null;
-  #options: NormalizedMTPClientOptions;
+  #options: NormalizedMTPClientOptions<Registry>;
+  readonly #protocol: MTPProtocol<Registry> | undefined;
  readonly #protectedReplayGuard = new InMemoryReplayGuard();
  readonly #relayReplayGuard = new InMemoryReplayGuard();
  readonly raw: MTPRaw;
@ -934,10 +971,17 @@ export class MTPClient {
  readonly encryptedSecretProvider: MTPEncryptedSecretProvider;

  private constructor(
-    options: NormalizedMTPClientOptions,
+    options: NormalizedMTPClientOptions<Registry>,
    client: RawBindings.WasmClient,
  ) {
    this.#options = options;
+    this.#protocol = options.schemas
+      ? new MTPProtocol({
+          schemas: options.schemas,
+          throwProtocolErrors: options.throwProtocolErrors,
+          onValidationError: options.onValidationError,
+        })
+      : undefined;
    this.#credentials = deserializeCredentials(options.credentials);
    this.raw = { client, bindings };
    this.encryptedSecretProvider =
@ -947,7 +991,11 @@ export class MTPClient {
    );
  }

-  static async create(options: MTPClientOptions): Promise<MTPClient> {
+  static async create<
+    const Registry extends MTPSchemaRegistry = MTPNoSchemas,
+  >(
+    options: MTPClientOptions<Registry>,
+  ): Promise<MTPClient<Registry>> {
    validateOptions(options);
    await MTPClient.init(options.wasm);

@ -968,7 +1016,7 @@ export class MTPClient {
      securityProfile: resolveSecurityProfile(options),
    };

-    let sdk: MTPClient | undefined;
+    let sdk: MTPClient<Registry> | undefined;
    const client = new WasmClient(
      (state) =>
        emit(normalizedOptions.logger, {
@ -1004,7 +1052,7 @@ export class MTPClient {
      setReceiveLimits.call(rawClient, normalizedOptions.receiveLimits);
    }

-    sdk = new MTPClient(normalizedOptions, client);
+    sdk = new MTPClient<Registry>(normalizedOptions, client);
    await sdk.#loadStoredCredentials();
    if (!sdk.#credentials) {
      sdk.#credentials = {
@ -1238,6 +1286,35 @@ export class MTPClient {
    };
  }

+  async #parseRequestData(
+    type: MTPCommunicationType,
+    data: unknown,
+  ): Promise<Record<string, unknown>> {
+    if (!this.#protocol || !this.#protocol.schemas[type]) {
+      return (data ?? {}) as Record<string, unknown>;
+    }
+    const parsed = await this.#protocol.parseRequest(
+      type as MTPMessageType<Registry>,
+      data as never,
+    );
+    return (parsed ?? {}) as Record<string, unknown>;
+  }
+
+  async #parseResponseData(
+    requestedType: MTPCommunicationType,
+    frame: ParsedFrame,
+    phase: "response" | "subscription" = "response",
+  ): Promise<MTPFrame<unknown>> {
+    if (!this.#protocol || !this.#protocol.schemas[requestedType]) {
+      return frame;
+    }
+    return await this.#protocol.parseResponse(
+      requestedType as MTPMessageType<Registry>,
+      frame,
+      phase,
+    );
+  }
+
  #buildFrame(typeOrFrame, data, options) {
    if (typeOrFrame instanceof Uint8Array) {
      if (
@ -1279,6 +1356,11 @@ export class MTPClient {
  }

  async send(message: Uint8Array): Promise<void>;
+  async send<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data?: MTPRequestData<Registry, Type>,
+    options?: MTPSendOptions,
+  ): Promise<void>;
  async send(
    type: MTPCommunicationType,
    data: Record<string, unknown>,
@ -1286,10 +1368,14 @@ export class MTPClient {
  ): Promise<void>;
  async send(
    typeOrFrame: Uint8Array | MTPCommunicationType,
-    data?: Record<string, unknown>,
+    data?: unknown,
    options?: MTPSendOptions,
  ): Promise<void> {
-    const message = this.#buildFrame(typeOrFrame, data, options);
+    const parsedData =
+      typeof typeOrFrame === "string"
+        ? await this.#parseRequestData(typeOrFrame, data)
+        : data;
+    const message = this.#buildFrame(typeOrFrame, parsedData, options);

    try {
      const frame = this.raw.bindings.parse_frame(message);
@ -1345,6 +1431,11 @@ export class MTPClient {
    data?: never,
    options?: MTPRequestOptions,
  ): Promise<ParsedFrame>;
+  async request<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data?: MTPRequestData<Registry, Type>,
+    options?: MTPRequestOptions,
+  ): Promise<MTPResponseFrame<Registry, Type>>;
  async request(
    type: MTPCommunicationType,
    data: Record<string, unknown>,
@ -1352,15 +1443,19 @@ export class MTPClient {
  ): Promise<ParsedFrame>;
  async request(
    typeOrFrame: Uint8Array | MTPCommunicationType,
-    data?: Record<string, unknown>,
+    data?: unknown,
    options: MTPRequestOptions = {},
-  ): Promise<ParsedFrame> {
+  ): Promise<ParsedFrame | MTPFrame<unknown>> {
    const timeoutMs =
      options.timeoutMs ?? this.#options.requestTimeoutMs ?? 30_000;
    if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) {
      throw new TypeError("request timeoutMs must be a positive safe integer");
    }
-    const frame = this.#buildFrame(typeOrFrame, data, options);
+    const parsedData =
+      typeof typeOrFrame === "string"
+        ? await this.#parseRequestData(typeOrFrame, data)
+        : data;
+    const frame = this.#buildFrame(typeOrFrame, parsedData, options);
    try {
      const parsed = this.raw.bindings.parse_frame(frame);
      emit(
@ -1391,16 +1486,29 @@ export class MTPClient {
    // The WASM client owns request expiry and its late-response tombstones.
    // Keeping a second Promise timer here can reject the SDK call while the
    // protocol request is still allowed to complete successfully.
-    return await this.raw.client.request(
+    const response = await this.raw.client.request(
      frame,
      options.responseType ?? null,
      timeoutMs,
    );
+    return typeof typeOrFrame === "string"
+      ? await this.#parseResponseData(typeOrFrame, response)
+      : response;
  }

+  subscribe<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    handler: (
+      message: MTPResponseFrame<Registry, Type>,
+    ) => void | Promise<void>,
+  ): Unsubscribe;
  subscribe(
    type: MTPCommunicationType,
-    handler: (message: ParsedFrame) => void,
+    handler: (message: ParsedFrame) => void | Promise<void>,
+  ): Unsubscribe;
+  subscribe(
+    type: MTPCommunicationType,
+    handler: (message: any) => void | Promise<void>,
  ): Unsubscribe {
    if (typeof type !== "string" || !type) {
      throw new TypeError("subscription type must be a non-empty string");
@ -1408,8 +1516,25 @@ export class MTPClient {
    if (typeof handler !== "function") {
      throw new TypeError("subscription handler must be a function");
    }
-    const id = this.raw.client.subscribe(type, handler);
-    return () => this.raw.client.unsubscribe(id);
+    let active = true;
+    const id = this.raw.client.subscribe(type, (message) => {
+      if (!this.#protocol || !this.#protocol.schemas[type]) {
+        void handler(message);
+        return;
+      }
+      void this.#parseResponseData(type, message, "subscription").then(
+        (parsed) => {
+          if (active) void handler(parsed);
+        },
+        (error) => {
+          this.#protocol?.reportValidationError(error);
+        },
+      );
+    });
+    return () => {
+      active = false;
+      this.raw.client.unsubscribe(id);
+    };
  }

  #handleFrame(frame) {
--- a/src/sdk/index.ts
+++ b/src/sdk/index.ts
@ -5,3 +5,4 @@
 * keeps the package's historical exports stable.
 */
 export * from "./client.js";
+export * from "./schema.js";
--- a/src/sdk/schema.ts
+++ b/src/sdk/schema.ts
@ -0,0 +1,234 @@
+import type { MTPRequestOptions, ParsedFrame, Unsubscribe } from "./client.js";
+import type { MTPCommunicationType } from "../type-map/index.js";
+
+export interface MTPSchema<Input = unknown, Output = Input> {
+  readonly _input: Input;
+  readonly _output: Output;
+  parseAsync(value: unknown): Promise<Output>;
+}
+
+export interface MTPSchemaPair<
+  Request extends MTPSchema = MTPSchema,
+  Response extends MTPSchema = MTPSchema,
+> {
+  request: Request;
+  response: Response;
+}
+
+export type MTPSchemaRegistry = Record<string, MTPSchemaPair>;
+export type MTPNoSchemas = Record<never, never>;
+
+export type MTPSchemaInput<Schema extends MTPSchema> = Schema["_input"];
+export type MTPSchemaOutput<Schema extends MTPSchema> = Schema["_output"];
+export type MTPMessageType<Registry extends MTPSchemaRegistry> =
+  keyof Registry & string;
+
+export type MTPFrame<Data = ParsedFrame["data"]> = {
+  id?: number;
+  type: string;
+  data: Data;
+  sender?: ParsedFrame["sender"];
+  receiver?: ParsedFrame["receiver"];
+  raw?: ParsedFrame["raw"];
+};
+
+export type MTPTypedFrame<Data = ParsedFrame["data"]> = MTPFrame<Data>;
+
+export type MTPResponseFrame<
+  Registry extends MTPSchemaRegistry,
+  Type extends MTPMessageType<Registry>,
+> = MTPTypedFrame<MTPSchemaOutput<Registry[Type]["response"]>>;
+
+export type MTPRequestData<
+  Registry extends MTPSchemaRegistry,
+  Type extends MTPMessageType<Registry>,
+> = MTPSchemaInput<Registry[Type]["request"]>;
+
+export type MTPRequestFunction<Registry extends MTPSchemaRegistry> = <
+  Type extends MTPMessageType<Registry>,
+>(
+  type: Type,
+  data?: MTPRequestData<Registry, Type>,
+  options?: MTPRequestOptions,
+) => Promise<MTPResponseFrame<Registry, Type>>;
+
+export type MTPSubscriptionFunction<Registry extends MTPSchemaRegistry> = <
+  Type extends MTPMessageType<Registry>,
+>(
+  type: Type,
+  handler: (message: MTPResponseFrame<Registry, Type>) => void | Promise<void>,
+) => Unsubscribe;
+
+export class MTPValidationError extends Error {
+  readonly phase: "request" | "response" | "subscription";
+  readonly messageType: string;
+  readonly frame?: MTPFrame;
+
+  constructor(
+    phase: MTPValidationError["phase"],
+    messageType: string,
+    cause: unknown,
+    frame?: MTPFrame,
+  ) {
+    super(`${phase} validation failed for ${messageType}`, { cause });
+    this.name = "MTPValidationError";
+    this.phase = phase;
+    this.messageType = messageType;
+    this.frame = frame;
+  }
+}
+
+export class MTPProtocolError extends Error {
+  readonly type: string;
+  readonly id: number | undefined;
+  readonly communicationType: string;
+  readonly requestId: number | undefined;
+  readonly errorType: string | undefined;
+  readonly frame: MTPFrame;
+
+  constructor(frame: MTPFrame) {
+    const errorType =
+      frame.data &&
+      typeof frame.data === "object" &&
+      !Array.isArray(frame.data) &&
+      typeof (frame.data as Record<string, unknown>).ErrorType === "string"
+        ? ((frame.data as Record<string, unknown>).ErrorType as string)
+        : undefined;
+    super(errorType ? `${frame.type}: ${errorType}` : frame.type);
+    this.name = "MTPProtocolError";
+    this.type = frame.type;
+    this.id = frame.id;
+    this.communicationType = frame.type;
+    this.requestId = frame.id;
+    this.errorType = errorType;
+    this.frame = frame;
+  }
+}
+
+export interface MTPProtocolOptions<Registry extends MTPSchemaRegistry> {
+  schemas: Registry;
+  throwProtocolErrors?: boolean;
+  onValidationError?: (error: MTPValidationError) => void;
+}
+
+function isErrorFrame(frame: MTPFrame): boolean {
+  return frame.type.startsWith("Error");
+}
+
+export class MTPProtocol<Registry extends MTPSchemaRegistry> {
+  readonly schemas: Registry;
+  readonly #throwProtocolErrors: boolean;
+  readonly #onValidationError:
+    | ((error: MTPValidationError) => void)
+    | undefined;
+
+  constructor(options: MTPProtocolOptions<Registry>) {
+    this.schemas = options.schemas;
+    this.#throwProtocolErrors = options.throwProtocolErrors ?? false;
+    this.#onValidationError = options.onValidationError;
+  }
+
+  async parseRequest<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data: MTPRequestData<Registry, Type> | undefined,
+  ): Promise<MTPSchemaOutput<Registry[Type]["request"]>> {
+    try {
+      return await this.schemas[type].request.parseAsync(data);
+    } catch (error) {
+      throw new MTPValidationError("request", type, error);
+    }
+  }
+
+  async parseResponse<Type extends MTPMessageType<Registry>>(
+    requestedType: Type,
+    frame: MTPFrame,
+    phase: "response" | "subscription" = "response",
+  ): Promise<MTPResponseFrame<Registry, Type>> {
+    if (isErrorFrame(frame)) {
+      if (phase === "response" && this.#throwProtocolErrors) {
+        throw new MTPProtocolError(frame);
+      }
+      return frame as MTPResponseFrame<Registry, Type>;
+    }
+
+    const schema =
+      this.schemas[frame.type]?.response ??
+      this.schemas[requestedType].response;
+    try {
+      const data = await schema.parseAsync(frame.data);
+      return { ...frame, data } as MTPResponseFrame<Registry, Type>;
+    } catch (error) {
+      throw new MTPValidationError(
+        phase,
+        frame.type || requestedType,
+        error,
+        frame,
+      );
+    }
+  }
+
+  reportValidationError(error: unknown): void {
+    if (error instanceof MTPValidationError) {
+      this.#onValidationError?.(error);
+    }
+  }
+}
+
+export interface MTPProxyAdapter {
+  request(
+    type: MTPCommunicationType,
+    data: Record<string, unknown>,
+    options?: MTPRequestOptions,
+  ): Promise<MTPFrame>;
+  subscribe(
+    type: MTPCommunicationType,
+    handler: (message: MTPFrame) => void,
+  ): Unsubscribe;
+}
+
+export class MTPProxyConnection<Registry extends MTPSchemaRegistry> {
+  readonly #adapter: MTPProxyAdapter;
+  readonly #protocol: MTPProtocol<Registry>;
+
+  constructor(adapter: MTPProxyAdapter, options: MTPProtocolOptions<Registry>) {
+    this.#adapter = adapter;
+    this.#protocol = new MTPProtocol(options);
+  }
+
+  async request<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data?: MTPRequestData<Registry, Type>,
+    options?: MTPRequestOptions,
+  ): Promise<MTPResponseFrame<Registry, Type>> {
+    const parsed = await this.#protocol.parseRequest(type, data);
+    const response = await this.#adapter.request(
+      type,
+      (parsed ?? {}) as Record<string, unknown>,
+      options,
+    );
+    return await this.#protocol.parseResponse(type, response);
+  }
+
+  subscribe<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    handler: (
+      message: MTPResponseFrame<Registry, Type>,
+    ) => void | Promise<void>,
+  ): Unsubscribe {
+    let active = true;
+    const unsubscribe = this.#adapter.subscribe(type, (message) => {
+      void this.#protocol.parseResponse(type, message, "subscription").then(
+        (parsed) => {
+          if (active) void handler(parsed);
+        },
+        (error) => {
+          this.#protocol.reportValidationError(error);
+        },
+      );
+    });
+    return () => {
+      active = false;
+      unsubscribe();
+    };
+  }
+}
--- a/transport/src/connection.rs
+++ b/transport/src/connection.rs
@ -4,6 +4,10 @@ use crate::framing::RetryClassifier;
 use crate::pipe::PipeReader;
 use mtp_codec::{CommunicationValue, DecodeError, DecodeLimits, EncodeLimits, TypeMap};
 use mtp_common::CommunicationError;
+#[cfg(feature = "pipes")]
+use mtp_common::{FirstFrameDisposition, classify_first_frame};
+#[cfg(feature = "pipes")]
+use std::collections::HashSet;
 use std::ops::Deref;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
@ -288,15 +292,15 @@ impl Sender {
            Ok(Ok(())) => Ok(()),
            Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => {
                warn!("[Sender] write failed: peer sent STOP_SENDING (error code {code})");
-                Err(CommunicationError::StreamClosed)
+                Err(CommunicationError::DeliveryUnknown)
            }
            Ok(Err(other)) => {
                warn!("[Sender] write failed: {other}");
-                Err(CommunicationError::StreamError)
+                Err(CommunicationError::DeliveryUnknown)
            }
            Err(_) => {
                warn!("[Sender] write timed out (len={})", bytes.len());
-                Err(CommunicationError::StreamError)
+                Err(CommunicationError::DeliveryUnknown)
            }
        }
    }
@ -398,15 +402,15 @@ impl Sender {
            Ok(Ok(())) => Ok(()),
            Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => {
                warn!("[Sender] finish failed: peer sent STOP_SENDING (error code {code})");
-                Err(CommunicationError::StreamClosed)
+                Err(CommunicationError::DeliveryUnknown)
            }
            Ok(Err(other)) => {
                warn!("[Sender] finish failed: {other}");
-                Err(CommunicationError::StreamError)
+                Err(CommunicationError::DeliveryUnknown)
            }
            Err(_) => {
                warn!("[Sender] finish timed out");
-                Err(CommunicationError::StreamError)
+                Err(CommunicationError::DeliveryUnknown)
            }
        }
    }
@ -745,6 +749,8 @@ struct ReceiverInner {
    max_message_size: Arc<AtomicU64>,
    type_map: Arc<RwLock<TypeMap>>,
    decode_rejections: Arc<DecodeRejectionCounters>,
+    #[cfg(feature = "pipes")]
+    expected_pipes: Arc<std::sync::Mutex<HashSet<u32>>>,
 }

 impl Clone for Receiver {
@ -831,6 +837,10 @@ impl Receiver {
        let accept_type_map = type_map.clone();
        let decode_rejections = Arc::new(DecodeRejectionCounters::default());
        let accept_decode_rejections = decode_rejections.clone();
+        #[cfg(feature = "pipes")]
+        let expected_pipes = Arc::new(std::sync::Mutex::new(HashSet::new()));
+        #[cfg(feature = "pipes")]
+        let accept_expected_pipes = expected_pipes.clone();
        let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1)));
        let accept_stream_limit = stream_limit.clone();
        debug!(
@ -900,6 +910,8 @@ impl Receiver {
                                let stream_max_message_size = accept_max_message_size.clone();
                                let stream_type_map = accept_type_map.clone();
                                let stream_decode_rejections = accept_decode_rejections.clone();
+                                #[cfg(feature = "pipes")]
+                                let stream_expected_pipes = accept_expected_pipes.clone();

                                tokio::spawn(async move {
                                    let _permit = permit;
@ -935,17 +947,32 @@ impl Receiver {

                                                #[cfg(feature = "pipes")]
                                                {
-                                                    if msg.is_type(mtp_codec::CommunicationType::PipeRequest)
-                                                        && frame_count == 1
-                                                    {
-                                                        let Some(pipe_id) = msg.id().filter(|id| *id != 0) else {
-                                                            let error = CommunicationError::Other(
-                                                                "PipeRequest frame must contain a non-zero id".into(),
+                                                    if frame_count == 1 {
+                                                        let is_pipe_request = msg.is_type(
+                                                            mtp_codec::CommunicationType::PipeRequest,
                                                        );
-                                                            let _ = msg_tx_stream.send(Err(error.clone())).await;
+                                                        let pipe_id = msg.id().filter(|id| *id != 0);
+                                                        let pipe_is_expected = is_pipe_request && pipe_id.is_some_and(|pipe_id| {
+                                                            stream_expected_pipes
+                                                                .lock()
+                                                                .is_ok_and(|mut expected| expected.remove(&pipe_id))
+                                                        });
+                                                        let disposition = match classify_first_frame(
+                                                            is_pipe_request,
+                                                            msg.id(),
+                                                            pipe_is_expected,
+                                                        ) {
+                                                            Ok(disposition) => disposition,
+                                                            Err(error) => {
+                                                                let _ = msg_tx_stream
+                                                                    .send(Err(error.clone()))
+                                                                    .await;
                                                                stream_handle.close(Some(error));
                                                                break;
+                                                            }
                                                        };
+
+                                                        if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
                                                            let description = msg
                                                                .get_str(mtp_codec::DataType::Description)
                                                                .unwrap_or("")
@ -969,6 +996,7 @@ impl Receiver {
                                                            break;
                                                        }
                                                    }
+                                                }

                                                let pong_sender = if msg.is_type(mtp_codec::CommunicationType::Ping) {
                                                    stream_ping_control
@ -1111,6 +1139,8 @@ impl Receiver {
                max_message_size,
                type_map,
                decode_rejections,
+                #[cfg(feature = "pipes")]
+                expected_pipes,
            }),
        }
    }
@ -1127,6 +1157,26 @@ impl Receiver {
        *self.inner.type_map.write().await = type_map.clone();
    }

+    #[cfg(feature = "pipes")]
+    pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
+        if pipe_id == 0 {
+            return Err(CommunicationError::Other("pipe id must be non-zero".into()));
+        }
+        self.inner
+            .expected_pipes
+            .lock()
+            .map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))?
+            .insert(pipe_id);
+        Ok(())
+    }
+
+    #[cfg(feature = "pipes")]
+    pub fn cancel_expected_pipe(&self, pipe_id: u32) {
+        if let Ok(mut expected) = self.inner.expected_pipes.lock() {
+            expected.remove(&pipe_id);
+        }
+    }
+
    /// Return local counts for frames rejected by the structured decoder.
    ///
    /// These counters are intentionally local-only; peers continue to receive
--- a/transport/src/framing.rs
+++ b/transport/src/framing.rs
@ -82,6 +82,10 @@ mod tests {
        async fn finish(&mut self) -> Result<(), CommunicationError> {
            Ok(())
        }
+
+        fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
+            Ok(())
+        }
    }

    #[tokio::test]
--- a/transport/src/generic_connection.rs
+++ b/transport/src/generic_connection.rs
@ -10,8 +10,12 @@ use crate::{
    framing::{RetryClassifier, write_frame},
 };
 use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap};
-use mtp_common::CommunicationError;
+use mtp_common::{CommunicationError, FirstFrameDisposition, classify_first_frame};
+#[cfg(feature = "pipes")]
+use std::collections::HashSet;
 use std::sync::Arc;
+#[cfg(feature = "pipes")]
+use std::sync::Mutex as StdMutex;
 use std::sync::atomic::{AtomicU64, Ordering};
 use tokio::sync::{Mutex, Notify, RwLock, Semaphore, mpsc};
 use tokio::time::{Instant, timeout, timeout_at};
@ -85,9 +89,10 @@ impl<C: TransportConnection> GenericSender<C> {
                )
                .await
                .map_err(|_| CommunicationError::StreamError)??;
-                timeout(self.policy.write_timeout, stream.finish())
-                    .await
-                    .map_err(|_| CommunicationError::StreamError)?
+                match timeout(self.policy.write_timeout, stream.finish()).await {
+                    Ok(Ok(())) => Ok(()),
+                    Ok(Err(_)) | Err(_) => Err(CommunicationError::DeliveryUnknown),
+                }
            }
            crate::SendMode::PersistentStream => {
                let mut stream = self.persistent.lock().await;
@ -204,6 +209,8 @@ pub struct GenericReceiver<C: TransportConnection> {
    type_map: Arc<RwLock<TypeMap>>,
    queue_notify: Arc<Notify>,
    decode_rejections: Arc<DecodeRejectionCounters>,
+    #[cfg(feature = "pipes")]
+    expected_pipes: Arc<StdMutex<HashSet<u32>>>,
    _accept_task: Arc<tokio::task::JoinHandle<()>>,
 }

@ -219,6 +226,8 @@ impl<C: TransportConnection> Clone for GenericReceiver<C> {
            type_map: self.type_map.clone(),
            queue_notify: self.queue_notify.clone(),
            decode_rejections: self.decode_rejections.clone(),
+            #[cfg(feature = "pipes")]
+            expected_pipes: self.expected_pipes.clone(),
            _accept_task: self._accept_task.clone(),
        }
    }
@ -254,6 +263,10 @@ impl<C: TransportConnection> GenericReceiver<C> {
        let task_queue_notify = queue_notify.clone();
        let decode_rejections = Arc::new(DecodeRejectionCounters::default());
        let task_decode_rejections = decode_rejections.clone();
+        #[cfg(feature = "pipes")]
+        let expected_pipes = Arc::new(StdMutex::new(HashSet::new()));
+        #[cfg(feature = "pipes")]
+        let task_expected_pipes = expected_pipes.clone();
        let task_accept_task_tx = tx.clone();
        #[cfg(feature = "pipes")]
        let task_accept_task_pipe_tx = pipe_tx.clone();
@ -312,6 +325,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
                let connection = task_connection.clone();
                let type_map = task_type_map.clone();
                let decode_rejections = task_decode_rejections.clone();
+                #[cfg(feature = "pipes")]
+                let expected_pipes = task_expected_pipes.clone();
                tokio::spawn(async move {
                    let _permit = permit;
                    let mut stream = stream;
@ -443,20 +458,33 @@ impl<C: TransportConnection> GenericReceiver<C> {

                        #[cfg(feature = "pipes")]
                        {
-                            if message.is_type(mtp_codec::CommunicationType::PipeRequest)
-                                && frames == 1
-                            {
-                                let Some(pipe_id) = message.id().filter(|id| *id != 0) else {
-                                    let error = CommunicationError::Other(
-                                        "PipeRequest frame must contain a non-zero id".into(),
-                                    );
+                            if frames == 1 {
+                                let is_pipe_request =
+                                    message.is_type(mtp_codec::CommunicationType::PipeRequest);
+                                let pipe_id = message.id().filter(|id| *id != 0);
+                                let pipe_is_expected = is_pipe_request
+                                    && pipe_id.is_some_and(|pipe_id| {
+                                        expected_pipes
+                                            .lock()
+                                            .is_ok_and(|mut expected| expected.remove(&pipe_id))
+                                    });
+                                let disposition = match classify_first_frame(
+                                    is_pipe_request,
+                                    message.id(),
+                                    pipe_is_expected,
+                                ) {
+                                    Ok(disposition) => disposition,
+                                    Err(error) => {
                                        let _ = tx.send(Err(error.clone())).await;
                                        connection.close(
                                            policy.application_close_code,
                                            b"pipe request missing id",
                                        );
                                        break;
+                                    }
                                };
+
+                                if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
                                    let description = message
                                        .get_str(mtp_codec::DataType::Description)
                                        .unwrap_or("")
@ -476,6 +504,7 @@ impl<C: TransportConnection> GenericReceiver<C> {
                                    return;
                                }
                            }
+                        }

                        if message.is_type(mtp_codec::CommunicationType::Ping) {
                            if let Some(sender) = ping_sender.read().await.clone() {
@ -517,6 +546,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
            type_map,
            queue_notify,
            decode_rejections,
+            #[cfg(feature = "pipes")]
+            expected_pipes,
            _accept_task: Arc::new(accept_task),
        }
    }
@ -524,6 +555,25 @@ impl<C: TransportConnection> GenericReceiver<C> {
        *self.ping_sender.write().await = Some(sender);
    }

+    #[cfg(feature = "pipes")]
+    pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
+        if pipe_id == 0 {
+            return Err(CommunicationError::Other("pipe id must be non-zero".into()));
+        }
+        self.expected_pipes
+            .lock()
+            .map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))?
+            .insert(pipe_id);
+        Ok(())
+    }
+
+    #[cfg(feature = "pipes")]
+    pub fn cancel_expected_pipe(&self, pipe_id: u32) {
+        if let Ok(mut expected) = self.expected_pipes.lock() {
+            expected.remove(&pipe_id);
+        }
+    }
+
    /// Switch from the handshake frame limit to the application frame limit.
    pub fn set_max_message_size(&self, max_message_size: u64) {
        self.max_message_size
--- a/transport/src/transport_traits.rs
+++ b/transport/src/transport_traits.rs
@ -17,6 +17,7 @@ use mtp_common::CommunicationError;
 pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync {
    async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError>;
    async fn finish(&mut self) -> Result<(), CommunicationError>;
+    fn reset(&mut self, code: u32) -> Result<(), CommunicationError>;
 }

 /// A readable unidirectional stream suitable for MTP frames.
@ -28,6 +29,9 @@ pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync {
 pub trait TransportRecvStream: tokio::io::AsyncRead + Send + Sync {
    async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError>;
    async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError>;
+    fn stop(self, code: u32) -> Result<(), CommunicationError>
+    where
+        Self: Sized;
 }

 /// A QUIC/WebTransport connection that provides MTP's unidirectional streams.
@ -47,7 +51,7 @@ impl TransportSendStream for wtransport::SendStream {
    async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError> {
        wtransport::SendStream::write_all(self, buf)
            .await
-            .map_err(|_| CommunicationError::StreamError)
+            .map_err(|_| CommunicationError::DeliveryUnknown)
    }

    async fn finish(&mut self) -> Result<(), CommunicationError> {
@ -55,14 +59,23 @@ impl TransportSendStream for wtransport::SendStream {
            .await
            .map_err(|_| CommunicationError::StreamError)
    }
+
+    fn reset(&mut self, code: u32) -> Result<(), CommunicationError> {
+        wtransport::SendStream::reset(self, wtransport::VarInt::from_u32(code))
+            .map_err(|_| CommunicationError::StreamClosed)
+    }
 }

 #[async_trait]
 impl TransportRecvStream for wtransport::RecvStream {
    async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> {
-        wtransport::RecvStream::read_exact(self, buf)
-            .await
-            .map_err(|_| CommunicationError::StreamError)
+        match wtransport::RecvStream::read_exact(self, buf).await {
+            Ok(()) => Ok(()),
+            Err(wtransport::error::StreamReadExactError::FinishedEarly(0)) => {
+                Err(CommunicationError::StreamClosed)
+            }
+            Err(_) => Err(CommunicationError::StreamError),
+        }
    }

    async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError> {
@ -76,6 +89,11 @@ impl TransportRecvStream for wtransport::RecvStream {
            Err(_) => Err(CommunicationError::StreamError),
        }
    }
+
+    fn stop(self, code: u32) -> Result<(), CommunicationError> {
+        wtransport::RecvStream::stop(self, wtransport::VarInt::from_u32(code));
+        Ok(())
+    }
 }

 #[async_trait]
--- a/transport/tests/generic_pipe.rs
+++ b/transport/tests/generic_pipe.rs
@ -4,7 +4,7 @@ use async_trait::async_trait;
 use mtp_codec::CommunicationValue;
 use mtp_common::CommunicationError;
 use mtp_transport::{
-    GenericReceiver, GenericSender, Policy, TransportConnection, TransportEvent,
+    GenericReceiver, GenericSender, Policy, SendMode, TransportConnection, TransportEvent,
    TransportRecvStream, TransportSendStream,
 };
 use std::sync::Arc;
@ -53,6 +53,10 @@ impl TransportSendStream for MockSendStream {
            .await
            .map_err(|_| CommunicationError::StreamError)
    }
+
+    fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
+        Ok(())
+    }
 }

 struct MockRecvStream {
@ -89,6 +93,10 @@ impl TransportRecvStream for MockRecvStream {
            Err(_) => Err(CommunicationError::StreamError),
        }
    }
+
+    fn stop(self, _code: u32) -> Result<(), CommunicationError> {
+        Ok(())
+    }
 }

 #[derive(Clone)]
@ -157,6 +165,7 @@ async fn test_open_pipe_and_receive_reader() -> Result<(), Box<dyn std::error::E
    let sender = GenericSender::new(conn_a, policy.clone());
    let receiver = GenericReceiver::new(conn_b, policy);

+    receiver.expect_pipe(42)?;
    let pipe_writer = sender.open_pipe(42, "test-pipe").await?;

    let pipe_reader = receiver.receive_pipe().await?;
@ -174,6 +183,7 @@ async fn test_pipe_raw_data_roundtrip() -> Result<(), Box<dyn std::error::Error>
    let sender = GenericSender::new(conn_a, policy.clone());
    let receiver = GenericReceiver::new(conn_b, policy);

+    receiver.expect_pipe(1)?;
    let mut pipe_writer = sender.open_pipe(1, "data-pipe").await?;

    let data = b"hello through the pipe";
@ -195,6 +205,7 @@ async fn test_pipe_large_payload() -> Result<(), Box<dyn std::error::Error>> {
    let sender = GenericSender::new(conn_a, policy.clone());
    let receiver = GenericReceiver::new(conn_b, policy);

+    receiver.expect_pipe(7)?;
    let mut pipe_writer = sender.open_pipe(7, "big-pipe").await?;

    let data: Vec<u8> = (0..256 * 1024).map(|i| (i % 256) as u8).collect();
@ -223,6 +234,7 @@ async fn test_receive_event_dispatches_pipe() -> Result<(), Box<dyn std::error::
    let sender = GenericSender::new(conn_a, policy.clone());
    let receiver = GenericReceiver::new(conn_b, policy);

+    receiver.expect_pipe(99)?;
    let mut pipe_writer = sender.open_pipe(99, "event-pipe").await?;

    match receiver.receive_event().await? {
@ -261,22 +273,23 @@ async fn test_try_receive_pipe_returns_none_when_empty() -> Result<(), Box<dyn s
 async fn test_regular_messages_still_work_alongside_pipes() -> Result<(), Box<dyn std::error::Error>>
 {
    let (conn_a, conn_b) = mock_connected_pair().await;
-    let policy = Arc::new(Policy::default());
+    let policy = Arc::new(Policy::default().with_send_mode(SendMode::SingleStreamPerMessage));
    let sender = GenericSender::new(conn_a, policy.clone());
    let receiver = GenericReceiver::new(conn_b, policy);

-    let msg = CommunicationValue::new(mtp_codec::CommunicationType::BadRequest);
-    sender.send(&msg).await?;
-
-    let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
+    let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest)
+        .with_id(1)
+        .add_typed_default(
+            mtp_codec::DataType::Description,
+            mtp_codec::DataValue::Str("mixed-pipe".into()),
+        );
+    sender.send(&request).await?;

    let received = receiver.receive().await?;
-    assert_eq!(
-        received.get_type(),
-        mtp_codec::CommunicationType::BadRequest
-            .try_to_id(&mtp_codec::TypeMap::latest())
-            .unwrap()
-    );
+    assert!(received.is_type(mtp_codec::CommunicationType::PipeRequest));
+
+    receiver.expect_pipe(1)?;
+    let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;

    let pipe_reader = receiver.receive_pipe().await?;
    assert_eq!(pipe_reader.pipe_id(), 1);
@ -291,6 +304,8 @@ async fn test_multiple_pipes() -> Result<(), Box<dyn std::error::Error>> {
    let sender = GenericSender::new(conn_a, policy.clone());
    let receiver = GenericReceiver::new(conn_b, policy);

+    receiver.expect_pipe(10)?;
+    receiver.expect_pipe(20)?;
    let mut pw1 = sender.open_pipe(10, "first").await?;
    let mut pw2 = sender.open_pipe(20, "second").await?;

--- a/type-map/Cargo.toml
+++ b/type-map/Cargo.toml
@ -16,3 +16,7 @@ pipes = []
 [build-dependencies]
 serde = { version = "1", features = ["derive"] }
 serde_yaml = "0.9"
+
+[package.metadata.cargo-machete]
+# cargo-machete does not inspect build.rs, where both build dependencies are used.
+ignored = ["serde", "serde_yaml"]
--- a/wasm/src/client/receive.rs
+++ b/wasm/src/client/receive.rs
@ -179,6 +179,7 @@ impl WasmClient {
        let expired_pipe_creations = self.expired_pipe_creations.clone();
        let pending_pipes = self.pending_pipes.clone();
        let loop_pending_pipes = pending_pipes.clone();
+        let expected_pending_pipes = pending_pipes.clone();
        let on_pipe_request = self.on_pipe_request.clone();
        let loop_pipe_creations = pending_pipe_creations.clone();
        let loop_expired_pipe_creations = expired_pipe_creations.clone();
@ -293,6 +294,12 @@ impl WasmClient {
                            let _ = entry.sender.send(Ok(pipe_reader));
                        }
                    },
+                    move |pipe_id| {
+                        expected_pending_pipes
+                            .borrow()
+                            .get(&pipe_id)
+                            .is_some_and(|entry| entry.generation == loop_generation)
+                    },
                )
                .await;
            if connection_generation.get() != generation {
--- a/wasm/src/pipe.rs
+++ b/wasm/src/pipe.rs
@ -1,9 +1,6 @@
-use wasm_bindgen::JsCast;
 use wasm_bindgen::prelude::*;
-use wasm_bindgen_futures::JsFuture;

-use crate::error::js_error;
-use crate::transport::release_writer_lock;
+use crate::transport::{BrowserRecvStream, BrowserSendStream, log_stream_error_code};

 #[wasm_bindgen(typescript_custom_section)]
 const PIPE_TS: &str = r#"
@ -23,54 +20,41 @@ export interface PipeReader {

 #[wasm_bindgen]
 pub struct PipeWriter {
-    writer: JsValue,
+    stream: BrowserSendStream,
    pipe_id: u32,
 }

 impl PipeWriter {
-    pub fn new(writer: JsValue, pipe_id: u32) -> Self {
-        Self { writer, pipe_id }
+    pub(crate) fn new(stream: BrowserSendStream, pipe_id: u32) -> Self {
+        Self { stream, pipe_id }
+    }
+}
+
+impl Drop for PipeWriter {
+    fn drop(&mut self) {
+        self.stream.release();
    }
 }

 #[wasm_bindgen]
 impl PipeWriter {
    pub async fn write(&mut self, data: &[u8]) -> Result<(), JsValue> {
-        let chunk = js_sys::Uint8Array::from(data);
-        let write_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write"))
-            .map_err(|_| js_error("missing write"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("write not a function"))?;
-        let write_promise = write_fn
-            .call1(&self.writer, &chunk)
-            .map_err(|e| js_error(format!("write failed: {:?}", e)))?;
-        JsFuture::from(write_promise.unchecked_into::<js_sys::Promise>()).await?;
-        Ok(())
+        self.stream.write_all(data).await
    }

-    pub async fn close(self) -> Result<(), JsValue> {
-        let close_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close"))
-            .map_err(|_| js_error("missing close"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("close not a function"))?;
-        let close_promise = close_fn
-            .call0(&self.writer)
-            .map_err(|e| js_error(format!("close failed: {:?}", e)))?;
-        if let Err(e) = JsFuture::from(close_promise.unchecked_into::<js_sys::Promise>()).await {
-            crate::transport::log_stream_error_code(&e, "pipe writer close");
+    pub async fn close(mut self) -> Result<(), JsValue> {
+        let result = self.stream.finish().await;
+        if let Err(error) = &result {
+            log_stream_error_code(error, "pipe writer close");
        }
-        release_writer_lock(&self.writer);
-        Ok(())
+        self.stream.release();
+        result
    }

    pub fn abort(&mut self) -> Result<(), JsValue> {
-        let abort_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort"))
-            .map_err(|_| js_error("missing abort"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("abort not a function"))?;
-        let _ = abort_fn.call0(&self.writer);
-        release_writer_lock(&self.writer);
-        Ok(())
+        let result = self.stream.reset(0);
+        self.stream.release();
+        result
    }

    pub fn pipe_id(&self) -> u32 {
@ -80,19 +64,26 @@ impl PipeWriter {

 #[wasm_bindgen]
 pub struct PipeReader {
-    reader: JsValue,
+    stream: BrowserRecvStream,
    description: String,
    pipe_id: u32,
    pending: Vec<u8>,
+    finished: bool,
 }

 impl PipeReader {
-    pub fn new(reader: JsValue, pipe_id: u32, description: String, pending: Vec<u8>) -> Self {
+    pub(crate) fn new(
+        stream: BrowserRecvStream,
+        pipe_id: u32,
+        description: String,
+        pending: Vec<u8>,
+    ) -> Self {
        Self {
-            reader,
+            stream,
            pipe_id,
            description,
            pending,
+            finished: false,
        }
    }
 }
@ -105,27 +96,18 @@ impl PipeReader {
            return Ok(js_sys::Uint8Array::from(&data[..]).into());
        }

-        let read_fn = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read"))
-            .map_err(|_| js_error("missing read"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("read not a function"))?;
-        let promise = read_fn
-            .call0(&self.reader)
-            .map_err(|_| js_error("read call failed"))?
-            .unchecked_into::<js_sys::Promise>();
-        let result = JsFuture::from(promise).await?;
-
-        let done = js_sys::Reflect::get(&result, &JsValue::from_str("done"))
-            .ok()
-            .and_then(|v| v.as_bool())
-            .unwrap_or(true);
-        if done {
+        if self.finished {
            return Ok(JsValue::NULL);
        }

-        let value = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
-            .map_err(|_| js_error("missing value"))?;
-        Ok(js_sys::Uint8Array::new(&value).into())
+        match self.stream.read_chunk().await? {
+            Some(value) => Ok(js_sys::Uint8Array::from(&value[..]).into()),
+            None => {
+                self.stream.release();
+                self.finished = true;
+                Ok(JsValue::NULL)
+            }
+        }
    }

    pub fn pipe_id(&self) -> u32 {
@ -136,3 +118,9 @@ impl PipeReader {
        self.description.clone()
    }
 }
+
+impl Drop for PipeReader {
+    fn drop(&mut self) {
+        self.stream.release();
+    }
+}
--- a/wasm/src/transport.rs
+++ b/wasm/src/transport.rs
@ -9,6 +9,7 @@ use wasm_bindgen_futures::JsFuture;
 use crate::error::js_error;
 use crate::frame::parse_frame_value_with_limits;
 use mtp_codec::{DecodeLimits, EncodeLimits, TypeMap};
+use mtp_common::{FirstFrameDisposition, classify_first_frame};

 const CLOSE_FRAME_LEN: u32 = u32::MAX;

@ -25,12 +26,6 @@ pub(crate) fn log_stream_error_code(error: &JsValue, context: &str) {
    let stream_error_code = js_sys::Reflect::get(error, &JsValue::from_str("streamErrorCode"))
        .ok()
        .and_then(|v| v.as_f64());
-    if matches!(stream_error_code, Some(0.0)) {
-        // WebTransport reports peer-driven stream shutdown as code 0 in this
-        // environment. For one-frame handshake streams, that is expected and
-        // should not be surfaced as a warning.
-        return;
-    }
    let message = error
        .as_string()
        .or_else(|| {
@ -76,6 +71,233 @@ fn resolve_stream_readable(recv_stream: &JsValue) -> Result<JsValue, JsValue> {
    }
 }

+#[derive(Clone)]
+struct BrowserConnection {
+    inner: JsValue,
+    incoming_reader: Rc<RefCell<Option<JsValue>>>,
+}
+
+pub(crate) struct BrowserSendStream {
+    writer: JsValue,
+}
+
+pub(crate) struct BrowserRecvStream {
+    reader: JsValue,
+}
+
+impl BrowserConnection {
+    async fn connect(url: &str, cert_hashes: Option<Vec<String>>) -> Result<Self, JsValue> {
+        let constructor =
+            js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))?
+                .dyn_into::<js_sys::Function>()
+                .map_err(|_| js_error("WebTransport not available"))?;
+        let args = js_sys::Array::new();
+        args.push(&JsValue::from_str(url));
+
+        if let Some(hashes) = cert_hashes {
+            let webtransport_hashes = js_sys::Array::new();
+            for hash in hashes {
+                let (algorithm, value) = hash.split_once(':').unwrap_or(("sha-256", hash.as_str()));
+                if let Ok(value) = hex::decode(value) {
+                    let entry = js_sys::Object::new();
+                    js_sys::Reflect::set(
+                        &entry,
+                        &JsValue::from_str("algorithm"),
+                        &JsValue::from_str(algorithm),
+                    )?;
+                    js_sys::Reflect::set(
+                        &entry,
+                        &JsValue::from_str("value"),
+                        &js_sys::Uint8Array::from(&value[..]),
+                    )?;
+                    webtransport_hashes.push(&entry);
+                }
+            }
+            if webtransport_hashes.length() > 0 {
+                let options = js_sys::Object::new();
+                js_sys::Reflect::set(
+                    &options,
+                    &JsValue::from_str("serverCertificateHashes"),
+                    &webtransport_hashes,
+                )?;
+                args.push(&options);
+            }
+        }
+
+        let inner = js_sys::Reflect::construct(&constructor, &args)?;
+        let ready = js_sys::Reflect::get(&inner, &JsValue::from_str("ready"))?
+            .dyn_into::<js_sys::Promise>()
+            .map_err(|_| js_error("WebTransport.ready is not a Promise"))?;
+        JsFuture::from(ready)
+            .await
+            .map_err(|error| js_error(format!("WebTransport ready failed: {error:?}")))?;
+        Ok(Self {
+            inner,
+            incoming_reader: Rc::new(RefCell::new(None)),
+        })
+    }
+
+    async fn open_uni(&self) -> Result<BrowserSendStream, JsValue> {
+        let create_stream = js_sys::Reflect::get(
+            &self.inner,
+            &JsValue::from_str("createUnidirectionalStream"),
+        )?
+        .dyn_into::<js_sys::Function>()
+        .map_err(|_| js_error("createUnidirectionalStream not a function"))?;
+        let stream_promise = create_stream
+            .call0(&self.inner)?
+            .dyn_into::<js_sys::Promise>()
+            .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?;
+        let stream = JsFuture::from(stream_promise).await?;
+        let writable = resolve_stream_writable(&stream)?;
+        let writer = js_sys::Reflect::get(&writable, &JsValue::from_str("getWriter"))
+            .map_err(|_| js_error("missing getWriter"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("getWriter not a function"))?
+            .call0(&writable)
+            .map_err(|_| js_error("getWriter call failed"))?;
+        Ok(BrowserSendStream { writer })
+    }
+
+    async fn accept_uni(&self) -> Result<Option<BrowserRecvStream>, JsValue> {
+        let streams_reader = if let Some(reader) = self.incoming_reader.borrow().clone() {
+            reader
+        } else {
+            let incoming = js_sys::Reflect::get(
+                &self.inner,
+                &JsValue::from_str("incomingUnidirectionalStreams"),
+            )?;
+            let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader"))
+                .map_err(|_| js_error("missing getReader"))?
+                .dyn_into::<js_sys::Function>()
+                .map_err(|_| js_error("getReader not a function"))?
+                .call0(&incoming)
+                .map_err(|_| js_error("getReader call failed"))?;
+            *self.incoming_reader.borrow_mut() = Some(reader.clone());
+            reader
+        };
+
+        let read = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read"))
+            .map_err(|_| js_error("missing read"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("read not a function"))?;
+        let promise = read
+            .call0(&streams_reader)
+            .map_err(|_| js_error("read call failed"))?
+            .unchecked_into::<js_sys::Promise>();
+        let result = JsFuture::from(promise).await.map_err(|error| {
+            log_stream_error_code(&error, "accept_uni");
+            js_error(format!("accept stream failed: {error:?}"))
+        })?;
+        if js_sys::Reflect::get(&result, &JsValue::from_str("done"))
+            .ok()
+            .and_then(|value| value.as_bool())
+            .unwrap_or(false)
+        {
+            return Ok(None);
+        }
+
+        let stream = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
+            .map_err(|_| js_error("missing value"))?;
+        let readable = resolve_stream_readable(&stream)?;
+        let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader"))
+            .map_err(|_| js_error("missing stream getReader"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("stream getReader not a function"))?
+            .call0(&readable)
+            .map_err(|_| js_error("stream getReader call failed"))?;
+        Ok(Some(BrowserRecvStream { reader }))
+    }
+
+    fn close(&self) {
+        if let Some(reader) = self.incoming_reader.borrow_mut().take() {
+            release_reader_lock(&reader);
+        }
+        if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close"))
+            .and_then(|value| value.dyn_into::<js_sys::Function>())
+        {
+            let _ = close.call1(&self.inner, &js_sys::Object::new());
+        }
+    }
+}
+
+impl BrowserSendStream {
+    pub(crate) async fn write_all(&mut self, bytes: &[u8]) -> Result<(), JsValue> {
+        let write = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write"))
+            .map_err(|_| js_error("missing write"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("write not a function"))?;
+        let promise = write
+            .call1(&self.writer, &js_sys::Uint8Array::from(bytes))
+            .map_err(|error| js_error(format!("write failed: {error:?}")))?
+            .unchecked_into::<js_sys::Promise>();
+        JsFuture::from(promise).await.map(|_| ())
+    }
+
+    pub(crate) async fn finish(&mut self) -> Result<(), JsValue> {
+        let close = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close"))
+            .map_err(|_| js_error("missing close"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("close not a function"))?;
+        let promise = close
+            .call0(&self.writer)
+            .map_err(|error| js_error(format!("close failed: {error:?}")))?
+            .unchecked_into::<js_sys::Promise>();
+        JsFuture::from(promise).await.map(|_| ())
+    }
+
+    pub(crate) fn reset(&mut self, code: u32) -> Result<(), JsValue> {
+        let abort = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort"))
+            .map_err(|_| js_error("missing abort"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("abort not a function"))?;
+        let _ = abort.call1(&self.writer, &JsValue::from_f64(code as f64))?;
+        Ok(())
+    }
+
+    pub(crate) fn release(&self) {
+        release_writer_lock(&self.writer);
+    }
+}
+
+impl BrowserRecvStream {
+    pub(crate) async fn read_chunk(&mut self) -> Result<Option<Vec<u8>>, JsValue> {
+        let read = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read"))
+            .map_err(|_| js_error("missing read"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("read not a function"))?;
+        let promise = read
+            .call0(&self.reader)
+            .map_err(|_| js_error("read call failed"))?
+            .unchecked_into::<js_sys::Promise>();
+        let result = JsFuture::from(promise).await?;
+        if js_sys::Reflect::get(&result, &JsValue::from_str("done"))
+            .ok()
+            .and_then(|value| value.as_bool())
+            .unwrap_or(true)
+        {
+            return Ok(None);
+        }
+        let value = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
+            .map_err(|_| js_error("missing value"))?;
+        Ok(Some(js_sys::Uint8Array::new(&value).to_vec()))
+    }
+
+    #[allow(dead_code)]
+    pub(crate) fn stop(self, code: u32) -> Result<(), JsValue> {
+        let cancel = js_sys::Reflect::get(&self.reader, &JsValue::from_str("cancel"))
+            .map_err(|_| js_error("missing cancel"))?
+            .dyn_into::<js_sys::Function>()
+            .map_err(|_| js_error("cancel not a function"))?;
+        let _ = cancel.call1(&self.reader, &JsValue::from_f64(code as f64))?;
+        Ok(())
+    }
+
+    pub(crate) fn release(&self) {
+        release_reader_lock(&self.reader);
+    }
+}
+
 /// Releases a writer's lock so an abandoned writer isn't treated as an abort (which sends STOP_SENDING).
 pub(crate) fn release_writer_lock(writer: &JsValue) {
    if let Ok(release) = js_sys::Reflect::get(writer, &JsValue::from_str("releaseLock"))
@ -118,12 +340,10 @@ enum FrameOutcome {
 */
 #[derive(Clone)]
 pub struct WasmTransport {
-    inner: JsValue,
+    connection: BrowserConnection,
    max_message_size: u32,
-    /// Reader over `incoming_unidirectional_streams()` (a singleton stream of streams).
-    streams_reader: Rc<RefCell<Option<JsValue>>>,
-    /// Reader over the host's current uni-directional stream, if one is open.
-    stream_reader: Rc<RefCell<Option<JsValue>>>,
+    /// Current incoming unidirectional stream, shared across handshake and receive loops.
+    stream_reader: Rc<RefCell<Option<BrowserRecvStream>>>,
    /// Bytes already read from the current stream but not yet consumed as a frame.
    buffer: Rc<RefCell<Vec<u8>>>,
    /// Set to `true` when `open_next_stream` succeeds; cleared after the first frame is parsed.
@ -149,61 +369,14 @@ impl WasmTransport {
        max_message_size: u32,
        configured_limits: Option<DecodeLimits>,
    ) -> Result<Self, JsValue> {
-        let ctor = js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("WebTransport not available"))?;
-        let args = js_sys::Array::new();
-        args.push(&JsValue::from_str(url));
-
-        if let Some(hashes) = cert_hashes {
-            let wt_hashes = js_sys::Array::new();
-            for h in hashes {
-                let (algo, hex_val) = match h.split_once(':') {
-                    Some((algo, hex_val)) => (algo, hex_val),
-                    None => ("sha-256", h.as_str()),
-                };
-
-                if let Ok(bytes) = hex::decode(hex_val) {
-                    let hash = js_sys::Object::new();
-                    js_sys::Reflect::set(
-                        &hash,
-                        &JsValue::from_str("algorithm"),
-                        &JsValue::from_str(algo),
-                    )?;
-                    js_sys::Reflect::set(
-                        &hash,
-                        &JsValue::from_str("value"),
-                        &js_sys::Uint8Array::from(&bytes[..]),
-                    )?;
-                    wt_hashes.push(&hash);
-                }
-            }
-            if wt_hashes.length() > 0 {
-                let opts = js_sys::Object::new();
-                js_sys::Reflect::set(
-                    &opts,
-                    &JsValue::from_str("serverCertificateHashes"),
-                    &wt_hashes,
-                )?;
-                args.push(&opts);
-            }
-        };
-
-        let transport = js_sys::Reflect::construct(&ctor, &args)?;
-        let ready = js_sys::Reflect::get(&transport, &JsValue::from_str("ready"))?
-            .dyn_into::<js_sys::Promise>()
-            .map_err(|_| js_error("WebTransport.ready is not a Promise"))?;
-        JsFuture::from(ready)
-            .await
-            .map_err(|e| js_error(format!("WebTransport ready failed: {:?}", e)))?;
+        let connection = BrowserConnection::connect(url, cert_hashes).await?;
        let transport_limits = DecodeLimits::for_transport_message_size(max_message_size as u64);
        let decode_limits = configured_limits
            .map(|limits| restrict_decode_limits(limits, transport_limits))
            .unwrap_or(transport_limits);
        Ok(Self {
-            inner: transport,
+            connection,
            max_message_size,
-            streams_reader: Rc::new(RefCell::new(None)),
            stream_reader: Rc::new(RefCell::new(None)),
            buffer: Rc::new(RefCell::new(Vec::new())),
            new_stream_frame: Rc::new(Cell::new(false)),
@ -214,7 +387,7 @@ impl WasmTransport {
    }

    pub fn inner(&self) -> &JsValue {
-        &self.inner
+        &self.connection.inner
    }

    pub fn set_type_map(&self, type_map: &TypeMap) {
@ -250,154 +423,55 @@ impl WasmTransport {
        // in accept_uni() until the authentication deadline. The bytes are
        // already the canonical MTP self-framed value, so no extra stream
        // length prefix is added here.
-        let create_stream = js_sys::Reflect::get(
-            &self.inner,
-            &JsValue::from_str("createUnidirectionalStream"),
-        )?
-        .dyn_into::<js_sys::Function>()
-        .map_err(|_| js_error("createUnidirectionalStream not a function"))?;
-        let stream_promise = create_stream
-            .call0(&self.inner)?
-            .dyn_into::<js_sys::Promise>()
-            .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?;
-        let stream = JsFuture::from(stream_promise).await?;
-        let writable_or_stream = resolve_stream_writable(&stream)?;
-        let writer_val = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter"))
-            .map_err(|_| js_error("missing getWriter"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("getWriter not a function"))?
-            .call0(&writable_or_stream)
-            .map_err(|_| js_error("getWriter call failed"))?;
-
-        let chunk = js_sys::Uint8Array::from(frame);
-
-        let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write"))
-            .map_err(|_| js_error("missing write"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("write not a function"))?;
-        let write_promise = write_fn
-            .call1(&writer_val, &chunk)
-            .map_err(|e| js_error(format!("write failed: {:?}", e)))?;
-        if let Err(e) = JsFuture::from(write_promise.unchecked_into::<js_sys::Promise>()).await {
+        let mut stream = self.connection.open_uni().await?;
+        if let Err(e) = stream.write_all(frame).await {
            log_stream_error_code(&e, "send_frame write");
-            release_writer_lock(&writer_val);
+            stream.release();
            return Err(e);
        }

-        let close_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("close"))
-            .map_err(|_| js_error("missing close"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("close not a function"))?;
-        let close_promise = close_fn
-            .call0(&writer_val)
-            .map_err(|e| js_error(format!("close failed: {:?}", e)))?;
-        if let Err(e) = JsFuture::from(close_promise.unchecked_into::<js_sys::Promise>()).await {
+        if let Err(e) = stream.finish().await {
            // The frame was already written; do not retry it merely because
            // FIN failed, as that would duplicate the MTP frame.
            log_stream_error_code(&e, "send_frame close");
        }
-        release_writer_lock(&writer_val);
+        stream.release();

        Ok(())
    }

-    /// Get (creating once) the reader over `incoming_unidirectional_streams()`.
-    fn ensure_streams_reader(&self) -> Result<JsValue, JsValue> {
-        if let Some(reader) = self.streams_reader.borrow().clone() {
-            return Ok(reader);
-        }
-        let incoming = js_sys::Reflect::get(
-            &self.inner,
-            &JsValue::from_str("incomingUnidirectionalStreams"),
-        )?;
-        let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader"))
-            .map_err(|_| js_error("missing getReader"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("getReader not a function"))?
-            .call0(&incoming)
-            .map_err(|_| js_error("getReader call failed"))?;
-        *self.streams_reader.borrow_mut() = Some(reader.clone());
-        Ok(reader)
-    }
-
    /// Accept the next incoming uni-directional stream and make it current.
    /// Returns `false` if the incoming-streams readable has ended.
    async fn open_next_stream(&self) -> Result<bool, JsValue> {
-        let streams_reader = self.ensure_streams_reader()?;
-
-        let read_fn = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read"))
-            .map_err(|_| js_error("missing read"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("read not a function"))?;
-        let promise = read_fn
-            .call0(&streams_reader)
-            .map_err(|_| js_error("read call failed"))?
-            .unchecked_into::<js_sys::Promise>();
-        let result = match JsFuture::from(promise).await {
-            Ok(r) => r,
-            Err(e) => {
-                log_stream_error_code(&e, "open_next_stream accept");
-                return Err(js_error(format!("accept stream failed: {:?}", e)));
-            }
+        let Some(stream) = self.connection.accept_uni().await? else {
+            return Ok(false);
        };

-        let done = js_sys::Reflect::get(&result, &JsValue::from_str("done"))
-            .ok()
-            .and_then(|v| v.as_bool())
-            .unwrap_or(false);
-        if done {
-            return Ok(false);
-        }
-
-        let recv_stream = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
-            .map_err(|_| js_error("missing value"))?;
-        let readable = resolve_stream_readable(&recv_stream)?;
-        let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader"))
-            .map_err(|_| js_error("missing stream getReader"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("stream getReader not a function"))?
-            .call0(&readable)
-            .map_err(|_| js_error("stream getReader call failed"))?;
-
-        *self.stream_reader.borrow_mut() = Some(reader);
+        *self.stream_reader.borrow_mut() = Some(stream);
        self.new_stream_frame.set(true);
        Ok(true)
    }

    /// Read one chunk from the current stream. `Ok(None)` means the stream ended.
    async fn read_chunk(&self) -> Result<Option<Vec<u8>>, JsValue> {
-        let reader = match self.stream_reader.borrow().clone() {
-            Some(r) => r,
+        let mut stream = match self.stream_reader.borrow_mut().take() {
+            Some(stream) => stream,
            None => return Ok(None),
        };
-
-        let read_fn = js_sys::Reflect::get(&reader, &JsValue::from_str("read"))
-            .map_err(|_| js_error("missing read"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("read not a function"))?;
-        let promise = read_fn
-            .call0(&reader)
-            .map_err(|_| js_error("read call failed"))?
-            .unchecked_into::<js_sys::Promise>();
-        let result = match JsFuture::from(promise).await {
-            Ok(r) => r,
+        let result = match stream.read_chunk().await {
+            Ok(result) => result,
            Err(e) => {
                log_stream_error_code(&e, "read_chunk");
+                stream.release();
                return Err(js_error(format!("read failed: {:?}", e)));
            }
        };
-
-        let done = js_sys::Reflect::get(&result, &JsValue::from_str("done"))
-            .ok()
-            .and_then(|v| v.as_bool())
-            .unwrap_or(true);
-        if done {
-            return Ok(None);
+        if result.is_some() {
+            *self.stream_reader.borrow_mut() = Some(stream);
+        } else {
+            stream.release();
        }
-
-        let value = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
-            .map_err(|_| js_error("missing value"))?;
-        Ok(Some(js_sys::Uint8Array::new(&value).to_vec()))
+        Ok(result)
    }

    /// Try to pull one complete frame out of the buffer without reading more.
@ -459,9 +533,7 @@ impl WasmTransport {
                }
                None => {
                    // Stream finished; release the reader's lock to avoid a spurious cancel.
-                    if let Some(reader) = self.stream_reader.borrow_mut().take() {
-                        release_reader_lock(&reader);
-                    }
+                    // `read_chunk` releases the raw stream lock on clean FIN.
                    // A frame is never allowed to span stream boundaries. The
                    // native persistent-stream sender packs frames on one
                    // stream, while the WASM sender uses one stream per frame;
@ -520,15 +592,17 @@ impl WasmTransport {
    /// Pipe-aware receive loop. Identical to `receive_loop` but detects
    /// `PipeRequest` as the first frame on a new incoming stream and routes
    /// the stream to `on_pipe` instead of `on_message`.
-    pub async fn receive_loop_with_pipes<F, G, H>(
+    pub async fn receive_loop_with_pipes<F, G, H, I>(
        &self,
        mut on_message: F,
        mut on_error: H,
        mut on_pipe: G,
+        mut pipe_is_expected: I,
    ) where
        F: FnMut(JsValue),
        G: FnMut(crate::pipe::PipeReader),
        H: FnMut(JsValue),
+        I: FnMut(u32) -> bool,
    {
        loop {
            match self.next_frame(self.max_message_size).await {
@ -550,16 +624,23 @@ impl WasmTransport {

                    if is_first {
                        self.new_stream_frame.set(false);
-                        if let Some(comm) = comm.as_ref()
-                            && Some(comm.get_type()) == pipe_request_type
+                        if let Some(comm) = comm.as_ref() {
+                            let is_pipe_request = Some(comm.get_type()) == pipe_request_type;
+                            let pipe_id = comm.id().filter(|id| *id != 0);
+                            let is_expected =
+                                is_pipe_request && pipe_id.is_some_and(&mut pipe_is_expected);
+                            let disposition =
+                                match classify_first_frame(is_pipe_request, comm.id(), is_expected)
                                {
-                            let Some(pipe_id) = comm.id().filter(|id| *id != 0) else {
-                                on_error(JsValue::from_str(
-                                    "PipeRequest frame must contain a non-zero id",
-                                ));
+                                    Ok(disposition) => disposition,
+                                    Err(error) => {
+                                        on_error(JsValue::from_str(&error.to_string()));
                                        self.close();
                                        break;
+                                    }
                                };
+
+                            if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
                                let description = comm
                                    .get_str(mtp_codec::DataType::Description)
                                    .unwrap_or("")
@ -582,6 +663,7 @@ impl WasmTransport {
                                continue;
                            }
                        }
+                    }

                    if let Some(comm) = comm.as_ref()
                        && Some(comm.get_type()) == pipe_response_type
@ -635,25 +717,7 @@ impl WasmTransport {
        description: &str,
    ) -> Result<crate::pipe::PipeWriter, JsValue> {
        let _send_guard = self.send_lock.lock().await;
-        let create_stream = js_sys::Reflect::get(
-            &self.inner,
-            &JsValue::from_str("createUnidirectionalStream"),
-        )?
-        .dyn_into::<js_sys::Function>()
-        .map_err(|_| js_error("createUnidirectionalStream not a function"))?;
-        let stream_promise = create_stream
-            .call0(&self.inner)?
-            .dyn_into::<js_sys::Promise>()
-            .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?;
-        let stream = JsFuture::from(stream_promise).await?;
-
-        let writable_or_stream = resolve_stream_writable(&stream)?;
-        let writer_val = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter"))
-            .map_err(|_| js_error("missing getWriter"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("getWriter not a function"))?
-            .call0(&writable_or_stream)
-            .map_err(|_| js_error("getWriter call failed"))?;
+        let mut stream = self.connection.open_uni().await?;

        let type_map = self.type_map();
        let request = mtp_codec::CommunicationValue::new_with_type_map(
@ -669,37 +733,21 @@ impl WasmTransport {
            .to_bytes()
            .map_err(|e| js_error(format!("encode failed: {}", e)))?;

-        let chunk = js_sys::Uint8Array::from(&frame_bytes[..]);
-        let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write"))
-            .map_err(|_| js_error("missing write"))?
-            .dyn_into::<js_sys::Function>()
-            .map_err(|_| js_error("write not a function"))?;
-        let write_promise = write_fn
-            .call1(&writer_val, &chunk)
-            .map_err(|e| js_error(format!("write failed: {:?}", e)))?;
-        if let Err(e) = JsFuture::from(write_promise.unchecked_into::<js_sys::Promise>()).await {
+        if let Err(e) = stream.write_all(&frame_bytes).await {
            log_stream_error_code(&e, "open_pipe write");
-            release_writer_lock(&writer_val);
+            stream.release();
            return Err(e);
        }

-        Ok(crate::pipe::PipeWriter::new(writer_val, pipe_id))
+        Ok(crate::pipe::PipeWriter::new(stream, pipe_id))
    }

    pub fn close(&self) {
        // Release reader locks before closing so they aren't treated as cancels.
        if let Some(reader) = self.stream_reader.borrow_mut().take() {
-            release_reader_lock(&reader);
-        }
-        if let Some(reader) = self.streams_reader.borrow_mut().take() {
-            release_reader_lock(&reader);
-        }
-
-        if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close"))
-            .and_then(|value| value.dyn_into::<js_sys::Function>())
-        {
-            let _ = close.call1(&self.inner, &js_sys::Object::new());
+            reader.release();
        }
+        self.connection.close();
    }
 }
Author	SHA1	Message	Date
Alex Emmet	f6a8b464e0	Merge remote-tracking branch 'refs/remotes/origin/master' All checks were successful CI / checks (push) Successful in 5m48s Details	2026-08-27 22:43:40 +02:00
Alex Emmet	9cea795d1c	ChaCha20	2026-08-27 22:42:32 +02:00
Alois	4493ed32cf	chore(deps): update chacha20 from 0.10.1 to 0.10.2 All checks were successful CI / checks (push) Successful in 5m53s Details	2026-08-27 22:33:20 +02:00
Alois	c7c7afe578	fix(example): web-client type Some checks failed CI / checks (push) Failing after 3m16s Details	2026-08-27 22:16:16 +02:00
Alois	8d94bc5498	feat(qol): add direnv Some checks failed CI / checks (push) Failing after 3m14s Details	2026-08-27 21:55:36 +02:00
Alois	22d13742ae	feat(qol): remove dup Some checks failed CI / checks (push) Failing after 4m44s Details	2026-08-27 21:10:37 +02:00
Alois	df75fd2830	feat(qol): remove dup Some checks failed CI / checks (push) Failing after 4m19s Details	2026-08-27 20:28:26 +02:00
Alois	6348d7884f	Fix thing Some checks failed CI / checks (push) Failing after 3m59s Details	2026-08-27 20:17:11 +02:00
Alois	7ef6ec9e88	Merge remote-tracking branch 'refs/remotes/origin/master' Some checks failed CI / checks (push) Failing after 2s Details	2026-08-27 19:30:03 +02:00
Alois	bd5547ae6f	feat(ts-sdk): add schemas	2026-08-27 19:29:53 +02:00
Alex Emmet	c30315af94	Clean Some checks failed CI / checks (push) Failing after 2s Details	2026-08-27 16:55:52 +02:00
Alois	e83cd132a2	feat(wasm, native, h3): make wasm, native and h3 use unified interface Some checks failed CI / checks (push) Failing after 2s Details	2026-08-27 15:31:55 +02:00
Alex Emmet	101b8322a1	[Fix] Policy overwrites Some checks failed CI / checks (push) Failing after 2s Details	2026-08-20 21:59:57 +02:00