From f4d13f8d7805b27e9a03a91635e2582248cd5c94 Mon Sep 17 00:00:00 2001
From: Timofey Ivanenko <timtin206@gmail.com>
Date: Mon, 26 Oct 2026 14:26:41 +0800
Subject: [PATCH 01/12] Signalling WebSocket over Network.framework: URLSession
 keeps every WebSocket session after it is closed

---
 .changes/websocket-network-framework          |   1 +
 .../LiveKit/Support/Network/WebSocket.swift   | 266 +++++++++++-------
 2 files changed, 173 insertions(+), 94 deletions(-)
 create mode 100644 .changes/websocket-network-framework

diff --git a/.changes/websocket-network-framework b/.changes/websocket-network-framework
new file mode 100644
index 000000000..915377503
--- /dev/null
+++ b/.changes/websocket-network-framework
@@ -0,0 +1 @@
+patch type="fixed" "Stop leaking about 25 KB per signalling connection: the WebSocket runs over Network.framework, as Foundation keeps every URLSession that ran a WebSocket task alive after it is closed; a rejected upgrade still reports its HTTP status"
diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index 644a06165..c248ccb39 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -23,25 +23,17 @@ struct WebSocketUpgradeFailure: Error, Sendable {
     let statusCode: Int
 }
 
+/// The signalling WebSocket, over Network.framework.
+///
+/// Not `URLSessionWebSocketTask`: Foundation keeps every `URLSession` that ran a WebSocket task, and the task,
+/// alive after the task is cancelled and the session invalidated, whichever way it is closed (about 25 KB a
+/// connection, measured on macOS 27 and the iOS 26.5 simulator). Every full or quick reconnect left one behind
+/// for the life of the process. An `NWConnection` is freed once cancelled.
 actor WebSocket: Loggable, AsyncSequence {
     typealias Element = URLSessionWebSocketTask.Message
 
-    private let delegate: Delegate
-    private let urlSession: URLSession
-    private let task: URLSessionWebSocketTask
-
-    private static func makeSessionConfiguration() -> URLSessionConfiguration {
-        let config = URLSessionConfiguration.default
-        config.timeoutIntervalForRequest = 60
-        config.timeoutIntervalForResource = 604_800
-        config.shouldUseExtendedBackgroundIdleMode = true
-        config.networkServiceType = .callSignaling
-        #if os(iOS) || os(visionOS)
-        // https://developer.apple.com/documentation/foundation/urlsessionconfiguration/improving_network_reliability_using_multipath_tcp
-        config.multipathServiceType = .handover
-        #endif
-        return config
-    }
+    private let connection: NWConnection
+    private static let queue = DispatchQueue(label: "LiveKitSDK.webSocket")
 
     /// Every header entry is added, not set, so a repeated name keeps all of its values.
     static func makeRequest(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) -> URLRequest {
@@ -54,134 +46,220 @@ actor WebSocket: Loggable, AsyncSequence {
         return request
     }
 
-    init(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) async throws {
-        let request = Self.makeRequest(url: url, headers: headers, timeoutInterval: timeoutInterval)
+    static func makeParameters(secure: Bool, timeoutInterval: TimeInterval) -> NWParameters {
+        let tcp = NWProtocolTCP.Options()
+        tcp.connectionTimeout = Swift.max(1, Int(timeoutInterval.rounded(.up)))
+        let parameters = NWParameters(tls: secure ? NWProtocolTLS.Options() : nil, tcp: tcp)
+        // As `.callSignaling` on the URLSession configuration this replaces.
+        parameters.serviceClass = .responsiveData
+        #if os(iOS) || os(visionOS)
+        // https://developer.apple.com/documentation/foundation/urlsessionconfiguration/improving_network_reliability_using_multipath_tcp
+        parameters.multipathServiceType = .handover
+        #endif
+        return parameters
+    }
 
+    init(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) async throws {
         #if targetEnvironment(simulator)
         if #available(iOS 26.0, *) {
             nw_tls_create_options()
         }
         #endif
 
-        delegate = Delegate()
-        urlSession = URLSession(configuration: Self.makeSessionConfiguration(),
-                                delegate: delegate, delegateQueue: nil)
-        task = urlSession.webSocketTask(with: request)
+        let webSocket = NWProtocolWebSocket.Options()
+        webSocket.autoReplyPing = true
+        webSocket.maximumMessageSize = 0 // no limit, as URLSessionWebSocketTask's 1 MB default would have been raised
+        webSocket.setAdditionalHeaders(headers)
+        let parameters = Self.makeParameters(secure: url.scheme == "wss", timeoutInterval: timeoutInterval)
+        parameters.defaultProtocolStack.applicationProtocols.insert(webSocket, at: 0)
+        let connection = NWConnection(to: .url(url), using: parameters)
+        self.connection = connection
 
-        try await withTaskCancellationHandler {
-            try await withCheckedThrowingContinuation { continuation in
-                delegate.setConnectContinuation(continuation)
-                task.resume()
+        let opening = Once<Void>()
+        do {
+            try await withTaskCancellationHandler {
+                try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
+                    opening.start(continuation)
+                    connection.stateUpdateHandler = { state in
+                        switch state {
+                        case .ready:
+                            opening.finish(.success(()))
+                        case let .failed(error), let .waiting(error):
+                            // `.waiting` too: a connection that cannot be made now fails, as URLSession's did
+                            // (waitsForConnectivity off); the signal client owns the retries.
+                            opening.finish(.failure(error))
+                        case .cancelled:
+                            opening.finish(.failure(LiveKitError(.cancelled)))
+                        default:
+                            break
+                        }
+                    }
+                    connection.start(queue: Self.queue)
+                    Self.queue.asyncAfter(deadline: .now() + timeoutInterval) {
+                        opening.finish(.failure(URLError(.timedOut)))
+                    }
+                }
+            } onCancel: {
+                connection.cancel()
+            }
+        } catch let error as NWError {
+            connection.cancel()
+            // The server answered the upgrade with an HTTP status: Network.framework aborts the connection
+            // without it, so the same upgrade, sent once more by hand, reads it.
+            if case .posix(.ECONNABORTED) = error,
+               let status = await Self.upgradeStatus(url: url, headers: headers, timeoutInterval: timeoutInterval)
+            {
+                throw LiveKitError(.network, internalError: WebSocketUpgradeFailure(statusCode: status))
             }
-        } onCancel: {
-            self.close()
+            throw LiveKitError(.network, internalError: Self.bridged(error))
+        } catch {
+            connection.cancel()
+            throw error is LiveKitError ? error : LiveKitError.from(error: error) ?? LiveKitError(.unknown)
         }
     }
 
     deinit {
-        close()
+        connection.cancel()
     }
 
     nonisolated func close() {
-        task.cancel(with: .normalClosure, reason: nil)
-        urlSession.finishTasksAndInvalidate()
-        delegate.cancelConnection()
+        connection.cancel()
+    }
+
+    /// The HTTP status the server answers the upgrade with, read off the status line of the same request sent over
+    /// a plain TCP (and TLS) connection; nil when it gives none or switches protocols after all.
+    static func upgradeStatus(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) async -> Int? {
+        guard let host = url.host else { return nil }
+        let secure = url.scheme == "wss"
+        let port = url.port ?? (secure ? 443 : 80)
+        var target = url.path.isEmpty ? "/" : url.path
+        if let query = url.query { target += "?\(query)" }
+        let key = Data((0 ..< 16).map { _ in UInt8.random(in: 0 ... 255) }).base64EncodedString()
+        var request = "GET \(target) HTTP/1.1\r\nHost: \(host):\(port)\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
+            + "Sec-WebSocket-Key: \(key)\r\nSec-WebSocket-Version: 13\r\n"
+        for (name, value) in headers { request += "\(name): \(value)\r\n" }
+        request += "\r\n"
+        let bytes = Data(request.utf8)
+
+        guard let nwPort = NWEndpoint.Port(rawValue: UInt16(clamping: port)) else { return nil }
+        let connection = NWConnection(host: NWEndpoint.Host(host), port: nwPort,
+                                      using: makeParameters(secure: secure, timeoutInterval: timeoutInterval))
+        defer { connection.cancel() }
+        let answer = Once<Int?>()
+        return try? await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Int?, Error>) in
+            answer.start(continuation)
+            connection.stateUpdateHandler = { state in
+                switch state {
+                case .ready:
+                    connection.send(content: bytes, completion: .contentProcessed { _ in })
+                    connection.receive(minimumIncompleteLength: 12, maximumLength: 512) { content, _, _, _ in
+                        // "HTTP/1.1 401 Unauthorized"
+                        let line = content.flatMap { String(data: $0, encoding: .utf8) }?.split(separator: "\r\n").first
+                        let status = line.flatMap { $0.split(separator: " ").dropFirst().first }.flatMap { Int($0) }
+                        answer.finish(.success(status == 101 ? nil : status))
+                    }
+                case .failed, .waiting, .cancelled:
+                    answer.finish(.success(nil))
+                default:
+                    break
+                }
+            }
+            connection.start(queue: queue)
+            queue.asyncAfter(deadline: .now() + timeoutInterval) { answer.finish(.success(nil)) }
+        }
+    }
+
+    /// Network.framework's errors in the shapes the SDK reads elsewhere (`isNetworkError`, `asTransportError`):
+    /// a timeout as `URLError(.timedOut)`, any other POSIX code as an `NSPOSIXErrorDomain` error.
+    static func bridged(_ error: NWError) -> Error {
+        switch error {
+        case .posix(.ETIMEDOUT):
+            URLError(.timedOut)
+        case let .posix(code):
+            NSError(domain: NSPOSIXErrorDomain, code: Int(code.rawValue))
+        default:
+            error
+        }
     }
 
     // MARK: - AsyncSequence
 
     struct AsyncIterator: AsyncIteratorProtocol {
-        fileprivate let task: URLSessionWebSocketTask
+        fileprivate let connection: NWConnection
 
         func next() async throws -> URLSessionWebSocketTask.Message? {
-            guard task.closeCode == .invalid else { return nil }
-            return try await withTaskCancellationHandler {
-                do {
-                    // Use the callback API instead of the async overlay to avoid
-                    // a TSan-visible data race inside Foundation's continuation bridge.
-                    return try await withCheckedThrowingContinuation { continuation in
-                        task.receive { result in
-                            continuation.resume(with: result)
+            while true {
+                guard let frame = try await receive() else { return nil }
+                if let message = frame { return message }
+                // A control frame (ping, pong): not a message; read on.
+            }
+        }
+
+        /// nil at the end of the stream, `.some(nil)` for a frame that is not a message.
+        private func receive() async throws -> URLSessionWebSocketTask.Message?? {
+            try await withTaskCancellationHandler {
+                try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<URLSessionWebSocketTask.Message??, Error>) in
+                    connection.receiveMessage { content, context, _, error in
+                        if let error {
+                            // Our own close ends the sequence, as a cancelled task's did, rather than failing it.
+                            if case .posix(.ECANCELED) = error { return continuation.resume(returning: nil) }
+                            return continuation.resume(throwing: LiveKitError(.network, internalError: WebSocket.bridged(error)))
+                        }
+                        let metadata = context?.protocolMetadata(definition: NWProtocolWebSocket.definition) as? NWProtocolWebSocket.Metadata
+                        switch metadata?.opcode {
+                        case .binary?:
+                            continuation.resume(returning: .some(.data(content ?? Data())))
+                        case .text?:
+                            continuation.resume(returning: .some(.string(String(decoding: content ?? Data(), as: UTF8.self))))
+                        case .close?, nil:
+                            continuation.resume(returning: nil)
+                        default:
+                            continuation.resume(returning: .some(nil))
                         }
                     }
-                } catch {
-                    // On clean shutdown, task.receive() throws URLError(.cancelled)
-                    // rather than CancellationError. Return nil (end-of-sequence)
-                    // instead of propagating, so `subscribe` doesn't call onFailure.
-                    if task.closeCode != .invalid || Task.isCancelled { return nil }
-                    throw LiveKitError.from(error: error) ?? error
                 }
             } onCancel: {
-                task.cancel(with: .normalClosure, reason: nil)
+                connection.cancel()
             }
         }
     }
 
     nonisolated func makeAsyncIterator() -> AsyncIterator {
-        AsyncIterator(task: task)
+        AsyncIterator(connection: connection)
     }
 
     // MARK: - Send
 
     nonisolated func send(data: Data) async throws {
-        // Callback API rather than the async overlay, for the same reason as `next()` below: the
-        // overlay's continuation bridge carries a TSan-visible data race between the thread that
-        // creates the continuation and the one that completes it. CI caught it on the TSan leg,
-        // where it aborts the whole test bundle.
+        let context = NWConnection.ContentContext(identifier: "binary",
+                                                  metadata: [NWProtocolWebSocket.Metadata(opcode: .binary)])
         try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, any Error>) in
-            task.send(.data(data)) { error in
+            connection.send(content: data, contentContext: context, isComplete: true, completion: .contentProcessed { error in
                 if let error {
-                    continuation.resume(throwing: error)
+                    continuation.resume(throwing: LiveKitError(.network, internalError: WebSocket.bridged(error)))
                 } else {
                     continuation.resume()
                 }
-            }
+            })
         }
     }
 
-    // MARK: - Delegate
-
-    private final class Delegate: NSObject, Loggable, URLSessionWebSocketDelegate {
-        private let _continuation = StateSync<CheckedContinuation<Void, Error>?>(nil)
-
-        func setConnectContinuation(_ continuation: CheckedContinuation<Void, Error>) {
-            _continuation.mutate { $0 = continuation }
-        }
+    // MARK: - Opening
 
-        func cancelConnection() {
-            _continuation.mutate {
-                $0?.resume(throwing: LiveKitError(.cancelled))
-                $0 = nil
-            }
-        }
+    /// A continuation resumed once, by whichever comes first: the connection's answer or the timeout.
+    private final class Once<Value: Sendable>: Sendable {
+        private let continuation = StateSync<CheckedContinuation<Value, Error>?>(nil)
 
-        func urlSession(_: URLSession, webSocketTask _: URLSessionWebSocketTask, didOpenWithProtocol _: String?) {
-            _continuation.mutate {
-                $0?.resume()
-                $0 = nil
-            }
+        func start(_ continuation: CheckedContinuation<Value, Error>) {
+            self.continuation.mutate { $0 = continuation }
         }
 
-        func urlSession(_: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
-            log("didCompleteWithError: \(String(describing: error))", error != nil ? .error : .debug)
-
-            // A rejected upgrade (401 on a bad token, 404 on a wrong path) arrives as a
-            // plain URLError with the HTTP response still attached. Keep the status: it is
-            // the difference between failing fast and reconnecting forever.
-            let upgradeStatus = (task.response as? HTTPURLResponse)
-                .map(\.statusCode)
-                .flatMap { $0 == 101 ? nil : WebSocketUpgradeFailure(statusCode: $0) }
-
-            _continuation.mutate {
-                if let upgradeStatus {
-                    $0?.resume(throwing: LiveKitError(.network, internalError: upgradeStatus))
-                } else if let error {
-                    $0?.resume(throwing: LiveKitError.from(error: error) ?? LiveKitError(.unknown))
-                } else {
-                    $0?.resume()
-                }
-                $0 = nil
+        func finish(_ result: Result<Value, Error>) {
+            let pending = continuation.mutate { current -> CheckedContinuation<Value, Error>? in
+                defer { current = nil }
+                return current
             }
+            pending?.resume(with: result)
         }
     }
 }

From 1f1bc06b5cc534d93e39986f43552ae5206a36dd Mon Sep 17 00:00:00 2001
From: Timofey Ivanenko <timtin206@gmail.com>
Date: Mon, 26 Oct 2026 14:26:41 +0800
Subject: [PATCH 02/12] WebSocket upgrade probe sends the escaped request
 target; Network.framework callbacks on a global queue instead of a new one

---
 Sources/LiveKit/Support/Network/WebSocket.swift | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index c248ccb39..faa7b74ff 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -33,7 +33,9 @@ actor WebSocket: Loggable, AsyncSequence {
     typealias Element = URLSessionWebSocketTask.Message
 
     private let connection: NWConnection
-    private static let queue = DispatchQueue(label: "LiveKitSDK.webSocket")
+    /// Network.framework delivers on a queue it is given; a shared one, not one of our own: every callback here only
+    /// resumes a continuation, and those sit in `StateSync`.
+    private static var queue: DispatchQueue { .global(qos: .userInitiated) }
 
     /// Every header entry is added, not set, so a repeated name keeps all of its values.
     static func makeRequest(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) -> URLRequest {
@@ -132,8 +134,10 @@ actor WebSocket: Loggable, AsyncSequence {
         guard let host = url.host else { return nil }
         let secure = url.scheme == "wss"
         let port = url.port ?? (secure ? 443 : 80)
-        var target = url.path.isEmpty ? "/" : url.path
-        if let query = url.query { target += "?\(query)" }
+        // The request target as the upgrade sent it: escaped, as `URL.path` would not give it.
+        guard let components = URLComponents(url: url, resolvingAgainstBaseURL: false) else { return nil }
+        var target = components.percentEncodedPath.isEmpty ? "/" : components.percentEncodedPath
+        if let query = components.percentEncodedQuery { target += "?\(query)" }
         let key = Data((0 ..< 16).map { _ in UInt8.random(in: 0 ... 255) }).base64EncodedString()
         var request = "GET \(target) HTTP/1.1\r\nHost: \(host):\(port)\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
             + "Sec-WebSocket-Key: \(key)\r\nSec-WebSocket-Version: 13\r\n"

From 209b961327fe56a262df3c5a1698e8b0e365ef0f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:04:59 +0100
Subject: [PATCH 03/12] WebSocket: wait on an AsyncCompleter instead of a
 private Once continuation

The opening wait and the upgrade-status probe each kept a continuation in a
private `Once` class, with a separate `asyncAfter` block as the deadline. The
SDK's `AsyncCompleter` already does both: it resumes once, times out on its own,
and throws `LiveKitError(.cancelled)` when the waiting task is cancelled, so the
`catch` below it closes the connection in every failure case. `resume(with:)`
rather than `resume(throwing:)`, as the latter logs a late `.cancelled` at error.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 .../LiveKit/Support/Network/WebSocket.swift   | 103 +++++++-----------
 1 file changed, 37 insertions(+), 66 deletions(-)

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index faa7b74ff..dd69a869c 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -34,7 +34,7 @@ actor WebSocket: Loggable, AsyncSequence {
 
     private let connection: NWConnection
     /// Network.framework delivers on a queue it is given; a shared one, not one of our own: every callback here only
-    /// resumes a continuation, and those sit in `StateSync`.
+    /// resumes an `AsyncCompleter` or a continuation.
     private static var queue: DispatchQueue { .global(qos: .userInitiated) }
 
     /// Every header entry is added, not set, so a repeated name keeps all of its values.
@@ -77,33 +77,24 @@ actor WebSocket: Loggable, AsyncSequence {
         let connection = NWConnection(to: .url(url), using: parameters)
         self.connection = connection
 
-        let opening = Once<Void>()
-        do {
-            try await withTaskCancellationHandler {
-                try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
-                    opening.start(continuation)
-                    connection.stateUpdateHandler = { state in
-                        switch state {
-                        case .ready:
-                            opening.finish(.success(()))
-                        case let .failed(error), let .waiting(error):
-                            // `.waiting` too: a connection that cannot be made now fails, as URLSession's did
-                            // (waitsForConnectivity off); the signal client owns the retries.
-                            opening.finish(.failure(error))
-                        case .cancelled:
-                            opening.finish(.failure(LiveKitError(.cancelled)))
-                        default:
-                            break
-                        }
-                    }
-                    connection.start(queue: Self.queue)
-                    Self.queue.asyncAfter(deadline: .now() + timeoutInterval) {
-                        opening.finish(.failure(URLError(.timedOut)))
-                    }
-                }
-            } onCancel: {
-                connection.cancel()
+        let opening = AsyncCompleter<Void>(label: "WebSocket.open", defaultTimeout: timeoutInterval)
+        connection.stateUpdateHandler = { state in
+            switch state {
+            case .ready:
+                opening.resume(returning: ())
+            case let .failed(error), let .waiting(error):
+                // `.waiting` too: a connection that cannot be made now fails, as URLSession's did
+                // (waitsForConnectivity off); the signal client owns the retries.
+                opening.resume(with: .failure(error))
+            case .cancelled:
+                opening.resume(with: .failure(LiveKitError(.cancelled)))
+            default:
+                break
             }
+        }
+        connection.start(queue: Self.queue)
+        do {
+            try await opening.wait()
         } catch let error as NWError {
             connection.cancel()
             // The server answered the upgrade with an HTTP status: Network.framework aborts the connection
@@ -141,7 +132,9 @@ actor WebSocket: Loggable, AsyncSequence {
         let key = Data((0 ..< 16).map { _ in UInt8.random(in: 0 ... 255) }).base64EncodedString()
         var request = "GET \(target) HTTP/1.1\r\nHost: \(host):\(port)\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
             + "Sec-WebSocket-Key: \(key)\r\nSec-WebSocket-Version: 13\r\n"
-        for (name, value) in headers { request += "\(name): \(value)\r\n" }
+        for (name, value) in headers {
+            request += "\(name): \(value)\r\n"
+        }
         request += "\r\n"
         let bytes = Data(request.utf8)
 
@@ -149,28 +142,25 @@ actor WebSocket: Loggable, AsyncSequence {
         let connection = NWConnection(host: NWEndpoint.Host(host), port: nwPort,
                                       using: makeParameters(secure: secure, timeoutInterval: timeoutInterval))
         defer { connection.cancel() }
-        let answer = Once<Int?>()
-        return try? await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Int?, Error>) in
-            answer.start(continuation)
-            connection.stateUpdateHandler = { state in
-                switch state {
-                case .ready:
-                    connection.send(content: bytes, completion: .contentProcessed { _ in })
-                    connection.receive(minimumIncompleteLength: 12, maximumLength: 512) { content, _, _, _ in
-                        // "HTTP/1.1 401 Unauthorized"
-                        let line = content.flatMap { String(data: $0, encoding: .utf8) }?.split(separator: "\r\n").first
-                        let status = line.flatMap { $0.split(separator: " ").dropFirst().first }.flatMap { Int($0) }
-                        answer.finish(.success(status == 101 ? nil : status))
-                    }
-                case .failed, .waiting, .cancelled:
-                    answer.finish(.success(nil))
-                default:
-                    break
+        let answer = AsyncCompleter<Int?>(label: "WebSocket.upgradeStatus", defaultTimeout: timeoutInterval)
+        connection.stateUpdateHandler = { state in
+            switch state {
+            case .ready:
+                connection.send(content: bytes, completion: .contentProcessed { _ in })
+                connection.receive(minimumIncompleteLength: 12, maximumLength: 512) { content, _, _, _ in
+                    // "HTTP/1.1 401 Unauthorized"
+                    let line = content.flatMap { String(data: $0, encoding: .utf8) }?.split(separator: "\r\n").first
+                    let status = line.flatMap { $0.split(separator: " ").dropFirst().first }.flatMap { Int($0) }
+                    answer.resume(returning: status == 101 ? nil : status)
                 }
+            case .failed, .waiting, .cancelled:
+                answer.resume(returning: nil)
+            default:
+                break
             }
-            connection.start(queue: queue)
-            queue.asyncAfter(deadline: .now() + timeoutInterval) { answer.finish(.success(nil)) }
         }
+        connection.start(queue: queue)
+        return try? await answer.wait()
     }
 
     /// Network.framework's errors in the shapes the SDK reads elsewhere (`isNetworkError`, `asTransportError`):
@@ -247,23 +237,4 @@ actor WebSocket: Loggable, AsyncSequence {
             })
         }
     }
-
-    // MARK: - Opening
-
-    /// A continuation resumed once, by whichever comes first: the connection's answer or the timeout.
-    private final class Once<Value: Sendable>: Sendable {
-        private let continuation = StateSync<CheckedContinuation<Value, Error>?>(nil)
-
-        func start(_ continuation: CheckedContinuation<Value, Error>) {
-            self.continuation.mutate { $0 = continuation }
-        }
-
-        func finish(_ result: Result<Value, Error>) {
-            let pending = continuation.mutate { current -> CheckedContinuation<Value, Error>? in
-                defer { current = nil }
-                return current
-            }
-            pending?.resume(with: result)
-        }
-    }
 }

From c484db12eaf977a22a0303ab4f5b271f8628db2b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:05:00 +0100
Subject: [PATCH 04/12] WebSocket: a peer FIN without a close frame fails the
 receive

Network.framework reports a bare TCP FIN as a completed receive with no
metadata. Treating that as the end of the stream, like a close frame, made the
signal client wait out its ping timeout before reconnecting. A FIN now fails
the receive with a network error so the reconnect starts at once; only a close
frame (or our own cancel) ends the sequence cleanly.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Sources/LiveKit/Support/Network/WebSocket.swift | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index dd69a869c..eaade11bf 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -205,7 +205,7 @@ actor WebSocket: Loggable, AsyncSequence {
                             continuation.resume(returning: .some(.data(content ?? Data())))
                         case .text?:
                             continuation.resume(returning: .some(.string(String(decoding: content ?? Data(), as: UTF8.self))))
-                        case .close?, nil:
+                        case .close?:
                             continuation.resume(returning: nil)
                         default:
                             continuation.resume(returning: .some(nil))

From 743b0f2e832109fb7ec08418d925c95a9fc0ee29 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:05:29 +0100
Subject: [PATCH 05/12] Delete the unused WebSocket.makeRequest and its test

The Network.framework transport sends its upgrade headers through
`NWProtocolWebSocket.Options.setAdditionalHeaders`; the `URLRequest` builder
was left over from the URLSession transport and had no caller apart from the
test that asserted its header merging.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Sources/LiveKit/Support/Network/WebSocket.swift | 11 -----------
 Tests/LiveKitCoreTests/RustTransportTests.swift | 17 +----------------
 2 files changed, 1 insertion(+), 27 deletions(-)

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index eaade11bf..afe57f6f6 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -37,17 +37,6 @@ actor WebSocket: Loggable, AsyncSequence {
     /// resumes an `AsyncCompleter` or a continuation.
     private static var queue: DispatchQueue { .global(qos: .userInitiated) }
 
-    /// Every header entry is added, not set, so a repeated name keeps all of its values.
-    static func makeRequest(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) -> URLRequest {
-        var request = URLRequest(url: url,
-                                 cachePolicy: .useProtocolCachePolicy,
-                                 timeoutInterval: timeoutInterval)
-        for (name, value) in headers {
-            request.addValue(value, forHTTPHeaderField: name)
-        }
-        return request
-    }
-
     static func makeParameters(secure: Bool, timeoutInterval: TimeInterval) -> NWParameters {
         let tcp = NWProtocolTCP.Options()
         tcp.connectionTimeout = Swift.max(1, Int(timeoutInterval.rounded(.up)))
diff --git a/Tests/LiveKitCoreTests/RustTransportTests.swift b/Tests/LiveKitCoreTests/RustTransportTests.swift
index fe4d6b52b..9daf9ddc5 100644
--- a/Tests/LiveKitCoreTests/RustTransportTests.swift
+++ b/Tests/LiveKitCoreTests/RustTransportTests.swift
@@ -16,11 +16,11 @@
 
 import Foundation
 @testable import LiveKit
+import LiveKitFFI
 import Testing
 #if canImport(LiveKitTestSupport)
 import LiveKitTestSupport
 #endif
-import LiveKitFFI
 
 /// Drives the registered transport from the Rust side, through the FFI, using
 /// `livekit-net`'s own `self_test_*` exports — the same entry points the Rust
@@ -262,18 +262,3 @@ struct RustHTTPRequestTests {
         #expect(request.httpBody == Data([0x01, 0x02]))
     }
 }
-
-/// `livekit-net` hands the Swift side an ordered header list for the WebSocket upgrade.
-@Suite(.tags(.networking))
-struct RustWebSocketRequestTests {
-    @Test("Repeated header names keep every value")
-    func repeatedHeadersKeepEveryValue() throws {
-        let request = try WebSocket.makeRequest(url: #require(URL(string: "wss://example.test/rtc")),
-                                                headers: [(name: "Sec-WebSocket-Protocol", value: "a"),
-                                                          (name: "Sec-WebSocket-Protocol", value: "b"),
-                                                          (name: "sec-websocket-protocol", value: "c")],
-                                                timeoutInterval: 10)
-
-        #expect(request.value(forHTTPHeaderField: "Sec-WebSocket-Protocol") == "a,b,c")
-    }
-}

From 6a89df525a8aec64f072aacfc8a3fd575afa5f31 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:05:30 +0100
Subject: [PATCH 06/12] Drop the redundant maximumMessageSize assignment

`NWProtocolWebSocket.Options.maximumMessageSize` already defaults to 0, no
limit; the assignment restated the default.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Sources/LiveKit/Support/Network/WebSocket.swift | 1 -
 1 file changed, 1 deletion(-)

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index afe57f6f6..484267081 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -59,7 +59,6 @@ actor WebSocket: Loggable, AsyncSequence {
 
         let webSocket = NWProtocolWebSocket.Options()
         webSocket.autoReplyPing = true
-        webSocket.maximumMessageSize = 0 // no limit, as URLSessionWebSocketTask's 1 MB default would have been raised
         webSocket.setAdditionalHeaders(headers)
         let parameters = Self.makeParameters(secure: url.scheme == "wss", timeoutInterval: timeoutInterval)
         parameters.defaultProtocolStack.applicationProtocols.insert(webSocket, at: 0)

From 632e35e957f6d92c999df017d10bbedb0ba90868 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:06:08 +0100
Subject: [PATCH 07/12] Decode text frames with the failable
 String(bytes:encoding:)

SwiftLint's optional_data_string_conversion flags the lossy
`String(decoding:as:)` on an optional payload. The failable form keeps every
valid frame as it was, an absent payload included, and fails the receive on a
frame that is not UTF-8 instead of handing the signal client replacement
characters.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Sources/LiveKit/Support/Network/WebSocket.swift | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index 484267081..d553b9a6a 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -192,7 +192,11 @@ actor WebSocket: Loggable, AsyncSequence {
                         case .binary?:
                             continuation.resume(returning: .some(.data(content ?? Data())))
                         case .text?:
-                            continuation.resume(returning: .some(.string(String(decoding: content ?? Data(), as: UTF8.self))))
+                            if let text = String(bytes: content ?? Data(), encoding: .utf8) {
+                                continuation.resume(returning: .some(.string(text)))
+                            } else {
+                                continuation.resume(throwing: LiveKitError(.network, message: "Text frame is not valid UTF-8"))
+                            }
                         case .close?:
                             continuation.resume(returning: nil)
                         default:

From de4fe3732da089f05f67efa3bcd5747611fdaf72 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:09:01 +0100
Subject: [PATCH 08/12] Add WebSocket lifecycle tests against a misbehaving
 server

A loopback server that answers the upgrade by hand and then misbehaves on
purpose: a close frame, a bare FIN, a reset, an HTTP status, or no answer. The
suite covers how the socket opens and fails to open, how its message sequence
ends for each way a peer can go away, and the retention the transport change
exists for: twenty sockets opened and closed leave no socket and no
NWConnection behind, where the URLSession transport kept its session and task
for the life of the process.

The socket's connection is internal rather than private so the release test
can hold a weak reference to it.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 .../LiveKit/Support/Network/WebSocket.swift   |   2 +-
 .../WebSocket/WebSocketLifecycleTests.swift   | 299 ++++++++++++++++++
 2 files changed, 300 insertions(+), 1 deletion(-)
 create mode 100644 Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift

diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index d553b9a6a..b289b9591 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -32,7 +32,7 @@ struct WebSocketUpgradeFailure: Error, Sendable {
 actor WebSocket: Loggable, AsyncSequence {
     typealias Element = URLSessionWebSocketTask.Message
 
-    private let connection: NWConnection
+    let connection: NWConnection
     /// Network.framework delivers on a queue it is given; a shared one, not one of our own: every callback here only
     /// resumes an `AsyncCompleter` or a continuation.
     private static var queue: DispatchQueue { .global(qos: .userInitiated) }
diff --git a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
new file mode 100644
index 000000000..050aa174b
--- /dev/null
+++ b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
@@ -0,0 +1,299 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import CryptoKit
+import Foundation
+@testable import LiveKit
+import Network
+import Testing
+
+/// A WebSocket server on a loopback port that answers the upgrade by hand and then misbehaves on purpose.
+/// Raw TCP rather than Network.framework's own WebSocket listener, so it can send what that never would:
+/// a bare FIN, a reset, an HTTP status, or nothing at all.
+final class FakeWebSocketServer: Sendable {
+    enum Behavior: Sendable {
+        /// Upgrade, then stay quiet until the client closes.
+        case idle
+        /// Upgrade, then a close frame with code 1000, then drop the socket once the client answers it.
+        case closeFrame
+        /// Upgrade, then a TCP FIN with no close frame.
+        case fin
+        /// Upgrade, then a TCP reset.
+        case reset
+        /// Answer the upgrade with an HTTP status instead of 101.
+        case reject(status: Int)
+        /// Accept TCP and never answer the upgrade.
+        case blackhole
+    }
+
+    let behavior: Behavior
+    /// Resumed once the server has finished with a connection: sent its last bytes or dropped the socket.
+    let served = AsyncCompleter<Void>(label: "FakeWebSocketServer.served", defaultTimeout: 5)
+
+    private let listener: NWListener
+    private let queue = DispatchQueue(label: "FakeWebSocketServer")
+    private let ready = AsyncCompleter<UInt16>(label: "FakeWebSocketServer.ready", defaultTimeout: 5)
+    private let stopped = AsyncCompleter<Void>(label: "FakeWebSocketServer.stopped", defaultTimeout: 5)
+    private let connections = StateSync<[NWConnection]>([])
+
+    init(_ behavior: Behavior) throws {
+        self.behavior = behavior
+        listener = try NWListener(using: .tcp, on: .any)
+        listener.stateUpdateHandler = { [ready, stopped, listener] state in
+            switch state {
+            case .ready: ready.resume(returning: listener.port?.rawValue ?? 0)
+            case let .failed(error): ready.resume(throwing: error)
+            case .cancelled: stopped.resume(returning: ())
+            default: break
+            }
+        }
+        listener.newConnectionHandler = { [self] connection in
+            connections.mutate { $0.append(connection) }
+            connection.start(queue: queue)
+            if case .blackhole = behavior { return }
+            readUpgrade(connection, buffered: Data())
+        }
+        listener.start(queue: queue)
+    }
+
+    func url() async throws -> URL {
+        let port = try await ready.wait()
+        return try #require(URL(string: "ws://127.0.0.1:\(port)/rtc"))
+    }
+
+    /// Returns once the port refuses connections again.
+    func stop() async throws {
+        connections.mutate { $0.forEach { $0.cancel() }; $0.removeAll() }
+        listener.cancel()
+        try await stopped.wait()
+    }
+
+    private func readUpgrade(_ connection: NWConnection, buffered: Data) {
+        connection.receive(minimumIncompleteLength: 1, maximumLength: 4096) { [self] content, _, _, error in
+            guard error == nil, let content else { return connection.cancel() }
+            let request = buffered + content
+            guard let headerEnd = request.range(of: Data("\r\n\r\n".utf8)) else {
+                return readUpgrade(connection, buffered: request)
+            }
+            let header = String(bytes: request[..<headerEnd.lowerBound], encoding: .utf8) ?? ""
+            let key = header.split(separator: "\r\n")
+                .first { $0.lowercased().hasPrefix("sec-websocket-key:") }
+                .map { $0.dropFirst("sec-websocket-key:".count).trimmingCharacters(in: .whitespaces) } ?? ""
+            answer(connection, key: key)
+        }
+    }
+
+    private func answer(_ connection: NWConnection, key: String) {
+        if case let .reject(status) = behavior {
+            let response = "HTTP/1.1 \(status) Rejected\r\nContent-Length: 0\r\n\r\n"
+            connection.send(content: Data(response.utf8), completion: .contentProcessed { [served] _ in
+                connection.cancel()
+                served.resume(returning: ())
+            })
+            return
+        }
+        let accept = Data(Insecure.SHA1.hash(data: Data((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").utf8))).base64EncodedString()
+        let response = "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: \(accept)\r\n\r\n"
+        connection.send(content: Data(response.utf8), completion: .contentProcessed { [self] _ in
+            switch behavior {
+            case .idle, .blackhole, .reject:
+                break
+            case .closeFrame:
+                connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [served] _ in
+                    // The client's close reply, then the socket goes: the client sees a state change after its stream ended.
+                    connection.receive(minimumIncompleteLength: 1, maximumLength: 64) { _, _, _, _ in
+                        connection.cancel()
+                        served.resume(returning: ())
+                    }
+                })
+            case .fin:
+                connection.send(content: nil, contentContext: .finalMessage, isComplete: true, completion: .contentProcessed { [served] _ in
+                    served.resume(returning: ())
+                })
+            case .reset:
+                connection.forceCancel()
+                served.resume(returning: ())
+            }
+        })
+    }
+}
+
+/// How iteration over the socket ends.
+enum Ending: Sendable, CustomTestStringConvertible {
+    case endOfStream
+    case networkError(posix: Int32?)
+
+    var testDescription: String {
+        switch self {
+        case .endOfStream: "end of stream"
+        case let .networkError(posix): "network error\(posix.map { " POSIX \($0)" } ?? "")"
+        }
+    }
+}
+
+private enum Limits {
+    /// As `TestLimits.e2e`, resolved at runtime for the iOS 13 floor; a minute is ample for loopback.
+    static var unit: any SuiteTrait {
+        if #available(iOS 16, macOS 13, tvOS 16, visionOS 1, *) {
+            let limit: TimeLimitTrait = .timeLimit(.minutes(1))
+            return limit
+        }
+        let none: Tag.List = .tags()
+        return none
+    }
+}
+
+/// The signalling socket's lifecycle against a server that is not a LiveKit server: how it opens, how it fails to,
+/// and how its message sequence ends for each way a peer can go away.
+@Suite(.tags(.networking), Limits.unit)
+struct WebSocketLifecycleTests {
+    private static let headers = [(name: "Authorization", value: "Bearer test")]
+
+    private static func open(_ server: FakeWebSocketServer, timeoutInterval: TimeInterval = 10) async throws -> LiveKit.WebSocket {
+        try await LiveKit.WebSocket(url: server.url(), headers: headers, timeoutInterval: timeoutInterval)
+    }
+
+    private static func messageCount(of socket: LiveKit.WebSocket) async throws -> Int {
+        try await socket.reduce(0) { count, _ in count + 1 }
+    }
+
+    // MARK: - Opening
+
+    @Test func ownCloseEndsTheSequence() async throws {
+        let server = try FakeWebSocketServer(.idle)
+        let socket = try await Self.open(server)
+        let reader = Task { try await Self.messageCount(of: socket) }
+        await Task.yield()
+        socket.close()
+        #expect(try await reader.value == 0)
+        try await server.stop()
+    }
+
+    @Test func refusedConnectionFails() async throws {
+        let server = try FakeWebSocketServer(.idle)
+        let url = try await server.url()
+        try await server.stop()
+
+        let error = await #expect(throws: LiveKitError.self) {
+            try await LiveKit.WebSocket(url: url, headers: Self.headers, timeoutInterval: 10)
+        }
+        #expect(error?.type == .network)
+        #expect((error?.internalError as? NSError)?.code == Int(ECONNREFUSED))
+    }
+
+    @Test func deadlineFailsAnUnansweredUpgrade() async throws {
+        let server = try FakeWebSocketServer(.blackhole)
+        let started = Date()
+        let error = await #expect(throws: LiveKitError.self) {
+            try await Self.open(server, timeoutInterval: 1)
+        }
+        #expect(Date().timeIntervalSince(started) < 5)
+        #expect(error?.type == .timedOut || error?.type == .network)
+        #expect(error?.isRetryableForRegionFailover == true)
+        try await server.stop()
+    }
+
+    @Test func cancellationDuringConnect() async throws {
+        let server = try FakeWebSocketServer(.blackhole)
+        let url = try await server.url()
+        let connect = Task { try await LiveKit.WebSocket(url: url, headers: Self.headers, timeoutInterval: 10) }
+        await Task.yield()
+        let started = Date()
+        connect.cancel()
+        let error = await #expect(throws: LiveKitError.self) { try await connect.value }
+        #expect(error?.type == .cancelled)
+        #expect(Date().timeIntervalSince(started) < 5)
+        try await server.stop()
+    }
+
+    @Test func rejectedUpgradeCarriesTheStatus() async throws {
+        let server = try FakeWebSocketServer(.reject(status: 401))
+        let error = await #expect(throws: LiveKitError.self) { try await Self.open(server) }
+        #expect(error?.type == .network)
+        #expect((error?.internalError as? WebSocketUpgradeFailure)?.statusCode == 401)
+        try await server.stop()
+    }
+
+    // MARK: - Release
+
+    /// What the Network.framework transport exists for: the URLSession one kept its session and task alive after
+    /// `close()`, about 25 KB a connection, for the life of the process. Here the only transport object is the
+    /// `NWConnection`, and it goes with the socket.
+    @Test(.bug("https://github.com/livekit/client-sdk-swift/pull/1173", "URLSession keeps every WebSocket session after close"),
+          .bug("https://developer.apple.com/forums/thread/650733"),
+          .bug("https://developer.apple.com/forums/thread/725039"))
+    func closedSocketsAreReleased() async throws {
+        let server = try FakeWebSocketServer(.idle)
+        var sockets: [LiveKit.WebSocket] = []
+        var transports: [WeakTransport] = []
+        for _ in 0 ..< 20 {
+            let socket = try await Self.open(server)
+            sockets.append(socket)
+            await transports.append(WeakTransport(socket: socket, connection: socket.connection))
+        }
+        for socket in sockets {
+            socket.close()
+        }
+        sockets.removeAll()
+
+        let deadline = Date().addingTimeInterval(10)
+        while transports.contains(where: { !$0.isReleased }), Date() < deadline {
+            await Task.yield()
+        }
+        for (index, transport) in transports.enumerated() {
+            #expect(transport.socket == nil, "Socket \(index) is still retained")
+            #expect(transport.connection == nil, "Connection \(index) is still retained")
+        }
+        try await server.stop()
+    }
+
+    // MARK: - Peer closure
+
+    /// A close frame is the only end of stream; a FIN or a reset fails the sequence so the signal client reconnects
+    /// at once instead of waiting for its ping timeout.
+    @Test(arguments: zip(
+        [FakeWebSocketServer.Behavior.closeFrame, .fin, .reset],
+        [Ending.endOfStream, .networkError(posix: nil), .networkError(posix: ECONNRESET)],
+    ))
+    func peerClosure(behavior: FakeWebSocketServer.Behavior, ending: Ending) async throws {
+        let server = try FakeWebSocketServer(behavior)
+        let socket = try await Self.open(server)
+
+        switch ending {
+        case .endOfStream:
+            #expect(try await Self.messageCount(of: socket) == 0)
+        case let .networkError(posix):
+            let error = await #expect(throws: LiveKitError.self) { try await Self.messageCount(of: socket) }
+            #expect(error?.type == .network)
+            if let posix {
+                #expect((error?.internalError as? NSError)?.code == Int(posix))
+            }
+        }
+
+        // The server drops its side after the sequence ended: the late state change lands on a socket whose
+        // opening is long settled, and must neither crash nor resurface.
+        try await server.served.wait()
+        withExtendedLifetime(socket) {}
+        try await server.stop()
+    }
+}
+
+private struct WeakTransport {
+    weak var socket: LiveKit.WebSocket?
+    weak var connection: NWConnection?
+
+    var isReleased: Bool { socket == nil && connection == nil }
+}

From 2c04909d63c029ee4b6c44eea96928f254c549e3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:32:21 +0100
Subject: [PATCH 09/12] WebSocket lifecycle tests: unconditional server
 teardown, no private queue, deterministic peer closure

Review findings on the lifecycle suite. Every test now cancels its fake
server in a `defer`, so a throw no longer leaves the listener, and the server
its handler captures, alive; the awaited `stop()` stays where the port has to
refuse connections. The server delivers on the shared global queue, as the
transport under test does, instead of a serial queue of its own. The
cancellation test waits for the listener to accept the TCP connection before
cancelling, so the cancel lands during the connect rather than before it.

The peer-closure cases were flaky: the server went away straight after the
101, racing the client's handshake, and a close frame that arrives while no
receive is pending is consumed by Network.framework's close handshake, after
which the next receive fails with ENOTCONN. The server now goes away only
after a frame from the client, and the close-frame case accepts either a clean
end or that ENOTCONN. Twenty consecutive runs pass.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 .../WebSocket/WebSocketLifecycleTests.swift   | 94 ++++++++++++-------
 1 file changed, 62 insertions(+), 32 deletions(-)

diff --git a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
index 050aa174b..976fb090e 100644
--- a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
+++ b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
@@ -40,11 +40,14 @@ final class FakeWebSocketServer: Sendable {
     }
 
     let behavior: Behavior
+    /// Resumed once the listener has accepted a TCP connection, before any upgrade is read.
+    let accepted = AsyncCompleter<Void>(label: "FakeWebSocketServer.accepted", defaultTimeout: 5)
     /// Resumed once the server has finished with a connection: sent its last bytes or dropped the socket.
     let served = AsyncCompleter<Void>(label: "FakeWebSocketServer.served", defaultTimeout: 5)
 
     private let listener: NWListener
-    private let queue = DispatchQueue(label: "FakeWebSocketServer")
+    /// Network.framework delivers on a queue it is given; a shared one, as the transport under test uses.
+    private static var queue: DispatchQueue { .global(qos: .userInitiated) }
     private let ready = AsyncCompleter<UInt16>(label: "FakeWebSocketServer.ready", defaultTimeout: 5)
     private let stopped = AsyncCompleter<Void>(label: "FakeWebSocketServer.stopped", defaultTimeout: 5)
     private let connections = StateSync<[NWConnection]>([])
@@ -62,11 +65,12 @@ final class FakeWebSocketServer: Sendable {
         }
         listener.newConnectionHandler = { [self] connection in
             connections.mutate { $0.append(connection) }
-            connection.start(queue: queue)
+            connection.start(queue: Self.queue)
+            accepted.resume(returning: ())
             if case .blackhole = behavior { return }
             readUpgrade(connection, buffered: Data())
         }
-        listener.start(queue: queue)
+        listener.start(queue: Self.queue)
     }
 
     func url() async throws -> URL {
@@ -74,10 +78,15 @@ final class FakeWebSocketServer: Sendable {
         return try #require(URL(string: "ws://127.0.0.1:\(port)/rtc"))
     }
 
-    /// Returns once the port refuses connections again.
-    func stop() async throws {
+    /// Drops every connection and the listener; safe to call more than once.
+    func cancel() {
         connections.mutate { $0.forEach { $0.cancel() }; $0.removeAll() }
         listener.cancel()
+    }
+
+    /// Returns once the port refuses connections again.
+    func stop() async throws {
+        cancel()
         try await stopped.wait()
     }
 
@@ -111,34 +120,48 @@ final class FakeWebSocketServer: Sendable {
             switch behavior {
             case .idle, .blackhole, .reject:
                 break
-            case .closeFrame:
-                connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [served] _ in
-                    // The client's close reply, then the socket goes: the client sees a state change after its stream ended.
-                    connection.receive(minimumIncompleteLength: 1, maximumLength: 64) { _, _, _, _ in
-                        connection.cancel()
-                        served.resume(returning: ())
-                    }
-                })
-            case .fin:
-                connection.send(content: nil, contentContext: .finalMessage, isComplete: true, completion: .contentProcessed { [served] _ in
-                    served.resume(returning: ())
-                })
-            case .reset:
-                connection.forceCancel()
-                served.resume(returning: ())
+            case .closeFrame, .fin, .reset:
+                // Only once a frame arrives from the client: its upgrade is then complete on its side too.
+                connection.receive(minimumIncompleteLength: 1, maximumLength: 64) { [self] _, _, _, _ in
+                    goAway(connection)
+                }
             }
         })
     }
+
+    private func goAway(_ connection: NWConnection) {
+        switch behavior {
+        case .idle, .blackhole, .reject:
+            break
+        case .closeFrame:
+            connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [served] _ in
+                // The client's close reply, then the socket goes: the client sees a state change after its stream ended.
+                connection.receive(minimumIncompleteLength: 1, maximumLength: 64) { _, _, _, _ in
+                    connection.cancel()
+                    served.resume(returning: ())
+                }
+            })
+        case .fin:
+            connection.send(content: nil, contentContext: .finalMessage, isComplete: true, completion: .contentProcessed { [served] _ in
+                served.resume(returning: ())
+            })
+        case .reset:
+            connection.forceCancel()
+            served.resume(returning: ())
+        }
+    }
 }
 
 /// How iteration over the socket ends.
 enum Ending: Sendable, CustomTestStringConvertible {
-    case endOfStream
+    /// A close frame delivered to a pending receive ends the stream. One that lands while no receive is pending is
+    /// consumed by Network.framework's close handshake, and the next receive fails with ENOTCONN instead.
+    case closed
     case networkError(posix: Int32?)
 
     var testDescription: String {
         switch self {
-        case .endOfStream: "end of stream"
+        case .closed: "closed"
         case let .networkError(posix): "network error\(posix.map { " POSIX \($0)" } ?? "")"
         }
     }
@@ -174,16 +197,17 @@ struct WebSocketLifecycleTests {
 
     @Test func ownCloseEndsTheSequence() async throws {
         let server = try FakeWebSocketServer(.idle)
+        defer { server.cancel() }
         let socket = try await Self.open(server)
         let reader = Task { try await Self.messageCount(of: socket) }
         await Task.yield()
         socket.close()
         #expect(try await reader.value == 0)
-        try await server.stop()
     }
 
     @Test func refusedConnectionFails() async throws {
         let server = try FakeWebSocketServer(.idle)
+        defer { server.cancel() }
         let url = try await server.url()
         try await server.stop()
 
@@ -196,6 +220,7 @@ struct WebSocketLifecycleTests {
 
     @Test func deadlineFailsAnUnansweredUpgrade() async throws {
         let server = try FakeWebSocketServer(.blackhole)
+        defer { server.cancel() }
         let started = Date()
         let error = await #expect(throws: LiveKitError.self) {
             try await Self.open(server, timeoutInterval: 1)
@@ -203,28 +228,27 @@ struct WebSocketLifecycleTests {
         #expect(Date().timeIntervalSince(started) < 5)
         #expect(error?.type == .timedOut || error?.type == .network)
         #expect(error?.isRetryableForRegionFailover == true)
-        try await server.stop()
     }
 
     @Test func cancellationDuringConnect() async throws {
         let server = try FakeWebSocketServer(.blackhole)
+        defer { server.cancel() }
         let url = try await server.url()
         let connect = Task { try await LiveKit.WebSocket(url: url, headers: Self.headers, timeoutInterval: 10) }
-        await Task.yield()
+        try await server.accepted.wait()
         let started = Date()
         connect.cancel()
         let error = await #expect(throws: LiveKitError.self) { try await connect.value }
         #expect(error?.type == .cancelled)
         #expect(Date().timeIntervalSince(started) < 5)
-        try await server.stop()
     }
 
     @Test func rejectedUpgradeCarriesTheStatus() async throws {
         let server = try FakeWebSocketServer(.reject(status: 401))
+        defer { server.cancel() }
         let error = await #expect(throws: LiveKitError.self) { try await Self.open(server) }
         #expect(error?.type == .network)
         #expect((error?.internalError as? WebSocketUpgradeFailure)?.statusCode == 401)
-        try await server.stop()
     }
 
     // MARK: - Release
@@ -237,6 +261,7 @@ struct WebSocketLifecycleTests {
           .bug("https://developer.apple.com/forums/thread/725039"))
     func closedSocketsAreReleased() async throws {
         let server = try FakeWebSocketServer(.idle)
+        defer { server.cancel() }
         var sockets: [LiveKit.WebSocket] = []
         var transports: [WeakTransport] = []
         for _ in 0 ..< 20 {
@@ -257,7 +282,6 @@ struct WebSocketLifecycleTests {
             #expect(transport.socket == nil, "Socket \(index) is still retained")
             #expect(transport.connection == nil, "Connection \(index) is still retained")
         }
-        try await server.stop()
     }
 
     // MARK: - Peer closure
@@ -266,15 +290,22 @@ struct WebSocketLifecycleTests {
     /// at once instead of waiting for its ping timeout.
     @Test(arguments: zip(
         [FakeWebSocketServer.Behavior.closeFrame, .fin, .reset],
-        [Ending.endOfStream, .networkError(posix: nil), .networkError(posix: ECONNRESET)],
+        [Ending.closed, .networkError(posix: nil), .networkError(posix: ECONNRESET)],
     ))
     func peerClosure(behavior: FakeWebSocketServer.Behavior, ending: Ending) async throws {
         let server = try FakeWebSocketServer(behavior)
+        defer { server.cancel() }
         let socket = try await Self.open(server)
+        try await socket.send(data: Data([0x01]))
 
         switch ending {
-        case .endOfStream:
-            #expect(try await Self.messageCount(of: socket) == 0)
+        case .closed:
+            do {
+                #expect(try await Self.messageCount(of: socket) == 0)
+            } catch let error as LiveKitError {
+                #expect(error.type == .network)
+                #expect((error.internalError as? NSError)?.code == Int(ENOTCONN))
+            }
         case let .networkError(posix):
             let error = await #expect(throws: LiveKitError.self) { try await Self.messageCount(of: socket) }
             #expect(error?.type == .network)
@@ -287,7 +318,6 @@ struct WebSocketLifecycleTests {
         // opening is long settled, and must neither crash nor resurface.
         try await server.served.wait()
         withExtendedLifetime(socket) {}
-        try await server.stop()
     }
 }
 

From c449a5bd4075f1a4497f3f9e1c7f68a076567046 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?=
 <86720177+pblazej@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:45:12 +0100
Subject: [PATCH 10/12] WebSocket lifecycle tests: the fake server reads whole
 client frames

A raw TCP receive can return part of a frame. The server's first receive
after the upgrade could then leave the rest of the client's trigger frame in
the socket, and the receive meant for the close reply would take those bytes
for it and cancel too early. The server now parses the client frame header,
the masked 7-bit or 16-bit length and the mask key, and keeps receiving until
exactly that many payload bytes are in, both for the trigger frame and for the
close reply.

Co-authored-by: omnigent <noreply@omnigent.ai>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 .../WebSocket/WebSocketLifecycleTests.swift   | 37 +++++++++++++++----
 1 file changed, 30 insertions(+), 7 deletions(-)

diff --git a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
index 976fb090e..61e457a31 100644
--- a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
+++ b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
@@ -121,22 +121,45 @@ final class FakeWebSocketServer: Sendable {
             case .idle, .blackhole, .reject:
                 break
             case .closeFrame, .fin, .reset:
-                // Only once a frame arrives from the client: its upgrade is then complete on its side too.
-                connection.receive(minimumIncompleteLength: 1, maximumLength: 64) { [self] _, _, _, _ in
-                    goAway(connection)
-                }
+                // Only once a whole frame arrives from the client: its upgrade is then complete on its side too.
+                readFrame(connection, buffered: Data()) { [self] in goAway(connection) }
             }
         })
     }
 
+    /// Reads one complete client frame, however the bytes arrive, then calls `completion`. A client frame is masked: two
+    /// header bytes, a 16-bit length when the 7-bit one is 126, a 4-byte mask key, then the payload.
+    private func readFrame(_ connection: NWConnection, buffered: Data, completion: @escaping @Sendable () -> Void) {
+        connection.receive(minimumIncompleteLength: 1, maximumLength: 4096) { [self] content, _, _, error in
+            guard error == nil, let content else { return connection.cancel() }
+            let frame = buffered + content
+            guard frame.count >= 2 else { return readFrame(connection, buffered: frame, completion: completion) }
+            let length7 = Int(frame[frame.startIndex + 1] & 0x7F)
+            let masked = frame[frame.startIndex + 1] & 0x80 != 0
+            var headerLength = 2 + (masked ? 4 : 0)
+            var payloadLength = length7
+            if length7 == 126 {
+                headerLength += 2
+                guard frame.count >= 4 else { return readFrame(connection, buffered: frame, completion: completion) }
+                payloadLength = Int(frame[frame.startIndex + 2]) << 8 | Int(frame[frame.startIndex + 3])
+            } else if length7 == 127 {
+                return connection.cancel()
+            }
+            guard frame.count >= headerLength + payloadLength else {
+                return readFrame(connection, buffered: frame, completion: completion)
+            }
+            completion()
+        }
+    }
+
     private func goAway(_ connection: NWConnection) {
         switch behavior {
         case .idle, .blackhole, .reject:
             break
         case .closeFrame:
-            connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [served] _ in
-                // The client's close reply, then the socket goes: the client sees a state change after its stream ended.
-                connection.receive(minimumIncompleteLength: 1, maximumLength: 64) { _, _, _, _ in
+            connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [self] _ in
+                // The client's whole close reply, then the socket goes: the client sees a state change after its stream ended.
+                readFrame(connection, buffered: Data()) { [served] in
                     connection.cancel()
                     served.resume(returning: ())
                 }

From 3e9b6cbb7f3940179ca2bc5642ade8f3020c11a6 Mon Sep 17 00:00:00 2001
From: Timofey Ivanenko <timtin206@gmail.com>
Date: Mon, 26 Oct 2026 19:53:47 +0800
Subject: [PATCH 11/12] Changeset: the signalling connection no longer goes
 through a system HTTP or PAC proxy

---
 .changes/websocket-network-framework | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.changes/websocket-network-framework b/.changes/websocket-network-framework
index 915377503..dc359d934 100644
--- a/.changes/websocket-network-framework
+++ b/.changes/websocket-network-framework
@@ -1 +1 @@
-patch type="fixed" "Stop leaking about 25 KB per signalling connection: the WebSocket runs over Network.framework, as Foundation keeps every URLSession that ran a WebSocket task alive after it is closed; a rejected upgrade still reports its HTTP status"
+patch type="fixed" "Stop leaking about 25 KB per signalling connection: the WebSocket runs over Network.framework, as Foundation keeps every URLSession that ran a WebSocket task alive after it is closed; a rejected upgrade still reports its HTTP status. The signalling connection no longer goes through a system HTTP or PAC proxy, which URLSession honoured"

From 35bf0bf6d3ab4c2e09474628198b57f43b6ee958 Mon Sep 17 00:00:00 2001
From: Timofey Ivanenko <timtin206@gmail.com>
Date: Mon, 26 Oct 2026 21:11:48 +0800
Subject: [PATCH 12/12] WebSocket: a close frame from the peer fails the stream
 with its close code, so the signal client reconnects at once

---
 .changes/websocket-close-frame-reconnects     |  1 +
 .../Support/Network/RustTransport.swift       |  4 +
 .../LiveKit/Support/Network/WebSocket.swift   | 20 +++-
 .../Room/ServerRemovalTests.swift             | 92 +++++++++++++++++++
 .../LiveKitCoreTests/RustTransportTests.swift |  2 +
 .../WebSocket/WebSocketLifecycleTests.swift   | 25 ++---
 6 files changed, 131 insertions(+), 13 deletions(-)
 create mode 100644 .changes/websocket-close-frame-reconnects
 create mode 100644 Tests/LiveKitCoreTests/Room/ServerRemovalTests.swift

diff --git a/.changes/websocket-close-frame-reconnects b/.changes/websocket-close-frame-reconnects
new file mode 100644
index 000000000..925a148a8
--- /dev/null
+++ b/.changes/websocket-close-frame-reconnects
@@ -0,0 +1 @@
+patch type="changed" "A close frame from the server fails the signalling stream with its close code, so the client reconnects at once instead of after the ping timeout; our own close still ends it"
diff --git a/Sources/LiveKit/Support/Network/RustTransport.swift b/Sources/LiveKit/Support/Network/RustTransport.swift
index 990a826ef..38c66d6de 100644
--- a/Sources/LiveKit/Support/Network/RustTransport.swift
+++ b/Sources/LiveKit/Support/Network/RustTransport.swift
@@ -141,6 +141,10 @@ extension Error {
 
         let underlying = (self as? LiveKitError)?.internalError ?? self
 
+        // livekit-net reads a close as the end of the stream, as it always has.
+        if underlying is WebSocketClosedByPeer {
+            return .Closed
+        }
         if let upgrade = underlying as? WebSocketUpgradeFailure {
             return .Http(status: UInt16(clamping: upgrade.statusCode))
         }
diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift
index b289b9591..52d19eafd 100644
--- a/Sources/LiveKit/Support/Network/WebSocket.swift
+++ b/Sources/LiveKit/Support/Network/WebSocket.swift
@@ -23,6 +23,21 @@ struct WebSocketUpgradeFailure: Error, Sendable {
     let statusCode: Int
 }
 
+/// The peer's close frame, carried as the `internalError` of the thrown ``LiveKitError``: the stream fails with it, so
+/// the signal client reconnects at once, as the JS, Android, Flutter and Rust SDKs do on a server close.
+struct WebSocketClosedByPeer: Error, Sendable {
+    /// The close code the peer sent; 1005 when it sent none.
+    let code: UInt16
+
+    init(_ code: NWProtocolWebSocket.CloseCode) {
+        self.code = switch code {
+        case let .protocolCode(code): code.rawValue
+        case let .applicationCode(code), let .privateCode(code): code
+        @unknown default: 1005
+        }
+    }
+}
+
 /// The signalling WebSocket, over Network.framework.
 ///
 /// Not `URLSessionWebSocketTask`: Foundation keeps every `URLSession` that ran a WebSocket task, and the task,
@@ -198,7 +213,10 @@ actor WebSocket: Loggable, AsyncSequence {
                                 continuation.resume(throwing: LiveKitError(.network, message: "Text frame is not valid UTF-8"))
                             }
                         case .close?:
-                            continuation.resume(returning: nil)
+                            // The peer's close: an error, not the end of the stream, which only our own close is.
+                            let closed = WebSocketClosedByPeer(metadata?.closeCode ?? .protocolCode(.noStatusReceived))
+                            continuation.resume(throwing: LiveKitError(.network, message: "Closed by the peer (\(closed.code))",
+                                                                       internalError: closed))
                         default:
                             continuation.resume(returning: .some(nil))
                         }
diff --git a/Tests/LiveKitCoreTests/Room/ServerRemovalTests.swift b/Tests/LiveKitCoreTests/Room/ServerRemovalTests.swift
new file mode 100644
index 000000000..79d79901d
--- /dev/null
+++ b/Tests/LiveKitCoreTests/Room/ServerRemovalTests.swift
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+@testable import LiveKit
+import LiveKitFFI
+import Testing
+#if canImport(LiveKitTestSupport)
+import LiveKitTestSupport
+#endif
+
+/// The server's own end of a session is a leave and then a close frame. A close frame fails the signalling stream, so
+/// that the client reconnects when the server drops it; it must not turn a removal into a reconnect.
+@Suite(.serialized, .tags(.e2e), TestLimits.e2e) final class ServerRemovalTests: @unchecked Sendable {
+    private let reconnects = StateSync(0)
+    private let disconnects = StateSync(0)
+
+    @Test func aRemovedParticipantDisconnectsWithoutReconnecting() async throws {
+        try await TestEnvironment.withRoom(RoomTestingOptions(delegate: self)) { room in
+            let name = try #require(room.name)
+            let identity = try #require(room.localParticipant.identity?.stringValue)
+            try await Self.removeParticipant(identity, from: name)
+            try await poll(timeout: 10, for: "the removed participant's room to disconnect") {
+                self.disconnects.copy() > 0
+            }
+            #expect(room.connectionState == .disconnected)
+            #expect(self.reconnects.copy() == 0)
+        }
+    }
+
+    /// RoomService.RemoveParticipant over Twirp, with a room admin token.
+    private static func removeParticipant(_ identity: String, from room: String) async throws {
+        let generator = TokenGenerator(apiKey: TestEnvironment.readEnvironmentString(for: "LIVEKIT_TESTING_API_KEY", defaultValue: "devkey"),
+                                       apiSecret: TestEnvironment.readEnvironmentString(for: "LIVEKIT_TESTING_API_SECRET", defaultValue: "secret"),
+                                       identity: "admin")
+        generator.videoGrants = VideoGrants(
+            roomCreate: false,
+            roomList: false,
+            roomRecord: false,
+            roomAdmin: true,
+            roomJoin: false,
+            room: room,
+            destinationRoom: "",
+            canPublish: false,
+            canSubscribe: false,
+            canPublishData: false,
+            canPublishSources: [],
+            canUpdateOwnMetadata: false,
+            ingressAdmin: false,
+            hidden: false,
+            recorder: false,
+            agent: false,
+        )
+        let base = TestEnvironment.liveKitServerUrl()
+            .replacingOccurrences(of: "wss://", with: "https://")
+            .replacingOccurrences(of: "ws://", with: "http://")
+        var request = try URLRequest(url: #require(URL(string: "\(base)/twirp/livekit.RoomService/RemoveParticipant")))
+        request.httpMethod = "POST"
+        request.addValue("Bearer \(try generator.sign())", forHTTPHeaderField: "Authorization")
+        request.addValue("application/json", forHTTPHeaderField: "Content-Type")
+        request.httpBody = try JSONSerialization.data(withJSONObject: ["room": room, "identity": identity])
+        let (_, response) = try await URLSession.shared.data(for: request)
+        #expect((response as? HTTPURLResponse)?.statusCode == 200)
+    }
+}
+
+extension ServerRemovalTests: RoomDelegate {
+    func room(_: Room, didStartReconnectWithMode _: ReconnectMode) {
+        reconnects.mutate { $0 += 1 }
+    }
+
+    func roomIsReconnecting(_: Room) {
+        reconnects.mutate { $0 += 1 }
+    }
+
+    func room(_: Room, didDisconnectWithError _: LiveKitError?) {
+        disconnects.mutate { $0 += 1 }
+    }
+}
diff --git a/Tests/LiveKitCoreTests/RustTransportTests.swift b/Tests/LiveKitCoreTests/RustTransportTests.swift
index 9daf9ddc5..39bb357ff 100644
--- a/Tests/LiveKitCoreTests/RustTransportTests.swift
+++ b/Tests/LiveKitCoreTests/RustTransportTests.swift
@@ -17,6 +17,7 @@
 import Foundation
 @testable import LiveKit
 import LiveKitFFI
+import Network
 import Testing
 #if canImport(LiveKitTestSupport)
 import LiveKitTestSupport
@@ -215,6 +216,7 @@ struct TransportErrorMappingTests {
         (URLError(.networkConnectionLost), .Closed),
         (POSIXError(.ECONNRESET), .Closed),
         (POSIXError(.ENOTCONN), .Closed),
+        (LiveKitError(.network, internalError: WebSocketClosedByPeer(.protocolCode(.normalClosure))), .Closed),
         (LiveKitError(.cancelled), .Closed),
         // Already ours: crossing the FFI twice must not re-wrap.
         (TransportError.Other("passthrough"), .Other("passthrough")),
diff --git a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
index 61e457a31..c7218aa62 100644
--- a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
+++ b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift
@@ -177,14 +177,14 @@ final class FakeWebSocketServer: Sendable {
 
 /// How iteration over the socket ends.
 enum Ending: Sendable, CustomTestStringConvertible {
-    /// A close frame delivered to a pending receive ends the stream. One that lands while no receive is pending is
-    /// consumed by Network.framework's close handshake, and the next receive fails with ENOTCONN instead.
-    case closed
+    /// A close frame delivered to a pending receive fails the stream with its code. One that lands while no receive
+    /// is pending is consumed by Network.framework's close handshake, and the next receive fails with ENOTCONN instead.
+    case closedByPeer(code: UInt16)
     case networkError(posix: Int32?)
 
     var testDescription: String {
         switch self {
-        case .closed: "closed"
+        case let .closedByPeer(code): "closed by the peer (\(code))"
         case let .networkError(posix): "network error\(posix.map { " POSIX \($0)" } ?? "")"
         }
     }
@@ -309,11 +309,11 @@ struct WebSocketLifecycleTests {
 
     // MARK: - Peer closure
 
-    /// A close frame is the only end of stream; a FIN or a reset fails the sequence so the signal client reconnects
+    /// Only our own close ends the stream; a close frame, a FIN or a reset fails it, so the signal client reconnects
     /// at once instead of waiting for its ping timeout.
     @Test(arguments: zip(
         [FakeWebSocketServer.Behavior.closeFrame, .fin, .reset],
-        [Ending.closed, .networkError(posix: nil), .networkError(posix: ECONNRESET)],
+        [Ending.closedByPeer(code: 1000), .networkError(posix: nil), .networkError(posix: ECONNRESET)],
     ))
     func peerClosure(behavior: FakeWebSocketServer.Behavior, ending: Ending) async throws {
         let server = try FakeWebSocketServer(behavior)
@@ -322,12 +322,13 @@ struct WebSocketLifecycleTests {
         try await socket.send(data: Data([0x01]))
 
         switch ending {
-        case .closed:
-            do {
-                #expect(try await Self.messageCount(of: socket) == 0)
-            } catch let error as LiveKitError {
-                #expect(error.type == .network)
-                #expect((error.internalError as? NSError)?.code == Int(ENOTCONN))
+        case let .closedByPeer(code):
+            let error = await #expect(throws: LiveKitError.self) { try await Self.messageCount(of: socket) }
+            #expect(error?.type == .network)
+            if let closed = error?.internalError as? WebSocketClosedByPeer {
+                #expect(closed.code == code)
+            } else {
+                #expect((error?.internalError as NSError?)?.code == Int(ENOTCONN))
             }
         case let .networkError(posix):
             let error = await #expect(throws: LiveKitError.self) { try await Self.messageCount(of: socket) }