refactor(liveobjects): rename async future helpers and cover their propagation

Rename the DefaultRealtimeObject async bridges for clarity: asyncApi -> asyncFuture
(generic, value-returning) and asyncVoidApi -> asyncVoidFuture (the Unit->Void adapter
required by the Java write APIs), updating all call sites. Behaviour is unchanged; the
thenApply { null } bridge stays because kotlinx's future builder cannot produce a
CompletableFuture<Void> (its type param is non-null, Void's only value is null).

Add DefaultRealtimeObjectAsyncTest asserting result and exception propagation for both
helpers, and - crucially - that a failed operation does not cancel the shared sequential
scope for subsequent operations, locking in the SupervisorJob failure-isolation guarantee.

Author: sacOO7

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

@@ -78,18 +78,25 @@ internal class DefaultRealtimeObject(
     incomingObjectsHandler = initializeHandlerForIncomingObjectMessages()
   }
 
-  override fun get(): CompletableFuture<LiveMapPathObject> {
-    throwIfInvalidAccessApiConfiguration() // RTO23a
-    return asyncApi { getRootAsync() }
-  }
-
   /**
    * Runs [block] on the sequential scope and exposes it as a CompletableFuture. Failures
    * complete the future exceptionally with the original AblyException.
    */
-  internal fun <T> asyncApi(block: suspend () -> T): CompletableFuture<T> =
+  internal fun <T> asyncFuture(block: suspend () -> T): CompletableFuture<T> =
     sequentialScope.future { block() }
 
+  /**
+   * Runs a mutating [block] on the sequential scope, exposed as a CompletableFuture<Void>.
+   * Used by the path/instance write APIs.
+   */
+  internal fun asyncVoidFuture(block: suspend () -> Unit): CompletableFuture<Void> =
+    asyncFuture(block).thenApply { null }
+
+  override fun get(): CompletableFuture<LiveMapPathObject> {
+    throwIfInvalidAccessApiConfiguration() // RTO23a
+    return asyncFuture { getRootAsync() }
+  }
+
   override fun on(event: ObjectStateEvent, listener: ObjectStateChange.Listener): Subscription {
     throwIfInvalidAccessApiConfiguration()
     return objectsManager.on(event, listener)
@@ -115,13 +122,6 @@ internal class DefaultRealtimeObject(
     return DefaultLiveMapPathObject(this, "")
   }
 
-  /**
-   * Runs a mutating [block] on the sequential scope, exposed as a CompletableFuture<Void>.
-   * Used by the path/instance write APIs.
-   */
-  internal fun asyncVoidApi(block: suspend () -> Unit): CompletableFuture<Void> =
-    asyncApi(block).thenApply { null }
-
   /**
    * Spec: RTO14
    */

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/instance/types/DefaultLiveCounterInstance.kt

@@ -41,15 +41,15 @@ internal class DefaultLiveCounterInstance(
 
   override fun increment(amount: Number): CompletableFuture<Void> {
     channelObject.throwIfInvalidWriteApiConfiguration() // RTINS14b
-    return channelObject.asyncVoidApi { counter.increment(amount) } // RTINS14c -> RTLC12
+    return channelObject.asyncVoidFuture { counter.increment(amount) } // RTINS14c -> RTLC12
   }
 
   // RTINS15a1 - default amount of 1; delegates to decrement(Number)
   override fun decrement(): CompletableFuture<Void> = decrement(1)
 
   override fun decrement(amount: Number): CompletableFuture<Void> {
     channelObject.throwIfInvalidWriteApiConfiguration() // RTINS15b
-    return channelObject.asyncVoidApi { counter.decrement(amount) } // RTINS15c -> RTLC13
+    return channelObject.asyncVoidFuture { counter.decrement(amount) } // RTINS15c -> RTLC13
   }
 
   override fun subscribe(listener: InstanceListener): Subscription {

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/instance/types/DefaultLiveMapInstance.kt

@@ -71,12 +71,12 @@ internal class DefaultLiveMapInstance(
 
   override fun set(key: String, value: LiveMapValue): CompletableFuture<Void> {
     channelObject.throwIfInvalidWriteApiConfiguration() // RTINS12b
-    return channelObject.asyncVoidApi { map.set(key, value) } // RTINS12c -> RTLM20
+    return channelObject.asyncVoidFuture { map.set(key, value) } // RTINS12c -> RTLM20
   }
 
   override fun remove(key: String): CompletableFuture<Void> {
     channelObject.throwIfInvalidWriteApiConfiguration() // RTINS13b
-    return channelObject.asyncVoidApi { map.remove(key) } // RTINS13c -> RTLM21
+    return channelObject.asyncVoidFuture { map.remove(key) } // RTINS13c -> RTLM21
   }
 
   override fun subscribe(listener: InstanceListener): Subscription {

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/path/types/DefaultLiveCounterPathObject.kt

@@ -36,7 +36,7 @@ internal class DefaultLiveCounterPathObject(
     if (resolvedValue !is ResolvedValue.CounterRef) {
       throw typeMismatchError("Cannot increment a non-LiveCounter object at path: \"$path\"") // RTPO17e
     }
-    return channelObject.asyncVoidApi { resolvedValue.counter.increment(amount) } // RTPO17d -> RTLC12
+    return channelObject.asyncVoidFuture { resolvedValue.counter.increment(amount) } // RTPO17d -> RTLC12
   }
 
   // RTPO18a1 - default amount of 1; delegates to decrement(Number)
@@ -48,6 +48,6 @@ internal class DefaultLiveCounterPathObject(
     if (resolvedValue !is ResolvedValue.CounterRef) {
       throw typeMismatchError("Cannot decrement a non-LiveCounter object at path: \"$path\"") // RTPO18e
     }
-    return channelObject.asyncVoidApi { resolvedValue.counter.decrement(amount) } // RTPO18d -> RTLC13
+    return channelObject.asyncVoidFuture { resolvedValue.counter.decrement(amount) } // RTPO18d -> RTLC13
   }
 }

liveobjects/src/main/kotlin/io/ably/lib/liveobjects/path/types/DefaultLiveMapPathObject.kt

@@ -67,7 +67,7 @@ internal class DefaultLiveMapPathObject(
     if (resolvedValue !is ResolvedValue.MapRef) {
       throw typeMismatchError("Cannot set a key on a non-LiveMap object at path: \"$path\"") // RTPO15e
     }
-    return channelObject.asyncVoidApi { resolvedValue.map.set(key, value) } // RTPO15d -> RTLM20
+    return channelObject.asyncVoidFuture { resolvedValue.map.set(key, value) } // RTPO15d -> RTLM20
   }
 
   override fun remove(key: String): CompletableFuture<Void> {
@@ -76,6 +76,6 @@ internal class DefaultLiveMapPathObject(
     if (resolvedValue !is ResolvedValue.MapRef) {
       throw typeMismatchError("Cannot remove a key from a non-LiveMap object at path: \"$path\"") // RTPO16e
     }
-    return channelObject.asyncVoidApi { resolvedValue.map.remove(key) } // RTPO16d -> RTLM21
+    return channelObject.asyncVoidFuture { resolvedValue.map.remove(key) } // RTPO16d -> RTLM21
   }
 }

liveobjects/src/test/kotlin/io/ably/lib/liveobjects/unit/DefaultRealtimeObjectAsyncTest.kt

@@ -0,0 +1,74 @@
+package io.ably.lib.liveobjects.unit
+
+import io.ably.lib.liveobjects.DefaultRealtimeObject
+import io.ably.lib.types.AblyException
+import io.ably.lib.types.ErrorInfo
+import io.mockk.unmockkAll
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertSame
+import org.junit.Test
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
+import kotlin.test.assertFailsWith
+
+/**
+ * Unit tests for [DefaultRealtimeObject.asyncFuture] / [DefaultRealtimeObject.asyncVoidFuture]:
+ * result and exception propagation, plus the SupervisorJob guarantee that a failed operation
+ * must not cancel the shared sequential scope for subsequent (sibling) operations.
+ */
+class DefaultRealtimeObjectAsyncTest {
+
+  private fun newObject(): DefaultRealtimeObject =
+    DefaultRealtimeObject("ch", getMockAblyClientAdapter())
+
+  private fun boom() = AblyException.fromErrorInfo(ErrorInfo("boom", 400, 40000))
+
+  @After
+  fun tearDown() = unmockkAll() // getMockAblyClientAdapter uses mockkStatic - clean up global state
+
+  @Test
+  fun asyncFutureCompletesWithResult() {
+    val result = newObject().asyncFuture { 42 }.get(2, TimeUnit.SECONDS)
+    assertEquals(42, result)
+  }
+
+  @Test
+  fun asyncVoidFutureCompletesWithNullOnSuccess() {
+    val result = newObject().asyncVoidFuture { /* no-op mutation */ }.get(2, TimeUnit.SECONDS)
+    assertNull(result) // Void's only inhabitant
+  }
+
+  @Test
+  fun asyncFuturePropagatesException() {
+    val boom = boom()
+    val ex = assertFailsWith<ExecutionException> {
+      newObject().asyncFuture<Unit> { throw boom }.get(2, TimeUnit.SECONDS)
+    }
+    assertSame(boom, ex.cause) // original exception surfaced to the caller
+  }
+
+  @Test
+  fun asyncVoidFuturePropagatesException() {
+    val boom = boom()
+    val ex = assertFailsWith<ExecutionException> {
+      newObject().asyncVoidFuture { throw boom }.get(2, TimeUnit.SECONDS)
+    }
+    // the thenApply { null } bridge must forward the failure, not swallow it
+    assertSame(boom, ex.cause)
+  }
+
+  @Test
+  fun failedOperationDoesNotCancelScopeForSiblings() {
+    val obj = newObject()
+    val boom = boom()
+    // a mutating op fails on the sequential scope...
+    assertFailsWith<ExecutionException> {
+      obj.asyncVoidFuture { throw boom }.get(2, TimeUnit.SECONDS)
+    }
+    // ...and a subsequent op on the same scope still succeeds - proves SupervisorJob isolation
+    // (a plain Job would have cancelled the scope, failing this call).
+    assertEquals(7, obj.asyncFuture { 7 }.get(2, TimeUnit.SECONDS))
+  }
+}