uts: port LiveObjects unit-test harness helpers into infra/unit

Package.swift

@@ -112,7 +112,10 @@ let package = Package(
             name: "UTS",
             dependencies: [
                 .byName(name: "Ably"),
-                .target(name: "_AblyPluginSupportPrivate")
+                .target(name: "_AblyPluginSupportPrivate"),
+                // The `objects` UTS module exercises the LiveObjects plugin, so the harness
+                // references its types (e.g. ProtocolTypes, DefaultInternalPlugin.ObjectMessageBox).
+                .target(name: "AblyLiveObjects")
             ],
             path: "Test/UTS",
             exclude: [

Test/UTS/infra/unit/MockWebSocket.swift

@@ -16,8 +16,13 @@ import Ably.Private
 /// (`respondWithSuccess()` + `sendToClient(...)`).
 final class MockWebSocketProvider: @unchecked Sendable {
     typealias ConnectionAttemptHandler = @Sendable (MockWebSocket) -> Void
+    /// Handler invoked with each `ARTProtocolMessage` the SDK sends towards the server (UTS
+    /// `onMessageFromClient`). Used by the LiveObjects `setup_synced_channel` pattern to respond to
+    /// `ATTACH` with `ATTACHED`+`OBJECT_SYNC` and to auto-`ACK` `OBJECT` publishes.
+    typealias MessageFromClientHandler = @Sendable (ARTProtocolMessage) -> Void
 
     private let onConnectionAttempt: ConnectionAttemptHandler?
+    private let onMessageFromClient: MessageFromClientHandler?
     private let lock = NSLock()
     private var latestConnection: MockWebSocket?
 
@@ -29,8 +34,10 @@ final class MockWebSocketProvider: @unchecked Sendable {
         return latestConnection
     }
 
-    init(onConnectionAttempt: ConnectionAttemptHandler? = nil) {
+    init(onConnectionAttempt: ConnectionAttemptHandler? = nil,
+         onMessageFromClient: MessageFromClientHandler? = nil) {
         self.onConnectionAttempt = onConnectionAttempt
+        self.onMessageFromClient = onMessageFromClient
     }
 
     fileprivate func register(_ socket: MockWebSocket) {
@@ -42,6 +49,10 @@ final class MockWebSocketProvider: @unchecked Sendable {
     fileprivate func fireConnectionAttempt(_ socket: MockWebSocket) {
         onConnectionAttempt?(socket)
     }
+
+    fileprivate func fireMessageFromClient(_ message: ARTProtocolMessage) {
+        onMessageFromClient?(message)
+    }
 }
 
 /// A single simulated WebSocket connection. Conforms to `ARTWebSocket` so the real
@@ -74,10 +85,23 @@ final class MockWebSocket: NSObject, ARTWebSocket, @unchecked Sendable {
         return sent
     }
     private var sent: [ARTProtocolMessage] = []
+
+    /// The raw frames the SDK has sent towards the server, in order. Object-message tests inspect the
+    /// generated wire form here (the decoded `ARTProtocolMessage.state` drops the outbound-only
+    /// `*WithObjectId` fields), mirroring ably-js's `captured.state` wire inspection.
+    var sentFrames: [Data] {
+        lock.lock(); defer { lock.unlock() }
+        return frames
+    }
+    private var frames: [Data] = []
     private let lock = NSLock()
 
     private let decoder: ARTEncoder
 
+    /// The provider that created this socket, notified of each client-sent message (UTS
+    /// `onMessageFromClient`). Weak to avoid a retain cycle; the provider outlives the socket.
+    fileprivate weak var provider: MockWebSocketProvider?
+
     init(request: URLRequest, decoder: ARTEncoder) {
         self.request = request
         self.decoder = decoder
@@ -103,7 +127,10 @@ final class MockWebSocket: NSObject, ARTWebSocket, @unchecked Sendable {
         guard let decoded = try? decoder.decodeProtocolMessage(data) else { return }
         lock.lock()
         sent.append(decoded)
+        frames.append(data)
         lock.unlock()
+        // UTS `onMessageFromClient`: let the test's simulated server react to what the client sent.
+        provider?.fireMessageFromClient(decoded)
     }
 
     // MARK: UTS server-side API
@@ -202,6 +229,7 @@ final class MockWebSocketFactory: NSObject, WebSocketFactory {
 
     func createWebSocket(with request: URLRequest, logger: InternalLog?) -> ARTWebSocket {
         let webSocket = MockWebSocket(request: request, decoder: decoder)
+        webSocket.provider = wsProvider
         wsProvider.register(webSocket)
         // Fire `onConnectionAttempt` on the work queue. This block is enqueued during the transport's
         // `setupWebSocket` turn, so it runs *after* the transport has wired up `delegate` and

Test/UTS/infra/unit/ProtocolMessage.swift

@@ -1,17 +1,30 @@
 import Foundation
 import Ably
 import Ably.Private
+import _AblyPluginSupportPrivate
+@testable import AblyLiveObjects
 
 /// A `Sendable` description of a server-to-client protocol message that a test injects via
 /// `MockWebSocket.sendToClient(_:)` / `sendToClientAndClose(_:)`.
+///
+/// The LiveObjects `OBJECT` / `OBJECT_SYNC` variants carry decoded ``ProtocolTypes/InboundObjectMessage``
+/// values (which are `Sendable`); the concrete `ARTProtocolMessage.state` — an array of the plugin's
+/// `ObjectMessageBox` boxes — is built at delivery time in `makeProtocolMessage()`, so no
+/// non-`Sendable` value crosses the queue hop. This mirrors what ably-cocoa's wire decoder produces
+/// (the `state` property holds already-decoded `APObjectMessageProtocol` boxes), so the mock can
+/// bypass wire encoding entirely.
 struct ProtocolMessage: Sendable {
 
     private enum Kind: Sendable {
-        case connected(connectionId: String, connectionKey: String, maxIdleInterval: TimeInterval, connectionStateTtl: TimeInterval)
-        case attached(channel: String, channelSerial: String)
+        case connected(connectionId: String, connectionKey: String, maxIdleInterval: TimeInterval, connectionStateTtl: TimeInterval, siteCode: String?, objectsGCGracePeriod: Double?)
+        case attached(channel: String, channelSerial: String, hasObjects: Bool)
         case error(code: Int, statusCode: Int, message: String)
-        case ack(msgSerial: Int, count: Int)
+        case ack(msgSerial: Int, count: Int, serials: [String?]?)
         case closed
+        case detached(channel: String)
+        case channelError(channel: String, code: Int, statusCode: Int, message: String)
+        case objectSync(channel: String, channelSerial: String?, state: [ProtocolTypes.InboundObjectMessage])
+        case object(channel: String, state: [ProtocolTypes.InboundObjectMessage])
     }
 
     private let kind: Kind
@@ -27,35 +40,61 @@ struct ProtocolMessage: Sendable {
     static func connected(connectionId: String,
                           connectionKey: String,
                           maxIdleInterval: TimeInterval = 15,
-                          connectionStateTtl: TimeInterval = 120) -> ProtocolMessage {
-        .init(kind: .connected(connectionId: connectionId, connectionKey: connectionKey, maxIdleInterval: maxIdleInterval, connectionStateTtl: connectionStateTtl))
+                          connectionStateTtl: TimeInterval = 120,
+                          siteCode: String? = nil,
+                          objectsGCGracePeriod: Double? = nil) -> ProtocolMessage {
+        .init(kind: .connected(connectionId: connectionId, connectionKey: connectionKey, maxIdleInterval: maxIdleInterval, connectionStateTtl: connectionStateTtl, siteCode: siteCode, objectsGCGracePeriod: objectsGCGracePeriod))
     }
 
-    /// An `ATTACHED` message for a channel (UTS `ProtocolMessage(action: ATTACHED, ...)`).
-    static func attached(channel: String, channelSerial: String) -> ProtocolMessage {
-        .init(kind: .attached(channel: channel, channelSerial: channelSerial))
+    /// An `ATTACHED` message for a channel (UTS `ProtocolMessage(action: ATTACHED, ...)`). Set
+    /// `hasObjects` to raise the `HAS_OBJECTS` flag (RTO4).
+    static func attached(channel: String, channelSerial: String, hasObjects: Bool = false) -> ProtocolMessage {
+        .init(kind: .attached(channel: channel, channelSerial: channelSerial, hasObjects: hasObjects))
     }
 
     /// An `ERROR` message (UTS `ProtocolMessage(action: ERROR, error: ErrorInfo(...))`).
     static func error(code: Int, statusCode: Int, message: String) -> ProtocolMessage {
         .init(kind: .error(code: code, statusCode: statusCode, message: message))
     }
 
-    /// An `ACK` message (UTS `ProtocolMessage(action: ACK, msgSerial: ..., count: ...)`).
-    static func ack(msgSerial: Int, count: Int) -> ProtocolMessage {
-        .init(kind: .ack(msgSerial: msgSerial, count: count))
+    /// An `ACK` message (UTS `ProtocolMessage(action: ACK, msgSerial: ..., count: ...)`). Pass
+    /// `serials` to populate the `res` array (UTS `build_ack_message`, one `res` entry carrying all
+    /// the object serials).
+    static func ack(msgSerial: Int, count: Int, serials: [String?]? = nil) -> ProtocolMessage {
+        .init(kind: .ack(msgSerial: msgSerial, count: count, serials: serials))
     }
 
     /// A `CLOSED` message (UTS `ProtocolMessage(action: CLOSED)`).
     static func closed() -> ProtocolMessage {
         .init(kind: .closed)
     }
 
+    /// A channel `DETACHED` message (UTS `ProtocolMessage(action: DETACHED, channel: ...)`).
+    static func detached(channel: String) -> ProtocolMessage {
+        .init(kind: .detached(channel: channel))
+    }
+
+    /// A channel-level `ERROR` message that transitions the channel to `FAILED`
+    /// (UTS `ProtocolMessage(action: ERROR, channel: ..., error: ...)`).
+    static func channelError(channel: String, code: Int, statusCode: Int, message: String) -> ProtocolMessage {
+        .init(kind: .channelError(channel: channel, code: code, statusCode: statusCode, message: message))
+    }
+
+    /// An `OBJECT_SYNC` message carrying object state (UTS `build_object_sync_message`).
+    static func objectSync(channel: String, channelSerial: String?, state: [ProtocolTypes.InboundObjectMessage]) -> ProtocolMessage {
+        .init(kind: .objectSync(channel: channel, channelSerial: channelSerial, state: state))
+    }
+
+    /// An `OBJECT` message carrying object operations (UTS `build_object_message`).
+    static func object(channel: String, state: [ProtocolTypes.InboundObjectMessage]) -> ProtocolMessage {
+        .init(kind: .object(channel: channel, state: state))
+    }
+
     /// Builds the concrete `ARTProtocolMessage`. Call on the delegate queue, at delivery time.
     func makeProtocolMessage() -> ARTProtocolMessage {
         let message = ARTProtocolMessage()
         switch kind {
-        case let .connected(connectionId, connectionKey, maxIdleInterval, connectionStateTtl):
+        case let .connected(connectionId, connectionKey, maxIdleInterval, connectionStateTtl, siteCode, objectsGCGracePeriod):
             message.action = .connected
             message.connectionId = connectionId
             message.connectionKey = connectionKey
@@ -68,23 +107,63 @@ struct ProtocolMessage: Sendable {
                 connectionStateTtl: connectionStateTtl,
                 serverId: "",
                 maxIdleInterval: maxIdleInterval,
-                objectsGCGracePeriod: nil,
-                siteCode: nil
+                objectsGCGracePeriod: objectsGCGracePeriod.map { NSNumber(value: $0) },
+                siteCode: siteCode
             )
-        case let .attached(channel, channelSerial):
+        case let .attached(channel, channelSerial, hasObjects):
             message.action = .attached
             message.channel = channel
             message.channelSerial = channelSerial
+            if hasObjects {
+                // ARTProtocolMessageFlagHasObjects = 1 << 7
+                message.flags |= (1 << 7)
+            }
         case let .error(code, statusCode, text):
             message.action = .error
             message.error = ARTErrorInfo.create(withCode: code, status: statusCode, message: text)
-        case let .ack(msgSerial, count):
+        case let .ack(msgSerial, count, serials):
             message.action = .ack
             message.msgSerial = NSNumber(value: msgSerial)
             message.count = Int32(count)
+            if let serials {
+                message.res = [ARTPublishResult(serials: serials.map { ARTPublishResultSerial(value: $0) })]
+            }
         case .closed:
             message.action = .closed
+        case let .detached(channel):
+            message.action = .detached
+            message.channel = channel
+        case let .channelError(channel, code, statusCode, text):
+            message.action = .error
+            message.channel = channel
+            message.error = ARTErrorInfo.create(withCode: code, status: statusCode, message: text)
+        case let .objectSync(channel, channelSerial, state):
+            message.action = .objectSync
+            message.channel = channel
+            message.channelSerial = channelSerial
+            message.setState(state)
+        case let .object(channel, state):
+            message.action = .object
+            message.channel = channel
+            message.setState(state)
         }
         return message
     }
 }
+
+private extension ARTProtocolMessage {
+    /// Sets the object `state` on the protocol message.
+    ///
+    /// `ARTProtocolMessage.state` is gated behind `#ifdef ABLY_SUPPORTS_PLUGINS` in ably-cocoa, a
+    /// define set only for ably-cocoa's own target — so the property is absent from the `Ably` Swift
+    /// interface for consumers. (Defining the macro for our import isn't viable: it unlocks other
+    /// gated headers, e.g. `ARTPluginAPI.h` → `ARTTypes.h`, that aren't on a consumer's header search
+    /// path, breaking the `Ably` module build.) The property exists in the compiled runtime, so we
+    /// set it via KVC. The boxes are the same `ObjectMessageProtocol` type the plugin emits and
+    /// receives, so ably-cocoa passes them straight back to the plugin's `handleObjectSync…` hook —
+    /// mirroring what the wire decoder would have produced, without needing wire encoding.
+    func setState(_ objectMessages: [ProtocolTypes.InboundObjectMessage]) {
+        let boxes = objectMessages.map { DefaultInternalPlugin.ObjectMessageBox(objectMessage: $0) }
+        setValue(boxes, forKey: "state")
+    }
+}