fix(xpc): honor request timeouts

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
                 "CAuditToken",
             ]
         ),
+        .testTarget(
+            name: "ContainerXPCTests",
+            dependencies: [
+                "ContainerXPC"
+            ]
+        ),
         .target(
             name: "ContainerOS",
             dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
             }
 
             group.addTask {
-                try await withCheckedThrowingContinuation { cont in
-                    xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
-                        do {
-                            let message = try self.parseReply(reply)
-                            cont.resume(returning: message)
-                        } catch {
-                            cont.resume(throwing: error)
+                let box = XPCReplyBox()
+                return try await withTaskCancellationHandler {
+                    try await withCheckedThrowingContinuation { cont in
+                        box.store(cont)
+                        xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+                            box.resume { try self.parseReply(reply) }
                         }
                     }
+                } onCancel: {
+                    // XPC reply callbacks do not observe task cancellation, so
+                    // resume the awaiting continuation when a timeout wins.
+                    box.resume { throw CancellationError() }
                 }
             }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// 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
+//
+//   https://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.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+    private let lock = NSLock()
+    private var continuation: CheckedContinuation<XPCMessage, Error>?
+    private var result: Result<XPCMessage, Error>?
+
+    func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+        lock.lock()
+        guard let result else {
+            self.continuation = continuation
+            lock.unlock()
+            return
+        }
+        lock.unlock()
+        continuation.resume(with: result)
+    }
+
+    func resume(_ body: () throws -> XPCMessage) {
+        let result = Result { try body() }
+        lock.lock()
+        guard self.result == nil else {
+            lock.unlock()
+            return
+        }
+        self.result = result
+        guard let continuation else {
+            lock.unlock()
+            return
+        }
+        self.continuation = nil
+        lock.unlock()
+        continuation.resume(with: result)
+    }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// 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
+//
+//   https://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.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+    @Test
+    func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+        let box = XPCReplyBox()
+        box.resume { throw CancellationError() }
+
+        do {
+            _ = try await withCheckedThrowingContinuation { continuation in
+                box.store(continuation)
+            }
+            Issue.record("expected the pending cancellation to be delivered")
+        } catch is CancellationError {
+            // Expected.
+        }
+    }
+
+    @Test
+    func responseTimeoutReturnsWithinBound() async throws {
+        let server = AnonymousXPCServer()
+        defer { server.close() }
+
+        let client = server.makeClient()
+        let clock = ContinuousClock()
+        let start = clock.now
+
+        do {
+            _ = try await client.send(
+                XPCMessage(route: "hang"),
+                responseTimeout: .milliseconds(100)
+            )
+            Issue.record("expected send to time out")
+        } catch let error as ContainerizationError {
+            #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+            #expect(start.duration(to: clock.now) < .seconds(2))
+        }
+    }
+
+    @Test
+    func callerCancellationReturnsWithinBound() async throws {
+        let server = AnonymousXPCServer()
+        defer { server.close() }
+
+        let client = server.makeClient()
+        let request = Task {
+            try await client.send(
+                XPCMessage(route: "hang"),
+                responseTimeout: .seconds(30)
+            )
+        }
+
+        try await Task.sleep(for: .milliseconds(50))
+        let clock = ContinuousClock()
+        let start = clock.now
+        request.cancel()
+
+        do {
+            _ = try await request.value
+            Issue.record("expected send to be cancelled")
+        } catch is CancellationError {
+            #expect(start.duration(to: clock.now) < .seconds(2))
+        }
+    }
+
+    @Test
+    func clientCanBeReusedAfterResponseTimeout() async throws {
+        let server = AnonymousXPCServer()
+        defer { server.close() }
+
+        let client = server.makeClient()
+
+        do {
+            _ = try await client.send(
+                XPCMessage(route: "hang"),
+                responseTimeout: .milliseconds(100)
+            )
+            Issue.record("expected send to time out")
+        } catch let error as ContainerizationError {
+            #expect(error.message.contains("XPC timeout"))
+        }
+
+        let response = try await client.send(
+            XPCMessage(route: "echo"),
+            responseTimeout: .seconds(1)
+        )
+        #expect(response.string(key: "result") == "ok")
+    }
+
+    @Test
+    func lateReplyAfterTimeoutIsIgnored() async throws {
+        let server = AnonymousXPCServer()
+        defer { server.close() }
+
+        let client = server.makeClient()
+
+        do {
+            _ = try await client.send(
+                XPCMessage(route: "hang"),
+                responseTimeout: .milliseconds(100)
+            )
+            Issue.record("expected send to time out")
+        } catch let error as ContainerizationError {
+            #expect(error.message.contains("XPC timeout"))
+        }
+
+        #expect(server.replyToPendingRequests())
+        try await Task.sleep(for: .milliseconds(50))
+
+        let response = try await client.send(
+            XPCMessage(route: "echo"),
+            responseTimeout: .seconds(1)
+        )
+        #expect(response.string(key: "result") == "ok")
+    }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+    private let listener: xpc_connection_t
+    private let lock = NSLock()
+    private var connections = [xpc_connection_t]()
+    private var pendingRequests = [xpc_object_t]()
+
+    init() {
+        listener = xpc_connection_create(nil, nil)
+        xpc_connection_set_event_handler(listener) { [weak self] object in
+            switch xpc_get_type(object) {
+            case XPC_TYPE_CONNECTION:
+                self?.accept(connection: object)
+            case XPC_TYPE_ERROR:
+                break
+            default:
+                fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+            }
+        }
+        xpc_connection_activate(listener)
+    }
+
+    func makeClient() -> XPCClient {
+        let endpoint = xpc_endpoint_create(listener)
+        let connection = xpc_connection_create_from_endpoint(endpoint)
+        return XPCClient(connection: connection, label: "test.container.xpc")
+    }
+
+    func close() {
+        xpc_connection_cancel(listener)
+        lock.withLock {
+            for connection in connections {
+                xpc_connection_cancel(connection)
+            }
+            connections.removeAll()
+            pendingRequests.removeAll()
+        }
+    }
+
+    func replyToPendingRequests() -> Bool {
+        let requests = lock.withLock {
+            let requests = pendingRequests
+            pendingRequests.removeAll()
+            return requests
+        }
+        guard let connection = lock.withLock({ connections.last }) else {
+            return false
+        }
+        for request in requests {
+            guard let reply = xpc_dictionary_create_reply(request) else {
+                continue
+            }
+            xpc_dictionary_set_string(reply, "result", "late")
+            xpc_connection_send_message(connection, reply)
+        }
+        return !requests.isEmpty
+    }
+
+    private func accept(connection: xpc_connection_t) {
+        lock.withLock {
+            connections.append(connection)
+        }
+        nonisolated(unsafe) let connection = connection
+        xpc_connection_set_event_handler(connection) { [weak self] object in
+            guard let self else {
+                return
+            }
+            switch xpc_get_type(object) {
+            case XPC_TYPE_DICTIONARY:
+                let message = XPCMessage(object: object)
+                switch message.string(key: XPCMessage.routeKey) {
+                case "echo":
+                    guard let reply = xpc_dictionary_create_reply(object) else {
+                        return
+                    }
+                    xpc_dictionary_set_string(reply, "result", "ok")
+                    xpc_connection_send_message(connection, reply)
+                default:
+                    self.lock.withLock {
+                        self.pendingRequests.append(object)
+                    }
+                }
+            case XPC_TYPE_ERROR:
+                break
+            default:
+                fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+            }
+        }
+        xpc_connection_activate(connection)
+    }
+}
+#endif