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

.envrc

@@ -0,0 +1 @@
+use flake

.forgejo/workflows/ci.yml

@@ -7,16 +7,12 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
-  NIX_CONFIG: experimental-features = nix-command flakes
 
 jobs:
   checks:
     name: checks
     runs-on: nixos
     steps:
-      - name: Install node
-        run: nix profile add nixpkgs#nodejs_24
-
       - name: Checkout
         uses: https://data.forgejo.org/actions/checkout@v7
 
@@ -33,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

.forgejo/workflows/release.yml

@@ -14,16 +14,10 @@ on:
         required: true
         type: string
 
-env:
-  NIX_CONFIG: experimental-features = nix-command flakes
-
 jobs:
   release:
     runs-on: nixos
     steps:
-      - name: Install node & bun
-        run: nix profile add nixpkgs#nodejs_24 nixpkgs#bun
-
       - name: Check out repo
         uses: https://data.forgejo.org/actions/checkout@v7
         with:
@@ -32,9 +26,6 @@ jobs:
       - name: Install dependencies
         run: bun install
 
-      - name: Install cc linker, sed & jq
-        run: nix profile add nixpkgs#stdenv.cc nixpkgs#gnused nixpkgs#jq
-
       - name: Build all
         run: bun build:all
 

.gitignore

@@ -6,3 +6,4 @@ dist/
 *.tgz
 wasm/pkg/
 web_client/
+.direnv

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",
 ]

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"]

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;

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",
 ]

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",

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",

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"))]

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",
 ]

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

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",
 ]

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>> {

flake.nix

@@ -5,24 +5,28 @@
     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:
+      eachSystem =
+        f:
         nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate { } (
           map (system: nixpkgs.lib.mapAttrs (_: value: { ${system} = value; }) (f system)) systems
         );
     in
     eachSystem (
-      system: let
+      system:
+      let
         overlays = [ rust-overlay.overlays.default ];
         pkgs = import nixpkgs { inherit system overlays; };
 
@@ -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";

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
+        );
+    }
 }

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

host/src/engine.rs

@@ -309,6 +309,12 @@ impl HandshakeEngine {
             ) || registration
                 || first_msg.get_data(DataType::PublicKeys).is_some()
                 || claimed_client_id.is_some_and(|client_id| client_id != 0);
+            tracing::info!(
+                claimed_client_id = ?claimed_client_id,
+                registration,
+                authentication_requested,
+                "classified MTP opening authentication mode"
+            );
             if authentication_requested {
                 let attempt = crate::config::AuthenticationAttempt {
                     peer_network_identity: _authentication_context.peer_network_identity.clone(),
@@ -566,6 +572,7 @@ impl HandshakeEngine {
             }
 
             // Unknown or zero ID: fall back to guest
+            tracing::info!("allocating MTP guest identity");
             let guest_id_lease = match self.assign_guest_id().await {
                 Ok(lease) => lease,
                 Err(error) => {
@@ -574,6 +581,7 @@ impl HandshakeEngine {
                 }
             };
             let guest_id = guest_id_lease.id;
+            tracing::info!(guest_id, "allocated MTP guest identity");
             send_accepted_generic(sender, &negotiated, tm, Some(guest_id))
                 .await
                 .map_err(AcceptError::Send)?;

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;

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 {

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",

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) {

src/sdk/index.ts

@@ -5,3 +5,4 @@
  * keeps the package's historical exports stable.
  */
 export * from "./client.js";
+export * from "./schema.js";

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();
+    };
+  }
+}

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

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]

transport/src/generic_connection.rs

@@ -9,9 +9,13 @@ use crate::{
     connection::{DecodeRejectionCounters, RuntimePolicy, classify_decode_error},
     framing::{RetryClassifier, write_frame},
 };
-use mtp_codec::{CommunicationValue, DecodeLimits, TypeMap};
-use mtp_common::CommunicationError;
+use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap};
+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};
@@ -67,6 +71,15 @@ impl<C: TransportConnection> GenericSender<C> {
         if self.connection.close_reason().is_some() {
             return Err(CommunicationError::StreamClosed);
         }
+        if let Some(version) = value.get_str(DataType::Version) {
+            tracing::debug!(
+                message_type = ?value.get_type(),
+                version,
+                connected = ?value.get_data(DataType::Connected),
+                client_id = ?value.get_data(DataType::Id),
+                "sending MTP handshake response frame"
+            );
+        }
         match self.policy.send_mode {
             crate::SendMode::SingleStreamPerMessage => {
                 let mut stream = self.open().await?;
@@ -76,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;
@@ -195,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<()>>,
 }
 
@@ -210,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(),
         }
     }
@@ -245,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();
@@ -303,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;
@@ -434,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("")
@@ -467,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() {
@@ -508,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),
         }
     }
@@ -515,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

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]

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?;
 

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"]

wasm/src/client/authentication.rs

@@ -99,10 +99,19 @@ impl WasmClient {
                 return Err(js_error(message));
             }
 
+            let missing_version = || {
+                js_error(format!(
+                    "host omitted a valid negotiated protocol version (response_type={:?}, connected={:?}, frame_len={})",
+                    outcome.get_type(),
+                    outcome.get_data(DataType::Connected),
+                    outcome_bytes.len(),
+                ))
+            };
+
             let negotiated_version = match outcome.get_data(DataType::Version) {
                 Some(DataValue::Str(version)) => mtp_codec::Version::parse(version)
-                    .ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?,
-                _ => return Err(js_error("host omitted a valid negotiated protocol version")),
+                    .ok_or_else(|| missing_version())?,
+                _ => return Err(missing_version()),
             };
             if negotiated_version != PROTOCOL_VERSION {
                 return Err(js_error(

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 {

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();
+    }
+}

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();
     }
 }