diff --git a/CHANGELOG.md b/CHANGELOG.md
index 56fdef005..d42e03051 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -86,6 +86,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Literal backticks in cloudflared, cloud-sql-proxy, SSH config, remote command and dump tool install messages.
 - Tunnel command preview showing port 0 or the wrong host when Port is blank or the connection uses a host list.
 - SSH tab host-list warning naming replica set failover for Redis and Kafka, and implying Sentinel works through a tunnel.
+- ClickHouse query timeout never reaching the server, and capped SELECTs failing at 60 seconds whatever it is set to.
+- Stop on a ClickHouse SELECT leaving the query running on the server.
+- ClickHouse connect errors such as a wrong password or unknown database shown as "Failed to establish connection".
+- ClickHouse client certificate and key never sent to servers that require mutual TLS.
 - iPhone and iPad reading a Safe Mode level they do not recognize from iCloud as Off.
 
 ### Security
diff --git a/Plugins/ClickHouseDriverPlugin/ClickHouseConnectFailure.swift b/Plugins/ClickHouseDriverPlugin/ClickHouseConnectFailure.swift
new file mode 100644
index 000000000..ce6fd25b1
--- /dev/null
+++ b/Plugins/ClickHouseDriverPlugin/ClickHouseConnectFailure.swift
@@ -0,0 +1,20 @@
+import Foundation
+import TableProPluginKit
+
+internal enum ClickHouseConnectFailure {
+    static func error(for failure: Error, tlsRefusal: SSLHandshakeError?) -> Error {
+        if let tlsRefusal {
+            return tlsRefusal
+        }
+        if let serverAnswer = failure as? ClickHouseError {
+            return serverAnswer
+        }
+        if let sslError = ClickHouseSSLClassifier.classifySSLError(failure) {
+            return sslError
+        }
+        return ClickHouseError(message: String(
+            format: String(localized: "Connection failed: %@"),
+            failure.localizedDescription
+        ))
+    }
+}
diff --git a/Plugins/ClickHouseDriverPlugin/ClickHouseHTTPChunks.swift b/Plugins/ClickHouseDriverPlugin/ClickHouseHTTPChunks.swift
index db1d77997..c42fa492c 100644
--- a/Plugins/ClickHouseDriverPlugin/ClickHouseHTTPChunks.swift
+++ b/Plugins/ClickHouseDriverPlugin/ClickHouseHTTPChunks.swift
@@ -83,9 +83,9 @@ internal final class ClickHouseHTTPChunks: NSObject, URLSessionDataDelegate, @un
 
     internal func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
         let failure = lock.withLock { () -> Error? in
-            guard failureStatusCode != nil else { return error }
+            guard let statusCode = failureStatusCode else { return error }
             let body = String(decoding: failureBody, as: UTF8.self) // swiftlint:disable:this optional_data_string_conversion
-            return ClickHouseError(message: body.trimmingCharacters(in: .whitespacesAndNewlines))
+            return ClickHouseError.httpFailure(statusCode: statusCode, body: body)
         }
         guard let failure else {
             continuation.finish()
diff --git a/Plugins/ClickHouseDriverPlugin/ClickHousePlugin.swift b/Plugins/ClickHouseDriverPlugin/ClickHousePlugin.swift
index a3358c92e..ca1c04d0a 100644
--- a/Plugins/ClickHouseDriverPlugin/ClickHousePlugin.swift
+++ b/Plugins/ClickHouseDriverPlugin/ClickHousePlugin.swift
@@ -7,6 +7,7 @@ import Foundation
 import os
 import TableProNumberFormatting
 import TableProPluginKit
+import TableProTLSClientIdentity
 
 final class ClickHousePlugin: NSObject, TableProPlugin, DriverPlugin {
     static let pluginName = "ClickHouse Driver"
@@ -129,11 +130,14 @@ struct ClickHouseError: Error, PluginDriverError {
     var pluginErrorMessage: String { message }
 
     static let notConnected = ClickHouseError(message: String(localized: "Not connected to database"))
-    static let connectionFailed = ClickHouseError(message: String(localized: "Failed to establish connection"))
     static let verifyCaNeedsCertificate = ClickHouseError(message: String(localized: """
         Verify CA needs a CA certificate. On the connection's Network tab, choose the CA certificate that signed \
         the server's certificate, or set SSL Mode to Verify Identity.
         """))
+    static let clientCertificateNeedsKey = ClickHouseError(message: String(localized: """
+        A client certificate needs its client key. On the connection's Network tab, choose the client key, or \
+        clear the client certificate.
+        """))
 
     static func unreadableCACertificate(at path: String) -> ClickHouseError {
         ClickHouseError(message: String(
@@ -141,6 +145,15 @@ struct ClickHouseError: Error, PluginDriverError {
             path
         ))
     }
+
+    static func httpFailure(statusCode: Int, body: String) -> ClickHouseError {
+        let serverText = body.trimmingCharacters(in: .whitespacesAndNewlines)
+        guard serverText.isEmpty else { return ClickHouseError(message: serverText) }
+        return ClickHouseError(message: String(
+            format: String(localized: "ClickHouse returned HTTP %lld."),
+            Int64(statusCode)
+        ))
+    }
 }
 
 // MARK: - Internal Query Result
@@ -169,6 +182,7 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
     var _currentDatabase: String
     var _lastQueryId: String?
     let _queryTimeout = HttpQueryTimeoutBox()
+    private(set) var acceptedExecutionTimeLimit: Int?
 
     static let logger = Logger(subsystem: "com.TablePro", category: "ClickHousePluginDriver")
 
@@ -253,13 +267,7 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
                 session = nil
             }
             Self.logger.error("Connection test failed: \(error.localizedDescription)")
-            if let refusal = tlsDelegate?.recordedRefusal {
-                throw refusal
-            }
-            if let sslError = ClickHouseSSLClassifier.classifySSLError(error) {
-                throw sslError
-            }
-            throw ClickHouseError.connectionFailed
+            throw ClickHouseConnectFailure.error(for: error, tlsRefusal: tlsDelegate?.recordedRefusal)
         }
 
         report(.preparingSession)
@@ -462,8 +470,10 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
 
     func applyQueryTimeout(_ seconds: Int) async throws {
         _queryTimeout.set(serverTimeoutSeconds: seconds)
+        lock.withLock { acceptedExecutionTimeLimit = nil }
         guard seconds > 0 else { return }
-        _ = try await execute(query: "SET max_execution_time = \(seconds)")
+        _ = try await sendStatement("SELECT 1", queryId: nil, params: nil, executionTimeLimit: seconds)
+        lock.withLock { acceptedExecutionTimeLimit = seconds }
     }
 
     // MARK: - Database Switching
@@ -497,7 +507,7 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
         lock.unlock()
         guard let connectedSession else { return }
 
-        let killConfig = URLSessionConfiguration.default
+        let killConfig = connectedSession.configuration
         killConfig.timeoutIntervalForRequest = 5
         let killSession = URLSession(
             configuration: killConfig,
@@ -549,14 +559,13 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
         rowCap: Int,
         continuation: AsyncThrowingStream<PluginStreamElement, Error>.Continuation
     ) async throws {
-        let (session, database) = try lock.withLock { () throws -> (URLSession, String) in
-            guard let session = self.session else { throw ClickHouseError.notConnected }
-            return (session, _currentDatabase)
-        }
-
+        let queryId = UUID().uuidString
+        let (session, database, executionTimeLimit) = try streamingContext(queryId: queryId)
         let request = try buildStreamRequest(
             query: Self.withoutTrailingSemicolons(query),
             database: database,
+            queryId: queryId,
+            executionTimeLimit: executionTimeLimit,
             rowCap: rowCap
         )
         try await streamTabSeparatedRows(
@@ -639,12 +648,14 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
         query: String,
         continuation: AsyncThrowingStream<PluginStreamElement, Error>.Continuation
     ) async throws {
-        let (session, database) = try lock.withLock { () throws -> (URLSession, String) in
-            guard let session = self.session else { throw ClickHouseError.notConnected }
-            return (session, _currentDatabase)
-        }
-
-        let request = try buildStreamRequest(query: Self.withoutTrailingSemicolons(query), database: database)
+        let queryId = UUID().uuidString
+        let (session, database, executionTimeLimit) = try streamingContext(queryId: queryId)
+        let request = try buildStreamRequest(
+            query: Self.withoutTrailingSemicolons(query),
+            database: database,
+            queryId: queryId,
+            executionTimeLimit: executionTimeLimit
+        )
         try await streamTabSeparatedRows(
             request: request,
             session: session,
@@ -653,7 +664,23 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
         )
     }
 
-    private func buildStreamRequest(query: String, database: String, rowCap: Int? = nil) throws -> URLRequest {
+    private func streamingContext(
+        queryId: String
+    ) throws -> (session: URLSession, database: String, executionTimeLimit: Int?) {
+        try lock.withLock {
+            guard let session = self.session else { throw ClickHouseError.notConnected }
+            _lastQueryId = queryId
+            return (session, _currentDatabase, acceptedExecutionTimeLimit)
+        }
+    }
+
+    private func buildStreamRequest(
+        query: String,
+        database: String,
+        queryId: String,
+        executionTimeLimit: Int?,
+        rowCap: Int? = nil
+    ) throws -> URLRequest {
         let useTLS = config.ssl.isEnabled
 
         var components = URLComponents()
@@ -666,10 +693,12 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
         if !database.isEmpty {
             queryItems.append(URLQueryItem(name: "database", value: database))
         }
+        queryItems.append(URLQueryItem(name: "query_id", value: queryId))
         queryItems.append(URLQueryItem(
             name: "default_format",
             value: ClickHouseResponseClassifier.requestedFormat
         ))
+        queryItems.append(contentsOf: Self.executionTimeLimitItems(executionTimeLimit))
         if let rowCap {
             /// The bound rides as an HTTP setting so the SQL in the body stays exactly what the
             /// user wrote. One row past the cap, so a full page can be told from a truncated one.
@@ -685,6 +714,7 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
 
         var request = URLRequest(url: url)
         request.httpMethod = "POST"
+        request.timeoutInterval = _queryTimeout.requestTimeoutInterval
 
         if let authorization = ClickHouseCredentials.basicAuthorizationHeader(
             username: config.username,
@@ -846,11 +876,13 @@ final class ClickHousePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
 
 final class ClickHouseTLSDelegate: NSObject, URLSessionDelegate, @unchecked Sendable {
     private enum Strategy {
+        case systemTrust
         case skipVerify
         case verifyChain(anchor: SecCertificate, checksHostname: Bool)
     }
 
     private let strategy: Strategy
+    private let clientCredential: URLCredential?
     private let lock = NSLock()
     private var refusal: SSLHandshakeError?
 
@@ -858,19 +890,30 @@ final class ClickHouseTLSDelegate: NSObject, URLSessionDelegate, @unchecked Send
         lock.withLock { refusal }
     }
 
-    private init(strategy: Strategy) {
+    private init(strategy: Strategy, clientCredential: URLCredential?) {
         self.strategy = strategy
+        self.clientCredential = clientCredential
     }
 
     static func make(for ssl: SSLConfiguration) throws -> ClickHouseTLSDelegate? {
+        guard ssl.isEnabled else { return nil }
+        let strategy = try serverTrustStrategy(for: ssl)
+        let clientCredential = try clientCredential(for: ssl)
+        if case .systemTrust = strategy, clientCredential == nil {
+            return nil
+        }
+        return ClickHouseTLSDelegate(strategy: strategy, clientCredential: clientCredential)
+    }
+
+    private static func serverTrustStrategy(for ssl: SSLConfiguration) throws -> Strategy {
         let caPath = ssl.caCertificatePath.trimmingCharacters(in: .whitespaces)
         switch ssl.mode {
         case .disabled:
-            return nil
+            return .systemTrust
         case .preferred, .required:
-            return ClickHouseTLSDelegate(strategy: .skipVerify)
+            return .skipVerify
         case .verifyIdentity:
-            guard !caPath.isEmpty else { return nil }
+            guard !caPath.isEmpty else { return .systemTrust }
             return try anchored(at: caPath, checksHostname: true)
         case .verifyCa:
             guard !caPath.isEmpty else { throw ClickHouseError.verifyCaNeedsCertificate }
@@ -878,11 +921,11 @@ final class ClickHouseTLSDelegate: NSObject, URLSessionDelegate, @unchecked Send
         }
     }
 
-    private static func anchored(at path: String, checksHostname: Bool) throws -> ClickHouseTLSDelegate {
+    private static func anchored(at path: String, checksHostname: Bool) throws -> Strategy {
         guard let anchor = loadAnchor(at: path) else {
             throw ClickHouseError.unreadableCACertificate(at: path)
         }
-        return ClickHouseTLSDelegate(strategy: .verifyChain(anchor: anchor, checksHostname: checksHostname))
+        return .verifyChain(anchor: anchor, checksHostname: checksHostname)
     }
 
     private static func loadAnchor(at path: String) -> SecCertificate? {
@@ -893,18 +936,48 @@ final class ClickHouseTLSDelegate: NSObject, URLSessionDelegate, @unchecked Send
         return SecCertificateCreateWithData(nil, der as CFData)
     }
 
+    private static func clientCredential(for ssl: SSLConfiguration) throws -> URLCredential? {
+        let certificatePath = ssl.clientCertificatePath.trimmingCharacters(in: .whitespaces)
+        let keyPath = ssl.clientKeyPath.trimmingCharacters(in: .whitespaces)
+        guard !certificatePath.isEmpty else { return nil }
+        guard !keyPath.isEmpty else { throw ClickHouseError.clientCertificateNeedsKey }
+        do {
+            return try TLSClientIdentity.credential(
+                certificateFile: URL(fileURLWithPath: certificatePath),
+                privateKeyFile: URL(fileURLWithPath: keyPath)
+            )
+        } catch let failure as TLSClientIdentityError {
+            throw ClickHouseError(message: failure.message(certificatePath: certificatePath, keyPath: keyPath))
+        }
+    }
+
     func urlSession(
         _ session: URLSession,
         didReceive challenge: URLAuthenticationChallenge,
         completionHandler: @escaping (URLSession.AuthChallengeDisposition, URLCredential?) -> Void
     ) {
-        guard challenge.protectionSpace.authenticationMethod == NSURLAuthenticationMethodServerTrust,
-              let serverTrust = challenge.protectionSpace.serverTrust else {
+        switch challenge.protectionSpace.authenticationMethod {
+        case NSURLAuthenticationMethodServerTrust:
+            answerServerTrust(challenge, completionHandler: completionHandler)
+        case NSURLAuthenticationMethodClientCertificate:
+            answerClientCertificate(completionHandler: completionHandler)
+        default:
+            completionHandler(.performDefaultHandling, nil)
+        }
+    }
+
+    private func answerServerTrust(
+        _ challenge: URLAuthenticationChallenge,
+        completionHandler: @escaping (URLSession.AuthChallengeDisposition, URLCredential?) -> Void
+    ) {
+        guard let serverTrust = challenge.protectionSpace.serverTrust else {
             completionHandler(.performDefaultHandling, nil)
             return
         }
 
         switch strategy {
+        case .systemTrust:
+            completionHandler(.performDefaultHandling, nil)
         case .skipVerify:
             completionHandler(.useCredential, URLCredential(trust: serverTrust))
         case .verifyChain(let anchor, let checksHostname):
@@ -921,6 +994,16 @@ final class ClickHouseTLSDelegate: NSObject, URLSessionDelegate, @unchecked Send
         }
     }
 
+    private func answerClientCertificate(
+        completionHandler: @escaping (URLSession.AuthChallengeDisposition, URLCredential?) -> Void
+    ) {
+        guard let clientCredential else {
+            completionHandler(.performDefaultHandling, nil)
+            return
+        }
+        completionHandler(.useCredential, clientCredential)
+    }
+
     static func refusal(for error: CFError?) -> SSLHandshakeError {
         let message = error.map { CFErrorCopyDescription($0) as String } ?? ""
         guard let error, CFErrorGetCode(error) == Int(errSecHostNameMismatch) else {
diff --git a/Plugins/ClickHouseDriverPlugin/ClickHousePluginDriver+Http.swift b/Plugins/ClickHouseDriverPlugin/ClickHousePluginDriver+Http.swift
index 5d2cbd513..f8f69546b 100644
--- a/Plugins/ClickHouseDriverPlugin/ClickHousePluginDriver+Http.swift
+++ b/Plugins/ClickHouseDriverPlugin/ClickHousePluginDriver+Http.swift
@@ -10,22 +10,24 @@ import TableProPluginKit
 extension ClickHousePluginDriver {
     // MARK: - Private HTTP Layer
 
-    func executeRaw(_ query: String, queryId: String? = nil) async throws -> CHQueryResult {
-        let (session, database) = try lock.withLock { () throws -> (URLSession, String) in
-            guard let session = self.session else { throw ClickHouseError.notConnected }
-            let database = _currentDatabase
-            if let queryId {
-                _lastQueryId = queryId
-            }
-            return (session, database)
-        }
+    var executionTimeLimit: Int? {
+        lock.withLock { acceptedExecutionTimeLimit }
+    }
 
-        var request = try buildRequest(query: query, database: database, queryId: queryId)
-        request.timeoutInterval = _queryTimeout.requestTimeoutInterval
-        return try await perform(request: request, session: session)
+    func executeRaw(_ query: String, queryId: String? = nil) async throws -> CHQueryResult {
+        try await sendStatement(query, queryId: queryId, params: nil, executionTimeLimit: executionTimeLimit)
     }
 
     func executeRawWithParams(_ query: String, params: [String: String?], queryId: String? = nil) async throws -> CHQueryResult {
+        try await sendStatement(query, queryId: queryId, params: params, executionTimeLimit: executionTimeLimit)
+    }
+
+    func sendStatement(
+        _ query: String,
+        queryId: String?,
+        params: [String: String?]?,
+        executionTimeLimit: Int?
+    ) async throws -> CHQueryResult {
         let (session, database) = try lock.withLock { () throws -> (URLSession, String) in
             guard let session = self.session else { throw ClickHouseError.notConnected }
             let database = _currentDatabase
@@ -35,11 +37,22 @@ extension ClickHousePluginDriver {
             return (session, database)
         }
 
-        var request = try buildRequest(query: query, database: database, queryId: queryId, params: params)
+        var request = try buildRequest(
+            query: query,
+            database: database,
+            queryId: queryId,
+            params: params,
+            executionTimeLimit: executionTimeLimit
+        )
         request.timeoutInterval = _queryTimeout.requestTimeoutInterval
         return try await perform(request: request, session: session)
     }
 
+    static func executionTimeLimitItems(_ seconds: Int?) -> [URLQueryItem] {
+        guard let seconds, seconds > 0 else { return [] }
+        return [URLQueryItem(name: "max_execution_time", value: String(seconds))]
+    }
+
     private func perform(request: URLRequest, session: URLSession) async throws -> CHQueryResult {
         let (data, response) = try await send(request: request, session: session)
 
@@ -50,7 +63,7 @@ extension ClickHousePluginDriver {
             let body = String(decoding: data, as: UTF8.self) // swiftlint:disable:this optional_data_string_conversion
             let exceptionCode = httpResponse.value(forHTTPHeaderField: "X-ClickHouse-Exception-Code") ?? "none"
             Self.logger.error("ClickHouse HTTP \(httpResponse.statusCode) exception \(exceptionCode): \(body)")
-            throw ClickHouseError(message: body.trimmingCharacters(in: .whitespacesAndNewlines))
+            throw ClickHouseError.httpFailure(statusCode: httpResponse.statusCode, body: body)
         }
 
         let headers = Self.headerFields(httpResponse)
@@ -107,7 +120,13 @@ extension ClickHousePluginDriver {
         return fields
     }
 
-    func buildRequest(query: String, database: String, queryId: String? = nil, params: [String: String?]? = nil) throws -> URLRequest {
+    func buildRequest(
+        query: String,
+        database: String,
+        queryId: String? = nil,
+        params: [String: String?]? = nil,
+        executionTimeLimit: Int? = nil
+    ) throws -> URLRequest {
         let useTLS = config.ssl.isEnabled
 
         var components = URLComponents()
@@ -123,6 +142,7 @@ extension ClickHousePluginDriver {
         if let queryId {
             queryItems.append(URLQueryItem(name: "query_id", value: queryId))
         }
+        queryItems.append(contentsOf: Self.executionTimeLimitItems(executionTimeLimit))
         queryItems.append(URLQueryItem(name: "send_progress_in_http_headers", value: "1"))
         queryItems.append(contentsOf: ClickHouseResponseClassifier.transportQueryItems(
             supportsWriteExceptionSetting: ClickHouseCapabilities.parse(serverVersion).hasWriteExceptionInOutputFormatSetting
diff --git a/TablePro/Resources/Localizable.xcstrings b/TablePro/Resources/Localizable.xcstrings
index 0535a15bc..70a35a6a7 100644
--- a/TablePro/Resources/Localizable.xcstrings
+++ b/TablePro/Resources/Localizable.xcstrings
@@ -184770,6 +184770,9 @@
     },
     "Next cursor: %@. The scan has not finished." : {
 
+    },
+    "ClickHouse returned HTTP %lld." : {
+
     }
   },
   "version" : "1.1"
diff --git a/TableProTests/Helpers/ClickHouseHTTPStub.swift b/TableProTests/Helpers/ClickHouseHTTPStub.swift
new file mode 100644
index 000000000..b70868c08
--- /dev/null
+++ b/TableProTests/Helpers/ClickHouseHTTPStub.swift
@@ -0,0 +1,114 @@
+import Foundation
+import TableProPluginKit
+
+struct ClickHouseStubRequest: Sendable {
+    let queryItems: [String: String]
+    let body: String
+    let timeoutInterval: TimeInterval
+
+    func item(_ name: String) -> String? {
+        queryItems[name]
+    }
+}
+
+struct ClickHouseStubReply: Sendable {
+    let statusCode: Int
+    let body: String
+
+    static let oneRow = ClickHouseStubReply(statusCode: 200, body: "n\nUInt8\n1\n")
+    static let empty = ClickHouseStubReply(statusCode: 200, body: "")
+}
+
+final class ClickHouseHTTPStubProtocol: URLProtocol, @unchecked Sendable {
+    typealias Responder = @Sendable (ClickHouseStubRequest) -> ClickHouseStubReply?
+
+    private static let lock = NSLock()
+    nonisolated(unsafe) private static var responders: [String: Responder] = [:]
+    nonisolated(unsafe) private static var recorded: [String: [ClickHouseStubRequest]] = [:]
+
+    static func register(host: String, responder: @escaping Responder) {
+        lock.withLock { responders[host] = responder }
+    }
+
+    static func requests(to host: String) -> [ClickHouseStubRequest] {
+        lock.withLock { recorded[host] ?? [] }
+    }
+
+    override class func canInit(with request: URLRequest) -> Bool { true }
+    override class func canonicalRequest(for request: URLRequest) -> URLRequest { request }
+
+    override func startLoading() {
+        guard let url = request.url, let host = url.host else { return }
+        let items = URLComponents(url: url, resolvingAgainstBaseURL: false)?.queryItems ?? []
+        let stubRequest = ClickHouseStubRequest(
+            queryItems: Dictionary(items.map { ($0.name, $0.value ?? "") }, uniquingKeysWith: { _, last in last }),
+            body: Self.bodyText(of: request),
+            timeoutInterval: request.timeoutInterval
+        )
+        let responder = Self.lock.withLock { () -> Responder? in
+            Self.recorded[host, default: []].append(stubRequest)
+            return Self.responders[host]
+        }
+        guard let responder,
+              let reply = responder(stubRequest),
+              let response = HTTPURLResponse(url: url, statusCode: reply.statusCode, httpVersion: nil, headerFields: nil)
+        else { return }
+        client?.urlProtocol(self, didReceive: response, cacheStoragePolicy: .notAllowed)
+        client?.urlProtocol(self, didLoad: Data(reply.body.utf8))
+        client?.urlProtocolDidFinishLoading(self)
+    }
+
+    override func stopLoading() {}
+
+    private static func bodyText(of request: URLRequest) -> String {
+        guard let stream = request.httpBodyStream else {
+            return String(bytes: request.httpBody ?? Data(), encoding: .utf8) ?? ""
+        }
+        stream.open()
+        defer { stream.close() }
+        var data = Data()
+        var buffer = [UInt8](repeating: 0, count: 1_024)
+        while stream.hasBytesAvailable {
+            let read = stream.read(&buffer, maxLength: buffer.count)
+            guard read > 0 else { break }
+            data.append(buffer, count: read)
+        }
+        return String(bytes: data, encoding: .utf8) ?? ""
+    }
+}
+
+struct ClickHouseStubServer {
+    let host = "clickhouse-\(UUID().uuidString.lowercased()).test"
+
+    init(responder: @escaping ClickHouseHTTPStubProtocol.Responder = { _ in .oneRow }) {
+        ClickHouseHTTPStubProtocol.register(host: host, responder: responder)
+    }
+
+    var requests: [ClickHouseStubRequest] {
+        ClickHouseHTTPStubProtocol.requests(to: host)
+    }
+
+    func connectedDriver() -> ClickHousePluginDriver {
+        let driver = ClickHousePluginDriver(config: DriverConnectionConfig(
+            host: host,
+            port: 8_123,
+            username: "default",
+            password: "",
+            database: "default"
+        ))
+        let configuration = URLSessionConfiguration.ephemeral
+        configuration.protocolClasses = [ClickHouseHTTPStubProtocol.self]
+        driver.session = URLSession(configuration: configuration)
+        return driver
+    }
+
+    func firstRequest(where matches: (ClickHouseStubRequest) -> Bool) async -> ClickHouseStubRequest? {
+        for _ in 0 ..< 500 {
+            if let request = requests.first(where: matches) {
+                return request
+            }
+            try? await Task.sleep(for: .milliseconds(10))
+        }
+        return nil
+    }
+}
diff --git a/TableProTests/Plugins/ClickHouseCancelQueryTests.swift b/TableProTests/Plugins/ClickHouseCancelQueryTests.swift
new file mode 100644
index 000000000..96bcf125c
--- /dev/null
+++ b/TableProTests/Plugins/ClickHouseCancelQueryTests.swift
@@ -0,0 +1,70 @@
+import Foundation
+import TableProPluginKit
+import Testing
+
+struct ClickHouseCancelQueryTests {
+    private static let heldStatement = "SELECT sleep(3)"
+
+    private static func holdingTheHeldStatement() -> ClickHouseStubServer {
+        ClickHouseStubServer { request in
+            request.body == heldStatement ? nil : .oneRow
+        }
+    }
+
+    private func stopKillsTheHeldStatement(
+        on server: ClickHouseStubServer,
+        driver: ClickHousePluginDriver
+    ) async throws {
+        let streamed = try #require(await server.firstRequest { $0.body == Self.heldStatement })
+        let queryId = try #require(streamed.item("query_id"))
+        #expect(!queryId.isEmpty)
+
+        try driver.cancelQuery()
+
+        let kill = try #require(await server.firstRequest { $0.body.hasPrefix("KILL QUERY") })
+        #expect(kill.body == "KILL QUERY WHERE query_id = '\(queryId)'")
+        #expect(kill.item("max_execution_time") == nil)
+    }
+
+    @Test("Stop on a capped query kills that query on the server, not the statement before it")
+    func stopKillsBoundedStream() async throws {
+        let server = Self.holdingTheHeldStatement()
+        let driver = server.connectedDriver()
+        try await driver.applyQueryTimeout(120)
+        _ = try await driver.execute(query: "SELECT 0")
+        let earlierId = try #require(server.requests.first { $0.body == "SELECT 0" }?.item("query_id"))
+
+        let query = Task { try await driver.executeBoundedQuery(query: Self.heldStatement, rowCap: 10) }
+        defer { query.cancel() }
+
+        try await stopKillsTheHeldStatement(on: server, driver: driver)
+        let kill = try #require(server.requests.first { $0.body.hasPrefix("KILL QUERY") })
+        #expect(!kill.body.contains(earlierId))
+    }
+
+    @Test("Stop on an unbounded streamed read kills that read on the server")
+    func stopKillsUnboundedStream() async throws {
+        let server = Self.holdingTheHeldStatement()
+        let driver = server.connectedDriver()
+
+        let read = Task {
+            for try await _ in driver.streamRows(query: Self.heldStatement) {}
+        }
+        defer { read.cancel() }
+
+        try await stopKillsTheHeldStatement(on: server, driver: driver)
+    }
+
+    @Test("Each streamed statement gets its own query id")
+    func streamedStatementsGetDistinctIds() async throws {
+        let server = ClickHouseStubServer()
+        let driver = server.connectedDriver()
+
+        _ = try await driver.executeBoundedQuery(query: "SELECT 1", rowCap: 10)
+        _ = try await driver.executeBoundedQuery(query: "SELECT 2", rowCap: 10)
+
+        let ids = server.requests.compactMap { $0.item("query_id") }
+        #expect(ids.count == 2)
+        #expect(Set(ids).count == 2)
+    }
+}
diff --git a/TableProTests/Plugins/ClickHouseConnectFailureTests.swift b/TableProTests/Plugins/ClickHouseConnectFailureTests.swift
new file mode 100644
index 000000000..b62c67780
--- /dev/null
+++ b/TableProTests/Plugins/ClickHouseConnectFailureTests.swift
@@ -0,0 +1,84 @@
+import Foundation
+import TableProPluginKit
+import Testing
+
+struct ClickHouseConnectFailureTests {
+    private func message(of error: Error) -> String? {
+        (error as? ClickHouseError)?.message
+    }
+
+    @Test("A wrong password shows the server's own explanation")
+    func authenticationFailureKeepsServerText() {
+        let serverText = "Code: 516. DB::Exception: default: Authentication failed: password is incorrect, "
+            + "or there is no user with such name. (AUTHENTICATION_FAILED)"
+
+        let mapped = ClickHouseConnectFailure.error(for: ClickHouseError(message: serverText), tlsRefusal: nil)
+
+        #expect(message(of: mapped) == serverText)
+    }
+
+    @Test("An unknown database shows the server's own explanation")
+    func unknownDatabaseKeepsServerText() {
+        let serverText = "Code: 81. DB::Exception: Database analytics does not exist. (UNKNOWN_DATABASE)"
+
+        let mapped = ClickHouseConnectFailure.error(for: ClickHouseError(message: serverText), tlsRefusal: nil)
+
+        #expect(message(of: mapped) == serverText)
+    }
+
+    @Test("A refused host says what the network said, not only that the connection failed")
+    func transportFailureNamesTheCause() throws {
+        let transport = URLError(.cannotConnectToHost)
+
+        let text = try #require(message(of: ClickHouseConnectFailure.error(for: transport, tlsRefusal: nil)))
+
+        #expect(text.contains(transport.localizedDescription))
+    }
+
+    @Test("A certificate the TLS delegate refused is reported as that refusal")
+    func recordedRefusalWins() {
+        let refusal = SSLHandshakeError.hostnameMismatch(serverMessage: "host name mismatch")
+
+        let mapped = ClickHouseConnectFailure.error(for: URLError(.cancelled), tlsRefusal: refusal)
+
+        guard case .hostnameMismatch = mapped as? SSLHandshakeError else {
+            Issue.record("Expected the recorded hostname mismatch, got \(mapped)")
+            return
+        }
+    }
+
+    @Test("An untrusted server certificate is still reported as a TLS failure")
+    func untrustedCertificateIsTLS() {
+        let mapped = ClickHouseConnectFailure.error(for: URLError(.serverCertificateUntrusted), tlsRefusal: nil)
+
+        guard case .untrustedCertificate = mapped as? SSLHandshakeError else {
+            Issue.record("Expected an untrusted certificate, got \(mapped)")
+            return
+        }
+    }
+
+    @Test("An error status with no body names the status instead of an empty message")
+    func emptyErrorBodyNamesTheStatus() async throws {
+        let server = ClickHouseStubServer { _ in ClickHouseStubReply(statusCode: 502, body: "") }
+        let driver = server.connectedDriver()
+
+        let error = await #expect(throws: ClickHouseError.self) {
+            try await driver.execute(query: "SELECT 1")
+        }
+
+        let mapped = try #require(error.map { ClickHouseConnectFailure.error(for: $0, tlsRefusal: nil) })
+        #expect(message(of: mapped)?.contains("502") == true)
+    }
+
+    @Test("A streamed read answered with an error status and no body names the status")
+    func emptyStreamedErrorBodyNamesTheStatus() async throws {
+        let server = ClickHouseStubServer { _ in ClickHouseStubReply(statusCode: 503, body: "  \n") }
+        let driver = server.connectedDriver()
+
+        let error = await #expect(throws: ClickHouseError.self) {
+            _ = try await driver.executeBoundedQuery(query: "SELECT 1", rowCap: 10)
+        }
+
+        #expect(error?.message.contains("503") == true)
+    }
+}
diff --git a/TableProTests/Plugins/ClickHouseQueryTimeoutTests.swift b/TableProTests/Plugins/ClickHouseQueryTimeoutTests.swift
new file mode 100644
index 000000000..33874c9a4
--- /dev/null
+++ b/TableProTests/Plugins/ClickHouseQueryTimeoutTests.swift
@@ -0,0 +1,71 @@
+import Foundation
+import TableProPluginKit
+import Testing
+
+struct ClickHouseQueryTimeoutTests {
+    @Test("The query timeout rides every statement as max_execution_time, streamed reads included")
+    func timeoutReachesEveryStatement() async throws {
+        let server = ClickHouseStubServer()
+        let driver = server.connectedDriver()
+
+        try await driver.applyQueryTimeout(120)
+        _ = try await driver.execute(query: "SELECT 1")
+        _ = try await driver.executeParameterized(query: "SELECT ?", parameters: [.text("a")])
+        _ = try await driver.executeBoundedQuery(query: "SELECT 2", rowCap: 10)
+
+        let requests = server.requests
+        #expect(requests.count >= 3)
+        for request in requests {
+            #expect(request.item("max_execution_time") == "120", "\(request.body)")
+            #expect(!request.body.uppercased().hasPrefix("SET "), "\(request.body)")
+        }
+    }
+
+    @Test("A streamed read waits as long as the query timeout allows, not URLRequest's 60 seconds")
+    func streamedReadTakesTheClientTimeout() async throws {
+        let server = ClickHouseStubServer()
+        let driver = server.connectedDriver()
+
+        try await driver.applyQueryTimeout(120)
+        _ = try await driver.executeBoundedQuery(query: "SELECT 2", rowCap: 10)
+        try await driver.applyQueryTimeout(0)
+        _ = try await driver.executeBoundedQuery(query: "SELECT 3", rowCap: 10)
+
+        let bounded = try #require(server.requests.first { $0.body == "SELECT 2" })
+        #expect(bounded.timeoutInterval == HttpQueryTimeout(serverTimeoutSeconds: 120).requestTimeoutInterval)
+        let unlimited = try #require(server.requests.first { $0.body == "SELECT 3" })
+        #expect(unlimited.timeoutInterval == HttpQueryTimeout(serverTimeoutSeconds: 0).requestTimeoutInterval)
+        #expect(unlimited.item("max_execution_time") == nil)
+    }
+
+    @Test("No query timeout sends no max_execution_time, so the server's own profile limit stands")
+    func noTimeoutSendsNoLimit() async throws {
+        let server = ClickHouseStubServer()
+        let driver = server.connectedDriver()
+
+        try await driver.applyQueryTimeout(0)
+        _ = try await driver.execute(query: "SELECT 1")
+
+        #expect(server.requests.allSatisfy { $0.item("max_execution_time") == nil })
+    }
+
+    @Test("A limit the server refuses is reported once and left off every later statement")
+    func refusedLimitIsNotSentAgain() async throws {
+        let refusal = "Code: 452. DB::Exception: Setting max_execution_time shouldn't be greater than 30."
+        let server = ClickHouseStubServer { request in
+            request.item("max_execution_time") == nil ? .oneRow : ClickHouseStubReply(statusCode: 500, body: refusal)
+        }
+        let driver = server.connectedDriver()
+
+        await #expect(throws: ClickHouseError.self) {
+            try await driver.applyQueryTimeout(120)
+        }
+        let result = try await driver.execute(query: "SELECT 10")
+        _ = try await driver.executeBoundedQuery(query: "SELECT 20", rowCap: 10)
+
+        #expect(result.rows.count == 1)
+        let later = server.requests.filter { $0.body == "SELECT 10" || $0.body == "SELECT 20" }
+        #expect(later.count == 2)
+        #expect(later.allSatisfy { $0.item("max_execution_time") == nil })
+    }
+}
diff --git a/TableProTests/Plugins/ClickHouseTLSConfigurationTests.swift b/TableProTests/Plugins/ClickHouseTLSConfigurationTests.swift
index a056e1819..f3018f8e7 100644
--- a/TableProTests/Plugins/ClickHouseTLSConfigurationTests.swift
+++ b/TableProTests/Plugins/ClickHouseTLSConfigurationTests.swift
@@ -75,6 +75,202 @@ struct ClickHouseTLSConfigurationTests {
         }
     }
 
+    private static let clientCertificatePEM = """
+        -----BEGIN CERTIFICATE-----
+        MIIBRjCB7AIJAI8IxGRVwB2sMAoGCCqGSM49BAMCMCoxKDAmBgNVBAMMH1RhYmxl
+        UHJvIENsaWNrSG91c2UgVGVzdCBDbGllbnQwIBcNMjYwOTMwMDgwODM5WhgPMjEy
+        NjA5MDYwODA4MzlaMCoxKDAmBgNVBAMMH1RhYmxlUHJvIENsaWNrSG91c2UgVGVz
+        dCBDbGllbnQwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAR4MsJgLIbNVUdTp1AI
+        +UI59GxEZNE+2GS37Dq8sEluUP+MK4lJmP5eamZoDI1FJgXuol/6/j99J3x9tWSo
+        2JCGMAoGCCqGSM49BAMCA0kAMEYCIQDo8UPcX3AbSM4vSga4ZmNz8l1yt7PI0i5+
+        pis0886smAIhALdWMO0Vi6SA4NEJUr3EeQyye3rrS9j43ZMTMhXxovX/
+        -----END CERTIFICATE-----
+        """
+
+    private static let clientKeyPEM = """
+        -----BEGIN PRIVATE KEY-----
+        MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgyzfBMM5MAZs5n7VW
+        ZAMWdxn0FdlZiRKhL2eWOwXfs96hRANCAAR4MsJgLIbNVUdTp1AI+UI59GxEZNE+
+        2GS37Dq8sEluUP+MK4lJmP5eamZoDI1FJgXuol/6/j99J3x9tWSo2JCG
+        -----END PRIVATE KEY-----
+        """
+
+    private struct ClientIdentityFiles {
+        let certificate: URL
+        let key: URL
+
+        init() throws {
+            let folder = FileManager.default.temporaryDirectory
+            certificate = folder.appendingPathComponent("client-\(UUID().uuidString).pem")
+            key = folder.appendingPathComponent("client-\(UUID().uuidString).key")
+            try Data(ClickHouseTLSConfigurationTests.clientCertificatePEM.utf8).write(to: certificate)
+            try Data(ClickHouseTLSConfigurationTests.clientKeyPEM.utf8).write(to: key)
+        }
+
+        func remove() {
+            try? FileManager.default.removeItem(at: certificate)
+            try? FileManager.default.removeItem(at: key)
+        }
+    }
+
+    private final class IgnoringChallengeSender: NSObject, URLAuthenticationChallengeSender {
+        func use(_ credential: URLCredential, for challenge: URLAuthenticationChallenge) {}
+        func continueWithoutCredential(for challenge: URLAuthenticationChallenge) {}
+        func cancel(_ challenge: URLAuthenticationChallenge) {}
+    }
+
+    private final class ServerTrustProtectionSpace: URLProtectionSpace, @unchecked Sendable {
+        private let trust: SecTrust
+
+        init(trust: SecTrust) {
+            self.trust = trust
+            super.init(
+                host: "clickhouse.example.com",
+                port: 8_443,
+                protocol: NSURLProtectionSpaceHTTPS,
+                realm: nil,
+                authenticationMethod: NSURLAuthenticationMethodServerTrust
+            )
+        }
+
+        required init?(coder: NSCoder) {
+            nil
+        }
+
+        override var serverTrust: SecTrust? { trust }
+    }
+
+    private func fixtureServerTrust() throws -> SecTrust {
+        let base64 = Self.clientCertificatePEM
+            .split(separator: "\n")
+            .filter { !$0.hasPrefix("-----") }
+            .joined()
+        let der = try #require(Data(base64Encoded: base64))
+        let certificate = try #require(SecCertificateCreateWithData(nil, der as CFData))
+        var trust: SecTrust?
+        let status = SecTrustCreateWithCertificates(
+            certificate,
+            SecPolicyCreateSSL(true, "clickhouse.example.com" as CFString),
+            &trust
+        )
+        try #require(status == errSecSuccess)
+        return try #require(trust)
+    }
+
+    private func answer(
+        _ authenticationMethod: String,
+        with delegate: ClickHouseTLSDelegate
+    ) -> (disposition: URLSession.AuthChallengeDisposition, credential: URLCredential?) {
+        answer(
+            URLProtectionSpace(
+                host: "clickhouse.example.com",
+                port: 8_443,
+                protocol: NSURLProtectionSpaceHTTPS,
+                realm: nil,
+                authenticationMethod: authenticationMethod
+            ),
+            with: delegate
+        )
+    }
+
+    private func answer(
+        _ protectionSpace: URLProtectionSpace,
+        with delegate: ClickHouseTLSDelegate
+    ) -> (disposition: URLSession.AuthChallengeDisposition, credential: URLCredential?) {
+        let challenge = URLAuthenticationChallenge(
+            protectionSpace: protectionSpace,
+            proposedCredential: nil,
+            previousFailureCount: 0,
+            failureResponse: nil,
+            error: nil,
+            sender: IgnoringChallengeSender()
+        )
+        var answer: (URLSession.AuthChallengeDisposition, URLCredential?) = (.rejectProtectionSpace, nil)
+        delegate.urlSession(URLSession.shared, didReceive: challenge) { disposition, credential in
+            answer = (disposition, credential)
+        }
+        return answer
+    }
+
+    @Test("A client certificate and key are presented when the server asks for one, in every TLS mode")
+    func clientIdentityAnswersTheCertificateChallenge() throws {
+        let files = try ClientIdentityFiles()
+        defer { files.remove() }
+        for mode in [SSLMode.verifyIdentity, .required, .preferred] {
+            let ssl = SSLConfiguration(
+                mode: mode,
+                clientCertificatePath: files.certificate.path,
+                clientKeyPath: files.key.path
+            )
+            let delegate = try #require(try ClickHouseTLSDelegate.make(for: ssl), "\(mode)")
+
+            let answered = answer(NSURLAuthenticationMethodClientCertificate, with: delegate)
+
+            #expect(answered.disposition == .useCredential, "\(mode)")
+            #expect(answered.credential?.identity != nil, "\(mode)")
+        }
+    }
+
+    @Test("Verify Identity with only a client identity still leaves the server's certificate to the system")
+    func clientIdentityKeepsSystemServerTrust() throws {
+        let files = try ClientIdentityFiles()
+        defer { files.remove() }
+        let trust = try fixtureServerTrust()
+        let identityOnly = try #require(try ClickHouseTLSDelegate.make(for: SSLConfiguration(
+            mode: .verifyIdentity,
+            clientCertificatePath: files.certificate.path,
+            clientKeyPath: files.key.path
+        )))
+        let skipVerify = try #require(try ClickHouseTLSDelegate.make(for: SSLConfiguration(
+            mode: .required,
+            clientCertificatePath: files.certificate.path,
+            clientKeyPath: files.key.path
+        )))
+
+        #expect(answer(ServerTrustProtectionSpace(trust: trust), with: skipVerify).disposition == .useCredential)
+        #expect(
+            answer(ServerTrustProtectionSpace(trust: trust), with: identityOnly).disposition == .performDefaultHandling
+        )
+    }
+
+    @Test("A certificate challenge with no client identity configured is left to the system")
+    func noClientIdentityLeavesTheChallenge() throws {
+        let delegate = try #require(try ClickHouseTLSDelegate.make(for: configuration(.required)))
+
+        let answered = answer(NSURLAuthenticationMethodClientCertificate, with: delegate)
+
+        #expect(answered.disposition == .performDefaultHandling)
+        #expect(answered.credential == nil)
+    }
+
+    @Test("SSL off presents no client identity even when the paths are filled in")
+    func disabledPresentsNothing() throws {
+        let files = try ClientIdentityFiles()
+        defer { files.remove() }
+        let ssl = SSLConfiguration(
+            mode: .disabled,
+            clientCertificatePath: files.certificate.path,
+            clientKeyPath: files.key.path
+        )
+
+        #expect(try ClickHouseTLSDelegate.make(for: ssl) == nil)
+    }
+
+    @Test("A client certificate without its key, or a key that does not load, is refused before any request")
+    func unusableClientIdentityIsRefused() throws {
+        let files = try ClientIdentityFiles()
+        defer { files.remove() }
+        let withoutKey = SSLConfiguration(mode: .verifyIdentity, clientCertificatePath: files.certificate.path)
+        let unreadableKey = SSLConfiguration(
+            mode: .verifyIdentity,
+            clientCertificatePath: files.certificate.path,
+            clientKeyPath: files.certificate.path
+        )
+
+        #expect(refusal(withoutKey)?.contains("client key") == true)
+        #expect(refusal(unreadableKey)?.contains(files.certificate.path) == true)
+    }
+
     @Test("A certificate that fails its host check is reported as a hostname mismatch, anything else as untrusted")
     func refusalKinds() {
         let hostname = CFErrorCreate(nil, NSOSStatusErrorDomain as CFString, CFIndex(errSecHostNameMismatch), nil)
diff --git a/project.yml b/project.yml
index b70891ebb..f6da374f8 100644
--- a/project.yml
+++ b/project.yml
@@ -448,6 +448,7 @@ targets:
       - Plugins/CSVImportPlugin/CSVImportPlugin.swift
       - Plugins/CSVImportPlugin/CSVImportText.swift
       - Plugins/ClickHouseDriverPlugin/ClickHouseCapabilities.swift
+      - Plugins/ClickHouseDriverPlugin/ClickHouseConnectFailure.swift
       - Plugins/ClickHouseDriverPlugin/ClickHouseCredentials.swift
       - Plugins/ClickHouseDriverPlugin/ClickHouseGeneratedColumnClassification.swift
       - Plugins/ClickHouseDriverPlugin/ClickHouseHTTPChunks.swift
@@ -994,6 +995,8 @@ targets:
     dependencies:
       - package: TableProCore
         product: TableProNumberFormatting
+      - package: TableProCore
+        product: TableProTLSClientIdentity
 
   RedisDriver:
     templates: [DriverPlugin]