diff --git a/rust/src/napi/convert.rs b/rust/src/napi/convert.rs
index b4ad945..6a12757 100644
--- a/rust/src/napi/convert.rs
+++ b/rust/src/napi/convert.rs
@@ -538,6 +538,26 @@ fn js_object_to_multipart_body(
     }))
 }
 
+fn js_object_to_stream_body(
+    cx: &mut FunctionContext,
+    obj: Handle<JsObject>,
+) -> NeonResult<Option<RequestBody>> {
+    let Some(value) = obj.get_opt::<JsValue, _, _>(cx, "bodyStream")? else {
+        return Ok(None);
+    };
+    let stream = value.downcast::<JsObject, _>(cx).or_throw(cx)?;
+    let handle_value: Handle<JsValue> = stream.get(cx, "uploadHandle")?;
+    let handle = js_value_to_safe_u64(cx, handle_value, "bodyStream.uploadHandle")?;
+    let length = stream
+        .get_opt::<JsValue, _, _>(cx, "length")?
+        .map(|value| js_value_to_safe_u64(cx, value, "bodyStream.length"))
+        .transpose()?;
+    let receiver =
+        take_upload_receiver(handle).or_else(|error| cx.throw_error(error.to_string()))?;
+
+    Ok(Some(RequestBody::Stream { receiver, length }))
+}
+
 pub(crate) fn js_object_to_request_options(
     cx: &mut FunctionContext,
     obj: Handle<JsObject>,
@@ -571,16 +591,22 @@ pub(crate) fn js_object_to_request_options(
         .map(|value| js_value_to_bytes(cx, value))
         .transpose()?;
     let multipart = js_object_to_multipart_body(cx, obj)?;
+    let stream = js_object_to_stream_body(cx, obj)?;
+
+    let body_count = usize::from(body_bytes.is_some())
+        + usize::from(multipart.is_some())
+        + usize::from(stream.is_some());
 
-    if body_bytes.is_some() && multipart.is_some() {
-        return cx.throw_type_error("body and multipart cannot both be provided");
+    if body_count > 1 {
+        return cx.throw_type_error("body, bodyStream, and multipart are mutually exclusive");
     }
 
-    let body = match (body_bytes, multipart) {
-        (Some(bytes), None) => Some(RequestBody::Bytes(bytes)),
-        (None, Some(multipart)) => Some(RequestBody::Multipart(multipart)),
-        (None, None) => None,
-        (Some(_), Some(_)) => unreachable!(),
+    let body = match (body_bytes, multipart, stream) {
+        (Some(bytes), None, None) => Some(RequestBody::Bytes(bytes)),
+        (None, Some(multipart), None) => Some(RequestBody::Multipart(multipart)),
+        (None, None, Some(stream)) => Some(stream),
+        (None, None, None) => None,
+        _ => unreachable!(),
     };
 
     let proxy = obj
diff --git a/rust/src/napi/websocket.rs b/rust/src/napi/websocket.rs
index bf8b4ef..a6d9f4b 100644
--- a/rust/src/napi/websocket.rs
+++ b/rust/src/napi/websocket.rs
@@ -1,21 +1,35 @@
 use crate::napi::convert::{js_object_to_websocket_options, websocket_to_js_object};
+use crate::store::runtime::runtime;
+use crate::store::websocket_connect_store::{
+    cancel_websocket_connect, insert_websocket_connect, remove_websocket_connect,
+};
 use crate::store::websocket_store::{
     close_websocket, read_websocket_message, send_websocket_binary, send_websocket_text,
+    terminate_websocket,
 };
-use crate::transport::{connect_websocket, types::WebSocketReadResult};
+use crate::transport::{make_websocket, types::WebSocketReadResult};
 use neon::prelude::*;
 use neon::types::buffer::TypedArray;
 use neon::types::JsBuffer;
 
-fn websocket_connect_js(mut cx: FunctionContext) -> JsResult<JsPromise> {
+fn websocket_connect_js(mut cx: FunctionContext) -> JsResult<JsObject> {
     let options_obj = cx.argument::<JsObject>(0)?;
     let options = js_object_to_websocket_options(&mut cx, options_obj)?;
 
     let channel = cx.channel();
     let (deferred, promise) = cx.promise();
+    let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
+    let handle = insert_websocket_connect(cancel_tx);
 
     std::thread::spawn(move || {
-        let result = connect_websocket(options);
+        let result = runtime().block_on(async move {
+            tokio::select! {
+                result = make_websocket(options) => result,
+                _ = cancel_rx => Err(anyhow::anyhow!("WebSocket connection aborted")),
+            }
+        });
+
+        remove_websocket_connect(handle);
 
         deferred.settle_with(&channel, move |mut cx| match result {
             Ok(websocket) => websocket_to_js_object(&mut cx, websocket),
@@ -23,7 +37,19 @@ fn websocket_connect_js(mut cx: FunctionContext) -> JsResult<JsPromise> {
         });
     });
 
-    Ok(promise)
+    let result = JsObject::new(&mut cx);
+    let handle_value = cx.number(handle as f64);
+
+    result.set(&mut cx, "handle", handle_value)?;
+    result.set(&mut cx, "promise", promise)?;
+
+    Ok(result)
+}
+
+fn websocket_cancel_connect_js(mut cx: FunctionContext) -> JsResult<JsBoolean> {
+    let handle = cx.argument::<JsNumber>(0)?.value(&mut cx) as u64;
+
+    Ok(cx.boolean(cancel_websocket_connect(handle)))
 }
 
 fn websocket_read_js(mut cx: FunctionContext) -> JsResult<JsPromise> {
@@ -140,11 +166,19 @@ fn websocket_close_js(mut cx: FunctionContext) -> JsResult<JsPromise> {
     Ok(promise)
 }
 
+fn websocket_terminate_js(mut cx: FunctionContext) -> JsResult<JsBoolean> {
+    let handle = cx.argument::<JsNumber>(0)?.value(&mut cx) as u64;
+
+    Ok(cx.boolean(terminate_websocket(handle)))
+}
+
 pub fn register(cx: &mut ModuleContext) -> NeonResult<()> {
     cx.export_function("websocketConnect", websocket_connect_js)?;
+    cx.export_function("websocketCancelConnect", websocket_cancel_connect_js)?;
     cx.export_function("websocketRead", websocket_read_js)?;
     cx.export_function("websocketSendText", websocket_send_text_js)?;
     cx.export_function("websocketSendBinary", websocket_send_binary_js)?;
     cx.export_function("websocketClose", websocket_close_js)?;
+    cx.export_function("websocketTerminate", websocket_terminate_js)?;
     Ok(())
 }
diff --git a/rust/src/store/mod.rs b/rust/src/store/mod.rs
index a1fd087..c5d4b3d 100644
--- a/rust/src/store/mod.rs
+++ b/rust/src/store/mod.rs
@@ -3,4 +3,5 @@ pub mod client_store;
 pub mod request_store;
 pub mod runtime;
 pub mod upload_store;
+pub mod websocket_connect_store;
 pub mod websocket_store;
diff --git a/rust/src/store/websocket_connect_store.rs b/rust/src/store/websocket_connect_store.rs
new file mode 100644
index 0000000..0f74c64
--- /dev/null
+++ b/rust/src/store/websocket_connect_store.rs
@@ -0,0 +1,42 @@
+use std::collections::HashMap;
+use std::sync::{
+    atomic::{AtomicU64, Ordering},
+    Mutex, OnceLock,
+};
+
+static NEXT_WEBSOCKET_CONNECT_HANDLE: AtomicU64 = AtomicU64::new(1);
+static WEBSOCKET_CONNECT_STORE: OnceLock<Mutex<HashMap<u64, tokio::sync::oneshot::Sender<()>>>> =
+    OnceLock::new();
+
+fn websocket_connect_store() -> &'static Mutex<HashMap<u64, tokio::sync::oneshot::Sender<()>>> {
+    WEBSOCKET_CONNECT_STORE.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+pub fn insert_websocket_connect(cancel: tokio::sync::oneshot::Sender<()>) -> u64 {
+    let handle = NEXT_WEBSOCKET_CONNECT_HANDLE.fetch_add(1, Ordering::Relaxed);
+
+    websocket_connect_store()
+        .lock()
+        .expect("websocket connect store poisoned")
+        .insert(handle, cancel);
+
+    handle
+}
+
+pub fn remove_websocket_connect(handle: u64) {
+    websocket_connect_store()
+        .lock()
+        .expect("websocket connect store poisoned")
+        .remove(&handle);
+}
+
+pub fn cancel_websocket_connect(handle: u64) -> bool {
+    let cancel = websocket_connect_store()
+        .lock()
+        .expect("websocket connect store poisoned")
+        .remove(&handle);
+
+    cancel
+        .map(|cancel| cancel.send(()).is_ok())
+        .unwrap_or(false)
+}
diff --git a/rust/src/store/websocket_store.rs b/rust/src/store/websocket_store.rs
index 328ff89..d98a5b5 100644
--- a/rust/src/store/websocket_store.rs
+++ b/rust/src/store/websocket_store.rs
@@ -8,17 +8,24 @@ use std::sync::{
 
 #[derive(Debug)]
 pub(crate) enum WebSocketCommand {
-    Text(String),
-    Binary(Vec<u8>),
+    Text {
+        text: String,
+        ack: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
+    },
+    Binary {
+        bytes: Vec<u8>,
+        ack: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
+    },
     Close {
         code: Option<u16>,
         reason: Option<String>,
+        ack: tokio::sync::oneshot::Sender<std::result::Result<(), String>>,
     },
 }
 
 #[derive(Debug)]
 pub(crate) struct StoredWebSocket {
-    pub commands: tokio::sync::mpsc::UnboundedSender<WebSocketCommand>,
+    pub commands: tokio::sync::mpsc::Sender<WebSocketCommand>,
     pub events: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<WebSocketReadResult>>,
 }
 
@@ -32,7 +39,7 @@ fn websocket_store() -> &'static Mutex<HashMap<u64, SharedWebSocket>> {
 }
 
 pub(crate) fn insert_websocket(
-    commands: tokio::sync::mpsc::UnboundedSender<WebSocketCommand>,
+    commands: tokio::sync::mpsc::Sender<WebSocketCommand>,
     events: tokio::sync::mpsc::UnboundedReceiver<WebSocketReadResult>,
 ) -> u64 {
     let handle = NEXT_WEBSOCKET_HANDLE.fetch_add(1, Ordering::Relaxed);
@@ -62,11 +69,16 @@ fn get_websocket(handle: u64) -> Result<SharedWebSocket> {
         .ok_or_else(|| anyhow::anyhow!("Unknown websocket handle: {}", handle))
 }
 
-pub(crate) fn remove_websocket(handle: u64) {
+pub(crate) fn remove_websocket(handle: u64) -> bool {
     websocket_store()
         .lock()
         .expect("websocket store poisoned")
-        .remove(&handle);
+        .remove(&handle)
+        .is_some()
+}
+
+pub fn terminate_websocket(handle: u64) -> bool {
+    remove_websocket(handle)
 }
 
 pub fn read_websocket_message(handle: u64) -> Result<WebSocketReadResult> {
@@ -87,27 +99,49 @@ pub fn read_websocket_message(handle: u64) -> Result<WebSocketReadResult> {
     result
 }
 
-fn send_websocket_command(handle: u64, command: WebSocketCommand) -> Result<()> {
+fn send_websocket_command(
+    handle: u64,
+    command: WebSocketCommand,
+    acknowledgement: tokio::sync::oneshot::Receiver<std::result::Result<(), String>>,
+) -> Result<()> {
     let websocket = get_websocket(handle)?;
+    let result = crate::store::runtime::runtime().block_on(async {
+        websocket
+            .commands
+            .send(command)
+            .await
+            .map_err(|_| anyhow::anyhow!("WebSocket is already closed"))?;
 
-    if websocket.commands.send(command).is_err() {
+        acknowledgement
+            .await
+            .map_err(|_| anyhow::anyhow!("WebSocket send acknowledgement was dropped"))?
+            .map_err(anyhow::Error::msg)
+    });
+
+    if result.is_err() {
         remove_websocket(handle);
-        anyhow::bail!("WebSocket is already closed");
     }
 
-    Ok(())
+    result
 }
 
 pub fn send_websocket_text(handle: u64, text: String) -> Result<()> {
-    send_websocket_command(handle, WebSocketCommand::Text(text))
+    let (ack, result) = tokio::sync::oneshot::channel();
+    send_websocket_command(handle, WebSocketCommand::Text { text, ack }, result)
 }
 
 pub fn send_websocket_binary(handle: u64, bytes: Vec<u8>) -> Result<()> {
-    send_websocket_command(handle, WebSocketCommand::Binary(bytes))
+    let (ack, result) = tokio::sync::oneshot::channel();
+    send_websocket_command(handle, WebSocketCommand::Binary { bytes, ack }, result)
 }
 
 pub fn close_websocket(handle: u64, code: Option<u16>, reason: Option<String>) -> Result<()> {
-    send_websocket_command(handle, WebSocketCommand::Close { code, reason })
+    let (ack, result) = tokio::sync::oneshot::channel();
+    send_websocket_command(
+        handle,
+        WebSocketCommand::Close { code, reason, ack },
+        result,
+    )
 }
 
 #[cfg(test)]
@@ -116,7 +150,7 @@ mod tests {
 
     #[test]
     fn removes_websocket_when_event_stream_closes() {
-        let (commands, _command_receiver) = tokio::sync::mpsc::unbounded_channel();
+        let (commands, _command_receiver) = tokio::sync::mpsc::channel(1);
         let (event_sender, events) = tokio::sync::mpsc::unbounded_channel();
         let handle = insert_websocket(commands, events);
 
@@ -133,7 +167,7 @@ mod tests {
 
     #[test]
     fn removes_websocket_when_command_stream_closes() {
-        let (commands, command_receiver) = tokio::sync::mpsc::unbounded_channel();
+        let (commands, command_receiver) = tokio::sync::mpsc::channel(1);
         let (_event_sender, events) = tokio::sync::mpsc::unbounded_channel();
         let handle = insert_websocket(commands, events);
 
diff --git a/rust/src/transport/mod.rs b/rust/src/transport/mod.rs
index f91ed73..7448c31 100644
--- a/rust/src/transport/mod.rs
+++ b/rust/src/transport/mod.rs
@@ -8,4 +8,4 @@ pub mod types;
 mod websocket;
 
 pub use request::make_request;
-pub use websocket::connect_websocket;
+pub(crate) use websocket::make_websocket;
diff --git a/rust/src/transport/request.rs b/rust/src/transport/request.rs
index cc625c3..17b09a2 100644
--- a/rust/src/transport/request.rs
+++ b/rust/src/transport/request.rs
@@ -106,6 +106,19 @@ pub async fn make_request(options: RequestOptions) -> Result<Response> {
     if let Some(body) = body {
         request = match body {
             RequestBody::Bytes(bytes) => request.body(bytes),
+            RequestBody::Stream { receiver, length } => {
+                if let Some(length) = length {
+                    let has_content_length = headers
+                        .iter()
+                        .any(|(name, _)| name.eq_ignore_ascii_case("content-length"));
+
+                    if !has_content_length {
+                        request = request.header("content-length", length);
+                    }
+                }
+
+                request.body(Body::wrap_stream(ReceiverStream::new(receiver)))
+            }
             RequestBody::Multipart(options) => request.multipart(build_multipart(options)?),
         };
     }
diff --git a/rust/src/transport/types.rs b/rust/src/transport/types.rs
index b8ed124..d807f79 100644
--- a/rust/src/transport/types.rs
+++ b/rust/src/transport/types.rs
@@ -77,6 +77,10 @@ pub enum ConnectionGroup {
 #[derive(Debug)]
 pub enum RequestBody {
     Bytes(Vec<u8>),
+    Stream {
+        receiver: UploadReceiver,
+        length: Option<u64>,
+    },
     Multipart(MultipartBodyOptions),
 }
 
diff --git a/rust/src/transport/websocket.rs b/rust/src/transport/websocket.rs
index 030f9d7..478d0df 100644
--- a/rust/src/transport/websocket.rs
+++ b/rust/src/transport/websocket.rs
@@ -8,25 +8,24 @@ use wreq::ws::message::{CloseCode, CloseFrame, Message};
 use wreq::ws::WebSocket;
 use wreq::Version;
 
-pub fn connect_websocket(options: WebSocketConnectOptions) -> Result<WebSocketConnection> {
-    runtime().block_on(make_websocket(options))
-}
-
 async fn run_websocket_task(
     mut websocket: WebSocket,
-    mut commands: tokio::sync::mpsc::UnboundedReceiver<WebSocketCommand>,
+    mut commands: tokio::sync::mpsc::Receiver<WebSocketCommand>,
     events: tokio::sync::mpsc::UnboundedSender<WebSocketReadResult>,
 ) {
-    let mut close_requested = false;
-    let mut requested_close_code = 1000;
-    let mut requested_close_reason = String::new();
-
     loop {
         tokio::select! {
             command = commands.recv() => {
                 match command {
-                    Some(WebSocketCommand::Text(text)) => {
-                        if websocket.send(Message::Text(text.into())).await.is_err() {
+                    Some(WebSocketCommand::Text { text, ack }) => {
+                        let result = websocket
+                            .send(Message::Text(text.into()))
+                            .await
+                            .map_err(|error| error.to_string());
+                        let failed = result.is_err();
+                        let _ = ack.send(result);
+
+                        if failed {
                             let _ = events.send(WebSocketReadResult::Close {
                                 code: 1006,
                                 reason: String::new(),
@@ -35,8 +34,15 @@ async fn run_websocket_task(
                             break;
                         }
                     }
-                    Some(WebSocketCommand::Binary(bytes)) => {
-                        if websocket.send(Message::Binary(bytes.into())).await.is_err() {
+                    Some(WebSocketCommand::Binary { bytes, ack }) => {
+                        let result = websocket
+                            .send(Message::Binary(bytes.into()))
+                            .await
+                            .map_err(|error| error.to_string());
+                        let failed = result.is_err();
+                        let _ = ack.send(result);
+
+                        if failed {
                             let _ = events.send(WebSocketReadResult::Close {
                                 code: 1006,
                                 reason: String::new(),
@@ -45,17 +51,23 @@ async fn run_websocket_task(
                             break;
                         }
                     }
-                    Some(WebSocketCommand::Close { code, reason }) => {
-                        close_requested = true;
-                        requested_close_code = code.unwrap_or(1000);
-                        requested_close_reason = reason.unwrap_or_default();
+                    Some(WebSocketCommand::Close { code, reason, ack }) => {
+                        let reason = reason.unwrap_or_default();
+
+                        let frame = if code.is_none() && reason.is_empty() {
+                            Message::Close(None)
+                        } else {
+                            Message::Close(Some(CloseFrame {
+                                code: CloseCode::from(code.unwrap_or(1000)),
+                                reason: reason.into(),
+                            }))
+                        };
 
-                        let frame = Message::Close(Some(CloseFrame {
-                            code: CloseCode::from(requested_close_code),
-                            reason: requested_close_reason.clone().into(),
-                        }));
+                        let result = websocket.send(frame).await.map_err(|error| error.to_string());
+                        let failed = result.is_err();
+                        let _ = ack.send(result);
 
-                        if websocket.send(frame).await.is_err() {
+                        if failed {
                             let _ = events.send(WebSocketReadResult::Close {
                                 code: 1006,
                                 reason: String::new(),
@@ -84,13 +96,7 @@ async fn run_websocket_task(
                     Some(Ok(Message::Close(frame))) => {
                         let (code, reason) = match frame {
                             Some(frame) => (u16::from(frame.code), frame.reason.to_string()),
-                            None => {
-                                if close_requested {
-                                    (requested_close_code, requested_close_reason.clone())
-                                } else {
-                                    (1005, String::new())
-                                }
-                            }
+                            None => (1005, String::new()),
                         };
 
                         let _ = events.send(WebSocketReadResult::Close {
@@ -111,17 +117,9 @@ async fn run_websocket_task(
                     }
                     None => {
                         let _ = events.send(WebSocketReadResult::Close {
-                            code: if close_requested {
-                                requested_close_code
-                            } else {
-                                1006
-                            },
-                            reason: if close_requested {
-                                requested_close_reason.clone()
-                            } else {
-                                String::new()
-                            },
-                            was_clean: close_requested,
+                            code: 1006,
+                            reason: String::new(),
+                            was_clean: false,
                         });
                         break;
                     }
@@ -131,7 +129,9 @@ async fn run_websocket_task(
     }
 }
 
-async fn make_websocket(options: WebSocketConnectOptions) -> Result<WebSocketConnection> {
+pub(crate) async fn make_websocket(
+    options: WebSocketConnectOptions,
+) -> Result<WebSocketConnection> {
     let client = websocket_client(&options).await?;
     let WebSocketConnectOptions {
         client_id: _,
@@ -232,7 +232,7 @@ async fn make_websocket(options: WebSocketConnectOptions) -> Result<WebSocketCon
         .and_then(|value| value.to_str().ok())
         .map(str::to_owned);
 
-    let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (command_tx, command_rx) = tokio::sync::mpsc::channel(1);
     let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel();
     runtime().spawn(run_websocket_task(websocket, command_rx, event_tx));
     let handle = insert_websocket(command_tx, event_rx);
diff --git a/scripts/postbuild.mjs b/scripts/postbuild.mjs
index a8bf75b..e96f205 100644
--- a/scripts/postbuild.mjs
+++ b/scripts/postbuild.mjs
@@ -11,6 +11,7 @@ const runtimeExports = [
   'getProfiles',
   'BROWSER_PROFILES',
   'Headers',
+  'Request',
   'Response',
   'RequestError',
   'HTTPError',
diff --git a/src/headers/index.ts b/src/headers/index.ts
index ff3f196..fac90f1 100644
--- a/src/headers/index.ts
+++ b/src/headers/index.ts
@@ -37,8 +37,14 @@ export class Headers implements Iterable<HeaderTuple> {
       return;
     }
 
-    if (Array.isArray(init) || isIterable<HeaderTuple>(init)) {
-      for (const [name, value] of init as Iterable<HeaderTuple>) {
+    if (Array.isArray(init) || isIterable<readonly string[]>(init)) {
+      for (const entry of init as Iterable<readonly string[]>) {
+        if (!Array.isArray(entry) || entry.length !== 2) {
+          throw new TypeError('Each header tuple must contain exactly two items');
+        }
+
+        const [name, value] = entry;
+
         this.append(name, value);
       }
 
@@ -179,6 +185,30 @@ export class Headers implements Iterable<HeaderTuple> {
     }
   }
 
+  /** Iterates over header names. */
+  *keys(): IterableIterator<string> {
+    for (const [name] of this.entries()) {
+      yield name;
+    }
+  }
+
+  /** Iterates over combined header values. */
+  *values(): IterableIterator<string> {
+    for (const [, value] of this.entries()) {
+      yield value;
+    }
+  }
+
+  /** Invokes `callback` once for every normalized header entry. */
+  forEach(
+    callback: (value: string, key: string, parent: Headers) => void,
+    thisArg?: unknown
+  ): void {
+    for (const [name, value] of this.entries()) {
+      callback.call(thisArg, value, name, this);
+    }
+  }
+
   /** Iterates over normalized `[name, value]` header entries. */
   [Symbol.iterator](): IterableIterator<HeaderTuple> {
     return this.entries();
diff --git a/src/http/body/bytes.ts b/src/http/body/bytes.ts
index 4bd756a..cbcb04d 100644
--- a/src/http/body/bytes.ts
+++ b/src/http/body/bytes.ts
@@ -1,4 +1,9 @@
-import type { BodyInit, NativeMultipartStreamPart, NativeMultipartUpload } from '../../types';
+import type {
+  BodyInit,
+  NativeBodyStreamUpload,
+  NativeMultipartStreamPart,
+  NativeMultipartUpload,
+} from '../../types';
 import { Buffer } from 'node:buffer';
 import { randomBytes } from 'node:crypto';
 import {
@@ -18,6 +23,84 @@ export function isFormDataBody(body: BodyInit | null | undefined): body is FormD
   return typeof FormData !== 'undefined' && body instanceof FormData;
 }
 
+export function isBlobBody(body: BodyInit | null | undefined): body is Blob {
+  return typeof Blob !== 'undefined' && body instanceof Blob;
+}
+
+export function isReadableStreamBody(
+  body: BodyInit | null | undefined
+): body is ReadableStream<Uint8Array> {
+  return typeof ReadableStream !== 'undefined' && body instanceof ReadableStream;
+}
+
+/** Converts Fetch-compatible iterable bodies into a web ReadableStream. */
+export function toReadableStreamBody(
+  body: BodyInit | null | undefined
+): ReadableStream<Uint8Array> | undefined {
+  if (isReadableStreamBody(body)) {
+    return body;
+  }
+
+  if (
+    body === undefined ||
+    body === null ||
+    typeof body === 'string' ||
+    isBlobBody(body) ||
+    isFormDataBody(body) ||
+    body instanceof URLSearchParams ||
+    body instanceof ArrayBuffer ||
+    ArrayBuffer.isView(body)
+  ) {
+    return undefined;
+  }
+
+  if (Symbol.asyncIterator in Object(body)) {
+    const iterator = (body as AsyncIterable<Uint8Array>)[Symbol.asyncIterator]();
+
+    return new ReadableStream<Uint8Array>(
+      {
+        pull: async (controller) => {
+          const result = await iterator.next();
+
+          if (result.done) {
+            controller.close();
+          } else {
+            controller.enqueue(result.value);
+          }
+        },
+        cancel: async (reason) => {
+          await iterator.return?.(reason);
+        },
+      },
+      { highWaterMark: 0 }
+    );
+  }
+
+  if (Symbol.iterator in Object(body)) {
+    const iterator = (body as Iterable<Uint8Array>)[Symbol.iterator]();
+
+    return new ReadableStream<Uint8Array>(
+      {
+        pull: (controller) => {
+          const result = iterator.next();
+
+          if (result.done) {
+            controller.close();
+          } else {
+            controller.enqueue(result.value);
+          }
+        },
+        cancel: (reason) => {
+          iterator.return?.(reason);
+        },
+      },
+      { highWaterMark: 0 }
+    );
+  }
+
+  return undefined;
+}
+
 export function cloneFormData(body: FormData): FormData {
   const cloned = new FormData();
 
@@ -327,6 +410,190 @@ export class MultipartBody {
   }
 }
 
+/** Lazily consumed raw request body backed by a Blob or ReadableStream. */
+export class StreamingBody {
+  readonly contentLength?: number;
+  readonly contentType?: string;
+  #source: Blob | ReadableStream<Uint8Array>;
+  #stream: ReadableStream<Uint8Array> | null = null;
+  #reader: ReadableStreamDefaultReader<Uint8Array> | null = null;
+  #used = false;
+
+  constructor(source: Blob | ReadableStream<Uint8Array>) {
+    this.#source = source;
+
+    if (isBlobBody(source)) {
+      this.contentLength = source.size;
+      this.contentType = source.type || undefined;
+    }
+  }
+
+  get bodyUsed(): boolean {
+    return this.#used;
+  }
+
+  get locked(): boolean {
+    return this.#stream?.locked === true;
+  }
+
+  get stream(): ReadableStream<Uint8Array> {
+    this.#stream ??= new ReadableStream<Uint8Array>(
+      {
+        pull: async (controller) => {
+          this.#used = true;
+          this.#reader ??= this.#openSource().getReader();
+
+          const result = await this.#reader.read();
+
+          if (result.done) {
+            controller.close();
+
+            return;
+          }
+
+          if (!(result.value instanceof Uint8Array)) {
+            throw new TypeError('Request body stream must produce Uint8Array chunks');
+          }
+
+          controller.enqueue(result.value);
+        },
+        cancel: async (reason) => {
+          this.#used = true;
+          this.#reader ??= this.#openSource().getReader();
+
+          await this.#reader.cancel(reason);
+        },
+      },
+      { highWaterMark: 0 }
+    );
+
+    return this.#stream;
+  }
+
+  clone(): StreamingBody {
+    if (isBlobBody(this.#source)) {
+      return new StreamingBody(this.#source.slice(0, this.#source.size, this.#source.type));
+    }
+
+    if (this.#used || this.locked || this.#reader) {
+      throw new TypeError('Request body is already used');
+    }
+
+    const [left, right] = this.#source.tee();
+
+    this.#source = left;
+
+    return new StreamingBody(right);
+  }
+
+  transfer(): StreamingBody {
+    if (this.#used || this.locked || this.#reader) {
+      throw new TypeError('Request body is already used');
+    }
+
+    const transferred = new StreamingBody(this.#source);
+
+    this.#used = true;
+
+    return transferred;
+  }
+
+  prepareNativeUpload(): NativeBodyStreamUpload {
+    const replayableBlob = isBlobBody(this.#source);
+
+    if (!replayableBlob && (this.#used || this.locked)) {
+      throw new TypeError('Request body is already used');
+    }
+
+    const handle = nativeCreateUpload();
+    let started = false;
+    let cancelled = false;
+    let reader: ReadableStreamDefaultReader<Uint8Array> | null = null;
+
+    const cancel = (reason?: unknown) => {
+      if (cancelled) {
+        return;
+      }
+
+      cancelled = true;
+      void reader?.cancel(reason).catch(() => undefined);
+      nativeFinishUpload(handle);
+    };
+
+    return {
+      body: {
+        uploadHandle: handle,
+        length: this.contentLength,
+      },
+      cancel,
+      start: async (signal?: AbortSignal | null) => {
+        if (started) {
+          throw new TypeError('Request body upload has already started');
+        }
+
+        started = true;
+
+        if (cancelled || signal?.aborted) {
+          cancel(signal?.reason);
+
+          return;
+        }
+
+        try {
+          if (replayableBlob) {
+            this.#used = true;
+            reader = (this.#source as Blob).stream().getReader();
+          } else {
+            reader = this.stream.getReader();
+          }
+
+          while (true) {
+            if (cancelled || signal?.aborted) {
+              cancel(signal?.reason);
+
+              return;
+            }
+
+            const result = await reader.read();
+
+            if (result.done) {
+              nativeFinishUpload(handle);
+
+              return;
+            }
+
+            await nativeWriteUploadChunk(handle, result.value);
+          }
+        } catch (error) {
+          if (cancelled) {
+            return;
+          }
+
+          try {
+            await nativeFailUpload(handle, error);
+          } catch {
+            // The native request may already have closed its receiver.
+          }
+
+          throw error;
+        } finally {
+          if (reader) {
+            try {
+              reader.releaseLock();
+            } catch {
+              // Cancellation may still be settling an outstanding read.
+            }
+          }
+        }
+      },
+    };
+  }
+
+  #openSource(): ReadableStream<Uint8Array> {
+    return isBlobBody(this.#source) ? this.#source.stream() : this.#source;
+  }
+}
+
 export function createMultipartRequest(body: FormData, boundary?: string): MultipartBody {
   if (typeof globalThis.Request === 'undefined') {
     throw new TypeError('multipart/form-data requests require global Request support');
@@ -379,6 +646,14 @@ export function cloneBodyInit(body: BodyInit | null | undefined): BodyInit | nul
     return cloneFormData(body);
   }
 
+  if (isBlobBody(body)) {
+    return body.slice(0, body.size, body.type);
+  }
+
+  if (isReadableStreamBody(body)) {
+    return body;
+  }
+
   if (typeof body === 'string') {
     return body;
   }
diff --git a/src/http/fetch.ts b/src/http/fetch.ts
index 6be194c..ad230d6 100644
--- a/src/http/fetch.ts
+++ b/src/http/fetch.ts
@@ -23,11 +23,21 @@ import {
   toRedirectEntry,
 } from './pipeline/redirects';
 import { runRetryDelay, shouldRetryRequest } from './pipeline/retries';
-import { cancelResponseBody } from './response';
+import { cancelResponseBody, Response } from './response';
 
 /** Performs an HTTP request using the native transport pipeline. */
-export async function fetch(input: RequestInput, init?: WreqInit) {
-  return fetchWithNativeClient(input, init);
+export function fetch(
+  input: string | URL | globalThis.Request,
+  init?: globalThis.RequestInit
+): Promise<Response>;
+
+export function fetch(input: RequestInput, init?: WreqInit): Promise<Response>;
+
+export async function fetch(
+  input: RequestInput,
+  init?: WreqInit | globalThis.RequestInit
+): Promise<Response> {
+  return fetchWithNativeClient(input, init as WreqInit | undefined);
 }
 
 /** @internal Performs a request using a reusable native client owner. */
@@ -145,16 +155,23 @@ export async function fetchWithNativeClient(
 
         const rewritten = rewriteRedirectMethod(normalizeMethod(request.method), response.status);
 
-        const nextRequest = rewritten.bodyDropped
-          ? request._replace({
-              url: nextUrl,
-              method: rewritten.method,
-              body: null,
-            })
-          : request._replace({
-              url: nextUrl,
-              method: rewritten.method,
-            });
+        let nextRequest;
+
+        try {
+          nextRequest = rewritten.bodyDropped
+            ? request._replace({
+                url: nextUrl,
+                method: rewritten.method,
+                body: null,
+              })
+            : request._replace({
+                url: nextUrl,
+                method: rewritten.method,
+              });
+        } catch (error) {
+          await cancelResponseBody(response);
+          throw error;
+        }
 
         stripRedirectSensitiveHeaders(
           nextRequest.headers,
diff --git a/src/http/pipeline/input.ts b/src/http/pipeline/input.ts
index bc073c0..47fec1b 100644
--- a/src/http/pipeline/input.ts
+++ b/src/http/pipeline/input.ts
@@ -1,5 +1,4 @@
 import type { RequestInput, WreqInit } from '../../types';
-import { Buffer } from 'node:buffer';
 import { RequestError } from '../../errors';
 import { Request } from '../request';
 
@@ -19,6 +18,15 @@ export async function mergeInputAndInit(
       throw new TypeError('Request body is already used');
     }
 
+    if (
+      input instanceof Request &&
+      init?.body === undefined &&
+      input.body !== null &&
+      ['GET', 'HEAD'].includes((init?.method ?? input.method).toUpperCase())
+    ) {
+      throw new TypeError('Request with GET/HEAD method cannot have body.');
+    }
+
     return {
       urlInput: input instanceof Request ? input.url : input,
       init:
@@ -28,7 +36,7 @@ export async function mergeInputAndInit(
               method: init?.method ?? input.method,
               headers: init?.headers ?? input.headers,
               signal: init?.signal ?? input.signal ?? undefined,
-              body: init?.body !== undefined ? init.body : (input._cloneBodyInit() ?? undefined),
+              body: init?.body !== undefined ? init.body : (input._takeBodyInit() ?? undefined),
               multipartBoundary:
                 init?.multipartBoundary ??
                 (init?.body === undefined ? input._getMultipartBoundary() : undefined),
@@ -45,7 +53,11 @@ export async function mergeInputAndInit(
     let body = init?.body;
 
     if (body === undefined && input.body !== null) {
-      body = Buffer.from(await input.arrayBuffer());
+      const transferred = new globalThis.Request(input, {
+        method: init?.method ?? input.method,
+      });
+
+      body = transferred.body as ReadableStream<Uint8Array>;
     }
 
     return {
diff --git a/src/http/pipeline/options.ts b/src/http/pipeline/options.ts
index 567bceb..ecc2d81 100644
--- a/src/http/pipeline/options.ts
+++ b/src/http/pipeline/options.ts
@@ -146,9 +146,12 @@ export async function buildNativeRequest(
   }
 
   const multipartUpload = request._prepareMultipartUpload();
+  const bodyStreamUpload = multipartUpload ? undefined : request._prepareBodyStreamUpload();
 
   try {
-    const body = multipartUpload ? undefined : await request._getBodyBytesForDispatch();
+    const body =
+      multipartUpload || bodyStreamUpload ? undefined : await request._getBodyBytesForDispatch();
+
     const nativeOptions: NativeRequestOptions = {
       url: request.url,
       method: normalizeMethod(request.method),
@@ -157,6 +160,8 @@ export async function buildNativeRequest(
       body: body ? Buffer.from(body) : undefined,
       multipart: multipartUpload?.body,
       multipartUpload,
+      bodyStream: bodyStreamUpload?.body,
+      bodyStreamUpload,
       ...browser,
       emulationJson: serializeEmulationOptions(options),
       proxy,
@@ -190,6 +195,7 @@ export async function buildNativeRequest(
     return nativeOptions;
   } catch (error) {
     multipartUpload?.cancel(error);
+    bodyStreamUpload?.cancel(error);
     throw error;
   }
 }
diff --git a/src/http/request.ts b/src/http/request.ts
index 46fa262..2c717ac 100644
--- a/src/http/request.ts
+++ b/src/http/request.ts
@@ -1,93 +1,170 @@
 import type { BodyInit, HeadersInit, NativeMultipartUpload, WreqInit } from '../types';
-import { Blob, Buffer } from 'node:buffer';
+import { Buffer } from 'node:buffer';
 import { ReadableStream } from 'node:stream/web';
 import { Headers } from '../headers';
+import { normalizeMethod } from '../native';
 import {
   cloneBodyInit,
   cloneBytes,
   createMultipartRequest,
+  isBlobBody,
   isFormDataBody,
+  isReadableStreamBody,
   MultipartBody,
+  StreamingBody,
+  toReadableStreamBody,
   toBodyBytes,
 } from './body/bytes';
+import { parseResponseFormData } from './body/form-data';
+
+function isGlobalRequest(value: unknown): value is globalThis.Request {
+  return (
+    typeof globalThis.Request !== 'undefined' &&
+    value instanceof globalThis.Request &&
+    !(value instanceof Request)
+  );
+}
 
 /** WHATWG-style request wrapper used by the public API. */
 export class Request {
   /** Fully resolved request URL. */
   readonly url: string;
-  /** Uppercased HTTP method. */
+  /** Normalized HTTP method. */
   readonly method: string;
   /** Request headers. */
   readonly headers: Headers;
   /** Abort signal associated with the request, if any. */
-  readonly signal: AbortSignal | null;
+  readonly signal: AbortSignal;
+  /** Cache mode exposed for Fetch API compatibility. */
+  readonly cache: globalThis.Request['cache'];
+  /** Credentials mode exposed for Fetch API compatibility. */
+  readonly credentials: globalThis.Request['credentials'];
+  /** Request destination exposed for Fetch API compatibility. */
+  readonly destination: globalThis.Request['destination'];
+  /** Subresource integrity metadata. */
+  readonly integrity: string;
+  /** Whether the request is eligible to outlive its initiating context. */
+  readonly keepalive: boolean;
+  /** Fetch mode exposed for compatibility. */
+  readonly mode: globalThis.Request['mode'];
+  /** Redirect mode used by the request. */
+  readonly redirect: globalThis.Request['redirect'];
+  /** Request referrer. */
+  readonly referrer: string;
+  /** Referrer policy used by the request. */
+  readonly referrerPolicy: globalThis.Request['referrerPolicy'];
+  /** Streaming request duplex mode. */
+  readonly duplex = 'half' as const;
   #bodyBytes: Uint8Array | null;
   #multipartBody: MultipartBody | null;
+  #streamingBody: StreamingBody | null;
   #bodyUsed = false;
   #stream: ReadableStream<Uint8Array> | null = null;
+  #transferredBodyReaders: ReadableStreamDefaultReader<Uint8Array>[] = [];
 
-  constructor(input: string | URL | Request, init: WreqInit = {}) {
-    if (input instanceof Request) {
-      if (input.bodyUsed) {
-        throw new TypeError('Request body is already used');
-      }
+  constructor(input: string | URL | Request | globalThis.Request, init: WreqInit = {}) {
+    const inputRequest = input instanceof Request || isGlobalRequest(input) ? input : undefined;
 
-      this.url = String(init.baseURL ? new URL(input.url, init.baseURL) : input.url);
-      this.method = (init.method ?? input.method).toUpperCase();
-      this.headers = new Headers(init.headers ?? input.headers);
-      this.signal = init.signal ?? input.signal ?? null;
-      this.#bodyBytes = null;
-      this.#multipartBody = null;
+    if (inputRequest?.bodyUsed && init.body === undefined) {
+      throw new TypeError('Request body is already used');
+    }
 
-      if (init.body !== undefined) {
-        this.#setBody(init.body, init.multipartBoundary);
-      } else {
-        this.#bodyBytes = cloneBytes(input.#bodyBytes);
-        this.#multipartBody = input.#multipartBody?.clone() ?? null;
+    const inputUrl = inputRequest?.url ?? String(input);
+    let parsedUrl: URL;
 
-        if (init.multipartBoundary && this.#multipartBody) {
-          this.#multipartBody = this.#multipartBody.withBoundary(init.multipartBoundary);
-          this.headers.set('content-type', this.#multipartBody.contentType);
-        }
-      }
+    try {
+      parsedUrl = init.baseURL ? new URL(inputUrl, init.baseURL) : new URL(inputUrl);
+    } catch {
+      throw new TypeError(`Invalid request URL: ${inputUrl}`);
+    }
 
-      return;
+    if (parsedUrl.username || parsedUrl.password) {
+      throw new TypeError('Request URL must not include credentials');
     }
 
-    this.url = String(init.baseURL ? new URL(String(input), init.baseURL) : input);
-    this.method = (init.method ?? 'GET').toUpperCase();
-    this.headers = new Headers(init.headers);
-    this.signal = init.signal ?? null;
+    this.url = parsedUrl.toString();
+    this.method = normalizeMethod(init.method === undefined ? inputRequest?.method : init.method);
+    this.headers = new Headers(init.headers ?? inputRequest?.headers);
+    this.signal = init.signal ?? inputRequest?.signal ?? new AbortController().signal;
+    this.cache = init.cache ?? inputRequest?.cache ?? 'default';
+    this.credentials = init.credentials ?? inputRequest?.credentials ?? 'same-origin';
+    this.destination = inputRequest?.destination ?? '';
+    this.integrity = init.integrity ?? inputRequest?.integrity ?? '';
+    this.keepalive = init.keepalive ?? inputRequest?.keepalive ?? false;
+    this.mode = init.mode ?? inputRequest?.mode ?? 'cors';
+    this.redirect = init.redirect ?? inputRequest?.redirect ?? 'follow';
+    this.referrer = init.referrer ?? inputRequest?.referrer ?? 'about:client';
+    this.referrerPolicy = init.referrerPolicy ?? inputRequest?.referrerPolicy ?? '';
     this.#bodyBytes = null;
     this.#multipartBody = null;
-    this.#setBody(init.body, init.multipartBoundary);
+    this.#streamingBody = null;
+
+    const inheritedBody =
+      init.body === undefined &&
+      (input instanceof Request ? input.#hasBody() : isGlobalRequest(input) && input.body !== null);
+
+    if (
+      (this.method === 'GET' || this.method === 'HEAD') &&
+      (init.body !== undefined && init.body !== null ? true : inheritedBody)
+    ) {
+      throw new TypeError('Request with GET/HEAD method cannot have body.');
+    }
+
+    if (init.body !== undefined) {
+      this.#setBody(init.body, init.multipartBoundary);
+    } else if (input instanceof Request) {
+      const boundary = init.multipartBoundary ?? input.#multipartBody?.boundary;
+
+      this.#setBody(input._takeBodyInit(), boundary, init.multipartBoundary !== undefined);
+    } else if (isGlobalRequest(input) && input.body) {
+      const transferred = new globalThis.Request(input, { method: this.method });
+
+      this.#setBody(transferred.body as ReadableStream<Uint8Array> | null, init.multipartBoundary);
+    }
   }
 
   /** Returns the request body as a readable byte stream. */
-  get body(): ReadableStream<Uint8Array> | null {
-    if (this.#bodyUsed || (this.#bodyBytes === null && this.#multipartBody === null)) {
+  get body(): globalThis.Request['body'] {
+    if (!this.#hasBody()) {
       return null;
     }
 
-    this.#bodyUsed = true;
-    this.#stream ??= new ReadableStream<Uint8Array>({
-      start: async (controller) => {
-        controller.enqueue(await this.#readBodyBytes());
-        controller.close();
+    if (this.#streamingBody) {
+      return this.#streamingBody.stream as globalThis.Request['body'];
+    }
+
+    let emitted = false;
+
+    this.#stream ??= new ReadableStream<Uint8Array>(
+      {
+        pull: async (controller) => {
+          this.#bodyUsed = true;
+
+          if (!emitted) {
+            emitted = true;
+            controller.enqueue(await this.#readBodyBytes());
+          }
+
+          controller.close();
+        },
+        cancel: () => {
+          this.#bodyUsed = true;
+        },
       },
-    });
+      { highWaterMark: 0 }
+    );
 
-    return this.#stream;
+    return this.#stream as globalThis.Request['body'];
   }
 
   /** Indicates whether the request body has already been consumed. */
   get bodyUsed(): boolean {
-    return this.#bodyUsed;
+    return this.#bodyUsed || this.#streamingBody?.bodyUsed === true;
   }
 
   /** Creates a clone whose body can be consumed independently. */
   clone(): Request {
-    if (this.#bodyUsed) {
+    if (this.bodyUsed || this.#stream?.locked || this.#streamingBody?.locked) {
       throw new TypeError('Request body is already used');
     }
 
@@ -95,10 +172,19 @@ export class Request {
       method: this.method,
       headers: this.headers,
       signal: this.signal ?? undefined,
+      cache: this.cache,
+      credentials: this.credentials,
+      integrity: this.integrity,
+      keepalive: this.keepalive,
+      mode: this.mode,
+      redirect: this.redirect,
+      referrer: this.referrer,
+      referrerPolicy: this.referrerPolicy,
     });
 
     cloned.#bodyBytes = cloneBytes(this.#bodyBytes);
     cloned.#multipartBody = this.#multipartBody?.clone() ?? null;
+    cloned.#streamingBody = this.#streamingBody?.clone() ?? null;
 
     return cloned;
   }
@@ -118,9 +204,16 @@ export class Request {
     return Uint8Array.from(await this.#consumeBytes()).buffer;
   }
 
+  /** Reads the request body as bytes. */
+  async bytes(): Promise<Uint8Array<ArrayBuffer>> {
+    return new Uint8Array(await this.#consumeBytes());
+  }
+
   /** Reads the request body as a `Blob`. */
   async blob(): Promise<Blob> {
-    return new Blob([await this.#consumeBytes()]);
+    return new globalThis.Blob([await this.#consumeBytes()], {
+      type: this.headers.get('content-type') ?? '',
+    });
   }
 
   /** Reads the request body as `FormData`. */
@@ -135,20 +228,9 @@ export class Request {
       return this.#multipartBody.clone().formData();
     }
 
-    const contentType = this.headers.get('content-type')?.toLowerCase() ?? '';
-
-    if (!contentType.includes('application/x-www-form-urlencoded')) {
-      throw new TypeError(`Request content-type is not form data: ${contentType || 'unknown'}`);
-    }
-
-    const formData = new FormData();
-    const searchParams = new URLSearchParams(await this.text());
+    const contentType = this.headers.get('content-type') ?? '';
 
-    for (const [name, value] of searchParams) {
-      formData.append(name, value);
-    }
-
-    return formData;
+    return parseResponseFormData(await this.#consumeBytes(), contentType);
   }
 
   /** Internal helper that clones the encoded request body bytes. */
@@ -164,13 +246,33 @@ export class Request {
     return new Uint8Array(await this.#multipartBody.clone().arrayBuffer());
   }
 
-  /** Internal helper that clones the source body without forcing multipart encoding. */
-  _cloneBodyInit(): BodyInit | null {
+  /** Internal helper that transfers ownership of the body to a new request. */
+  _takeBodyInit(): BodyInit | null {
+    if (this.bodyUsed || this.#stream?.locked || this.#streamingBody?.locked) {
+      throw new TypeError('Request body is already used');
+    }
+
+    let body: BodyInit | null = null;
+
     if (this.#bodyBytes !== null) {
-      return cloneBytes(this.#bodyBytes);
+      body = cloneBytes(this.#bodyBytes);
+    } else if (this.#multipartBody) {
+      body = this.#multipartBody.cloneFormData();
+    } else if (this.#streamingBody) {
+      body = this.#streamingBody.transfer().stream;
+    }
+
+    if (body !== null) {
+      const originalStream = this.body;
+
+      this.#bodyUsed = true;
+
+      if (originalStream) {
+        this.#transferredBodyReaders.push(originalStream.getReader());
+      }
     }
 
-    return this.#multipartBody?.cloneFormData() ?? null;
+    return body;
   }
 
   /** Internal helper that returns the explicit multipart boundary, when applicable. */
@@ -183,16 +285,18 @@ export class Request {
     return this.#multipartBody?.prepareNativeUpload();
   }
 
-  /** Internal helper that prepares body bytes for native dispatch. */
-  async _getBodyBytesForDispatch(): Promise<Uint8Array | undefined> {
-    return (await this._cloneBodyBytes()) ?? undefined;
+  /** Internal helper that prepares a native streaming raw-body upload. */
+  _prepareBodyStreamUpload(): import('../types').NativeBodyStreamUpload | undefined {
+    return this.#streamingBody?.prepareNativeUpload();
   }
 
-  /** Internal helper that marks the request body as consumed. */
-  _markBodyUsed(): void {
-    if (this.#bodyBytes !== null || this.#multipartBody !== null) {
-      this.#bodyUsed = true;
+  /** Internal helper that prepares body bytes for native dispatch. */
+  async _getBodyBytesForDispatch(): Promise<Uint8Array | undefined> {
+    if (this.#streamingBody) {
+      return undefined;
     }
+
+    return (await this._cloneBodyBytes()) ?? undefined;
   }
 
   /** Internal helper that creates a modified request copy. */
@@ -213,19 +317,25 @@ export class Request {
     if (!hasBodyOverride) {
       next.#bodyBytes = cloneBytes(this.#bodyBytes);
       next.#multipartBody = this.#multipartBody?.clone() ?? null;
+      next.#streamingBody = this.#streamingBody?.clone() ?? null;
     }
 
     return next;
   }
 
-  #setBody(body: BodyInit | null | undefined, multipartBoundary?: string): void {
-    const nextBody = cloneBodyInit(body);
+  #setBody(
+    body: BodyInit | null | undefined,
+    multipartBoundary?: string,
+    overwriteMultipartContentType = multipartBoundary !== undefined
+  ): void {
+    const nextBody = toReadableStreamBody(body) ?? cloneBodyInit(body);
 
     this.#stream = null;
 
     if (nextBody === null) {
       this.#bodyBytes = null;
       this.#multipartBody = null;
+      this.#streamingBody = null;
 
       return;
     }
@@ -235,14 +345,40 @@ export class Request {
 
       this.#bodyBytes = null;
       this.#multipartBody = multipartBody;
+      this.#streamingBody = null;
 
-      this.headers.set('content-type', multipartBody.contentType);
+      if (overwriteMultipartContentType || !this.headers.has('content-type')) {
+        this.headers.set('content-type', multipartBody.contentType);
+      }
+
+      return;
+    }
+
+    if (isBlobBody(nextBody) || isReadableStreamBody(nextBody)) {
+      const streamingBody = new StreamingBody(nextBody);
+
+      this.#bodyBytes = null;
+      this.#multipartBody = null;
+      this.#streamingBody = streamingBody;
+
+      if (streamingBody.contentType && !this.headers.has('content-type')) {
+        this.headers.set('content-type', streamingBody.contentType);
+      }
 
       return;
     }
 
     this.#bodyBytes = toBodyBytes(nextBody, 'Unsupported request body type');
     this.#multipartBody = null;
+    this.#streamingBody = null;
+
+    if (!this.headers.has('content-type')) {
+      if (typeof nextBody === 'string') {
+        this.headers.set('content-type', 'text/plain;charset=UTF-8');
+      } else if (nextBody instanceof URLSearchParams) {
+        this.headers.set('content-type', 'application/x-www-form-urlencoded;charset=UTF-8');
+      }
+    }
   }
 
   async #readBodyBytes(): Promise<Uint8Array> {
@@ -254,16 +390,45 @@ export class Request {
       return new Uint8Array(await this.#multipartBody.clone().arrayBuffer());
     }
 
+    if (this.#streamingBody) {
+      return this.#readStreamBytes(this.#streamingBody.stream);
+    }
+
     return new Uint8Array();
   }
 
   async #consumeBytes(): Promise<Uint8Array> {
-    if (this.#bodyUsed) {
+    if (this.bodyUsed || this.#stream?.locked || this.#streamingBody?.locked) {
       throw new TypeError('Request body is already used');
     }
 
+    if (this.#streamingBody) {
+      return this.#readStreamBytes(this.#streamingBody.stream);
+    }
+
     this.#bodyUsed = true;
 
     return this.#readBodyBytes();
   }
+
+  #hasBody(): boolean {
+    return this.#bodyBytes !== null || this.#multipartBody !== null || this.#streamingBody !== null;
+  }
+
+  async #readStreamBytes(stream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
+    const reader = stream.getReader();
+    const chunks: Uint8Array[] = [];
+
+    while (true) {
+      const result = await reader.read();
+
+      if (result.done) {
+        break;
+      }
+
+      chunks.push(result.value);
+    }
+
+    return new Uint8Array(Buffer.concat(chunks.map((chunk) => Buffer.from(chunk))));
+  }
 }
diff --git a/src/http/response.ts b/src/http/response.ts
index e8106b1..137f159 100644
--- a/src/http/response.ts
+++ b/src/http/response.ts
@@ -1,20 +1,19 @@
 import type {
   BodyInit,
-  HeadersInit,
   NativeResponse,
   RedirectEntry,
   RequestTimings,
   TlsPeerInfo,
   WreqResponseMeta,
 } from '../types';
-import { Blob, Buffer } from 'node:buffer';
+import { Buffer } from 'node:buffer';
 import { STATUS_CODES } from 'node:http';
 import { ReadableStream } from 'node:stream/web';
 import { TextDecoder } from 'node:util';
 import { RequestError, TimeoutError } from '../errors';
 import { Headers } from '../headers';
 import { nativeCancelBody, nativeForbidBodyRecycle, nativeReadBodyChunk } from '../native/index';
-import { cloneBytes, toBodyBytes } from './body/bytes';
+import { cloneBytes } from './body/bytes';
 import { parseResponseFormData } from './body/form-data';
 import { ResponseMeta } from './response-meta';
 
@@ -43,22 +42,6 @@ function decodeText(bytes: Uint8Array, contentType: string | null): string {
   }
 }
 
-function toHeadersInit(headers: ResponseInit['headers'] | undefined): HeadersInit | undefined {
-  if (headers === undefined) {
-    return undefined;
-  }
-
-  if (headers instanceof Headers) {
-    return new Headers(headers);
-  }
-
-  if (typeof globalThis.Headers !== 'undefined' && headers instanceof globalThis.Headers) {
-    return new Headers(Array.from(headers.entries()));
-  }
-
-  return headers as unknown as HeadersInit;
-}
-
 function isNativeResponse(value: unknown): value is NativeResponse {
   return (
     typeof value === 'object' &&
@@ -116,7 +99,7 @@ export class Response {
   /** Response headers. */
   readonly headers: Headers;
   /** Response type exposed for Fetch API compatibility. */
-  readonly type = 'basic' as const;
+  readonly type: globalThis.Response['type'];
   /** Extra transport metadata exposed by node-wreq. */
   readonly wreq: WreqResponseMeta;
   /** Internal cookie map used to build `response.wreq.cookies`. */
@@ -140,11 +123,14 @@ export class Response {
   #bodyFinalizerToken = {};
 
   constructor(body?: BodyInit | NativeResponse | null, init: ResponseInitWithUrl = {}) {
+    this.#streamSource = null;
+
     if (isNativeResponse(body)) {
       this.status = body.status;
       this.statusText = body.statusText ?? STATUS_CODES[body.status] ?? '';
       this.url = body.url;
       this.headers = new Headers(body.headers);
+      this.type = 'basic';
       this._cookies = { ...body.cookies };
       this._setCookies = [...(body.setCookies ?? [])];
       this._timings = body.timings ? { ...body.timings } : undefined;
@@ -155,24 +141,34 @@ export class Response {
       this.#bodyHandle = body.bodyHandle ?? null;
       this.#stream = null;
     } else {
-      this.status = init.status ?? 200;
-      this.statusText = init.statusText ?? STATUS_CODES[this.status] ?? '';
+      const nativeResponse = new globalThis.Response(
+        body as ConstructorParameters<typeof globalThis.Response>[0],
+        {
+          status: init.status,
+          statusText: init.statusText,
+          headers: init.headers,
+        }
+      );
+
+      this.status = nativeResponse.status;
+      this.statusText = nativeResponse.statusText;
       this.url = init.url ?? '';
-      this.headers = new Headers(toHeadersInit(init.headers));
+      this.headers = new Headers(nativeResponse.headers);
+      this.type = nativeResponse.type;
       this._cookies = {};
       this._setCookies = [];
       this._timings = undefined;
       this._redirectChain = [];
       this._tls = undefined;
       this.redirected = false;
-      this.#payloadBytes = toBodyBytes(body ?? null, 'Unsupported response body type');
+      this.#payloadBytes = null;
       this.#bodyHandle = null;
       this.#stream = null;
+      this.#streamSource = nativeResponse.body as ReadableStream<Uint8Array> | null;
     }
 
     this.ok = this.status >= 200 && this.status < 300;
     this.#bodyUsed = false;
-    this.#streamSource = null;
     this.wreq = new ResponseMeta(this);
     this.#orphanedStreamReaders = [];
 
@@ -211,8 +207,8 @@ export class Response {
   }
 
   /** Returns the response body as a readable byte stream. */
-  get body(): ReadableStream<Uint8Array> | null {
-    return this.#ensureStream();
+  get body(): globalThis.Response['body'] {
+    return this.#ensureStream() as globalThis.Response['body'];
   }
 
   /** Reads the response body as text, honoring the declared charset when possible. */
@@ -230,9 +226,16 @@ export class Response {
     return Uint8Array.from(await this.#consumeBytes()).buffer;
   }
 
+  /** Reads the response body as bytes. */
+  async bytes(): Promise<Uint8Array<ArrayBuffer>> {
+    return new Uint8Array(await this.#consumeBytes());
+  }
+
   /** Reads the response body as a `Blob`. */
   async blob(): Promise<Blob> {
-    return new Blob([await this.#consumeBytes()]);
+    return new globalThis.Blob([await this.#consumeBytes()], {
+      type: this.headers.get('content-type') ?? '',
+    });
   }
 
   /** Reads the response body as `FormData`. */
diff --git a/src/native/binding.ts b/src/native/binding.ts
index a35ef88..ab5fba3 100644
--- a/src/native/binding.ts
+++ b/src/native/binding.ts
@@ -19,11 +19,16 @@ export type NativeBinding = {
   failUpload: (handle: number, message: string) => Promise<void>;
   finishUpload: (handle: number) => boolean;
   releaseClient: (clientId: number) => boolean;
-  websocketConnect: (options: NativeWebSocketConnectOptions) => Promise<NativeWebSocketConnection>;
+  websocketConnect: (options: NativeWebSocketConnectOptions) => {
+    handle: number;
+    promise: Promise<NativeWebSocketConnection>;
+  };
+  websocketCancelConnect: (handle: number) => boolean;
   websocketRead: (handle: number) => Promise<NativeWebSocketReadResult>;
   websocketSendText: (handle: number, text: string) => Promise<void>;
   websocketSendBinary: (handle: number, data: Buffer) => Promise<void>;
   websocketClose: (handle: number, code?: number, reason?: string) => Promise<void>;
+  websocketTerminate: (handle: number) => boolean;
   readBodyChunk: (
     handle: number,
     size?: number
diff --git a/src/native/index.ts b/src/native/index.ts
index c9cd3fb..8b194a4 100644
--- a/src/native/index.ts
+++ b/src/native/index.ts
@@ -13,9 +13,11 @@ export {
 } from './request';
 
 export {
+  nativeWebSocketCancelConnect,
   nativeWebSocketClose,
   nativeWebSocketConnect,
   nativeWebSocketRead,
   nativeWebSocketSendBinary,
   nativeWebSocketSendText,
+  nativeWebSocketTerminate,
 } from './websocket';
diff --git a/src/native/methods.ts b/src/native/methods.ts
index f2c5457..a0cf0b2 100644
--- a/src/native/methods.ts
+++ b/src/native/methods.ts
@@ -1,11 +1,21 @@
 import type { HttpMethod } from '../types';
 
+const METHOD_PATTERN = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;
+const FORBIDDEN_METHODS = new Set(['CONNECT', 'TRACE', 'TRACK']);
+const NORMALIZED_METHODS = new Set(['DELETE', 'GET', 'HEAD', 'OPTIONS', 'POST', 'PUT']);
+
 export function normalizeMethod(method?: string): HttpMethod {
-  const normalized = (method ?? 'GET').toUpperCase();
+  const value = String(method === undefined ? 'GET' : method);
+
+  if (!METHOD_PATTERN.test(value)) {
+    throw new TypeError(`Invalid HTTP method: ${value}`);
+  }
+
+  const upper = value.toUpperCase();
 
-  if (normalized.length === 0) {
-    throw new Error(`Unsupported HTTP method: ${method}`);
+  if (FORBIDDEN_METHODS.has(upper)) {
+    throw new TypeError(`Forbidden HTTP method: ${value}`);
   }
 
-  return normalized;
+  return NORMALIZED_METHODS.has(upper) ? upper : value;
 }
diff --git a/src/native/request.ts b/src/native/request.ts
index f0c0372..4406c5e 100644
--- a/src/native/request.ts
+++ b/src/native/request.ts
@@ -7,32 +7,39 @@ export async function nativeRequest(
   options: NativeRequestOptions,
   signal?: AbortSignal | null
 ): Promise<NativeResponse> {
+  const upload = options.multipartUpload ?? options.bodyStreamUpload;
+
   if (signal?.aborted) {
-    options.multipartUpload?.cancel(signal.reason);
+    upload?.cancel(signal.reason);
     throw new AbortError(undefined, { cause: signal.reason });
   }
 
-  const { multipartUpload, ...nativeOptions } = options;
+  const {
+    multipartUpload: _multipartUpload,
+    bodyStreamUpload: _bodyStreamUpload,
+    ...nativeOptions
+  } = options;
+
   let task: ReturnType<ReturnType<typeof getBinding>['request']>;
 
   try {
     task = getBinding().request(nativeOptions);
   } catch (error) {
-    multipartUpload?.cancel(error);
+    upload?.cancel(error);
     throw error;
   }
 
-  if (multipartUpload) {
+  if (upload) {
     // Upload errors are forwarded through the native body stream so the request
     // retains the original failure instead of being replaced with a generic abort.
-    void multipartUpload.start(signal).catch(() => undefined);
+    void upload.start(signal).catch(() => undefined);
   }
 
   if (!signal) {
     try {
       return await task.promise;
     } finally {
-      multipartUpload?.cancel();
+      upload?.cancel();
     }
   }
 
@@ -50,7 +57,7 @@ export async function nativeRequest(
 
       settled = true;
       cleanup();
-      multipartUpload?.cancel(signal.reason);
+      upload?.cancel(signal.reason);
       getBinding().cancelRequest(task.handle);
       reject(new AbortError(undefined, { cause: signal.reason }));
     };
@@ -65,7 +72,7 @@ export async function nativeRequest(
 
         settled = true;
         cleanup();
-        multipartUpload?.cancel();
+        upload?.cancel();
         resolve(response);
       },
       (error) => {
@@ -75,7 +82,7 @@ export async function nativeRequest(
 
         settled = true;
         cleanup();
-        multipartUpload?.cancel(error);
+        upload?.cancel(error);
         reject(error);
       }
     );
diff --git a/src/native/websocket.ts b/src/native/websocket.ts
index 975bc32..bbb31f8 100644
--- a/src/native/websocket.ts
+++ b/src/native/websocket.ts
@@ -6,12 +6,17 @@ import type {
 import { Buffer } from 'node:buffer';
 import { getBinding } from './binding';
 
-export async function nativeWebSocketConnect(
-  options: NativeWebSocketConnectOptions
-): Promise<NativeWebSocketConnection> {
+export function nativeWebSocketConnect(options: NativeWebSocketConnectOptions): {
+  handle: number;
+  promise: Promise<NativeWebSocketConnection>;
+} {
   return getBinding().websocketConnect(options);
 }
 
+export function nativeWebSocketCancelConnect(handle: number): boolean {
+  return getBinding().websocketCancelConnect(handle);
+}
+
 export async function nativeWebSocketRead(handle: number): Promise<NativeWebSocketReadResult> {
   return getBinding().websocketRead(handle);
 }
@@ -31,3 +36,7 @@ export async function nativeWebSocketClose(
 ): Promise<void> {
   return getBinding().websocketClose(handle, code, reason);
 }
+
+export function nativeWebSocketTerminate(handle: number): boolean {
+  return getBinding().websocketTerminate(handle);
+}
diff --git a/src/test/helpers/local-server.ts b/src/test/helpers/local-server.ts
index 6cd30cc..8a08078 100644
--- a/src/test/helpers/local-server.ts
+++ b/src/test/helpers/local-server.ts
@@ -58,6 +58,7 @@ export function setupLocalTestServer() {
   let localServer: Server | undefined;
   let wsServer: WebSocketServer | undefined;
   const retryAttempts = new Map<string, number>();
+  const websocketMessages = new Map<string, string[]>();
   const connectionIds = new WeakMap<object, number>();
   let nextConnectionId = 1;
 
@@ -77,6 +78,11 @@ export function setupLocalTestServer() {
     wsServer.on('connection', (socket: WsPeer, request: IncomingMessage) => {
       const cookie = readCookieHeader(request);
       const url = new URL(request.url ?? '/', 'http://127.0.0.1');
+      const captureKey = url.searchParams.get('capture');
+
+      if (captureKey) {
+        websocketMessages.set(captureKey, []);
+      }
 
       socket.send(
         JSON.stringify({
@@ -89,6 +95,10 @@ export function setupLocalTestServer() {
       );
 
       socket.on('message', (data: Buffer, isBinary: boolean) => {
+        if (captureKey) {
+          websocketMessages.get(captureKey)?.push(data.toString());
+        }
+
         if (!isBinary && data.toString() === 'close-me') {
           socket.close(1000, 'done');
 
@@ -103,6 +113,14 @@ export function setupLocalTestServer() {
       void (async () => {
         const url = new URL(request.url ?? '/', 'http://127.0.0.1');
 
+        if (url.pathname === '/ws/messages') {
+          sendJson(response, 200, {
+            messages: websocketMessages.get(url.searchParams.get('key') ?? '') ?? [],
+          });
+
+          return;
+        }
+
         if (url.pathname === '/connection/id') {
           let connectionId = connectionIds.get(request.socket);
 
@@ -472,9 +490,23 @@ export function setupLocalTestServer() {
         return;
       }
 
-      wsServer?.handleUpgrade(request, socket, head, (websocketSocket: WsPeer) => {
-        wsServer?.emit('connection', websocketSocket, request);
-      });
+      const upgrade = () => {
+        if (socket.destroyed) {
+          return;
+        }
+
+        wsServer?.handleUpgrade(request, socket, head, (websocketSocket: WsPeer) => {
+          wsServer?.emit('connection', websocketSocket, request);
+        });
+      };
+
+      const upgradeDelay = Number(url.searchParams.get('upgradeDelay') ?? '0');
+
+      if (upgradeDelay > 0) {
+        setTimeout(upgrade, upgradeDelay);
+      } else {
+        upgrade();
+      }
     });
 
     await new Promise<void>((resolve) => {
diff --git a/src/test/hooks-retries.spec.ts b/src/test/hooks-retries.spec.ts
index b5a819f..35cbb42 100644
--- a/src/test/hooks-retries.spec.ts
+++ b/src/test/hooks-retries.spec.ts
@@ -43,7 +43,7 @@ describe('hooks and retries', () => {
         beforeRequest: [
           () =>
             new WreqResponse(JSON.stringify({ shortCircuited: true }), {
-              status: 204,
+              status: 200,
               headers: { 'content-type': 'application/json' },
               url: 'https://local/short-circuit',
             }),
@@ -51,7 +51,7 @@ describe('hooks and retries', () => {
       },
     });
 
-    assert.strictEqual(response.status, 204);
+    assert.strictEqual(response.status, 200);
     assert.deepStrictEqual(await response.json(), { shortCircuited: true });
     assert.strictEqual(response.url, 'https://local/short-circuit');
     assert.deepStrictEqual(response.wreq.timings, {
diff --git a/src/test/node-wreq.spec.ts b/src/test/node-wreq.spec.ts
index e71456f..699ab47 100644
--- a/src/test/node-wreq.spec.ts
+++ b/src/test/node-wreq.spec.ts
@@ -2,6 +2,7 @@ import './cookies-redirects.spec';
 import './hooks-retries.spec';
 import './http-client.spec';
 import './mtls.spec';
+import './package-exports.spec';
 import './response.spec';
 import './transport-features.spec';
 import './websocket.spec';
diff --git a/src/test/package-exports.spec.ts b/src/test/package-exports.spec.ts
new file mode 100644
index 0000000..266a0cf
--- /dev/null
+++ b/src/test/package-exports.spec.ts
@@ -0,0 +1,20 @@
+import assert from 'node:assert';
+import { spawnSync } from 'node:child_process';
+import { resolve } from 'node:path';
+import { test } from 'node:test';
+import { pathToFileURL } from 'node:url';
+
+test('should expose Request as a named ESM export', () => {
+  const entry = pathToFileURL(resolve(__dirname, '../node-wreq.mjs')).href;
+  const script = `
+    import { Request } from ${JSON.stringify(entry)};
+    const request = new Request('https://example.com/');
+    if (request.url !== 'https://example.com/') process.exit(1);
+  `;
+
+  const result = spawnSync(process.execPath, ['--input-type=module', '--eval', script], {
+    encoding: 'utf8',
+  });
+
+  assert.strictEqual(result.status, 0, result.stderr || result.stdout);
+});
diff --git a/src/test/response.spec.ts b/src/test/response.spec.ts
index 4db469a..a46aa05 100644
--- a/src/test/response.spec.ts
+++ b/src/test/response.spec.ts
@@ -86,6 +86,39 @@ describe('response behavior', () => {
     assert.strictEqual(right, JSON.stringify({ cloned: true }));
   });
 
+  test('should accept standard streaming BodyInit values in synthetic responses', async () => {
+    const blobResponse = new WreqResponse(
+      new Blob(['blob response'], { type: 'text/x-node-wreq-test' })
+    );
+
+    assert.strictEqual(blobResponse.headers.get('content-type'), 'text/x-node-wreq-test');
+    assert.deepStrictEqual([...(await blobResponse.bytes())], [...Buffer.from('blob response')]);
+
+    const streamResponse = new WreqResponse(
+      new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(Buffer.from('stream response'));
+          controller.close();
+        },
+      })
+    );
+
+    assert.strictEqual(await streamResponse.text(), 'stream response');
+  });
+
+  test('should preserve the response MIME type when reading a Blob', async () => {
+    const response = new WreqResponse('typed response', {
+      headers: {
+        'content-type': 'Text/Plain; Charset=UTF-8',
+      },
+    });
+
+    const blob = await response.blob();
+
+    assert.strictEqual(blob.type, 'text/plain; charset=utf-8');
+    assert.strictEqual(await blob.text(), 'typed response');
+  });
+
   test('should reject clone and convenience readers while the body stream is locked', async () => {
     const response = new WreqResponse('locked', {
       status: 200,
diff --git a/src/test/transport-features.spec.ts b/src/test/transport-features.spec.ts
index e34f804..1c8e389 100644
--- a/src/test/transport-features.spec.ts
+++ b/src/test/transport-features.spec.ts
@@ -5,6 +5,11 @@ import { createClient, fetch, Request } from '../node-wreq';
 import { setupLocalTestServer } from './helpers/local-server';
 import { setupProxyTestServer } from './helpers/proxy-server';
 
+async function* createAsyncBodyChunks(): AsyncIterable<Uint8Array> {
+  yield Buffer.from('async ');
+  yield Buffer.from('iterable');
+}
+
 describe('transport features', () => {
   const { getBaseUrl } = setupLocalTestServer();
   const proxyServer = setupProxyTestServer();
@@ -55,6 +60,304 @@ describe('transport features', () => {
     );
   });
 
+  test('should preserve an explicit content-type for FormData bodies', async () => {
+    const formData = new FormData();
+
+    formData.append('alpha', 'explicit-header');
+
+    const response = await fetch(`${getBaseUrl()}/body/echo`, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/x-node-wreq-test',
+      },
+      body: formData,
+    });
+
+    const body = await response.json<{ body: string; headers: Record<string, string> }>();
+
+    assert.strictEqual(body.headers['content-type'], 'application/x-node-wreq-test');
+    assert.ok(body.body.includes('name="alpha"'));
+    assert.ok(body.body.includes('explicit-header'));
+  });
+
+  test('should reject GET and HEAD request bodies like the Fetch API', () => {
+    for (const method of ['GET', 'HEAD']) {
+      assert.throws(
+        () =>
+          new Request('https://node-wreq.invalid/', {
+            method,
+            body: 'not allowed',
+          }),
+        (error: unknown) =>
+          error instanceof TypeError && error.message.includes('GET/HEAD method cannot have body')
+      );
+    }
+  });
+
+  test('should validate and normalize Request URLs and methods like Fetch', () => {
+    const request = new Request('https://EXAMPLE.com:443/a/../b', {
+      method: 'patch',
+    });
+
+    assert.strictEqual(request.url, 'https://example.com/b');
+    assert.strictEqual(request.method, 'patch');
+    assert.strictEqual(new Request('https://example.com/', { method: 123 as never }).method, '123');
+    assert.strictEqual(
+      new Request('https://example.com/', { method: null as never }).method,
+      'null'
+    );
+
+    assert.throws(() => new Request('not a URL'), TypeError);
+    assert.throws(() => new Request('https://user:pass@example.com/'), TypeError);
+    assert.throws(() => new Request('https://example.com/', { method: 'CONNECT' }), TypeError);
+  });
+
+  test('should preserve the request MIME type when reading a Blob', async () => {
+    const request = new Request('https://node-wreq.invalid/', {
+      method: 'POST',
+      headers: {
+        'content-type': 'Application/JSON; Charset=UTF-8',
+      },
+      body: '{"typed":true}',
+    });
+
+    const blob = await request.blob();
+
+    assert.strictEqual(blob.type, 'application/json; charset=utf-8');
+  });
+
+  test('should not disturb Request.body until the stream is read', async () => {
+    const request = new Request('https://node-wreq.invalid/', {
+      method: 'POST',
+      body: 'stream lifecycle',
+    });
+
+    const first = request.body;
+    const second = request.body;
+
+    assert.ok(first);
+    assert.strictEqual(first, second);
+    assert.strictEqual(request.bodyUsed, false);
+
+    const reader = first.getReader();
+    const chunk = await reader.read();
+
+    assert.strictEqual(chunk.done, false);
+    assert.strictEqual(request.bodyUsed, true);
+    assert.strictEqual(Buffer.from(chunk.value).toString(), 'stream lifecycle');
+  });
+
+  test('should transfer a Request body when constructing or fetching from it', async () => {
+    const source = new Request('https://node-wreq.invalid/', {
+      method: 'POST',
+      body: 'constructor transfer',
+    });
+
+    const sourceBody = source.body;
+    const transferred = new Request(source);
+
+    assert.strictEqual(source.bodyUsed, true);
+    assert.strictEqual(sourceBody?.locked, true);
+    assert.strictEqual(transferred.bodyUsed, false);
+    assert.strictEqual(await transferred.text(), 'constructor transfer');
+
+    const fetchSource = new Request(`${getBaseUrl()}/body/echo`, {
+      method: 'POST',
+      body: 'fetch transfer',
+    });
+
+    const response = await fetch(fetchSource);
+    const body = await response.json<{ body: string }>();
+
+    assert.strictEqual(fetchSource.bodyUsed, true);
+    assert.strictEqual(fetchSource.body?.locked, true);
+    assert.strictEqual(body.body, 'fetch transfer');
+  });
+
+  test('should stream raw Blob bodies without calling Blob.arrayBuffer()', async () => {
+    const blob = new Blob(['streamed blob'], { type: 'text/x-node-wreq-test' });
+    const originalArrayBuffer = Blob.prototype.arrayBuffer;
+
+    Blob.prototype.arrayBuffer = async () => {
+      throw new Error('Blob.arrayBuffer() must not be used by request dispatch');
+    };
+
+    try {
+      const response = await fetch(`${getBaseUrl()}/body/echo`, {
+        method: 'POST',
+        body: blob,
+      });
+
+      const body = await response.json<{ body: string; headers: Record<string, string> }>();
+
+      assert.strictEqual(body.body, 'streamed blob');
+      assert.strictEqual(body.headers['content-type'], 'text/x-node-wreq-test');
+      assert.strictEqual(Number(body.headers['content-length']), blob.size);
+    } finally {
+      Blob.prototype.arrayBuffer = originalArrayBuffer;
+    }
+  });
+
+  test('should replay streaming Blob bodies for retries and preserving redirects', async () => {
+    const blob = new Blob(['replayable blob'], { type: 'text/plain' });
+    const key = `blob-${Date.now()}-${Math.random()}`;
+    const retried = await fetch(
+      `${getBaseUrl()}/retry/body?key=${encodeURIComponent(key)}&failCount=1`,
+      {
+        method: 'POST',
+        body: blob,
+        retry: {
+          limit: 1,
+          methods: ['POST'],
+          statusCodes: [503],
+        },
+      }
+    );
+
+    const retryBody = await retried.json<{ attempt: number; body: string }>();
+
+    assert.strictEqual(retryBody.attempt, 2);
+    assert.strictEqual(retryBody.body, 'replayable blob');
+
+    const redirected = await fetch(`${getBaseUrl()}/redirect/preserve-body`, {
+      method: 'POST',
+      body: blob,
+    });
+
+    const redirectBody = await redirected.json<{ body: string; method: string }>();
+
+    assert.strictEqual(redirectBody.method, 'POST');
+    assert.strictEqual(redirectBody.body, 'replayable blob');
+  });
+
+  test('should stream ReadableStream request bodies through the native transport', async () => {
+    const chunks = ['streamed ', 'without ', 'buffering'];
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        const chunk = chunks.shift();
+
+        if (chunk === undefined) {
+          controller.close();
+
+          return;
+        }
+
+        controller.enqueue(Buffer.from(chunk));
+      },
+    });
+
+    const response = await fetch(`${getBaseUrl()}/body/echo`, {
+      method: 'POST',
+      body: stream,
+    });
+
+    const body = await response.json<{ body: string }>();
+
+    assert.strictEqual(body.body, 'streamed without buffering');
+  });
+
+  test('should stream async iterable request bodies', async () => {
+    const response = await fetch(`${getBaseUrl()}/body/echo`, {
+      method: 'POST',
+      body: createAsyncBodyChunks(),
+    });
+
+    const body = await response.json<{ body: string }>();
+
+    assert.strictEqual(body.body, 'async iterable');
+  });
+
+  test('should cancel an active raw request stream when aborted', async () => {
+    let cancelled = false;
+    const controller = new AbortController();
+    const stream = new ReadableStream<Uint8Array>({
+      async pull(streamController) {
+        await new Promise((resolve) => setTimeout(resolve, 25));
+        streamController.enqueue(Buffer.alloc(256 * 1024));
+      },
+      cancel() {
+        cancelled = true;
+      },
+    });
+
+    const pending = fetch(`${getBaseUrl()}/body/echo`, {
+      method: 'POST',
+      body: stream,
+      signal: controller.signal,
+    });
+
+    setTimeout(() => controller.abort(new Error('stop raw upload')), 10);
+
+    await assert.rejects(
+      pending,
+      (error: unknown) => error instanceof Error && error.name === 'AbortError'
+    );
+
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    assert.strictEqual(cancelled, true);
+  });
+
+  test('should propagate raw request stream failures', async () => {
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        controller.error(new Error('broken raw source'));
+      },
+    });
+
+    await assert.rejects(
+      fetch(`${getBaseUrl()}/body/echo`, {
+        method: 'POST',
+        body: stream,
+      }),
+      (error: unknown) => error instanceof Error && error.message.includes('broken raw source')
+    );
+  });
+
+  test('should reject preserving redirects for non-replayable request streams', async () => {
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(Buffer.from('one-shot stream'));
+        controller.close();
+      },
+    });
+
+    await assert.rejects(
+      fetch(`${getBaseUrl()}/redirect/preserve-body`, {
+        method: 'POST',
+        body: stream,
+      }),
+      (error: unknown) => error instanceof Error && error.message.includes('already used')
+    );
+  });
+
+  test('should consume a global Request body as a stream', async () => {
+    let arrayBufferCalled = false;
+    const input = new globalThis.Request(`${getBaseUrl()}/body/echo`, {
+      method: 'POST',
+      body: new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(Buffer.from('global request stream'));
+          controller.close();
+        },
+      }),
+      duplex: 'half',
+    });
+
+    Object.defineProperty(input, 'arrayBuffer', {
+      value: () => {
+        arrayBufferCalled = true;
+        throw new Error('global Request.arrayBuffer() must not be used by request dispatch');
+      },
+    });
+
+    const response = await fetch(input);
+    const body = await response.json<{ body: string }>();
+
+    assert.strictEqual(body.body, 'global request stream');
+    assert.strictEqual(arrayBufferCalled, false);
+    assert.strictEqual(input.bodyUsed, true);
+  });
+
   test('should stream FormData files without calling Blob.arrayBuffer()', async () => {
     const originalArrayBuffer = Blob.prototype.arrayBuffer;
     const formData = new FormData();
diff --git a/src/test/type-compatibility.ts b/src/test/type-compatibility.ts
new file mode 100644
index 0000000..b0dd005
--- /dev/null
+++ b/src/test/type-compatibility.ts
@@ -0,0 +1,18 @@
+import type {
+  fetch as wreqFetch,
+  Headers as WreqHeaders,
+  Request as WreqRequest,
+  Response as WreqResponse,
+  WebSocket as WreqWebSocket,
+} from '../node-wreq';
+
+type Assert<T extends true> = T;
+type IsAssignable<From, To> = [From] extends [To] ? true : false;
+
+export type WhatwgTypeCompatibility = [
+  Assert<IsAssignable<typeof wreqFetch, typeof globalThis.fetch>>,
+  Assert<IsAssignable<WreqHeaders, globalThis.Headers>>,
+  Assert<IsAssignable<WreqRequest, globalThis.Request>>,
+  Assert<IsAssignable<WreqResponse, globalThis.Response>>,
+  Assert<IsAssignable<WreqWebSocket, globalThis.WebSocket>>,
+];
diff --git a/src/test/websocket.spec.ts b/src/test/websocket.spec.ts
index 0b94640..28dc68e 100644
--- a/src/test/websocket.spec.ts
+++ b/src/test/websocket.spec.ts
@@ -5,6 +5,7 @@ import {
   CloseEvent as WreqCloseEvent,
   WebSocket as WreqWebSocket,
   createClient,
+  fetch,
   websocket,
 } from '../node-wreq';
 import { onceEvent, setupLocalTestServer } from './helpers/local-server';
@@ -50,6 +51,112 @@ describe('websocket', () => {
     assert.strictEqual(socket.readyState, WreqWebSocket.CLOSED);
   });
 
+  test('should support the standard constructor protocols argument and HTTP URL conversion', async () => {
+    const socket = new WreqWebSocket(`${getBaseUrl()}/ws`, 'chat');
+
+    await socket.opened;
+
+    assert.strictEqual(socket.url, getBaseUrl().replace('http://', 'ws://') + '/ws');
+    assert.strictEqual(socket.protocol, 'chat');
+
+    const connectedEvent = await onceEvent<MessageEvent>(socket, 'message');
+
+    assert.strictEqual(connectedEvent.origin, new URL(socket.url).origin);
+
+    const closePromise = onceEvent<WreqCloseEvent>(socket, 'close');
+
+    socket.close(1000, 'done');
+
+    await closePromise;
+
+    assert.doesNotThrow(() => socket.send('discarded after close'));
+    assert.strictEqual(socket.bufferedAmount, Buffer.byteLength('discarded after close'));
+  });
+
+  test('should fail promptly when close is called while connecting', async () => {
+    const socket = new WreqWebSocket(
+      getBaseUrl().replace('http://', 'ws://') + '/ws?upgradeDelay=500'
+    );
+
+    const events: string[] = [];
+    const closePromise = onceEvent<WreqCloseEvent>(socket, 'close');
+
+    socket.addEventListener('open', () => events.push('open'));
+    socket.addEventListener('error', () => events.push('error'));
+    socket.addEventListener('close', () => events.push('close'));
+
+    await new Promise((resolve) => setTimeout(resolve, 10));
+
+    const startTime = Date.now();
+
+    socket.close();
+
+    await assert.rejects(socket.opened, /closed before opening/);
+
+    const closeEvent = await closePromise;
+
+    assert.ok(Date.now() - startTime < 250, 'close should not wait for the pending upgrade');
+    assert.deepStrictEqual(events, ['error', 'close']);
+    assert.strictEqual(closeEvent.code, 1006);
+    assert.strictEqual(closeEvent.wasClean, false);
+    assert.strictEqual(socket.readyState, WreqWebSocket.CLOSED);
+  });
+
+  test('should surface cookie jar failures through error and close events', async () => {
+    const socket = new WreqWebSocket(getBaseUrl().replace('http://', 'ws://') + '/ws', {
+      cookieJar: {
+        getCookies() {
+          throw new Error('cookie lookup failed');
+        },
+        setCookie() {},
+      },
+    });
+
+    const errorPromise = onceEvent<Event & { error?: Error }>(socket, 'error');
+    const closePromise = onceEvent<WreqCloseEvent>(socket, 'close');
+
+    await assert.rejects(socket.opened, /cookie lookup failed/);
+
+    const [errorEvent, closeEvent] = await Promise.all([errorPromise, closePromise]);
+
+    assert.match(errorEvent.error?.message ?? '', /cookie lookup failed/);
+    assert.strictEqual(closeEvent.code, 1006);
+    assert.strictEqual(closeEvent.wasClean, false);
+  });
+
+  test('should report an empty clean close with the reserved 1005 event code', async () => {
+    const socket = await websocket(getBaseUrl().replace('http://', 'ws://') + '/ws');
+    const closePromise = onceEvent<WreqCloseEvent>(socket, 'close');
+
+    socket.close();
+
+    const closeEvent = await closePromise;
+
+    assert.strictEqual(closeEvent.code, 1005);
+    assert.strictEqual(closeEvent.reason, '');
+    assert.strictEqual(closeEvent.wasClean, true);
+  });
+
+  test('should use WHATWG URL and binaryType validation semantics', async () => {
+    for (const url of ['not a URL', 'ftp://example.com/socket']) {
+      assert.throws(
+        () => new WreqWebSocket(url),
+        (error: unknown) => error instanceof DOMException && error.name === 'SyntaxError'
+      );
+    }
+
+    const socket = await websocket(getBaseUrl().replace('http://', 'ws://') + '/ws');
+
+    (socket as { binaryType: string }).binaryType = 'invalid';
+    assert.strictEqual(socket.binaryType, 'blob');
+
+    const closePromise = onceEvent<WreqCloseEvent>(socket, 'close');
+
+    socket.close(1000);
+
+    await closePromise;
+  });
+
   test('should support binary messages and arraybuffer binaryType', async () => {
     const socket = new WreqWebSocket(getBaseUrl().replace('http://', 'ws://') + '/ws', {
       binaryType: 'arraybuffer',
@@ -60,7 +167,10 @@ describe('websocket', () => {
 
     const replyPromise = onceEvent<MessageEvent>(socket, 'message');
 
-    socket.send(new Uint8Array([1, 2, 3]));
+    const payload = new Uint8Array([1, 2, 3]);
+
+    socket.send(payload);
+    payload[0] = 9;
 
     const replyEvent = await replyPromise;
 
@@ -125,11 +235,40 @@ describe('websocket', () => {
           protocols: ['chat', 'chat'],
         }),
       (error: unknown) =>
-        error instanceof SyntaxError && error.message.includes('Duplicate WebSocket subprotocol'),
+        error instanceof DOMException &&
+        error.name === 'SyntaxError' &&
+        error.message.includes('Duplicate WebSocket subprotocol'),
       'duplicate websocket subprotocols should be rejected'
     );
   });
 
+  test('should flush queued messages in order before closing', async () => {
+    const capture = `queued-${Date.now()}-${Math.random()}`;
+    const socket = await websocket(
+      getBaseUrl().replace('http://', 'ws://') + `/ws?capture=${encodeURIComponent(capture)}`
+    );
+
+    await onceEvent<MessageEvent>(socket, 'message');
+
+    const expected = Array.from({ length: 20 }, (_, index) => String(index));
+
+    for (const message of expected) {
+      socket.send(message);
+    }
+
+    const closePromise = onceEvent<WreqCloseEvent>(socket, 'close');
+
+    socket.close(1000, 'queued messages sent');
+
+    const closeEvent = await closePromise;
+    const response = await fetch(`${getBaseUrl()}/ws/messages?key=${encodeURIComponent(capture)}`);
+    const body = await response.json<{ messages: string[] }>();
+
+    assert.strictEqual(closeEvent.wasClean, true);
+    assert.deepStrictEqual(body.messages, expected);
+    assert.strictEqual(socket.bufferedAmount, 0);
+  });
+
   test('should reject invalid websocket size limits', async () => {
     const socket = new WreqWebSocket(getBaseUrl().replace('http://', 'ws://') + '/ws', {
       maxFrameSize: 0,
diff --git a/src/types/http.ts b/src/types/http.ts
index a61eeb8..24c4c34 100644
--- a/src/types/http.ts
+++ b/src/types/http.ts
@@ -87,7 +87,12 @@ export interface RequestStats {
 }
 
 /** Request options accepted by `fetch`, `Request`, and client helpers. */
-export interface WreqInit {
+export interface WreqInit extends Omit<
+  globalThis.RequestInit,
+  'body' | 'headers' | 'method' | 'redirect' | 'signal'
+> {
+  /** Cache mode retained for compatibility with the browser Request API. */
+  cache?: globalThis.Request['cache'];
   /** HTTP method used for the request. */
   method?: string;
   /** Request headers. */
diff --git a/src/types/native.ts b/src/types/native.ts
index c3772bc..7f894ab 100644
--- a/src/types/native.ts
+++ b/src/types/native.ts
@@ -99,12 +99,27 @@ export interface NativeMultipartBody {
 }
 
 /** JS-side upload pump paired with a native multipart body. */
-export interface NativeMultipartUpload {
-  body: NativeMultipartBody;
+export interface NativeUploadPump {
   start(signal?: AbortSignal | null): Promise<void>;
   cancel(reason?: unknown): void;
 }
 
+/** JS-side upload pump paired with a native multipart body. */
+export interface NativeMultipartUpload extends NativeUploadPump {
+  body: NativeMultipartBody;
+}
+
+/** Raw request stream consumed by the native transport. */
+export interface NativeBodyStream {
+  uploadHandle: number;
+  length?: number;
+}
+
+/** JS-side upload pump paired with a raw native request stream. */
+export interface NativeBodyStreamUpload extends NativeUploadPump {
+  body: NativeBodyStream;
+}
+
 /** Fully normalized native request payload. */
 export interface NativeRequestOptions {
   /** Fully resolved request URL. */
@@ -121,6 +136,10 @@ export interface NativeRequestOptions {
   multipart?: NativeMultipartBody;
   /** Internal JS upload pump; never read by the native options converter. */
   multipartUpload?: NativeMultipartUpload;
+  /** Raw body stream consumed by the native transport. */
+  bodyStream?: NativeBodyStream;
+  /** Internal JS upload pump for `bodyStream`; never read by the native converter. */
+  bodyStreamUpload?: NativeBodyStreamUpload;
   /** Browser fingerprint profile used by the native transport. */
   browser?: BrowserProfile;
   /** Browser profile selection strategy used by the native transport. */
diff --git a/src/types/shared.ts b/src/types/shared.ts
index 902a870..0ab397a 100644
--- a/src/types/shared.ts
+++ b/src/types/shared.ts
@@ -48,12 +48,24 @@ export type TlsDataInput = string | TlsBinaryInput;
 
 /** Header input accepted by `fetch`, `Request`, and `WebSocket` helpers. */
 export type HeadersInit =
-  | Record<string, string | number | boolean | null | undefined>
+  | globalThis.Headers
+  | Record<string, string | readonly string[] | number | boolean | null | undefined>
   | HeaderTuple[]
+  | string[][]
   | Iterable<HeaderTuple>;
 
 /** Request and response body input supported by the library. */
-export type BodyInit = string | URLSearchParams | FormData | Buffer | ArrayBuffer | ArrayBufferView;
+export type BodyInit =
+  | string
+  | URLSearchParams
+  | FormData
+  | Blob
+  | Buffer
+  | ArrayBuffer
+  | ArrayBufferView
+  | ReadableStream<Uint8Array>
+  | AsyncIterable<Uint8Array>
+  | Iterable<Uint8Array>;
 
 /** DNS overrides applied by the native transport. */
 export interface DnsOptions {
diff --git a/src/websocket/index.ts b/src/websocket/index.ts
index d6fd632..51429b5 100644
--- a/src/websocket/index.ts
+++ b/src/websocket/index.ts
@@ -16,14 +16,21 @@ import { loadCookiesIntoHeaders } from '../http/pipeline/cookies';
 import { createNativeClientCacheKey } from '../native/client-cache';
 import {
   nativeWebSocketClose,
+  nativeWebSocketCancelConnect,
   nativeWebSocketConnect,
   nativeWebSocketRead,
   nativeWebSocketSendBinary,
   nativeWebSocketSendText,
+  nativeWebSocketTerminate,
   normalizeBrowserEmulation,
 } from '../native/index';
 import { CloseEvent } from './close-event';
-import { getSendByteLength, normalizeSendData, toMessageEventData } from './send-data';
+import {
+  getSendByteLength,
+  normalizeSendData,
+  snapshotSendData,
+  toMessageEventData,
+} from './send-data';
 import {
   normalizeHeaders,
   normalizeProtocols,
@@ -93,6 +100,7 @@ type ErrorHandler = ((event: Event) => void) | null;
 const NATIVE_CLIENT_ID = Symbol('node-wreq.nativeClientId');
 
 type InternalWebSocketInit = WebSocketInit & { [NATIVE_CLIENT_ID]?: number };
+type WebSocketConstructorOptions = WebSocketInit | string | string[];
 
 export { CloseEvent };
 
@@ -126,20 +134,29 @@ export class WebSocket extends EventTarget {
   #rejectOpened!: (reason?: unknown) => void;
   #readyState = WebSocket.CONNECTING;
   #handle?: number;
+  #connectHandle?: number;
   #protocol = '';
   #binaryType: WebSocketBinaryType;
   #bufferedAmount = 0;
   #sendQueue = Promise.resolve();
   #settled = false;
+  #openSettled = false;
+  #closeEnqueued = false;
+  #closeRequest: { code?: number; reason?: string } | null = null;
   #onopen: OpenHandler = null;
   #onmessage: MessageHandler = null;
   #onclose: CloseHandler = null;
   #onerror: ErrorHandler = null;
 
   /** Creates and starts connecting a new WebSocket instance. */
-  constructor(url: string | URL, init: WebSocketInit = {}) {
+  constructor(url: string | URL, protocolsOrInit: WebSocketConstructorOptions = {}) {
     super();
 
+    const init: WebSocketInit =
+      typeof protocolsOrInit === 'string' || Array.isArray(protocolsOrInit)
+        ? { protocols: protocolsOrInit }
+        : protocolsOrInit;
+
     this.url = resolveWebSocketUrl(url, init);
     normalizeBrowserEmulation(init.browser);
 
@@ -160,6 +177,8 @@ export class WebSocket extends EventTarget {
       this.#rejectOpened = reject;
     });
 
+    void this.opened.catch(() => undefined);
+
     void this.#connect(init, headers, protocols, (init as InternalWebSocketInit)[NATIVE_CLIENT_ID]);
   }
 
@@ -181,7 +200,7 @@ export class WebSocket extends EventTarget {
   /** Updates the binary representation used for incoming binary messages. */
   set binaryType(value: WebSocketBinaryType) {
     if (value !== 'blob' && value !== 'arraybuffer') {
-      throw new TypeError(`Invalid WebSocket binaryType: ${value}`);
+      return;
     }
 
     this.#binaryType = value;
@@ -238,28 +257,32 @@ export class WebSocket extends EventTarget {
 
   /** Queues a text or binary message for sending. */
   send(data: string | Blob | ArrayBuffer | ArrayBufferView): void {
-    if (this.#readyState !== WebSocket.OPEN || this.#handle === undefined) {
+    const queuedBytes = getSendByteLength(data);
+
+    if (this.#readyState === WebSocket.CONNECTING) {
       throw new DOMException('WebSocket is not open', 'InvalidStateError');
     }
 
-    const queuedBytes = getSendByteLength(data);
-
     this.#bufferedAmount += queuedBytes;
+
+    if (this.#readyState !== WebSocket.OPEN || this.#handle === undefined) {
+      return;
+    }
+
+    const handle = this.#handle;
+    const queuedData = snapshotSendData(data);
+
     this.#sendQueue = this.#sendQueue
       .then(async () => {
-        const normalized = await normalizeSendData(data);
-
-        if (this.#readyState !== WebSocket.OPEN || this.#handle === undefined) {
-          throw new DOMException('WebSocket is not open', 'InvalidStateError');
-        }
+        const normalized = await normalizeSendData(queuedData);
 
         if (normalized.type === 'text') {
-          await nativeWebSocketSendText(this.#handle, normalized.data);
+          await nativeWebSocketSendText(handle, normalized.data);
 
           return;
         }
 
-        await nativeWebSocketSendBinary(this.#handle, normalized.data);
+        await nativeWebSocketSendBinary(handle, normalized.data);
       })
       .catch((error: unknown) => {
         this.#handleError(error);
@@ -270,43 +293,58 @@ export class WebSocket extends EventTarget {
   }
 
   /** Starts the closing handshake. */
-  close(code?: number, reason = ''): void {
+  close(code?: number, reason?: string): void {
+    const normalizedReason = reason ?? '';
+
     if (code !== undefined) {
       validateCloseCode(code);
     }
 
-    validateCloseReason(reason);
+    validateCloseReason(normalizedReason);
 
     if (this.#readyState === WebSocket.CLOSING || this.#readyState === WebSocket.CLOSED) {
       return;
     }
 
+    const wasConnecting = this.#readyState === WebSocket.CONNECTING;
+
     this.#readyState = WebSocket.CLOSING;
+    this.#closeRequest = {
+      code,
+      reason: code === undefined && normalizedReason === '' ? undefined : normalizedReason,
+    };
 
-    if (this.#handle === undefined) {
-      return;
-    }
+    if (wasConnecting) {
+      const error = new WebSocketError('WebSocket was closed before opening');
 
-    const handle = this.#handle;
+      this.#rejectOpen(error);
 
-    this.#handle = undefined;
+      if (this.#connectHandle !== undefined) {
+        nativeWebSocketCancelConnect(this.#connectHandle);
+        this.#connectHandle = undefined;
+      }
+
+      queueMicrotask(() => {
+        if (this.#settled) {
+          return;
+        }
 
-    void nativeWebSocketClose(handle, code, reason)
-      .then(() => {
-        this.#finalizeClose({
-          code: code ?? 1000,
-          reason,
-          wasClean: true,
-        });
-      })
-      .catch((error: unknown) => {
         this.#handleError(error);
         this.#finalizeClose({
-          code: code ?? 1006,
-          reason,
+          code: 1006,
+          reason: '',
           wasClean: false,
         });
       });
+
+      return;
+    }
+
+    if (this.#handle === undefined) {
+      return;
+    }
+
+    this.#enqueueClose(this.#handle);
   }
 
   async #connect(
@@ -315,9 +353,15 @@ export class WebSocket extends EventTarget {
     protocols: string[],
     clientId?: number
   ): Promise<void> {
-    await loadCookiesIntoHeaders(init.cookieJar, this.url, headers);
+    let connectHandle: number | undefined;
 
     try {
+      await loadCookiesIntoHeaders(init.cookieJar, this.url, headers);
+
+      if (this.#settled || this.#readyState !== WebSocket.CONNECTING) {
+        return;
+      }
+
       const localBind = normalizeLocalBindOptions(init);
       const { proxy, disableSystemProxy } = normalizeProxyOptions(init.proxy);
       const browser = normalizeBrowserEmulation(init.browser);
@@ -362,7 +406,18 @@ export class WebSocket extends EventTarget {
         nativeOptions.clientCacheKey = createNativeClientCacheKey(nativeOptions);
       }
 
-      const connection = await nativeWebSocketConnect(nativeOptions);
+      const connectTask = nativeWebSocketConnect(nativeOptions);
+
+      connectHandle = connectTask.handle;
+      this.#connectHandle = connectHandle;
+
+      const connection = await connectTask.promise;
+
+      if (this.#settled || this.#readyState !== WebSocket.CONNECTING) {
+        nativeWebSocketTerminate(connection.handle);
+
+        return;
+      }
 
       this.#handle = connection.handle;
       this.#protocol = connection.protocol ?? '';
@@ -372,22 +427,37 @@ export class WebSocket extends EventTarget {
       }
 
       (this as { extensions: string }).extensions = connection.extensions ?? '';
+
       this.#readyState = WebSocket.OPEN;
-      this.#resolveOpened();
+      this.#resolveOpen();
       this.dispatchEvent(new Event('open'));
       void this.#pumpMessages();
     } catch (error) {
+      if (this.#settled) {
+        return;
+      }
+
       this.#handleError(error);
+
+      if (this.#handle !== undefined) {
+        nativeWebSocketTerminate(this.#handle);
+        this.#handle = undefined;
+      }
+
       this.#finalizeClose({
         code: 1006,
         reason: '',
         wasClean: false,
       });
+    } finally {
+      if (connectHandle !== undefined && this.#connectHandle === connectHandle) {
+        this.#connectHandle = undefined;
+      }
     }
   }
 
   async #pumpMessages(): Promise<void> {
-    while (this.#readyState === WebSocket.OPEN && this.#handle !== undefined) {
+    while (!this.#settled && this.#handle !== undefined) {
       try {
         const result = await nativeWebSocketRead(this.#handle);
 
@@ -398,11 +468,14 @@ export class WebSocket extends EventTarget {
           return;
         }
 
-        this.dispatchEvent(
-          new MessageEvent('message', {
-            data: toMessageEventData(result, this.#binaryType),
-          })
-        );
+        if (this.#readyState === WebSocket.OPEN) {
+          this.dispatchEvent(
+            new MessageEvent('message', {
+              data: toMessageEventData(result, this.#binaryType),
+              origin: new URL(this.url).origin,
+            })
+          );
+        }
       } catch (error) {
         this.#handleError(error);
         this.#handle = undefined;
@@ -441,13 +514,51 @@ export class WebSocket extends EventTarget {
       writable: false,
     });
 
-    if (this.#readyState === WebSocket.CONNECTING) {
-      this.#rejectOpened(error);
-    }
+    this.#rejectOpen(error);
 
     this.dispatchEvent(event);
   }
 
+  #resolveOpen(): void {
+    if (this.#openSettled) {
+      return;
+    }
+
+    this.#openSettled = true;
+    this.#resolveOpened();
+  }
+
+  #rejectOpen(error: unknown): void {
+    if (this.#openSettled) {
+      return;
+    }
+
+    this.#openSettled = true;
+    this.#rejectOpened(error);
+  }
+
+  #enqueueClose(handle: number): void {
+    if (this.#closeEnqueued) {
+      return;
+    }
+
+    this.#closeEnqueued = true;
+
+    const close = this.#closeRequest ?? {};
+
+    this.#sendQueue = this.#sendQueue
+      .then(() => nativeWebSocketClose(handle, close.code, close.reason))
+      .catch((error: unknown) => {
+        this.#handleError(error);
+        this.#handle = undefined;
+        this.#finalizeClose({
+          code: 1006,
+          reason: '',
+          wasClean: false,
+        });
+      });
+  }
+
   #finalizeClose(init: { code: number; reason: string; wasClean: boolean }): void {
     if (this.#settled) {
       return;
@@ -460,9 +571,7 @@ export class WebSocket extends EventTarget {
       this.#handle = undefined;
     }
 
-    if (init.wasClean === false) {
-      this.#rejectOpened(new WebSocketError('WebSocket connection closed before opening'));
-    }
+    this.#rejectOpen(new WebSocketError('WebSocket connection closed before opening'));
 
     this.dispatchEvent(new CloseEvent('close', init));
   }
diff --git a/src/websocket/send-data.ts b/src/websocket/send-data.ts
index 24e64cb..2de3ceb 100644
--- a/src/websocket/send-data.ts
+++ b/src/websocket/send-data.ts
@@ -1,6 +1,23 @@
 import type { NativeWebSocketReadResult, WebSocketBinaryType } from '../types';
 import { Buffer } from 'node:buffer';
 
+type QueuedSendData = string | Blob | ArrayBuffer | Uint8Array;
+
+/** Captures mutable buffer inputs at send() call time. */
+export function snapshotSendData(
+  data: string | Blob | ArrayBuffer | ArrayBufferView
+): QueuedSendData {
+  if (ArrayBuffer.isView(data)) {
+    return Uint8Array.from(new Uint8Array(data.buffer, data.byteOffset, data.byteLength));
+  }
+
+  if (data instanceof ArrayBuffer) {
+    return data.slice(0);
+  }
+
+  return data;
+}
+
 export function getSendByteLength(data: string | Blob | ArrayBuffer | ArrayBufferView): number {
   if (typeof data === 'string') {
     return Buffer.byteLength(data);
@@ -18,12 +35,10 @@ export function getSendByteLength(data: string | Blob | ArrayBuffer | ArrayBuffe
     return data.byteLength;
   }
 
-  return 0;
+  throw new TypeError('Unsupported WebSocket message type');
 }
 
-export async function normalizeSendData(
-  data: string | Blob | ArrayBuffer | ArrayBufferView
-): Promise<
+export async function normalizeSendData(data: QueuedSendData): Promise<
   | {
       type: 'text';
       data: string;
diff --git a/src/websocket/validation.ts b/src/websocket/validation.ts
index 31dec6d..6f7849b 100644
--- a/src/websocket/validation.ts
+++ b/src/websocket/validation.ts
@@ -1,6 +1,5 @@
 import type { HeadersInit, WebSocketInit } from '../types';
 import { Buffer } from 'node:buffer';
-import { WebSocketError } from '../errors';
 import { Headers } from '../headers';
 
 const SUBPROTOCOL_PATTERN = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;
@@ -28,12 +27,25 @@ function appendQuery(url: URL, query: WebSocketInit['query']): void {
 }
 
 export function resolveWebSocketUrl(rawUrl: string | URL, init?: WebSocketInit): string {
-  const url = init?.baseURL ? new URL(String(rawUrl), init.baseURL) : new URL(String(rawUrl));
+  let url: URL;
+
+  try {
+    url = init?.baseURL ? new URL(String(rawUrl), init.baseURL) : new URL(String(rawUrl));
+  } catch (error) {
+    throw new DOMException(
+      error instanceof Error ? error.message : 'Invalid WebSocket URL',
+      'SyntaxError'
+    );
+  }
 
   appendQuery(url, init?.query);
 
-  if (url.protocol !== 'ws:' && url.protocol !== 'wss:') {
-    throw new WebSocketError(`Invalid WebSocket URL protocol: ${url.protocol}`);
+  if (url.protocol === 'http:') {
+    url.protocol = 'ws:';
+  } else if (url.protocol === 'https:') {
+    url.protocol = 'wss:';
+  } else if (url.protocol !== 'ws:' && url.protocol !== 'wss:') {
+    throw new DOMException(`Invalid WebSocket URL protocol: ${url.protocol}`, 'SyntaxError');
   }
 
   if (url.hash) {
@@ -65,11 +77,11 @@ export function normalizeProtocols(protocols?: string | string[]): string[] {
 
   for (const value of values) {
     if (!SUBPROTOCOL_PATTERN.test(value)) {
-      throw new SyntaxError(`Invalid WebSocket subprotocol: ${value}`);
+      throw new DOMException(`Invalid WebSocket subprotocol: ${value}`, 'SyntaxError');
     }
 
     if (seen.has(value)) {
-      throw new SyntaxError(`Duplicate WebSocket subprotocol: ${value}`);
+      throw new DOMException(`Duplicate WebSocket subprotocol: ${value}`, 'SyntaxError');
     }
 
     seen.add(value);