fix(uts): deterministic smoke test, contract-correct mocks, run-to-quiescence FakeClock

Fixes the CI-red UnitInfraSmokeTest race and lands the review/spec-alignment
round on the shared UTS infra:

- Root cause of the CI flake: FakeClock.waitOn performs a real timed wait, so
  the disconnected-retry fires on wall-clock regardless of advance() — the
  "no attempt before advance" assertion was unassertable. The smoke test now
  owns attempt #2 via the buffered awaitConnectionAttempt() (32/32 green incl.
  CPU-saturation runs) and README §6.4/§9 teach the true semantics.
- FakeClock: advance() now runs due work to quiescence (cascades and timers
  created mid-advance fire within the same advance — the spec's Fake-time
  semantics Guarantee); timers/pending hardened against SDK-thread races.
  The waitOn advisory seam is unchanged. New cascade smoke test covers it.
- Mock contract fixes from review triage (verified against the UTS docs):
  transport cancel() now delivers listener.onClose; respondWith honors the
  headers param and JSON-serializes non-String bodies; SandboxApp checks HTTP
  status before parsing; delivery executor shutdown; @volatile channel fields;
  await helpers unregister listeners on success; AtomicReference for the
  cross-thread query-params capture.
- Docs: uts/README rewritten claims verified against sources; stale
  "reflection" wording fixed in the skill's objects-mapping notes.

Author: sacOO7

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

@@ -401,8 +401,9 @@ delivered to subscription listeners) map to ably-java interfaces with getters (p
 > obtain an `ObjectMessage` from a subscription event (`event.getMessage()`, §8); there is no public
 > factory. The spec's explicit construction-from-wire (`PublicObjectMessage.fromObjectMessage(source,
 > channel)` / `PublicObjectOperation.fromObjectOperation(op)`, `PAOM3`/`PAOOP3`, in
-> `public_object_message.md`) is `internal` to `:liveobjects` — but the unit helpers expose it **by
-> reflection** as `buildPublicObjectMessage(wireJson, channelName)` (§13). So `public_object_message.md` is
+> `public_object_message.md`) is `internal` to `:liveobjects` — but the unit helpers expose it as
+> `buildPublicObjectMessage(wireMessage, channelName)` (§13), a direct `toPublicMessage` call (no reflection,
+> since `:liveobjects`'s internals are visible to its own tests). So `public_object_message.md` is
 > translatable: build the source with the op builders (`buildMapSet(...)`, `buildCounterInc(...)`, …) and
 > assert the public getters on the result.
 
@@ -515,7 +516,7 @@ Several **unit** specs assert on the **internal CRDT graph**, not the public API
   is public and maps via §2/§12, but `publish` / `publishAndApply` (`RTO15`/`RTO20`, marked `internal` in the
   IDL) and the OBJECT/ACK wire assertions are internal.
 - `public_object_message.md` — **translatable** via the `buildPublicObjectMessage` helper (below), which
-  reflectively performs the `PAOM3`/`PAOOP3` construction (`WireObjectMessage` → `DefaultObjectMessage`)
+  performs the `PAOM3`/`PAOOP3` construction (`WireObjectMessage` → `DefaultObjectMessage`)
   that is otherwise `internal`. Build the source with the op builders and assert the public getters (§11).
 
 In ably-java these are **not public**. They live in the `:liveobjects` module as `Internal*` / `Default*` / `Wire*` /

java/build.gradle.kts

@@ -40,7 +40,9 @@ dependencies {
 }
 
 // kotlin-stdlib guardrail (invariant I5): the Kotlin plugin auto-adds kotlin-stdlib to the module's
-// main dependency scope, which would leak into :java's published POM/runtime. :java is Kotlin-free at
+// main dependency scope, which would leak into :java's published POM/runtime. The leak comes from the
+// PLUGIN, NOT from the testImplementation(project(":uts")) dependency — test scopes never enter the
+// POM (verified: removing the :uts dep leaves the leak identical). :java is Kotlin-free at
 // runtime, so strip it from the main artifact scopes. kotlin-stdlib still reaches the TEST classpath
 // transitively (via :uts's kotlin-test-junit5), so the UTS Kotlin suites compile and run.
 // Verified empirically on Kotlin 2.1.10: the plugin adds stdlib lazily (it does not appear in any

lib/src/test/kotlin/io/ably/lib/uts/unit/realtime/ConnectionRecoveryTest.kt

@@ -12,6 +12,7 @@ import io.ably.lib.uts.infra.pollUntil
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.atomic.AtomicReference
 import kotlin.test.*
 import kotlin.time.Duration.Companion.seconds
 
@@ -322,10 +323,12 @@ class ConnectionRecoveryTest {
    */
   @Test
   fun `RTN16f1 - Malformed recoveryKey logs error and connects normally`() = runTest {
-    var capturedQueryParams: Map<String, String>? = null
+    // Written on the SDK transport thread and read from the coroutine dispatcher, so use an
+    // AtomicReference to publish it safely across threads (avoids a visibility race).
+    val capturedQueryParams = AtomicReference<Map<String, String>?>(null)
     val mock = MockWebSocket {
       onConnectionAttempt = { conn ->
-        capturedQueryParams = conn.queryParams
+        capturedQueryParams.set(conn.queryParams)
         conn.respondWithSuccess(ProtocolMessage().apply {
           action = ProtocolMessage.Action.connected
           connectionId = "fresh-conn"
@@ -349,8 +352,8 @@ class ConnectionRecoveryTest {
     assertEquals(ConnectionState.connected, client.connection.state)
     assertEquals("fresh-conn", client.connection.id)
     assertEquals("fresh-key", client.connection.key)
-    assertNull(capturedQueryParams!!["recover"])
-    assertNull(capturedQueryParams!!["resume"])
+    assertNull(capturedQueryParams.get()!!["recover"])
+    assertNull(capturedQueryParams.get()!!["resume"])
     assertEquals(1, mock.events.filterIsInstance<MockEvent.ConnectionAttempt>().size)
 
     client.close()

liveobjects/src/test/kotlin/io/ably/lib/liveobjects/uts/unit/Helpers.kt

@@ -252,7 +252,8 @@ internal fun buildPublicObjectMessage(objectMessage: WireObjectMessage, channelN
     objectMessage.toPublicMessage(channelName)
 
 // `provision_objects_via_rest(...)` is intentionally not here — it's REST fixture provisioning for
-// *integration* tests and belongs with the :uts integration tier.
+// *integration* tests and lives with this module's integration tier
+// (io.ably.lib.liveobjects.uts.integration.Helpers).
 
 // ---------------------------------------------------------------------------
 // STANDARD_POOL_OBJECTS — the fixed tree shared by all objects unit specs

uts/README.md

@@ -214,18 +214,18 @@ dependencies {
     // :liveobjects.
     api(project(":java"))                    // the SDK + its types (DebugOptions, ProtocolMessage, …)
     api(project(":network-client-core"))     // HttpEngine / WebSocketEngine SPIs the mocks implement
-    implementation(libs.coroutine.core)
-    implementation(libs.ktor.client.core)    // sandbox/proxy control HTTP
+    implementation(libs.ktor.client.core)    // proxy infra uses ktor internally — must NOT leak to consumers
     implementation(libs.ktor.client.cio)
 
-    // :uts's own tests — the tier smoke tests. No :liveobjects edge (I1).
-    testImplementation(kotlin("test"))
-    testImplementation(platform(libs.junit.bom))
-    testImplementation(libs.junit.jupiter.params)   // @ParameterizedTest / @ValueSource
-    testImplementation(libs.coroutine.core)
-    testImplementation(libs.coroutine.test)         // runTest, virtual time
-    testImplementation(libs.ktor.client.core)
-    testImplementation(libs.ktor.client.cio)
+    // The UTS test toolkit — exported (api) so any module consuming the infra via
+    // testImplementation(project(":uts")) transitively gets JUnit 5, the kotlin.test Jupiter binding,
+    // and coroutines (runTest etc.). :uts's own smoke tests inherit it from main's api — nothing to declare.
+    api(platform(libs.junit.bom))
+    api(libs.junit.jupiter)
+    api(libs.junit.jupiter.params)                  // @ParameterizedTest / @ValueSource
+    api(kotlin("test-junit5"))
+    api(libs.coroutine.core)
+    api(libs.coroutine.test)                        // runTest, virtual time
 }
 
 tasks.withType<Test>().configureEach {
@@ -462,9 +462,16 @@ responses back — all without a socket.
 
 ### 6.4 `FakeClock` — deterministic time
 `FakeClock` implements the SDK's `Clock`. Time is frozen until you call `advance(ms)`; on each
-advance it fires any due virtual timers **synchronously**, and wakes any `waitOn` sleepers. This is
+advance it fires any due virtual timers **synchronously**, and wakes any `waitOn` sleepers. `advance`
+runs due work **to quiescence** — it re-scans until a full pass fires nothing, so cascades due within
+the advanced interval (a zero-delay reschedule, or a timer created by fired work) also run in that same
+`advance` (the spec run-to-quiescence Guarantee; see §9.3). This is
 how the unit test drives reconnection backoff and `connectionStateTtl` expiry **without real
-sleeping**:
+sleeping**. Caveat: `waitOn(target, timeout)` still performs a real `target.wait(timeout)`, so a
+sleeper also wakes once `timeout` ms of wall-clock elapse — `advance()` makes it wake *sooner*, but is
+not a hard gate (the **advisory** model of the spec's `mock_websocket.md` §Fake-time semantics).
+Drive transitions by owning the resulting attempt (`awaitConnectionAttempt()`), never
+by asserting a wait has *not* yet returned.
 ```kotlin
 val fakeClock = FakeClock()
 val client = TestRealtimeClient { enableFakeTimers(fakeClock); … }
@@ -595,7 +602,8 @@ future unit-tier UTS test should take.
 > (`lib/src/test/kotlin/io/ably/lib/uts/unit/realtime/`, e.g. `ConnectionRecoveryTest`) and
 > `:liveobjects` — see §13.
 
-It has **two** `@Test` methods; between them they exercise every teaching point of §5–§8.
+It has **three** `@Test` methods: two end-to-end transport tests (§9.1, §9.2) that between them exercise
+every teaching point of §5–§8, plus a focused `FakeClock` run-to-quiescence acceptance test (§9.3).
 
 ### 9.1 `unit infra drives the full mock-WebSocket connection lifecycle` — await style throughout
 One long **await-style** test that walks the SDK through the whole transport lifecycle:
@@ -632,12 +640,15 @@ One long **await-style** test that walks the SDK through the whole transport lif
    ```
 3. **Publish**, asserting the full MESSAGE frame (`action`, `channel`, `messages[0].name`/`data`) again
    via `awaitNextMessageFromClient()`.
-4. **Disconnect + negative check.** `simulateDisconnect()`, await DISCONNECTED, then assert **no**
-   reconnect has happened yet — exactly one `ConnectionAttempt` is recorded — because the retry is
-   blocked in `FakeClock.waitOn` until the clock advances.
+4. **Disconnect.** `simulateDisconnect()`, await DISCONNECTED, and assert the drop was recorded. Note
+   we do **not** snapshot the `ConnectionAttempt` count here: `FakeClock.waitOn(target, timeout)` does a
+   real `target.wait(timeout)`, so the disconnected-retry fires on its own after ~`disconnectedRetryTimeout`
+   ms of wall-clock even without an `advance()`. `advance()` only wins that race sooner — it is not a
+   hard gate — so a "still exactly one attempt" assertion would be racy on a loaded runner. Ownership
+   of attempt #2 belongs to the next step, which gates on it deterministically.
 5. **FakeClock-driven reconnect.** A coroutine loops `fakeClock.advance(2.seconds)` then answers the
-   next attempt with a short-TTL CONNECTED; the test awaits CONNECTED again and asserts a second
-   `ConnectionAttempt`.
+   next attempt (received via the buffered `awaitConnectionAttempt()`, so it cannot be missed) with a
+   short-TTL CONNECTED; the test awaits CONNECTED again and asserts a second `ConnectionAttempt`.
 6. **Refuse → SUSPENDED (the centrepiece).** After another `simulateDisconnect()`, a `refuseJob`
    coroutine advances the clock and `respondWithRefused()`s every reconnection attempt until the short
    `connectionStateTtl` (800 ms, from the short-lived CONNECTED) expires and the client gives up to
@@ -688,6 +699,14 @@ frames, `events` / `awaitNextMessageFromClient` for inspecting client output, an
 connect→request two-phase flow. The `RUN_DEVIATIONS` env-gated deviation pattern is **not** here (the
 smoke tests carry no deviations) — that teaching lives in §12.
 
+### 9.3 `FakeClock advance runs cascaded work to quiescence in one call` — the run-to-quiescence Guarantee
+A focused, SDK-free test that pins the `FakeClock` contract §6.4 depends on: a single `advance(ms)` runs
+**all** work due within the advanced interval, including cascades. It schedules a task that reschedules
+itself at zero delay and a task that creates a brand-new timer mid-advance, then asserts one `advance`
+fires the whole cascade (not just the first pass). This is the infra-level guarantee the reconnect/backoff
+walkthroughs in §9.1 rely on; it exercises `FakeClock` directly because the Guarantee is about `advance`
+alone reaching quiescence.
+
 ---
 
 ## 10. Walkthrough: the Direct-Sandbox Smoke Test (`IntegrationInfraSmokeTest`)
@@ -1110,7 +1129,7 @@ implicit.
 | File | Key public surface | Role |
 |------|--------------------|------|
 | `infra/Utils.kt` | `awaitState(client,target,timeout=5s)`, `awaitChannelState(channel,target,timeout=5s)`, `pollUntil(timeout=15s,interval=100ms){ }` | Shared wall-clock coroutine waits (package `io.ably.lib.uts.infra`); listener registered before state check. |
-| `unit/UnitInfraSmokeTest.kt` | 2 `@Test`s: full mock-WS lifecycle (await style), token-auth via mock HTTP (callback WS) | Unit-tier infra acceptance (`io.ably.lib.uts.unit`) — MockWebSocket/MockHttpClient/FakeClock end-to-end. **No** `@UTS`. |
+| `unit/UnitInfraSmokeTest.kt` | 3 `@Test`s: full mock-WS lifecycle (await style), token-auth via mock HTTP (callback WS), FakeClock run-to-quiescence | Unit-tier infra acceptance (`io.ably.lib.uts.unit`) — MockWebSocket/MockHttpClient/FakeClock end-to-end. **No** `@UTS`. |
 | `integration/standard/IntegrationInfraSmokeTest.kt` | 1 `@ParameterizedTest` × {JSON, msgpack} | Direct-sandbox infra acceptance (`io.ably.lib.uts.integration.standard`) — SandboxApp + realtime/REST round-trip, awaited publish + `pollUntil` on `history()`. **No** `@UTS`. |
 | `integration/proxy/ProxyInfraSmokeTest.kt` | 2 `@Test`s: late imperative disconnect, declarative ws-frame rule | Proxy infra acceptance (`io.ably.lib.uts.integration.proxy`) — ProxyManager + ProxySession, both fault-injection styles, proxy-log asserts. **No** `@UTS`. |
 

uts/src/main/kotlin/io/ably/lib/uts/infra/Utils.kt

@@ -28,11 +28,18 @@ suspend fun awaitState(
   withContext(Dispatchers.Default.limitedParallelism(1)) {
     withTimeout(timeout) {
       suspendCancellableCoroutine { cont ->
-        val listener = ConnectionStateListener { change ->
-          if (change.current == target && cont.isActive) cont.resume(Unit)
+        lateinit var listener: ConnectionStateListener
+        listener = ConnectionStateListener { change ->
+          if (change.current == target && cont.isActive) {
+            client.connection.off(listener)
+            cont.resume(Unit)
+          }
         }
         client.connection.on(listener)
-        if (client.connection.state == target && cont.isActive) cont.resume(Unit)
+        if (client.connection.state == target && cont.isActive) {
+          client.connection.off(listener)
+          cont.resume(Unit)
+        }
         cont.invokeOnCancellation { client.connection.off(listener) }
       }
     }
@@ -80,11 +87,18 @@ suspend fun awaitChannelState(
   withContext(Dispatchers.Default.limitedParallelism(1)) {
     withTimeout(timeout) {
       suspendCancellableCoroutine { cont ->
-        val listener = ChannelStateListener { change ->
-          if (change.current == target && cont.isActive) cont.resume(Unit)
+        lateinit var listener: ChannelStateListener
+        listener = ChannelStateListener { change ->
+          if (change.current == target && cont.isActive) {
+            channel.off(listener)
+            cont.resume(Unit)
+          }
         }
         channel.on(listener)
-        if (channel.state == target && cont.isActive) cont.resume(Unit)
+        if (channel.state == target && cont.isActive) {
+          channel.off(listener)
+          cont.resume(Unit)
+        }
         cont.invokeOnCancellation { channel.off(listener) }
       }
     }

uts/src/main/kotlin/io/ably/lib/uts/infra/integration/SandboxApp.kt

@@ -93,7 +93,11 @@ class SandboxApp private constructor(
         contentType(ContentType.Application.Json)
         setBody(loadAppCreationJson().toString())
       }
-      val body = JsonParser.parseString(response.bodyAsText()).asJsonObject
+      val responseBody = response.bodyAsText()
+      check(response.status.isSuccess()) {
+        "sandbox app provisioning failed: HTTP ${response.status} — $responseBody"
+      }
+      val body = JsonParser.parseString(responseBody).asJsonObject
       val keys = body["keys"].asJsonArray.map { it.asJsonObject["keyStr"].asString }
       return SandboxApp(
         appId = body["appId"].asString,

uts/src/main/kotlin/io/ably/lib/uts/infra/unit/DefaultPendingConnection.kt

@@ -32,6 +32,8 @@ internal class DefaultPendingConnection(
         // Async delivery per spec: the library must store the WS reference before processing CONNECTED.
         val encoded = Serialisation.gson.toJson(message)
         deliveryExecutor.submit { listener.onMessage(encoded) }
+        // One-shot delivery: release the daemon thread once the single message is queued.
+        deliveryExecutor.shutdown()
     }
 
     override fun respondWithRefused() = listener.onError(IOException("Connection refused to $host:$port"))

uts/src/main/kotlin/io/ably/lib/uts/infra/unit/DefaultPendingRequest.kt

@@ -4,6 +4,7 @@ import io.ably.lib.network.FailedConnectionException
 import io.ably.lib.network.HttpBody
 import io.ably.lib.network.HttpRequest
 import io.ably.lib.network.HttpResponse
+import io.ably.lib.util.Serialisation
 import kotlin.time.Duration
 import kotlinx.coroutines.CompletableDeferred
 import java.net.SocketTimeoutException
@@ -20,14 +21,15 @@ internal class DefaultPendingRequest(
     override fun respondWith(status: Int, body: Any, headers: Map<String, String>) {
         val bytes = when (body) {
             is ByteArray -> body
-            else -> body.toString().toByteArray(Charsets.UTF_8)
+            is String -> body.toByteArray(Charsets.UTF_8)
+            else -> Serialisation.gson.toJson(body).toByteArray(Charsets.UTF_8)
         }
         deferred.complete(
             HttpResponse.builder()
                 .code(status)
                 .message("")
                 .body(HttpBody("application/json", bytes))
-                .headers(emptyMap())
+                .headers(headers.mapValues { listOf(it.value) })
                 .build()
         )
     }