.envrc

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

.forgejo/workflows/ci.yml

@@ -7,12 +7,16 @@ 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
 
@@ -29,6 +33,8 @@ 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,10 +14,16 @@ 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:
@@ -26,6 +32,9 @@ 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,4 +6,3 @@ dist/
 *.tgz
 wasm/pkg/
 web_client/
-.direnv

Cargo.lock

@@ -256,9 +256,9 @@ dependencies = [
 
 [[package]]
 name = "chacha20"
-version = "0.10.2"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
 dependencies = [
  "cfg-if",
  "cpufeatures 0.3.0",
@@ -1326,6 +1326,9 @@ dependencies = [
  "mtp-transport",
  "mtp-type-map",
  "mtp-webserver",
+ "rand",
+ "rcgen",
+ "tokio",
 ]
 
 [[package]]
@@ -1360,6 +1363,7 @@ name = "mtp-common"
 version = "0.3.0"
 dependencies = [
  "quinn",
+ "rustls",
  "thiserror 2.0.20",
  "wtransport",
 ]
@@ -1800,7 +1804,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "chacha20 0.10.2",
+ "chacha20 0.10.1",
  "getrandom 0.4.3",
  "rand_core 0.10.1",
 ]

Cargo.toml

@@ -113,5 +113,10 @@ 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,41 +78,9 @@ 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 {
@@ -124,10 +92,6 @@ 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;
@@ -151,10 +115,7 @@ impl PipeRequest {
 
         let timeout = self.dispatcher.policy.read_timeout;
         match tokio::time::timeout(timeout, pipe_rx).await {
-            Ok(Ok(reader)) => {
+            Ok(Ok(reader)) => Ok(reader),
-                expected_pipe.disarm();
-                Ok(reader)
-            }
             Ok(Err(_)) => {
                 self.dispatcher
                     .pending_pipes
@@ -452,7 +413,6 @@ 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.2"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
 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.2",
+ "chacha20 0.10.1",
  "getrandom 0.4.3",
  "rand_core 0.10.1",
 ]

common/Cargo.lock

@@ -139,9 +139,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
 
 [[package]]
 name = "chacha20"
-version = "0.10.2"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
 dependencies = [
  "cfg-if",
  "cpufeatures",

common/Cargo.toml

@@ -15,6 +15,7 @@ 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,9 +164,6 @@ 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),
@@ -185,38 +182,6 @@ 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 {
@@ -247,7 +212,6 @@ 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.2"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
 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.2",
+ "chacha20 0.10.1",
  "getrandom 0.4.3",
  "rand_core 0.10.1",
 ]

docs/WASM-CLIENT.md

@@ -104,9 +104,6 @@ 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. |
@@ -474,60 +471,6 @@ 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.2"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
 dependencies = [
  "cfg-if",
  "cpufeatures 0.3.0",
@@ -1305,6 +1305,7 @@ name = "mtp-common"
 version = "0.3.0"
 dependencies = [
  "quinn",
+ "rustls",
  "thiserror 2.0.20",
  "wtransport",
 ]
@@ -1701,7 +1702,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "chacha20 0.10.2",
+ "chacha20 0.10.1",
  "getrandom 0.4.3",
  "rand_core 0.10.1",
 ]

example/server/src/main.rs

@@ -38,7 +38,6 @@ 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,30 +5,26 @@
     rust-overlay.url = "github:oxalica/rust-overlay";
   };
 
-  outputs =
+  outputs = {
-    {
+    self,
-      self,
+    nixpkgs,
-      nixpkgs,
+    rust-overlay,
-      rust-overlay,
+  }: let
-    }:
+    systems = [
-    let
+      "aarch64-darwin"
-      systems = [
+      "aarch64-linux"
-        "aarch64-darwin"
+      "x86_64-darwin"
-        "aarch64-linux"
+      "x86_64-linux"
-        "x86_64-darwin"
+    ];
-        "x86_64-linux"
+    eachSystem = f:
-      ];
+      nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate {} (
-      eachSystem =
+        map (system: nixpkgs.lib.mapAttrs (_: value: {${system} = value;}) (f system)) systems
-        f:
+      );
-        nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate { } (
+  in
-          map (system: nixpkgs.lib.mapAttrs (_: value: { ${system} = value; }) (f system)) systems
-        );
-    in
     eachSystem (
-      system:
+      system: let
-      let
+        overlays = [rust-overlay.overlays.default];
-        overlays = [ rust-overlay.overlays.default ];
+        pkgs = import nixpkgs {inherit system overlays;};
-        pkgs = import nixpkgs { inherit system overlays; };
 
         rustToolchain = pkgs.rust-bin.stable.latest.default.override {
           extensions = [
@@ -36,12 +32,12 @@
             "clippy"
             "rustfmt"
           ];
-          targets = [ "wasm32-unknown-unknown" ];
+          targets = ["wasm32-unknown-unknown"];
         };
 
         clippyCheck = pkgs.writeShellApplication {
           name = "mtp-clippy";
-          runtimeInputs = [ rustToolchain ];
+          runtimeInputs = [rustToolchain];
           text = ''
             export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}"
             cargo clippy --workspace --exclude mtp-wasm --all-targets --all-features -- -D warnings -W unreachable-pub
@@ -50,7 +46,7 @@
 
         macheteCheck = pkgs.writeShellApplication {
           name = "mtp-machete";
-          runtimeInputs = [ pkgs.cargo-machete ];
+          runtimeInputs = [pkgs.cargo-machete];
           text = ''
             cargo machete "$@"
           '';
@@ -58,15 +54,7 @@
 
         buildAll = pkgs.writeShellApplication {
           name = "mtp-build-all";
-          runtimeInputs = [
+          runtimeInputs = [rustToolchain pkgs.cargo-deny pkgs.wasm-pack pkgs.pnpm pkgs.coreutils clippyCheck macheteCheck];
-            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}"
 
@@ -78,6 +66,7 @@
               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
@@ -90,17 +79,13 @@
 
         healthCheck = pkgs.writeShellApplication {
           name = "mtp-health";
-          runtimeInputs = [
+          runtimeInputs = [clippyCheck macheteCheck];
-            clippyCheck
-            macheteCheck
-          ];
           text = ''
             mtp-clippy
             mtp-machete
           '';
         };
-      in
+      in {
-      {
         devShells = {
           default = pkgs.mkShell {
             name = "mtp-dev";

host/src/config.rs

@@ -251,8 +251,6 @@ 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,
@@ -290,8 +288,6 @@ 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,
@@ -345,9 +341,7 @@ impl HostConfig {
         get_existing_client: GetExistingClient,
         complete_register: CompleteRegister,
     ) -> Self {
-        if !self.authentication_policy_explicit {
+        self.authentication_policy = AuthenticationPolicy::ForceAuthentication;
-            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);
@@ -357,7 +351,6 @@ 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
     }
 
@@ -448,78 +441,4 @@ 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, R, P>>>,
+    pub(crate) pipe_req_rx: Mutex<mpsc::Receiver<PipeRequest<S, 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, R, P>, CommunicationError> {
+    pub async fn receive_pipe(&self) -> Result<PipeRequest<S, P>, CommunicationError> {
         self.pipe_req_rx
             .lock()
             .await

host/src/engine.rs

@@ -309,12 +309,6 @@ 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(),
@@ -572,7 +566,6 @@ 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) => {
@@ -581,7 +574,6 @@ 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,10 +27,6 @@ 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;
@@ -56,14 +52,6 @@ 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> {
@@ -99,14 +87,6 @@ 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> {
@@ -172,60 +152,16 @@ where
     }
 }
 
-pub struct PipeRequest<S, R, P> {
+pub struct PipeRequest<S, P> {
     pub(crate) pipe_id: u32,
     pub(crate) description: String,
     pub(crate) sender: S,
-    pub(crate) receiver: R,
     pub(crate) dispatcher: Arc<PipeDispatcher<P>>,
 }
 
-struct ExpectedPipeGuard<R, P>
+impl<S, P> PipeRequest<S, 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 {
@@ -237,10 +173,6 @@ 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
@@ -264,10 +196,7 @@ where
         }
 
         match tokio::time::timeout(self.dispatcher.policy.read_timeout, pipe_rx).await {
-            Ok(Ok(reader)) => {
+            Ok(Ok(reader)) => Ok(reader),
-                expected_pipe.disarm();
-                Ok(reader)
-            }
             Ok(Err(_)) => {
                 self.dispatcher
                     .pending_pipes
@@ -432,7 +361,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, R, P>>,
+    pipe_req_tx: mpsc::Sender<PipeRequest<S, P>>,
     dispatcher: Arc<PipeDispatcher<P>>,
 ) where
     S: PipeSender,
@@ -459,7 +388,6 @@ 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;
@@ -518,7 +446,6 @@ 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::DeliveryUnknown)?;
+            .map_err(|_| CommunicationError::StreamError)?;
         // 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::DeliveryUnknown)
+            .map_err(|_| CommunicationError::StreamError)
     }
 
     async fn finish(&mut self) -> Result<(), CommunicationError> {
@@ -73,11 +73,6 @@ 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]
@@ -145,11 +140,6 @@ 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,6 +61,7 @@
     "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,15 +10,6 @@ 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 {
@@ -258,9 +249,7 @@ export interface MTPPublicKeyBundleKeys {
   sigClPublicKey: Uint8Array;
 }
 
-export interface MTPClientOptions<
+export interface MTPClientOptions {
-  Registry extends MTPSchemaRegistry = MTPNoSchemas,
-> {
   url: string;
   descriptor?: string;
   hostPublicKey?: MTPKeyMaterialInput;
@@ -293,12 +282,6 @@ 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 {
@@ -579,8 +562,8 @@ export interface MTPAcceptEncryptedPipeOptions {
   signaturePolicy?: MTPSignatureVerificationPolicy;
 }
 
-type NormalizedMTPClientOptions<Registry extends MTPSchemaRegistry> = Omit<
+type NormalizedMTPClientOptions = Omit<
-  MTPClientOptions<Registry>,
+  MTPClientOptions,
   "hostPublicKey" | "receiveLimits"
 > & {
   hostPublicKey?: Uint8Array;
@@ -931,34 +914,14 @@ 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<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
+export class MTPClient {
   static readonly crypto = crypto;
   static readonly codec = codec;
 
   #credentials: InternalCredentials | null;
-  #options: NormalizedMTPClientOptions<Registry>;
+  #options: NormalizedMTPClientOptions;
-  readonly #protocol: MTPProtocol<Registry> | undefined;
   readonly #protectedReplayGuard = new InMemoryReplayGuard();
   readonly #relayReplayGuard = new InMemoryReplayGuard();
   readonly raw: MTPRaw;
@@ -971,17 +934,10 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
   readonly encryptedSecretProvider: MTPEncryptedSecretProvider;
 
   private constructor(
-    options: NormalizedMTPClientOptions<Registry>,
+    options: NormalizedMTPClientOptions,
     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 =
@@ -991,11 +947,7 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
     );
   }
 
-  static async create<
+  static async create(options: MTPClientOptions): Promise<MTPClient> {
-    const Registry extends MTPSchemaRegistry = MTPNoSchemas,
-  >(
-    options: MTPClientOptions<Registry>,
-  ): Promise<MTPClient<Registry>> {
     validateOptions(options);
     await MTPClient.init(options.wasm);
 
@@ -1016,7 +968,7 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
       securityProfile: resolveSecurityProfile(options),
     };
 
-    let sdk: MTPClient<Registry> | undefined;
+    let sdk: MTPClient | undefined;
     const client = new WasmClient(
       (state) =>
         emit(normalizedOptions.logger, {
@@ -1052,7 +1004,7 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
       setReceiveLimits.call(rawClient, normalizedOptions.receiveLimits);
     }
 
-    sdk = new MTPClient<Registry>(normalizedOptions, client);
+    sdk = new MTPClient(normalizedOptions, client);
     await sdk.#loadStoredCredentials();
     if (!sdk.#credentials) {
       sdk.#credentials = {
@@ -1286,35 +1238,6 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
     };
   }
 
-  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 (
@@ -1356,11 +1279,6 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
   }
 
   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>,
@@ -1368,14 +1286,10 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
   ): Promise<void>;
   async send(
     typeOrFrame: Uint8Array | MTPCommunicationType,
-    data?: unknown,
+    data?: Record<string, unknown>,
     options?: MTPSendOptions,
   ): Promise<void> {
-    const parsedData =
+    const message = this.#buildFrame(typeOrFrame, data, options);
-      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);
@@ -1431,11 +1345,6 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
     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>,
@@ -1443,19 +1352,15 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
   ): Promise<ParsedFrame>;
   async request(
     typeOrFrame: Uint8Array | MTPCommunicationType,
-    data?: unknown,
+    data?: Record<string, unknown>,
     options: MTPRequestOptions = {},
-  ): Promise<ParsedFrame | MTPFrame<unknown>> {
+  ): Promise<ParsedFrame> {
     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 parsedData =
+    const frame = this.#buildFrame(typeOrFrame, data, options);
-      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(
@@ -1486,29 +1391,16 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
     // 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.
-    const response = await this.raw.client.request(
+    return 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 | Promise<void>,
+    handler: (message: ParsedFrame) => 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");
@@ -1516,25 +1408,8 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
     if (typeof handler !== "function") {
       throw new TypeError("subscription handler must be a function");
     }
-    let active = true;
+    const id = this.raw.client.subscribe(type, handler);
-    const id = this.raw.client.subscribe(type, (message) => {
+    return () => this.raw.client.unsubscribe(id);
-      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,4 +5,3 @@
  * keeps the package's historical exports stable.
  */
 export * from "./client.js";
-export * from "./schema.js";

src/sdk/schema.ts

@@ -1,234 +0,0 @@
-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,10 +4,6 @@ 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};
@@ -292,15 +288,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::DeliveryUnknown)
+                Err(CommunicationError::StreamClosed)
             }
             Ok(Err(other)) => {
                 warn!("[Sender] write failed: {other}");
-                Err(CommunicationError::DeliveryUnknown)
+                Err(CommunicationError::StreamError)
             }
             Err(_) => {
                 warn!("[Sender] write timed out (len={})", bytes.len());
-                Err(CommunicationError::DeliveryUnknown)
+                Err(CommunicationError::StreamError)
             }
         }
     }
@@ -402,15 +398,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::DeliveryUnknown)
+                Err(CommunicationError::StreamClosed)
             }
             Ok(Err(other)) => {
                 warn!("[Sender] finish failed: {other}");
-                Err(CommunicationError::DeliveryUnknown)
+                Err(CommunicationError::StreamError)
             }
             Err(_) => {
                 warn!("[Sender] finish timed out");
-                Err(CommunicationError::DeliveryUnknown)
+                Err(CommunicationError::StreamError)
             }
         }
     }
@@ -749,8 +745,6 @@ 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 {
@@ -837,10 +831,6 @@ 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!(
@@ -910,8 +900,6 @@ 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;
@@ -947,54 +935,38 @@ impl Receiver {
 
                                                 #[cfg(feature = "pipes")]
                                                 {
-                                                    if frame_count == 1 {
+                                                    if msg.is_type(mtp_codec::CommunicationType::PipeRequest)
-                                                        let is_pipe_request = msg.is_type(
+                                                        && frame_count == 1
-                                                            mtp_codec::CommunicationType::PipeRequest,
+                                                    {
-                                                        );
+                                                        let Some(pipe_id) = msg.id().filter(|id| *id != 0) else {
-                                                        let pipe_id = msg.id().filter(|id| *id != 0);
+                                                            let error = CommunicationError::Other(
-                                                        let pipe_is_expected = is_pipe_request && pipe_id.is_some_and(|pipe_id| {
+                                                                "PipeRequest frame must contain a non-zero id".into(),
-                                                            stream_expected_pipes
+                                                            );
-                                                                .lock()
+                                                            let _ = msg_tx_stream.send(Err(error.clone())).await;
-                                                                .is_ok_and(|mut expected| expected.remove(&pipe_id))
+                                                            stream_handle.close(Some(error));
-                                                        });
+                                                            break;
-                                                        let disposition = match classify_first_frame(
+                                                        };
-                                                            is_pipe_request,
+                                                        let description = msg
-                                                            msg.id(),
+                                                            .get_str(mtp_codec::DataType::Description)
-                                                            pipe_is_expected,
+                                                            .unwrap_or("")
-                                                        ) {
+                                                            .to_string();
-                                                            Ok(disposition) => disposition,
+
-                                                            Err(error) => {
+                                                        let pipe_reader = crate::pipe::PipeReader {
-                                                                let _ = msg_tx_stream
+                                                            stream: s,
-                                                                    .send(Err(error.clone()))
+                                                            description,
-                                                                    .await;
+                                                            pipe_id,
-                                                                stream_handle.close(Some(error));
-                                                                break;
-                                                            }
                                                         };
 
-                                                        if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
+                                                        if pipe_tx_stream
-                                                            let description = msg
+                                                            .send(pipe_reader)
-                                                                .get_str(mtp_codec::DataType::Description)
+                                                            .await
-                                                                .unwrap_or("")
+                                                            .is_err()
-                                                                .to_string();
+                                                        {
-
+                                                            stream_handle.close(Some(
-                                                            let pipe_reader = crate::pipe::PipeReader {
+                                                                CommunicationError::StreamClosed,
-                                                                stream: s,
+                                                            ));
-                                                                description,
-                                                                pipe_id,
-                                                            };
-
-                                                            if pipe_tx_stream
-                                                                .send(pipe_reader)
-                                                                .await
-                                                                .is_err()
-                                                            {
-                                                                stream_handle.close(Some(
-                                                                    CommunicationError::StreamClosed,
-                                                                ));
-                                                            }
-                                                            break;
                                                         }
+                                                        break;
                                                     }
                                                 }
 
@@ -1139,8 +1111,6 @@ impl Receiver {
                 max_message_size,
                 type_map,
                 decode_rejections,
-                #[cfg(feature = "pipes")]
-                expected_pipes,
             }),
         }
     }
@@ -1157,26 +1127,6 @@ 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,10 +82,6 @@ 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,13 +9,9 @@ use crate::{
     connection::{DecodeRejectionCounters, RuntimePolicy, classify_decode_error},
     framing::{RetryClassifier, write_frame},
 };
-use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap};
+use mtp_codec::{CommunicationValue, DecodeLimits, TypeMap};
-use mtp_common::{CommunicationError, FirstFrameDisposition, classify_first_frame};
+use mtp_common::CommunicationError;
-#[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};
@@ -71,15 +67,6 @@ 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?;
@@ -89,10 +76,9 @@ impl<C: TransportConnection> GenericSender<C> {
                 )
                 .await
                 .map_err(|_| CommunicationError::StreamError)??;
-                match timeout(self.policy.write_timeout, stream.finish()).await {
+                timeout(self.policy.write_timeout, stream.finish())
-                    Ok(Ok(())) => Ok(()),
+                    .await
-                    Ok(Err(_)) | Err(_) => Err(CommunicationError::DeliveryUnknown),
+                    .map_err(|_| CommunicationError::StreamError)?
-                }
             }
             crate::SendMode::PersistentStream => {
                 let mut stream = self.persistent.lock().await;
@@ -209,8 +195,6 @@ 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<()>>,
 }
 
@@ -226,8 +210,6 @@ 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(),
         }
     }
@@ -263,10 +245,6 @@ 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();
@@ -325,8 +303,6 @@ 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;
@@ -458,51 +434,37 @@ impl<C: TransportConnection> GenericReceiver<C> {
 
                         #[cfg(feature = "pipes")]
                         {
-                            if frames == 1 {
+                            if message.is_type(mtp_codec::CommunicationType::PipeRequest)
-                                let is_pipe_request =
+                                && frames == 1
-                                    message.is_type(mtp_codec::CommunicationType::PipeRequest);
+                            {
-                                let pipe_id = message.id().filter(|id| *id != 0);
+                                let Some(pipe_id) = message.id().filter(|id| *id != 0) else {
-                                let pipe_is_expected = is_pipe_request
+                                    let error = CommunicationError::Other(
-                                    && pipe_id.is_some_and(|pipe_id| {
+                                        "PipeRequest frame must contain a non-zero id".into(),
-                                        expected_pipes
+                                    );
-                                            .lock()
+                                    let _ = tx.send(Err(error.clone())).await;
-                                            .is_ok_and(|mut expected| expected.remove(&pipe_id))
+                                    connection.close(
-                                    });
+                                        policy.application_close_code,
-                                let disposition = match classify_first_frame(
+                                        b"pipe request missing id",
-                                    is_pipe_request,
+                                    );
-                                    message.id(),
+                                    break;
-                                    pipe_is_expected,
+                                };
-                                ) {
+                                let description = message
-                                    Ok(disposition) => disposition,
+                                    .get_str(mtp_codec::DataType::Description)
-                                    Err(error) => {
+                                    .unwrap_or("")
-                                        let _ = tx.send(Err(error.clone())).await;
+                                    .to_string();
-                                        connection.close(
+
-                                            policy.application_close_code,
+                                let pipe_reader = PipeReader {
-                                            b"pipe request missing id",
+                                    stream,
-                                        );
+                                    description,
-                                        break;
+                                    pipe_id,
-                                    }
                                 };
 
-                                if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
+                                tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream");
-                                    let description = message
-                                        .get_str(mtp_codec::DataType::Description)
-                                        .unwrap_or("")
-                                        .to_string();
 
-                                    let pipe_reader = PipeReader {
+                                if pipe_tx.send(pipe_reader).await.is_err() {
-                                        stream,
+                                    break;
-                                        description,
-                                        pipe_id,
-                                    };
-
-                                    tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream");
-
-                                    if pipe_tx.send(pipe_reader).await.is_err() {
-                                        break;
-                                    }
-                                    return;
                                 }
+                                return;
                             }
                         }
 
@@ -546,8 +508,6 @@ impl<C: TransportConnection> GenericReceiver<C> {
             type_map,
             queue_notify,
             decode_rejections,
-            #[cfg(feature = "pipes")]
-            expected_pipes,
             _accept_task: Arc::new(accept_task),
         }
     }
@@ -555,25 +515,6 @@ 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,7 +17,6 @@ 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.
@@ -29,9 +28,6 @@ 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.
@@ -51,7 +47,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::DeliveryUnknown)
+            .map_err(|_| CommunicationError::StreamError)
     }
 
     async fn finish(&mut self) -> Result<(), CommunicationError> {
@@ -59,23 +55,14 @@ 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> {
-        match wtransport::RecvStream::read_exact(self, buf).await {
+        wtransport::RecvStream::read_exact(self, buf)
-            Ok(()) => Ok(()),
+            .await
-            Err(wtransport::error::StreamReadExactError::FinishedEarly(0)) => {
+            .map_err(|_| CommunicationError::StreamError)
-                Err(CommunicationError::StreamClosed)
-            }
-            Err(_) => Err(CommunicationError::StreamError),
-        }
     }
 
     async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError> {
@@ -89,11 +76,6 @@ 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, SendMode, TransportConnection, TransportEvent,
+    GenericReceiver, GenericSender, Policy, TransportConnection, TransportEvent,
     TransportRecvStream, TransportSendStream,
 };
 use std::sync::Arc;
@@ -53,10 +53,6 @@ impl TransportSendStream for MockSendStream {
             .await
             .map_err(|_| CommunicationError::StreamError)
     }
-
-    fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
-        Ok(())
-    }
 }
 
 struct MockRecvStream {
@@ -93,10 +89,6 @@ impl TransportRecvStream for MockRecvStream {
             Err(_) => Err(CommunicationError::StreamError),
         }
     }
-
-    fn stop(self, _code: u32) -> Result<(), CommunicationError> {
-        Ok(())
-    }
 }
 
 #[derive(Clone)]
@@ -165,7 +157,6 @@ 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?;
@@ -183,7 +174,6 @@ 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";
@@ -205,7 +195,6 @@ 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();
@@ -234,7 +223,6 @@ 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? {
@@ -273,23 +261,22 @@ 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().with_send_mode(SendMode::SingleStreamPerMessage));
+    let policy = Arc::new(Policy::default());
     let sender = GenericSender::new(conn_a, policy.clone());
     let receiver = GenericReceiver::new(conn_b, policy);
 
-    let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest)
+    let msg = CommunicationValue::new(mtp_codec::CommunicationType::BadRequest);
-        .with_id(1)
+    sender.send(&msg).await?;
-        .add_typed_default(
+
-            mtp_codec::DataType::Description,
+    let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
-            mtp_codec::DataValue::Str("mixed-pipe".into()),
-        );
-    sender.send(&request).await?;
 
     let received = receiver.receive().await?;
-    assert!(received.is_type(mtp_codec::CommunicationType::PipeRequest));
+    assert_eq!(
-
+        received.get_type(),
-    receiver.expect_pipe(1)?;
+        mtp_codec::CommunicationType::BadRequest
-    let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
+            .try_to_id(&mtp_codec::TypeMap::latest())
+            .unwrap()
+    );
 
     let pipe_reader = receiver.receive_pipe().await?;
     assert_eq!(pipe_reader.pipe_id(), 1);
@@ -304,8 +291,6 @@ 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,7 +16,3 @@ 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,19 +99,10 @@ 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(|| missing_version())?,
+                    .ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?,
-                _ => return Err(missing_version()),
+                _ => return Err(js_error("host omitted a valid negotiated protocol version")),
             };
             if negotiated_version != PROTOCOL_VERSION {
                 return Err(js_error(

wasm/src/client/receive.rs

@@ -179,7 +179,6 @@ 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();
@@ -294,12 +293,6 @@ 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,6 +1,9 @@
+use wasm_bindgen::JsCast;
 use wasm_bindgen::prelude::*;
+use wasm_bindgen_futures::JsFuture;
 
-use crate::transport::{BrowserRecvStream, BrowserSendStream, log_stream_error_code};
+use crate::error::js_error;
+use crate::transport::release_writer_lock;
 
 #[wasm_bindgen(typescript_custom_section)]
 const PIPE_TS: &str = r#"
@@ -20,41 +23,54 @@ export interface PipeReader {
 
 #[wasm_bindgen]
 pub struct PipeWriter {
-    stream: BrowserSendStream,
+    writer: JsValue,
     pipe_id: u32,
 }
 
 impl PipeWriter {
-    pub(crate) fn new(stream: BrowserSendStream, pipe_id: u32) -> Self {
+    pub fn new(writer: JsValue, pipe_id: u32) -> Self {
-        Self { stream, pipe_id }
+        Self { writer, 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> {
-        self.stream.write_all(data).await
+        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(())
     }
 
-    pub async fn close(mut self) -> Result<(), JsValue> {
+    pub async fn close(self) -> Result<(), JsValue> {
-        let result = self.stream.finish().await;
+        let close_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close"))
-        if let Err(error) = &result {
+            .map_err(|_| js_error("missing close"))?
-            log_stream_error_code(error, "pipe writer 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");
         }
-        self.stream.release();
+        release_writer_lock(&self.writer);
-        result
+        Ok(())
     }
 
     pub fn abort(&mut self) -> Result<(), JsValue> {
-        let result = self.stream.reset(0);
+        let abort_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort"))
-        self.stream.release();
+            .map_err(|_| js_error("missing abort"))?
-        result
+            .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(())
     }
 
     pub fn pipe_id(&self) -> u32 {
@@ -64,26 +80,19 @@ impl PipeWriter {
 
 #[wasm_bindgen]
 pub struct PipeReader {
-    stream: BrowserRecvStream,
+    reader: JsValue,
     description: String,
     pipe_id: u32,
     pending: Vec<u8>,
-    finished: bool,
 }
 
 impl PipeReader {
-    pub(crate) fn new(
+    pub fn new(reader: JsValue, pipe_id: u32, description: String, pending: Vec<u8>) -> Self {
-        stream: BrowserRecvStream,
-        pipe_id: u32,
-        description: String,
-        pending: Vec<u8>,
-    ) -> Self {
         Self {
-            stream,
+            reader,
             pipe_id,
             description,
             pending,
-            finished: false,
         }
     }
 }
@@ -96,18 +105,27 @@ impl PipeReader {
             return Ok(js_sys::Uint8Array::from(&data[..]).into());
         }
 
-        if self.finished {
+        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 {
             return Ok(JsValue::NULL);
         }
 
-        match self.stream.read_chunk().await? {
+        let value = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
-            Some(value) => Ok(js_sys::Uint8Array::from(&value[..]).into()),
+            .map_err(|_| js_error("missing value"))?;
-            None => {
+        Ok(js_sys::Uint8Array::new(&value).into())
-                self.stream.release();
-                self.finished = true;
-                Ok(JsValue::NULL)
-            }
-        }
     }
 
     pub fn pipe_id(&self) -> u32 {
@@ -118,9 +136,3 @@ impl PipeReader {
         self.description.clone()
     }
 }
-
-impl Drop for PipeReader {
-    fn drop(&mut self) {
-        self.stream.release();
-    }
-}

wasm/src/transport.rs

@@ -9,7 +9,6 @@ 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;
 
@@ -26,6 +25,12 @@ 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(|| {
@@ -71,233 +76,6 @@ 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"))
@@ -340,10 +118,12 @@ enum FrameOutcome {
  */
 #[derive(Clone)]
 pub struct WasmTransport {
-    connection: BrowserConnection,
+    inner: JsValue,
     max_message_size: u32,
-    /// Current incoming unidirectional stream, shared across handshake and receive loops.
+    /// Reader over `incoming_unidirectional_streams()` (a singleton stream of streams).
-    stream_reader: Rc<RefCell<Option<BrowserRecvStream>>>,
+    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>>>,
     /// 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.
@@ -369,14 +149,61 @@ impl WasmTransport {
         max_message_size: u32,
         configured_limits: Option<DecodeLimits>,
     ) -> Result<Self, JsValue> {
-        let connection = BrowserConnection::connect(url, cert_hashes).await?;
+        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 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 {
-            connection,
+            inner: transport,
             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)),
@@ -387,7 +214,7 @@ impl WasmTransport {
     }
 
     pub fn inner(&self) -> &JsValue {
-        &self.connection.inner
+        &self.inner
     }
 
     pub fn set_type_map(&self, type_map: &TypeMap) {
@@ -423,55 +250,154 @@ 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 mut stream = self.connection.open_uni().await?;
+        let create_stream = js_sys::Reflect::get(
-        if let Err(e) = stream.write_all(frame).await {
+            &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 {
             log_stream_error_code(&e, "send_frame write");
-            stream.release();
+            release_writer_lock(&writer_val);
             return Err(e);
         }
 
-        if let Err(e) = stream.finish().await {
+        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 {
             // 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");
         }
-        stream.release();
+        release_writer_lock(&writer_val);
 
         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 Some(stream) = self.connection.accept_uni().await? else {
+        let streams_reader = self.ensure_streams_reader()?;
-            return Ok(false);
+
+        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)));
+            }
         };
 
-        *self.stream_reader.borrow_mut() = Some(stream);
+        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.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 mut stream = match self.stream_reader.borrow_mut().take() {
+        let reader = match self.stream_reader.borrow().clone() {
-            Some(stream) => stream,
+            Some(r) => r,
             None => return Ok(None),
         };
-        let result = match stream.read_chunk().await {
+
-            Ok(result) => result,
+        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,
             Err(e) => {
                 log_stream_error_code(&e, "read_chunk");
-                stream.release();
                 return Err(js_error(format!("read failed: {:?}", e)));
             }
         };
-        if result.is_some() {
+
-            *self.stream_reader.borrow_mut() = Some(stream);
+        let done = js_sys::Reflect::get(&result, &JsValue::from_str("done"))
-        } else {
+            .ok()
-            stream.release();
+            .and_then(|v| v.as_bool())
+            .unwrap_or(true);
+        if done {
+            return Ok(None);
         }
-        Ok(result)
+
+        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()))
     }
 
     /// Try to pull one complete frame out of the buffer without reading more.
@@ -533,7 +459,9 @@ impl WasmTransport {
                 }
                 None => {
                     // Stream finished; release the reader's lock to avoid a spurious cancel.
-                    // `read_chunk` releases the raw stream lock on clean FIN.
+                    if let Some(reader) = self.stream_reader.borrow_mut().take() {
+                        release_reader_lock(&reader);
+                    }
                     // 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;
@@ -592,17 +520,15 @@ 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, I>(
+    pub async fn receive_loop_with_pipes<F, G, H>(
         &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 {
@@ -624,44 +550,36 @@ impl WasmTransport {
 
                     if is_first {
                         self.new_stream_frame.set(false);
-                        if let Some(comm) = comm.as_ref() {
+                        if let Some(comm) = comm.as_ref()
-                            let is_pipe_request = Some(comm.get_type()) == pipe_request_type;
+                            && Some(comm.get_type()) == pipe_request_type
-                            let pipe_id = comm.id().filter(|id| *id != 0);
+                        {
-                            let is_expected =
+                            let Some(pipe_id) = comm.id().filter(|id| *id != 0) else {
-                                is_pipe_request && pipe_id.is_some_and(&mut pipe_is_expected);
+                                on_error(JsValue::from_str(
-                            let disposition =
+                                    "PipeRequest frame must contain a non-zero id",
-                                match classify_first_frame(is_pipe_request, comm.id(), is_expected)
+                                ));
-                                {
+                                self.close();
-                                    Ok(disposition) => disposition,
+                                break;
-                                    Err(error) => {
+                            };
-                                        on_error(JsValue::from_str(&error.to_string()));
+                            let description = comm
-                                        self.close();
+                                .get_str(mtp_codec::DataType::Description)
-                                        break;
+                                .unwrap_or("")
-                                    }
+                                .to_string();
-                                };
 
-                            if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
+                            let pending = {
-                                let description = comm
+                                let mut buf = self.buffer.borrow_mut();
-                                    .get_str(mtp_codec::DataType::Description)
+                                std::mem::take(&mut *buf)
-                                    .unwrap_or("")
+                            };
-                                    .to_string();
 
-                                let pending = {
+                            if let Some(reader) = self.stream_reader.borrow_mut().take() {
-                                    let mut buf = self.buffer.borrow_mut();
+                                let pipe_reader = crate::pipe::PipeReader::new(
-                                    std::mem::take(&mut *buf)
+                                    reader,
-                                };
+                                    pipe_id,
-
+                                    description,
-                                if let Some(reader) = self.stream_reader.borrow_mut().take() {
+                                    pending,
-                                    let pipe_reader = crate::pipe::PipeReader::new(
+                                );
-                                        reader,
+                                on_pipe(pipe_reader);
-                                        pipe_id,
-                                        description,
-                                        pending,
-                                    );
-                                    on_pipe(pipe_reader);
-                                }
-                                continue;
                             }
+                            continue;
                         }
                     }
 
@@ -717,7 +635,25 @@ impl WasmTransport {
         description: &str,
     ) -> Result<crate::pipe::PipeWriter, JsValue> {
         let _send_guard = self.send_lock.lock().await;
-        let mut stream = self.connection.open_uni().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 type_map = self.type_map();
         let request = mtp_codec::CommunicationValue::new_with_type_map(
@@ -733,21 +669,37 @@ impl WasmTransport {
             .to_bytes()
             .map_err(|e| js_error(format!("encode failed: {}", e)))?;
 
-        if let Err(e) = stream.write_all(&frame_bytes).await {
+        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 {
             log_stream_error_code(&e, "open_pipe write");
-            stream.release();
+            release_writer_lock(&writer_val);
             return Err(e);
         }
 
-        Ok(crate::pipe::PipeWriter::new(stream, pipe_id))
+        Ok(crate::pipe::PipeWriter::new(writer_val, 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() {
-            reader.release();
+            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());
         }
-        self.connection.close();
     }
 }