diff --git a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerHttpRequestFraming.swift b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerHttpRequestFraming.swift index 3cc1a186fa..1a80468c1c 100644 --- a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerHttpRequestFraming.swift +++ b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerHttpRequestFraming.swift @@ -51,6 +51,7 @@ enum RunnerHttpRequestRejection: Equatable { struct RunnerHttpAdmittedHead: Equatable { let bodyOffset: Int let contentLength: Int + let keepsConnectionAlive: Bool } enum RunnerHttpHeadVerdict: Equatable { @@ -121,10 +122,16 @@ enum RunnerHttpRequestFraming { else { return .rejected(.bodyTooLarge) } + let connectionOptions = (headers["connection"] ?? []).flatMap { + $0.split(separator: ",").map { $0.trimmingCharacters(in: .whitespaces).lowercased() } + } + let keepsConnectionAlive = !connectionOptions.contains("close") + && (requestLine[2] == "HTTP/1.1" || connectionOptions.contains("keep-alive")) return .admitted( RunnerHttpAdmittedHead( bodyOffset: terminator.upperBound - buffer.startIndex, - contentLength: contentLength + contentLength: contentLength, + keepsConnectionAlive: keepsConnectionAlive ) ) } diff --git a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests+Transport.swift b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests+Transport.swift index 612a2bb0e8..7deeea8b19 100644 --- a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests+Transport.swift +++ b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests+Transport.swift @@ -8,20 +8,25 @@ extension RunnerTests { /// loopback peer may idle, so its clock starts at its first byte. static let requestTimeout: TimeInterval = 30 - /// Starts one connection and reads one request from it once its path is known. Every refusal - /// answers and closes that connection only; the listener keeps serving. + /// An unused loopback connection watches listener lifetime. Used connections have bounded idle time. + static let connectionIdleTimeout: TimeInterval = 30 + + /// Every request is authenticated, including subsequent requests on an admitted connection. func handle(connection: NWConnection) { guard let token = listenerToken else { connection.cancel() return } + transportConnections[ObjectIdentifier(connection)] = connection connection.stateUpdateHandler = { [weak self] state in switch state { case .ready: - connection.stateUpdateHandler = nil self?.admit(connection, token: token) case .failed: connection.cancel() + case .cancelled: + connection.stateUpdateHandler = nil + self?.transportConnections.removeValue(forKey: ObjectIdentifier(connection)) default: break } @@ -60,6 +65,8 @@ extension RunnerTests { } private func consume(_ data: Data, for inbound: RunnerInboundRequest, over connection: NWConnection) { + inbound.idleDeadline?.cancel() + inbound.idleDeadline = nil if inbound.deadline == nil { armDeadline(for: inbound, over: connection) } @@ -89,8 +96,9 @@ extension RunnerTests { } inbound.close() let body = inbound.buffer.subdata(in: head.bodyOffset.. Void = {} + closeConnection: Bool, + afterSend: @escaping @Sendable (Bool) -> Void = { _ in } ) { - connection.send(content: response, isComplete: true, completion: .contentProcessed { error in + // Network delivers send completions on the connection's transport queue. + connection.send(content: response.encoded(closeConnection: closeConnection), completion: .contentProcessed { error in if let error { NSLog("AGENT_DEVICE_RUNNER_SEND_FAILED=%@", String(describing: error)) } - connection.cancel() - afterSend() + if closeConnection || error != nil { connection.cancel() } + afterSend(error == nil) }) } private func handleRequestBody( _ body: Data, - completion: @escaping @Sendable ((data: Data, shouldFinish: Bool)) -> Void + completion: @escaping @Sendable ((response: RunnerHttpResponse, shouldFinish: Bool)) -> Void ) { guard String(data: body, encoding: .utf8) != nil else { completion(( @@ -257,7 +293,7 @@ extension RunnerTests { /// false so the caller enqueues the (single) execution. func attachToInFlightCommandIfNeeded( command: Command, - completion: @escaping @Sendable ((data: Data, shouldFinish: Bool)) -> Void + completion: @escaping @Sendable ((response: RunnerHttpResponse, shouldFinish: Bool)) -> Void ) -> Bool { guard let commandId = command.commandId?.trimmedNonEmpty else { return false } inFlightCommandLock.lock() @@ -278,10 +314,10 @@ extension RunnerTests { func deliverCommandResult( command: Command, - result: (data: Data, shouldFinish: Bool), - completion: ((data: Data, shouldFinish: Bool)) -> Void + result: (response: RunnerHttpResponse, shouldFinish: Bool), + completion: ((response: RunnerHttpResponse, shouldFinish: Bool)) -> Void ) { - var waiters: [@Sendable ((data: Data, shouldFinish: Bool)) -> Void] = [] + var waiters: [@Sendable ((response: RunnerHttpResponse, shouldFinish: Bool)) -> Void] = [] if let commandId = command.commandId?.trimmedNonEmpty { inFlightCommandLock.lock() inFlightCommandIds.remove(commandId) @@ -292,11 +328,15 @@ extension RunnerTests { for waiter in waiters { waiter(result) } + // Register after every waiter starts its write, so a fast response cannot close slower peers. + if result.shouldFinish { + shutdownResponseWrites.notify(queue: transportQueue) { self.finish() } + } } // MARK: - Response Encoding - private func jsonResponse(status: Int, response: Response) -> Data { + private func jsonResponse(status: Int, response: Response) -> RunnerHttpResponse { // Stamp the gesture-clock uptime at the END of command handling, just before the HTTP // write, so the warm snapshot and recordStart responses carry the anchor for free. This // runs AFTER commandJournal.finish, so journal-stored lifecycleResponseJson stays @@ -310,7 +350,7 @@ extension RunnerTests { : response let encoder = JSONEncoder() let body = (try? encoder.encode(stamped)).flatMap { String(data: $0, encoding: .utf8) } ?? "{}" - return httpResponse(status: status, body: body) + return RunnerHttpResponse(status: status, body: body) } private func errorResponse(code: String, message: String, hint: String? = nil) -> Response { @@ -343,53 +383,30 @@ extension RunnerTests { ) } - private func httpResponse(status: Int, body: String) -> Data { - let headers = [ - "HTTP/1.1 \(status) \(Self.reasonPhrase(status))", - "Content-Type: application/json", - "Content-Length: \(body.utf8.count)", - "\(Self.authEnforcedResponseHeader): required", - "Connection: close", - "", - body - ].joined(separator: "\r\n") - return Data(headers.utf8) - } - /// Present on every response so the daemon can refuse a runner build that predates the token. static let authEnforcedResponseHeader = "X-Agent-Device-Runner-Auth" - private static func reasonPhrase(_ status: Int) -> String { - switch status { - case 200: return "OK" - case 400: return "Bad Request" - case 401: return "Unauthorized" - case 404: return "Not Found" - case 405: return "Method Not Allowed" - case 408: return "Request Timeout" - case 411: return "Length Required" - case 413: return "Content Too Large" - case 431: return "Request Header Fields Too Large" - default: return "Internal Server Error" - } + func closeTransport() { + listener?.cancel() + listener = nil + for connection in transportConnections.values { connection.cancel() } + transportConnections.removeAll() } private func finish() { - listener?.cancel() - listener = nil - // Guard against double-fulfill: coalesced shutdown sends deliver one result to - // multiple waiters, each of which may ask to finish. + closeTransport() doneExpectation?.fulfill() doneExpectation = nil } } -/// One connection's request, read on the transport queue. +/// One request on a connection, read on the transport queue. final class RunnerInboundRequest { let token: RunnerListenerToken var buffer = Data() var head: RunnerHttpAdmittedHead? var deadline: DispatchWorkItem? + var idleDeadline: DispatchWorkItem? private(set) var closed = false init(token: RunnerListenerToken) { @@ -402,5 +419,41 @@ final class RunnerInboundRequest { closed = true deadline?.cancel() deadline = nil + idleDeadline?.cancel() + idleDeadline = nil + } +} + +/// Coalesced commands share their response, but each connection owns its persistence policy. +struct RunnerHttpResponse: Equatable, Sendable { + let status: Int + let body: String + + func encoded(closeConnection: Bool) -> Data { + let headers = [ + "HTTP/1.1 \(status) \(Self.reasonPhrase(status))", + "Content-Type: application/json", + "Content-Length: \(body.utf8.count)", + "\(RunnerTests.authEnforcedResponseHeader): required", + "Connection: \(closeConnection ? "close" : "keep-alive")", + "", + body + ].joined(separator: "\r\n") + return Data(headers.utf8) + } + + private static func reasonPhrase(_ status: Int) -> String { + switch status { + case 200: return "OK" + case 400: return "Bad Request" + case 401: return "Unauthorized" + case 404: return "Not Found" + case 405: return "Method Not Allowed" + case 408: return "Request Timeout" + case 411: return "Length Required" + case 413: return "Content Too Large" + case 431: return "Request Header Fields Too Large" + default: return "Internal Server Error" + } } } diff --git a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests.swift b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests.swift index 12146eb9d7..d152fa5cda 100644 --- a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests.swift +++ b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests.swift @@ -60,6 +60,8 @@ final class RunnerTests: XCTestCase { var listener: NWListener? var listenerToken: RunnerListenerToken? var doneExpectation: XCTestExpectation? + var transportConnections: [ObjectIdentifier: NWConnection] = [:] + let shutdownResponseWrites = DispatchGroup() let transportQueue = DispatchQueue(label: "agent-device.runner.transport") let commandExecutionQueue = DispatchQueue(label: "agent-device.runner.commands") let app = XCUIApplication() @@ -106,7 +108,7 @@ final class RunnerTests: XCTestCase { // execution instead of enqueueing them again behind it (#1105 capture pileup). let inFlightCommandLock = NSLock() var inFlightCommandIds: Set = [] - var inFlightCommandWaiters: [String: [@Sendable ((data: Data, shouldFinish: Bool)) -> Void]] = [:] + var inFlightCommandWaiters: [String: [@Sendable ((response: RunnerHttpResponse, shouldFinish: Bool)) -> Void]] = [:] // Tracks main-queue work abandoned by the execution watchdog (runMainThreadWork). While any is // outstanding the main thread is occupied: new main-thread commands fail fast as busy instead // of queueing behind work that cannot be cancelled, capture plans skip XCTest-backed tiers, @@ -247,6 +249,11 @@ final class RunnerTests: XCTestCase { continueAfterFailure = true } + override func tearDown() { + transportQueue.sync { closeTransport() } + super.tearDown() + } + /// The XCTest fetch wordings whose recorded issue carries the AX server's own error text: the /// element snapshot fetch and query resolution (`allElementsBoundByIndex`, recorded once per /// element type by the query-sweep tier). Both read the target's tree through testmanagerd and diff --git a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+HttpRequestFramingTests.swift b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+HttpRequestFramingTests.swift index 9f49aa0283..df64858db1 100644 --- a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+HttpRequestFramingTests.swift +++ b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+HttpRequestFramingTests.swift @@ -24,10 +24,30 @@ extension RunnerTests { let head = framingHead(authenticatedPost + ["Content-Length: 2"], body: "{}") XCTAssertEqual( admit(authenticatedPost + ["Content-Length: 2"], body: "{}"), - .admitted(RunnerHttpAdmittedHead(bodyOffset: head.count - 2, contentLength: 2)) + .admitted(RunnerHttpAdmittedHead(bodyOffset: head.count - 2, contentLength: 2, keepsConnectionAlive: true)) ) } + func testHttpFramingResolvesConnectionPersistence() { + let cases: [(String, [String], Bool)] = [ + ("HTTP/1.1", [], true), + ("HTTP/1.1", ["Connection: close"], false), + ("HTTP/1.1", ["Connection: upgrade, CLOSE"], false), + ("HTTP/1.1", ["Connection: keep-alive", "connection: Close"], false), + ("HTTP/1.0", [], false), + ("HTTP/1.0", ["Connection: Keep-Alive"], true), + ("HTTP/1.0", ["Connection: keep-alive, close"], false), + ] + for (version, connection, expected) in cases { + let verdict = admit(["POST /command \(version)", authenticatedPost[1], "Content-Length: 0"] + connection) + guard case .admitted(let head) = verdict else { + XCTFail("unexpected refusal: \(verdict)") + continue + } + XCTAssertEqual(head.keepsConnectionAlive, expected, "\(version) \(connection)") + } + } + func testHttpFramingWaitsForTheWholeHead() { let partial = Data("POST /command HTTP/1.1\r\nContent-Length: 2\r\n".utf8) XCTAssertEqual( @@ -87,7 +107,7 @@ extension RunnerTests { guard firstVerdict == .incomplete else { continue } XCTAssertEqual( RunnerHttpRequestFraming.admitHead(request, token: framingListenerToken, alreadyScanned: split), - .admitted(RunnerHttpAdmittedHead(bodyOffset: request.count, contentLength: 0)), + .admitted(RunnerHttpAdmittedHead(bodyOffset: request.count, contentLength: 0, keepsConnectionAlive: true)), "split at \(split)" ) } diff --git a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+TransportTests.swift b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+TransportTests.swift index c13745ad2d..59ebfcca2f 100644 --- a/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+TransportTests.swift +++ b/apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/UnitTests/RunnerTests+TransportTests.swift @@ -199,24 +199,276 @@ extension RunnerTests { /// Sends raw bytes and returns the status the runner answered with. private func rawTransportStatus(port: NWEndpoint.Port, request: String) throws -> Int? { - let connection = NWConnection(host: .ipv4(.loopback), port: port, using: .tcp) - let answered = expectation(description: "raw response") - final class Box: @unchecked Sendable { var data = Data() } - let box = Box() - connection.start(queue: .global()) - connection.send(content: Data(request.utf8), completion: .contentProcessed { _ in }) - func read() { + let socket = TransportSocket(port: port) + defer { socket.connection.cancel() } + try socket.send(request) + let status = try socket.response().status + try socket.expectEOF() + return status + } + + func testHttpConnectionServesSequentialRequestsAndCommandErrors() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let socket = try TransportSocket(port: XCTUnwrap(listener.port)) + defer { socket.connection.cancel() } + for (payload, status) in [(#"{"command":"uptime"}"#, 200), ("not-json", 400), (#"{"command":"uptime"}"#, 200)] { + try socket.send(rawCommand(payload)) + let response = try socket.response() + XCTAssertEqual(response.status, status) + XCTAssertTrue(response.head.contains("Connection: keep-alive")) + XCTAssertTrue(response.head.contains("X-Agent-Device-Runner-Auth: required")) + XCTAssertEqual(try response.envelope().ok, status == 200) + } + } + + func testHttpConnectionPreservesPipelinedRequestsAndResponseOrder() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let socket = try TransportSocket(port: XCTUnwrap(listener.port)) + defer { socket.connection.cancel() } + let queued = rawCommand(#"{"command":"appState","commandId":"pipeline-first"}"#) + let inline = rawCommand(#"{"command":"uptime"}"#) + // The inline command must not overtake queued work, and a partial third head must survive. + try socket.send(queued + inline + String(inline.prefix(13))) + let first = try socket.response() + XCTAssertEqual(first.status, 200) + XCTAssertFalse(try first.envelope().ok) + XCTAssertEqual(try first.envelope().error?.message, "appState requires appBundleId") + let second = try socket.response() + XCTAssertNotNil(try second.envelope().data?.currentUptimeMs) + try socket.send(String(inline.dropFirst(13))) + let third = try socket.response() + XCTAssertEqual(third.status, 200) + XCTAssertNotNil(try third.envelope().data?.currentUptimeMs) + } + + func testHttpConnectionAuthenticatesEveryRequestAndClosesOnRefusal() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let port = try XCTUnwrap(listener.port) + let socket = TransportSocket(port: port) + defer { socket.connection.cancel() } + try socket.send(rawCommand(#"{"command":"uptime"}"#)) + XCTAssertEqual(try socket.response().status, 200) + try socket.send( + rawCommand(#"{"command":"appState","commandId":"unauthorized-followup"}"#, token: "wrong") + + rawCommand(#"{"command":"appState","commandId":"after-refusal"}"#) + ) + let denied = try socket.response() + XCTAssertEqual(denied.status, 401) + XCTAssertEqual(try denied.envelope().error?.code, "UNAUTHORIZED") + XCTAssertTrue(denied.head.contains("Connection: close")) + try socket.expectEOF() + for id in ["unauthorized-followup", "after-refusal"] { + XCTAssertEqual(commandJournal.status(normalizedCommandId: id).lifecycleState, RunnerCommandLifecycleState.notAccepted.rawValue) + } + let other = TransportSocket(port: port) + defer { other.connection.cancel() } + try other.send(rawCommand(#"{"command":"uptime"}"#)) + XCTAssertEqual(try other.response().status, 200) + } + + func testHttpConnectionRefusesMalformedFollowupBeforeReadingItsBody() throws { + try withWarmTransportSocket { socket in + try socket.send("POST /command HTTP/1.1\r\nX-Agent-Device-Runner-Token: \(Self.transportToken)\r\nContent-Length: 4\r\nContent-Length: 4\r\n\r\n") + XCTAssertEqual(try socket.response().status, 411) + try socket.expectEOF() + } + } + + func testHttpPartialFollowupGetsAnArrivalDeadline() throws { + try withWarmTransportSocket { socket in + let request = rawCommand(#"{"command":"uptime"}"#) + try socket.send(String(request.dropLast(2))) + let timedOut = try socket.response(timeout: Self.requestTimeout + 3) + XCTAssertEqual(timedOut.status, 408) + XCTAssertTrue(timedOut.head.contains("Connection: close")) + try socket.expectEOF() + } + } + + private func withWarmTransportSocket(_ body: (TransportSocket) throws -> Void) throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let socket = try TransportSocket(port: XCTUnwrap(listener.port)) + defer { socket.connection.cancel() } + try socket.send(rawCommand(#"{"command":"uptime"}"#)) + XCTAssertEqual(try socket.response().status, 200) + try body(socket) + } + + func testHttpConnectionClosesForHttp10AndExplicitClose() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + for (version, headers) in [("HTTP/1.0", ""), ("HTTP/1.1", "Connection: keep-alive\r\nConnection: CLOSE\r\n")] { + let socket = try TransportSocket(port: XCTUnwrap(listener.port)) + defer { socket.connection.cancel() } + try socket.send(rawCommand(#"{"command":"uptime"}"#, version: version, headers: headers) + + rawCommand(#"{"command":"appState","commandId":"after-close"}"#)) + let response = try socket.response() + XCTAssertEqual(response.status, 200) + XCTAssertTrue(response.head.contains("Connection: close")) + try socket.expectEOF() + XCTAssertEqual(commandJournal.status(normalizedCommandId: "after-close").lifecycleState, RunnerCommandLifecycleState.notAccepted.rawValue) + } + } + + func testHttp10ConnectionReusesWhenExplicitlyRequested() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let socket = try TransportSocket(port: XCTUnwrap(listener.port)) + defer { socket.connection.cancel() } + for _ in 0..<2 { + try socket.send(rawCommand(#"{"command":"uptime"}"#, version: "HTTP/1.0", headers: "Connection: keep-alive\r\n")) + XCTAssertEqual(try socket.response().status, 200) + } + } + + func testHttpShutdownClosesOtherConnectionsAndDropsPipelineTail() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let port = try XCTUnwrap(listener.port) + let idle = TransportSocket(port: port) + let shutdown = TransportSocket(port: port) + defer { idle.connection.cancel(); shutdown.connection.cancel() } + try idle.send(rawCommand(#"{"command":"uptime"}"#)) + XCTAssertEqual(try idle.response().status, 200) + try shutdown.send(rawCommand(#"{"command":"shutdown"}"#) + + rawCommand(#"{"command":"appState","commandId":"after-shutdown"}"#)) + let response = try shutdown.response() + XCTAssertEqual(response.status, 200) + XCTAssertTrue(response.head.contains("Connection: close")) + try shutdown.expectEOF() + try idle.expectEOF() + XCTAssertEqual(commandJournal.status(normalizedCommandId: "after-shutdown").lifecycleState, RunnerCommandLifecycleState.notAccepted.rawValue) + } + + func testCoalescedShutdownDeliversEveryResponseBeforeClosingTransport() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let port = try XCTUnwrap(listener.port) + let sockets = (0..<32).map { _ in TransportSocket(port: port) } + defer { sockets.forEach { $0.connection.cancel() } } + + let releaseCommand = DispatchSemaphore(value: 0) + commandExecutionQueue.async { releaseCommand.wait() } + defer { releaseCommand.signal() } + for socket in sockets { + try socket.send(rawCommand(#"{"command":"shutdown","commandId":"shared-shutdown"}"#)) + } + let deadline = Date().addingTimeInterval(3) + var waiterCount = 0 + repeat { + inFlightCommandLock.lock() + waiterCount = inFlightCommandWaiters["shared-shutdown"]?.count ?? 0 + inFlightCommandLock.unlock() + if waiterCount == sockets.count - 1 { break } + Thread.sleep(forTimeInterval: 0.01) + } while Date() < deadline + XCTAssertEqual(waiterCount, sockets.count - 1) + releaseCommand.signal() + + for socket in sockets { + let reply = try socket.response() + XCTAssertEqual(reply.status, 200) + XCTAssertTrue(try reply.envelope().ok) + XCTAssertTrue(reply.head.contains("Connection: close")) + try socket.expectEOF() + } + } + + func testHttpIdleExpiryPreservesUnusedLoopbackListenerWatch() throws { + let listener = try startTransportListener() + defer { listener.cancel() } + let port = try XCTUnwrap(listener.port) + let watch = TransportSocket(port: port) + let used = TransportSocket(port: port) + defer { watch.connection.cancel(); used.connection.cancel() } + try used.send(rawCommand(#"{"command":"uptime"}"#)) + XCTAssertEqual(try used.response().status, 200) + try used.expectEOF(timeout: Self.connectionIdleTimeout + 3) + try watch.send(rawCommand(#"{"command":"uptime"}"#)) + XCTAssertEqual(try watch.response().status, 200) + } + + private func rawCommand( + _ payload: String, + token: String = RunnerTests.transportToken, + version: String = "HTTP/1.1", + headers: String = "" + ) -> String { + "POST /command \(version)\r\nX-Agent-Device-Runner-Token: \(token)\r\nContent-Length: \(payload.utf8.count)\r\n\(headers)\r\n\(payload)" + } + + private final class TransportSocket: @unchecked Sendable { + let connection: NWConnection + var buffer = Data() + var ended = false + var error: Error? + + init(port: NWEndpoint.Port) { + connection = NWConnection(host: .ipv4(.loopback), port: port, using: .tcp) + connection.start(queue: DispatchQueue(label: "agent-device.test.http-client")) + } + + func send(_ text: String) throws { + let sent = XCTestExpectation(description: "raw request sent") + connection.send(content: Data(text.utf8), completion: .contentProcessed { error in + self.error = error + sent.fulfill() + }) + guard XCTWaiter.wait(for: [sent], timeout: 3) == .completed else { throw SocketError.timedOut } + if let error { throw error } + } + + struct Reply { + let head: String + let body: Data + var status: Int? { Int(head.split(separator: " ")[1]) } + func envelope() throws -> Response { try JSONDecoder().decode(Response.self, from: body) } + } + + enum SocketError: Error { case timedOut, prematureEOF, missingLength } + + func response(timeout: TimeInterval = 3) throws -> Reply { + while true { + if let split = buffer.range(of: Data("\r\n\r\n".utf8)) { + let head = String(decoding: buffer[..= end { + let body = Data(buffer[split.upperBound.. 1 ? Int(parts[1]) : nil } func testDuplicateCommandIdCoalescesOntoInFlightExecution() throws { @@ -225,8 +477,8 @@ extension RunnerTests { from: Data(#"{"command":"snapshot","commandId":"snapshot-coalesce"}"#.utf8) ) final class Delivered { - var primaryData: Data? - var waiterData: Data? + var primaryData: RunnerHttpResponse? + var waiterData: RunnerHttpResponse? } let delivered = Delivered() defer { @@ -236,21 +488,21 @@ extension RunnerTests { XCTAssertFalse( attachToInFlightCommandIfNeeded(command: command) { result in - delivered.primaryData = result.data + delivered.primaryData = result.response } ) XCTAssertTrue( attachToInFlightCommandIfNeeded(command: command) { result in - delivered.waiterData = result.data + delivered.waiterData = result.response } ) - let result = Data("single-result".utf8) + let result = RunnerHttpResponse(status: 200, body: "single-result") deliverCommandResult( command: command, result: (result, false) ) { result in - delivered.primaryData = result.data + delivered.primaryData = result.response } XCTAssertEqual(delivered.primaryData, result) diff --git a/apple/runner/RUNNER_PROTOCOL.md b/apple/runner/RUNNER_PROTOCOL.md index 2c0a8394a1..c5a9e91743 100644 --- a/apple/runner/RUNNER_PROTOCOL.md +++ b/apple/runner/RUNNER_PROTOCOL.md @@ -8,10 +8,16 @@ The Apple runner speaks a small internal HTTP+JSON protocol between the TypeScri - Content type: `application/json` - Request body: one JSON command object - Response body: one JSON envelope -- One request per connection; every response carries `Connection: close`. +- HTTP/1.1 connections persist by default. `Connection: close` ends them after the response. +- HTTP/1.0 connections persist only when the request includes `Connection: keep-alive`. +- Pipelined requests run in order. Each request repeats token, route, and length validation. +- Shutdown and framing or authentication refusals close the connection. Shutdown closes other open connections after all coalesced shutdown responses finish sending. The daemon sends to `http://127.0.0.1:/command` for simulator and macOS runners. A physical device is reached over usbmux when cabled, otherwise at its CoreDevice tunnel address; it is never dialed at this Mac's loopback. +Loopback and CoreDevice tunnel requests use the global `fetch` connection pool to reuse TCP +connections. The usbmux transport sends `Connection: close` and opens a connection per request. + ### Listener binding Simulators and macOS bind `127.0.0.1` only. A physical device listens on every interface: a cabled @@ -58,6 +64,11 @@ accepted, except for a loopback peer, whose clock starts at its first byte: the watch holds an idle loopback connection to observe the runner's lifetime. Each refusal closes only its own connection; the listener keeps serving. +After a response, a reusable connection may idle for 30 s before it closes silently. The next +request gets a fresh 30 s arrival deadline starting at its first byte. Already-buffered requests +start that deadline when processing advances to them. The initial unused loopback listener watch +remains exempt from idle expiry. Command execution time does not consume the arrival or idle budget. + A runner reply without `X-Agent-Device-Runner-Auth: required` is session-fatal on the host: the session is invalidated, which stops the runner, and the command is neither resent nor replayed. Until a session has answered, every mutation is preceded by an `uptime` probe, so a runner that does