[AIT-1103] fix(liveobjects): Fix UTS unit tests

.claude/skills/uts-to-kotlin/references/objects-mapping.md

@@ -104,7 +104,7 @@ live in **`io.ably.lib.liveobjects.path.types`** (not `.path`) — import them f
 | Spec / ably-js | ably-java |
 |---|---|
 | `pathObj.path()` | `pathObj.path(): String` |
-| `pathObj.instance()` | `pathObj.instance(): Instance?` (null if path resolves to a primitive, or doesn't resolve) |
+| `pathObj.instance()` | `pathObj.instance(): Instance?` (wraps any resolved value incl. primitives per `RTPO8f`; null only when the path doesn't resolve) |
 | `pathObj.compactJson()` | `pathObj.compactJson(): JsonElement?` |
 | `pathObj.compact()` | **Not implemented in ably-java** (`RTTS3f`: typed SDKs need not implement `compact`). Use `compactJson()` for snapshot assertions; if a spec genuinely needs the non-JSON `compact()` shape, that's a deviation — flag it. |
 | `pathObj.subscribe(listener[, opts])` | `pathObj.subscribe(PathObjectListener[, PathObjectSubscriptionOptions]): Subscription` (§8) |
@@ -473,7 +473,8 @@ Spec assertions like `FAILS WITH error code 92007` map to ably-java exceptions:
 | write where path doesn't resolve | `AblyException` 400 / code `92005` |
 | write where value isn't the required type | `AblyException` 400 / code `92007` |
 | `get()` / op when channel lacks the object mode (`RTO23a`/`RTO2a2`) | `AblyException` 400 / code `40024` |
-| `get()` / access when channel is DETACHED or FAILED (`RTO23b`/`RTO25`) | `AblyException` 400 / code `90001` |
+| access methods (`on`/`off`/`offAll`) when channel is DETACHED or FAILED (`RTO25b`) | `AblyException` 400 / code `90001` |
+| `get()` when channel is FAILED (`RTO23e`/`RTL33c`) | `AblyException` 400 / code `90001`. **But `get()` on a DETACHED channel does *not* throw** — ensure-active-channel re-attaches and `get()` resolves (`RTO23e`/`RTL33b`); only the access methods above gate on DETACHED |
 | channel enters DETACHED/SUSPENDED/FAILED while awaiting SYNCED (`RTO20e`/`RTO23c`) | `AblyException` 400 / code `92008` |
 | write while `echoMessages` is false (`RTO26c`) | `AblyException` 400 / code `40000` |
 
@@ -554,15 +555,14 @@ setup or message JSON.**
 | `build_public_object_message(msg, channel)` | `buildPublicObjectMessage(wireJson, channel)` (reflective; §11) |
 | `STANDARD_POOL_OBJECTS` | `STANDARD_POOL_OBJECTS` |
 | inline ObjectData / map-entry / state fragments | `dataString` / `dataNumber` / `dataBoolean` / `dataObjectId` / `dataBytes` / `dataJson`, `mapEntry`, `mapState`, `counterState`, `mapCreateOp`, `counterCreateOp` |
+| Canonical Constants: `POOL_SERIAL`, `ack_serial(m, i)`, `remote_serial(i)`, `below_ack_serial(i)` | `POOL_SERIAL` (`"t:0"`), `ackSerial(msgSerial, i)`, `remoteSerial(i)`, `belowAckSerial(i)` — use these, never hand-rolled `"t:N"` literals (serials are compared as strings, so ad-hoc values silently sort wrong) |
 
 `mock_ws.send_to_client(...)` is the existing `mockWs.sendToClient(...)` (§ mock API in the main skill). The
 wire `action` / `semantics` are integer enum codes — the builders emit the codes for you.
 
-> **Runtime caveat:** `setupSyncedChannel` returns only once `RealtimeObject.get()` resolves, which needs
-> the `:liveobjects` SDK's OBJECT_SYNC processing. Until that lands the helpers **compile** and the test
-> structure is correct, but the setup throws at runtime — i.e. translate-only today, runnable once the SDK
-> is implemented. (`buildPublicObjectMessage` does *not* depend on this — the message/operation layer is
-> implemented, so those tests can run now.)
+> **Runtime status:** the `:liveobjects` SDK's OBJECT_SYNC processing and `RealtimeObject.get()` are
+> implemented — `setupSyncedChannel` and the full objects unit suite run green, so translate **and
+> evaluate** (don't stop at compile-only).
 
 (For the **integration** tier's REST fixture helper — `provision_objects_via_rest` — see §14.)
 
@@ -600,8 +600,8 @@ reflection and no `:liveobjects` dependency — it compiles and runs against `:j
 > (`sandbox.realtime.ably-nonprod.net`) — the same nonprod host `SandboxApp` and the realtime clients use,
 > **not** `environment="sandbox"` (which resolves to the legacy `sandbox-rest.ably.io`, and
 > can't be combined with `restHost` per `Hosts.java` TO3k2/TO3k3). The REST call hits the live sandbox
-> today; the realtime client it provisions for only *observes* the data once the SDK's OBJECT_SYNC +
-> `RealtimeObject.get()` land.
+> today, and the realtime client observes the provisioned data through the SDK's OBJECT_SYNC +
+> `RealtimeObject.get()` (both implemented).
 
 ---
 

lib/src/main/java/io/ably/lib/liveobjects/path/PathObject.java

@@ -79,16 +79,14 @@ public interface PathObject {
     @NotNull String path();
 
     /**
-     * Resolves this path and returns a {@link Instance} wrapping the underlying
-     * value if it is a {@code LiveMap} or {@code LiveCounter}.
+     * Resolves this path and returns a {@link Instance} wrapping the resolved value,
+     * whether it is a {@code LiveMap}, {@code LiveCounter} or a primitive (RTPO8c/RTPO8f).
      *
-     * <p>Returns {@code null} when the resolved value is a primitive (LiveObjects with
-     * no object id), when the path does not resolve, or when called on primitive
-     * {@code *PathObject} sub-types.
+     * <p>Returns {@code null} when the path does not resolve (RTPO8e).
      *
      * <p>Spec: RTPO8 / RTTS3b
      *
-     * @return a {@link Instance} wrapping the resolved live object, or {@code null}
+     * @return a {@link Instance} wrapping the resolved value, or {@code null}
      */
     @Nullable Instance instance();
 

lib/src/main/java/io/ably/lib/liveobjects/path/types/BinaryPathObject.java

@@ -7,8 +7,8 @@
  * A {@link PathObject} whose underlying value is expected to be a binary blob
  * (a {@code byte[]}).
  *
- * <p>This is a terminal type. {@link PathObject#instance()} returns {@code null}
- * because a primitive resolution does not produce a wrapped LiveObject; navigation
+ * <p>This is a terminal type. {@link PathObject#instance()} returns an {@code Instance}
+ * wrapping the resolved primitive value per RTPO8f; navigation
  * via {@code at(...)} is not available here because it is only defined on
  * {@code LiveMapPathObject}. Only {@link #value()} and the inherited read APIs are
  * useful here.

lib/src/main/java/io/ably/lib/liveobjects/path/types/BooleanPathObject.java

@@ -6,8 +6,8 @@
 /**
  * A {@link PathObject} whose underlying value is expected to be a {@code Boolean}.
  *
- * <p>This is a terminal type. {@link PathObject#instance()} returns {@code null}
- * because a primitive resolution does not produce a wrapped LiveObject; navigation
+ * <p>This is a terminal type. {@link PathObject#instance()} returns an {@code Instance}
+ * wrapping the resolved primitive value per RTPO8f; navigation
  * via {@code at(...)} is not available here because it is only defined on
  * {@code LiveMapPathObject}. Only {@link #value()} and the inherited read APIs are
  * useful here.

lib/src/main/java/io/ably/lib/liveobjects/path/types/JsonArrayPathObject.java

@@ -7,8 +7,8 @@
 /**
  * A {@link PathObject} whose underlying value is expected to be a {@link JsonArray}.
  *
- * <p>This is a terminal type. {@link PathObject#instance()} returns {@code null}
- * because this resolution does not produce a wrapped LiveObject instance; navigation
+ * <p>This is a terminal type. {@link PathObject#instance()} returns an {@code Instance}
+ * wrapping the resolved primitive value per RTPO8f; navigation
  * via {@code at(...)} is not available here because it is only defined on
  * {@code LiveMapPathObject}. Only {@link #value()} and the inherited read APIs are
  * useful here.

lib/src/main/java/io/ably/lib/liveobjects/path/types/JsonObjectPathObject.java

@@ -7,8 +7,8 @@
 /**
  * A {@link PathObject} whose underlying value is expected to be a {@link JsonObject}.
  *
- * <p>This is a terminal type. {@link PathObject#instance()} returns {@code null}
- * because this resolution does not produce a wrapped LiveObject instance; navigation
+ * <p>This is a terminal type. {@link PathObject#instance()} returns an {@code Instance}
+ * wrapping the resolved primitive value per RTPO8f; navigation
  * via {@code at(...)} is not available here because it is only defined on
  * {@code LiveMapPathObject}. Only {@link #value()} and the inherited read APIs are
  * useful here.

lib/src/main/java/io/ably/lib/liveobjects/path/types/NumberPathObject.java

@@ -6,8 +6,8 @@
 /**
  * A {@link PathObject} whose underlying value is expected to be a {@code Number}.
  *
- * <p>This is a terminal type. {@link PathObject#instance()} returns {@code null}
- * because a primitive resolution does not produce a wrapped LiveObject; navigation
+ * <p>This is a terminal type. {@link PathObject#instance()} returns an {@code Instance}
+ * wrapping the resolved primitive value per RTPO8f; navigation
  * via {@code at(...)} is not available here because it is only defined on
  * {@code LiveMapPathObject}. Only {@link #value()} and the inherited read APIs are
  * useful here.

lib/src/main/java/io/ably/lib/liveobjects/path/types/StringPathObject.java

@@ -6,8 +6,8 @@
 /**
  * A {@link PathObject} whose underlying value is expected to be a {@code String}.
  *
- * <p>This is a terminal type. {@link PathObject#instance()} returns {@code null}
- * because a primitive resolution does not produce a wrapped LiveObject; navigation
+ * <p>This is a terminal type. {@link PathObject#instance()} returns an {@code Instance}
+ * wrapping the resolved primitive value per RTPO8f; navigation
  * via {@code at(...)} is not available here because it is only defined on
  * {@code LiveMapPathObject}. Only {@link #value()} and the inherited read APIs are
  * useful here.

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/DefaultRealtimeObject.kt

@@ -42,10 +42,12 @@ internal class DefaultRealtimeObject(
    */
   internal val objectsPool = ObjectsPool(this)
 
+  // Non-volatile by design: written and read only on [sequentialScope].
   internal var state = ObjectsState.Initialized
 
   /**
    * Set of serials for operations applied locally upon ACK, awaiting deduplication of the server echo.
+   * Unsynchronized: accessed only on [sequentialScope].
    * @spec RTO7b, RTO7b1
    */
   internal val appliedOnAckSerials = mutableSetOf<String>()
@@ -62,7 +64,12 @@ internal class DefaultRealtimeObject(
   internal val pathObjectSubscriptionRegister = PathObjectSubscriptionRegister(this)
 
   /**
-   *  Coroutine scope for running sequential operations on a single thread, used to avoid concurrency issues.
+   * Coroutine scope running all objects work one task at a time. `limitedParallelism(1)` is a view
+   * over the shared Default pool - no dedicated thread is created or blocked, and the single slot is
+   * released at every suspension point (e.g. while awaiting a publish ACK). It is the safety contract
+   * for the unsynchronized state in this class and in ObjectsManager/ObjectsStateCoordinator
+   * (e.g. [state], [appliedOnAckSerials], pendingSyncWaiters), and its FIFO order preserves
+   * submission-order publishing for un-awaited mutations.
    */
   private val sequentialScope =
     CoroutineScope(Dispatchers.Default.limitedParallelism(1) + CoroutineName(channelName) + SupervisorJob())
@@ -81,6 +88,16 @@ internal class DefaultRealtimeObject(
   /**
    * Runs [block] on the sequential scope and exposes it as a CompletableFuture. Failures
    * complete the future exceptionally with the original AblyException.
+   *
+   * Deliberately kept on [sequentialScope] rather than a parallel dispatcher, at no performance
+   * cost: blocks suspend almost immediately on network I/O (releasing the slot), so only
+   * microseconds of CPU work per operation are serialized, and all publishes funnel into the
+   * single connection anyway. In return the FIFO order keeps un-awaited back-to-back mutations
+   * publishing in submission order (deterministic LWW outcomes), [dispose] can cancel in-flight
+   * operations via this scope's children, and [get]'s sync wait stays race-free (see [get]).
+   *
+   * Caveat: the returned future completes on this scope, so synchronous dependents (thenApply etc.)
+   * run on the lane - a blocking callback there stalls objects processing for the channel.
    */
   internal fun <T> asyncFuture(block: suspend () -> T): CompletableFuture<T> =
     sequentialScope.future { block() }
@@ -93,7 +110,11 @@ internal class DefaultRealtimeObject(
     asyncFuture(block).thenApply { null }
 
   override fun get(): CompletableFuture<LiveMapPathObject> {
-    throwIfInvalidAccessApiConfiguration() // RTO23a
+    // RTO23a - get() checks only the object_subscribe MODE here;
+    throwIfMissingObjectSubscribeMode()
+    // MUST run on sequentialScope: ensureSynced()'s state check + SYNCED-listener registration is
+    // atomic only there (SYNCED transitions run on the same scope). On a parallel dispatcher the
+    // SYNCED event could fire between the two and this future would hang forever (lost wakeup).
     return asyncFuture { getRootAsync() }
   }
 
@@ -179,7 +200,10 @@ internal class DefaultRealtimeObject(
     }
     if (syntheticMessages.isEmpty()) return
 
-    // RTO20e, RTO20f - dispatch to sequential scope for ordering
+    // RTO20e, RTO20f - dispatch to sequential scope for ordering. This hop is also the thread-safety
+    // boundary of the write path: everything above it touches only thread-safe/local state, while
+    // applyAckResult touches unsynchronized state (sync waiters, objects state, pool mutation) whose
+    // safety contract is the sequential scope.
     withContext(sequentialScope.coroutineContext) {
       objectsManager.applyAckResult(syntheticMessages) // suspends if SYNCING (RTO20e), applies on SYNCED (RTO20f)
     }
@@ -277,7 +301,7 @@ internal class DefaultRealtimeObject(
             ObjectHttpStatusCode.BadRequest,
             cause = errorReason?.let { AblyException.fromErrorInfo(it) }
           )
-          objectsManager.failBufferedAcks(error) // RTO20e1
+          objectsManager.failSyncWaiters(error) // RTO20e1
           if (state != ChannelState.suspended) {
             // do not emit data update events as the actual current state of Objects data is unknown when we're in these channel states
             objectsPool.clearObjectsData(false)
@@ -305,6 +329,9 @@ internal class DefaultRealtimeObject(
   /** Validates the channel is configured for access (read/subscribe) operations. Spec: RTO25 */
   internal fun throwIfInvalidAccessApiConfiguration() = adapter.throwIfInvalidAccessApiConfiguration(channelName)
 
+  /** Validates only the object_subscribe channel mode. Spec: RTO23a */
+  internal fun throwIfMissingObjectSubscribeMode() = adapter.throwIfMissingObjectSubscribeMode(channelName)
+
   /** Validates the channel is configured for write (mutation) operations. Spec: RTO26 */
   internal fun throwIfInvalidWriteApiConfiguration() = adapter.throwIfInvalidWriteApiConfiguration(channelName)
 

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/Helpers.kt

@@ -42,6 +42,18 @@ internal fun AblyClientAdapter.throwIfInvalidAccessApiConfiguration(channelName:
   throwIfMissingChannelMode(channelName, ChannelMode.object_subscribe)
 }
 
+/**
+ * Validates only the `object_subscribe` channel mode (no channel-state check), for
+ * `RealtimeObject.get()` (RTO23a). Unlike the access methods, `get()` delegates channel-state handling to
+ * ensure-active-channel ([ensureAttached], RTO23e / RTL33), which re-attaches a DETACHED channel and
+ * rejects only FAILED — so `get()` must not pre-empt that with a state gate.
+ *
+ * Spec: RTO23a
+ */
+internal fun AblyClientAdapter.throwIfMissingObjectSubscribeMode(channelName: String) {
+  throwIfMissingChannelMode(channelName, ChannelMode.object_subscribe)
+}
+
 /**
  * Validates that the channel is configured for the write (mutation) API: message echo must be
  * enabled, the channel must be usable (not detached/failed/suspended) and have the

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/ObjectsManager.kt

@@ -10,9 +10,7 @@ import io.ably.lib.liveobjects.value.ObjectUpdate
 import io.ably.lib.liveobjects.value.livecounter.InternalLiveCounter
 import io.ably.lib.liveobjects.value.livemap.InternalLiveMap
 import io.ably.lib.liveobjects.value.livemap.isEntryOrRefTombstoned
-import io.ably.lib.types.AblyException
 import io.ably.lib.util.Log
-import kotlinx.coroutines.CompletableDeferred
 
 /**
  * @spec RTO5 - Processes OBJECT and OBJECT_SYNC messages during sync sequences
@@ -29,7 +27,6 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    * @spec RTO7 - Buffered object operations during sync
    */
   private val bufferedObjectOperations = mutableListOf<WireObjectMessage>() // RTO7a
-  private var syncCompletionWaiter: CompletableDeferred<Unit>? = null
 
   /**
    * Handles object messages (non-sync messages).
@@ -85,7 +82,6 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
 
     syncObjectsPool.clear() // RTO5a2a
     currentSyncId = syncId
-    syncCompletionWaiter = CompletableDeferred()
     stateChange(ObjectsState.Syncing)
   }
 
@@ -102,33 +98,25 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
     syncObjectsPool.clear()                                                        // RTO5c4
     currentSyncId = null                                                           // RTO5c3
     realtimeObjects.appliedOnAckSerials.clear()                                    // RTO5c9
-    stateChange(ObjectsState.Synced)                              // RTO5c8
-    syncCompletionWaiter?.complete(Unit)
-    syncCompletionWaiter = null
+    stateChange(ObjectsState.Synced)                              // RTO5c8 - emits SYNCED, resolving any
+    // pending applyAckResult waiters (RTO20e) via awaitSyncCompletion's one-shot SYNCED listener
   }
 
   /**
    * Called from publishAndApply (via withContext sequentialScope).
    * If SYNCED: apply immediately with LOCAL source.
-   * If not SYNCED: suspend until endSync transitions to SYNCED (RTO20e), then apply.
+   * If not SYNCED: suspend until objects transition to SYNCED (RTO20e), then apply. If the channel leaves a
+   * usable state while waiting, [awaitSyncCompletion] throws the 92008 error and the apply fails (RTO20e1).
    */
   internal suspend fun applyAckResult(messages: List<WireObjectMessage>) {
+    // MUST run on the sequential scope: the state check + waiter registration in awaitSyncCompletion
+    // is atomic only there (same lost-wakeup hazard as ensureSynced).
     if (realtimeObjects.state != ObjectsState.Synced) {
-      if (syncCompletionWaiter == null) syncCompletionWaiter = CompletableDeferred()
-      syncCompletionWaiter?.await() // suspends; resumes after endSync transitions to SYNCED (RTO20e1)
+      awaitSyncCompletion() // suspends until SYNCED (RTO20e); throws 92008 on channel state change (RTO20e1)
     }
     applyObjectMessages(messages, ObjectsOperationSource.LOCAL) // RTO20f
   }
 
-  /**
-   * Fails all pending apply waiters.
-   * Called when the channel enters DETACHED/SUSPENDED/FAILED (RTO20e1).
-   */
-  internal fun failBufferedAcks(error: AblyException) {
-    syncCompletionWaiter?.completeExceptionally(error)
-    syncCompletionWaiter = null
-  }
-
   /**
    * Clears the sync objects pool.
    * Used by DefaultRealtimeObjects.handleStateChange.
@@ -352,7 +340,8 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
   }
 
   internal fun dispose() {
-    syncCompletionWaiter?.cancel()
+    // pending applyAckResult waiters are cancelled via the sequential scope teardown in
+    // DefaultRealtimeObject.dispose (their coroutines are cancelled, removing them from the waiter set)
     syncObjectsPool.clear()
     bufferedObjectOperations.clear()
     disposeObjectsStateListeners()