objects/uts: observe transient connection states via record-and-verify; pin fresh-channel pipeline readiness

uts/docs/writing-derived-tests.md

@@ -38,8 +38,9 @@ UTS specs use generic pseudocode. You need to map this onto the SDK's actual API
 | `install_mock(mock_http)` | How mocks are injected (DI, platform patching, etc.) |
 | `enable_fake_timers()` | Timer control mechanism |
 | `ADVANCE_TIME(ms)` | Fake timer tick method |
-| `AWAIT_STATE(connection, "connected")` | State waiting helper |
+| `AWAIT_STATE(connection, "connected")` | State waiting helper. Transient states (DISCONNECTED/SUSPENDED after a drop): never await post-stimulus — use the record-and-verify pattern (writing-test-specs.md, "Verifying Transient States"). |
 | `poll_until(condition, ...)` | Shared polling helper (wall-clock deadline — see below) |
+| `state_changes = []` / `events = []` (recording lists) | Event-recording collection for the record-and-verify pattern. Appended from SDK callback threads while `poll_until`/final asserts read it from the test thread — multithreaded SDKs must render it as a thread-safe list. |
 | `poll_until_success(condition)` | Error-tolerant polling helper (see the pseudocode conventions in `uts/README.md`) |
 
 Check the SDK's existing test infrastructure and conventions before writing anything. Reuse existing helpers, mock classes, and patterns.

uts/objects/integration/proxy/objects_faults.md

@@ -112,18 +112,27 @@ channel = client.channels.get(channel_name, { modes: ["OBJECT_SUBSCRIBE", "OBJEC
 ### Test Steps
 
 ```pseudo
+// Record connection states BEFORE the disconnect stimulus (channel.attach(), below).
+// DISCONNECTED is transient (RTN15a reconnects immediately); a post-stimulus AWAIT_STATE
+// can miss it, so this listener MUST precede the stimulus — reordering breaks the test.
+// (see docs/writing-test-specs.md, "Verifying Transient States")
+state_changes = []
+client.connection.on((change) => { state_changes.append(change.current) })
+
 client.connect()
 AWAIT_STATE client.connection.state == CONNECTED
   WITH timeout: 15 seconds
 
-// First attach triggers sync; proxy disconnects mid-sync
+// First attach triggers sync; proxy disconnects mid-sync, then the client auto-reconnects.
 channel.attach()
-AWAIT_STATE client.connection.state == DISCONNECTED
-  WITH timeout: 15 seconds
-
-// Client auto-reconnects; re-attach triggers fresh sync
+// Gate on the RECORDED list before the final wait: the drop lands only after the sync frame
+// round-trips, so an immediate AWAIT_STATE CONNECTED would no-op (still connected) and race it.
+poll_until(state_changes CONTAINS DISCONNECTED, timeout: 30s)
+// Final wait targets CONNECTED, a sticky state — safe for AWAIT_STATE.
 AWAIT_STATE client.connection.state == CONNECTED
   WITH timeout: 30 seconds
+// CONTAINS_IN_ORDER is a subsequence match, so the leading initial-connect states are fine.
+ASSERT state_changes CONTAINS_IN_ORDER [DISCONNECTED, CONNECTING, CONNECTED]
 
 // get() waits for SYNCED — will only resolve if re-sync completes
 root = AWAIT channel.object.get()
@@ -200,12 +209,21 @@ root_b = AWAIT channel_b.object.get()
   WITH timeout: 15 seconds
 poll_until_success(root_b.get("key1").value() == "initial")
 
+// Record B's connection states BEFORE the disconnect stimulus (trigger_action, below).
+// DISCONNECTED is transient (RTN15a reconnects immediately); a post-stimulus AWAIT_STATE
+// can miss it, so this listener MUST precede the stimulus — reordering breaks the test.
+// (see docs/writing-test-specs.md, "Verifying Transient States")
+state_changes = []
+client_b.connection.on((change) => { state_changes.append(change.current) })
+
 // Disconnect client B
 session.trigger_action({ type: "disconnect" })
-AWAIT_STATE client_b.connection.state == DISCONNECTED
-  WITH timeout: 15 seconds
+// Mid-test gate: poll the RECORDED list (not AWAIT_STATE on live state, which could miss
+// the transient DISCONNECTED). This blocks A's publish until B has observed the drop.
+poll_until(state_changes CONTAINS DISCONNECTED, timeout: 15s)
 
-// While B is disconnected, A publishes a mutation
+// A publishes while B is down. Best-effort: RTN15a may reconnect/re-sync B before this
+// round-trips (then it tests plain delivery, not RTO7/RTO8); the final poll tolerates both.
 AWAIT root_a.set("key1", "updated_during_disconnect")
 
 // Client B reconnects and re-syncs; the mutation should be visible

uts/objects/unit/realtime_object.md

@@ -1799,6 +1799,9 @@ scenarios = [
       install_mock(mock_ws)
       client = Realtime(options: { key: "fake:key", autoConnect: true })
       channel = client.channels.get("test", { modes: ["OBJECT_SUBSCRIBE", "OBJECT_PUBLISH"] })
+      // Let the fresh channel's objects message pipeline finish subscribing before attach()
+      // (see process_pending_events in uts/README.md)
+      process_pending_events()
       // NOTE: channel is NOT yet attached/synced here — listeners must be wired
       // by the loop before scenario.trigger() calls attach().
       RETURN { client, channel, mock_ws }