test(android): consolidate Wear bridge fixtures (#118559)

apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt

@@ -54,11 +54,8 @@ class WearProxyBridgeTest {
   @Test
   fun validRequestRegistersPeerAndReturnsCorrelatedResponse() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        backgroundScope.recordingBridge(
           handleRequest = { _, request ->
             WearMessage.Response(
               requestId = request.requestId,
@@ -74,7 +71,7 @@ class WearProxyBridgeTest {
       assertEquals(1, sent.size)
       assertEquals("watch-1", sent.single().nodeId)
       assertEquals(WearProtocol.RESPONSE_PATH, sent.single().path)
-      val response = (WearProtocolCodec.decode(sent.single().data) as WearDecodeResult.Success).message as WearMessage.Response
+      val response = sent.single().response()
       assertEquals("req-1", response.requestId)
       assertTrue(!response.eventStreamId.isNullOrBlank())
       assertEquals(0L, response.eventSequence)
@@ -83,13 +80,7 @@ class WearProxyBridgeTest {
   @Test
   fun malformedMessageDoesNotRegisterPeer() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
-          handleRequest = { _, _ -> error("must not run") },
-        )
+      val (bridge, sent) = backgroundScope.recordingBridge(handleRequest = { _, _ -> error("must not run") })
 
       bridge.handleMessage("watch-1", "not-json".encodeToByteArray())
 
@@ -103,8 +94,7 @@ class WearProxyBridgeTest {
       var cancelFirstResponse = true
       val sent = mutableListOf<SentWearMessage>()
       val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
+        backgroundScope.testBridge(
           sender =
             WearMessageSender { nodeId, path, data ->
               if (cancelFirstResponse) {
@@ -113,7 +103,6 @@ class WearProxyBridgeTest {
               }
               sent += SentWearMessage(nodeId, path, data)
             },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
       assertTrue(bridge.handleMessage("watch-1", WearProtocolCodec.encode(request("req-1"))))
@@ -130,8 +119,7 @@ class WearProxyBridgeTest {
       var cancelFirstEvent = true
       val sent = mutableListOf<SentWearMessage>()
       val bridge =
-        WearProxyBridge(
-          scope = actorScope,
+        actorScope.testBridge(
           sender =
             WearMessageSender { nodeId, path, data ->
               if (path == WearProtocol.EVENT_PATH && cancelFirstEvent) {
@@ -142,7 +130,6 @@ class WearProxyBridgeTest {
             },
           peerResolver = WearPeerResolver { setOf("watch-1") },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
       bridge.publishConnection(connected = true, status = "Connected")
@@ -156,11 +143,8 @@ class WearProxyBridgeTest {
   fun discoveryTaskCancellationDoesNotTerminateEventActor() =
     withActorScope { actorScope ->
       var cancelFirstDiscovery = true
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = actorScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        actorScope.recordingBridge(
           peerResolver =
             WearPeerResolver {
               if (cancelFirstDiscovery) {
@@ -170,7 +154,6 @@ class WearProxyBridgeTest {
               setOf("watch-1")
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
       bridge.publishConnection(connected = true, status = "first")
@@ -183,20 +166,16 @@ class WearProxyBridgeTest {
   @Test
   fun resyncInvalidatesTheWatchSnapshot() =
     withActorScope { actorScope ->
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = actorScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        actorScope.recordingBridge(
           peerResolver = WearPeerResolver { setOf("watch-1") },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
       bridge.publishResync()
       bridge.awaitIdleForTests()
 
-      val event = (WearProtocolCodec.decode(sent.single().data) as WearDecodeResult.Success).message as WearMessage.Event
+      val event = sent.single().event()
       assertEquals(WearProtocol.EVENT_PATH, sent.single().path)
       assertEquals(WearEventType.Resync, event.event)
       assertEquals(1L, event.sequence)
@@ -206,13 +185,10 @@ class WearProxyBridgeTest {
   @Test
   fun historyResponseWatermarkExcludesEventsQueuedBehindIt() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
       val requestStarted = CompletableDeferred<Unit>()
       val finishRequest = CompletableDeferred<Unit>()
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        backgroundScope.recordingBridge(
           handleRequest = { _, request ->
             requestStarted.complete(Unit)
             finishRequest.await()
@@ -223,13 +199,7 @@ class WearProxyBridgeTest {
       val requestJob = async { bridge.handleMessage("watch-1", WearProtocolCodec.encode(request("req-1"))) }
       runCurrent()
       requestStarted.await()
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "delta")
-          put("deltaText", "new")
-        },
-      )
+      bridge.publishChat(chatPayload(state = "delta", text = "new"))
       runCurrent()
       assertTrue(sent.isEmpty())
 
@@ -238,8 +208,8 @@ class WearProxyBridgeTest {
       runCurrent()
 
       assertEquals(listOf(WearProtocol.RESPONSE_PATH, WearProtocol.EVENT_PATH), sent.map { it.path })
-      val response = (WearProtocolCodec.decode(sent[0].data) as WearDecodeResult.Success).message as WearMessage.Response
-      val event = (WearProtocolCodec.decode(sent[1].data) as WearDecodeResult.Success).message as WearMessage.Event
+      val response = sent[0].response()
+      val event = sent[1].event()
       assertEquals(response.eventStreamId, event.streamId)
       assertEquals(0L, response.eventSequence)
       assertEquals(1L, event.sequence)
@@ -248,46 +218,9 @@ class WearProxyBridgeTest {
   @Test
   fun chatStreamProjectionCarriesCanonicalTextAndFallbackCompleteness() {
     val projector = WearChatStreamProjector()
-    val first =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("runId", "run-1")
-            put("state", "delta")
-            put("deltaText", "Hel")
-            put(
-              "message",
-              buildJsonObject {
-                put("role", "assistant")
-                put("content", "Hel")
-              },
-            )
-          },
-        ),
-      )
-    val continued =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("runId", "run-1")
-            put("state", "delta")
-            put("deltaText", "lo")
-          },
-        ),
-      )
-    val unknownPrefix =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("runId", "run-2")
-            put("state", "delta")
-            put("deltaText", "tail")
-          },
-        ),
-      )
+    val first = projectStreamEvent(projector, state = "delta", runId = "run-1", text = "Hel", message = "Hel")
+    val continued = projectStreamEvent(projector, state = "delta", runId = "run-1", text = "lo")
+    val unknownPrefix = projectStreamEvent(projector, state = "delta", runId = "run-2", text = "tail")
 
     assertEquals("Hel", first.getValue("streamText").jsonPrimitive.content)
     assertEquals("true", first.getValue("streamTextComplete").jsonPrimitive.content)
@@ -298,34 +231,22 @@ class WearProxyBridgeTest {
   }
 
   @Test
-  fun foreignFinalPreservesTheActiveWatchStream() {
-    assertForeignTerminalPreservesActiveStream("final")
-  }
+  fun foreignFinalPreservesTheActiveWatchStream() = assertForeignTerminalPreservesStream("final", runId = "active-run")
 
   @Test
-  fun foreignAbortPreservesTheActiveWatchStream() {
-    assertForeignTerminalPreservesActiveStream("aborted")
-  }
+  fun foreignAbortPreservesTheActiveWatchStream() = assertForeignTerminalPreservesStream("aborted", runId = "active-run")
 
   @Test
-  fun foreignErrorPreservesTheActiveWatchStream() {
-    assertForeignTerminalPreservesActiveStream("error")
-  }
+  fun foreignErrorPreservesTheActiveWatchStream() = assertForeignTerminalPreservesStream("error", runId = "active-run")
 
   @Test
-  fun foreignFinalPreservesAnAnonymousWatchStream() {
-    assertForeignTerminalPreservesAnonymousStream("final")
-  }
+  fun foreignFinalPreservesAnAnonymousWatchStream() = assertForeignTerminalPreservesStream("final")
 
   @Test
-  fun foreignAbortPreservesAnAnonymousWatchStream() {
-    assertForeignTerminalPreservesAnonymousStream("aborted")
-  }
+  fun foreignAbortPreservesAnAnonymousWatchStream() = assertForeignTerminalPreservesStream("aborted")
 
   @Test
-  fun foreignErrorPreservesAnAnonymousWatchStream() {
-    assertForeignTerminalPreservesAnonymousStream("error")
-  }
+  fun foreignErrorPreservesAnAnonymousWatchStream() = assertForeignTerminalPreservesStream("error")
 
   @Test
   fun identifiedTerminalClearsItsOwnStreamWithoutErasingAnotherRun() {
@@ -359,57 +280,11 @@ class WearProxyBridgeTest {
   @Test
   fun runIdLessDeltasUseSessionSnapshotAndKeepExactAppendSemantics() {
     val projector = WearChatStreamProjector()
-
-    val first =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("state", "delta")
-            put("deltaText", "a")
-          },
-        ),
-      )
-    val second =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("state", "delta")
-            put("deltaText", "a")
-          },
-        ),
-      )
-    val identified =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("runId", "run-now-known")
-            put("state", "delta")
-            put("deltaText", "a")
-          },
-        ),
-      )
-    checkNotNull(
-      projector.project(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("runId", "run-now-known")
-          put("state", "final")
-        },
-      ),
-    )
-    val afterFinal =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("state", "delta")
-            put("deltaText", "a")
-          },
-        ),
-      )
+    val first = projectStreamEvent(projector, state = "delta", text = "a")
+    val second = projectStreamEvent(projector, state = "delta", text = "a")
+    val identified = projectStreamEvent(projector, state = "delta", runId = "run-now-known", text = "a")
+    projectStreamEvent(projector, state = "final", runId = "run-now-known")
+    val afterFinal = projectStreamEvent(projector, state = "delta", text = "a")
 
     assertEquals("a", first.getValue("streamText").jsonPrimitive.content)
     assertEquals("aa", second.getValue("streamText").jsonPrimitive.content)
@@ -420,21 +295,9 @@ class WearProxyBridgeTest {
   @Test
   fun runIdLessDeltaContinuesTheActiveIdentifiedStream() {
     val projector = WearChatStreamProjector()
-
-    fun delta(
-      text: String,
-      runId: String? = null,
-    ): JsonObject =
-      buildJsonObject {
-        put("sessionKey", "main")
-        runId?.let { put("runId", it) }
-        put("state", "delta")
-        put("deltaText", text)
-      }
-
-    checkNotNull(projector.project(delta("H", runId = "run-1")))
-    val anonymous = checkNotNull(projector.project(delta("i")))
-    val identified = checkNotNull(projector.project(delta("!", runId = "run-1")))
+    projectStreamEvent(projector, state = "delta", runId = "run-1", text = "H")
+    val anonymous = projectStreamEvent(projector, state = "delta", text = "i")
+    val identified = projectStreamEvent(projector, state = "delta", runId = "run-1", text = "!")
 
     assertEquals("Hi", anonymous.getValue("streamText").jsonPrimitive.content)
     assertEquals("Hi!", identified.getValue("streamText").jsonPrimitive.content)
@@ -443,27 +306,10 @@ class WearProxyBridgeTest {
   @Test
   fun connectionResetClearsRunIdLessStreamState() {
     val projector = WearChatStreamProjector()
-    checkNotNull(
-      projector.project(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "delta")
-          put("deltaText", "stale")
-        },
-      ),
-    )
+    projectStreamEvent(projector, state = "delta", text = "stale")
 
     projector.reset()
-    val next =
-      checkNotNull(
-        projector.project(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("state", "delta")
-            put("deltaText", "fresh")
-          },
-        ),
-      )
+    val next = projectStreamEvent(projector, state = "delta", text = "fresh")
 
     assertEquals("fresh", next.getValue("streamText").jsonPrimitive.content)
   }
@@ -471,22 +317,13 @@ class WearProxyBridgeTest {
   @Test
   fun coldBridgeDiscoversReachableWatchBeforeBackgroundEvent() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        backgroundScope.recordingBridge(
           peerResolver = WearPeerResolver { setOf("watch-1") },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "final")
-        },
-      )
+      bridge.publishChat(chatPayload(state = "final"))
       runCurrent()
 
       assertEquals("watch-1", sent.single().nodeId)
@@ -498,8 +335,7 @@ class WearProxyBridgeTest {
     runTest {
       var resolutions = 0
       val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
+        backgroundScope.testBridge(
           sender = WearMessageSender { _, _, _ -> error("must not send") },
           peerResolver =
             WearPeerResolver {
@@ -507,17 +343,10 @@ class WearProxyBridgeTest {
               emptySet()
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
       repeat(20) { index ->
-        bridge.publishChat(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("state", "delta")
-            put("deltaText", "$index")
-          },
-        )
+        bridge.publishChat(chatPayload(state = "delta", text = "$index"))
       }
       runCurrent()
 
@@ -527,68 +356,44 @@ class WearProxyBridgeTest {
   @Test
   fun terminalEventBypassesCachedEmptyPeerDiscovery() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
       var resolutions = 0
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        backgroundScope.recordingBridge(
           peerResolver =
             WearPeerResolver {
               resolutions += 1
               if (resolutions == 1) emptySet() else setOf("watch-1")
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
 
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "delta")
-          put("deltaText", "working")
-        },
-      )
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "final")
-        },
-      )
+      bridge.publishChat(chatPayload(state = "delta", text = "working"))
+      bridge.publishChat(chatPayload(state = "final"))
       runCurrent()
 
       assertEquals(2, resolutions)
       assertEquals("watch-1", sent.single().nodeId)
-      val event = (WearProtocolCodec.decode(sent.single().data) as WearDecodeResult.Success).message as WearMessage.Event
+      val event = sent.single().event()
       assertEquals(2L, event.sequence)
     }
 
   @Test
   fun terminalEventDiscoversAnotherWatchWhileRememberedPeerIsHealthy() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
       var resolutions = 0
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        backgroundScope.recordingBridge(
           peerResolver =
             WearPeerResolver {
               resolutions += 1
               setOf("watch-1", "watch-2")
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
       bridge.handleMessage("watch-1", WearProtocolCodec.encode(request("req-1")))
       sent.clear()
 
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "final")
-        },
-      )
+      bridge.publishChat(chatPayload(state = "final"))
       bridge.awaitIdleForTests()
 
       assertEquals(1, resolutions)
@@ -602,8 +407,7 @@ class WearProxyBridgeTest {
       val attempts = mutableListOf<SentWearMessage>()
       var resolutions = 0
       val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
+        backgroundScope.testBridge(
           sender =
             WearMessageSender { nodeId, path, data ->
               attempts += SentWearMessage(nodeId, path, data)
@@ -617,17 +421,11 @@ class WearProxyBridgeTest {
               setOf("watch-current")
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
       bridge.handleMessage("watch-stale", WearProtocolCodec.encode(request("req-1")))
       attempts.clear()
 
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "final")
-        },
-      )
+      bridge.publishChat(chatPayload(state = "final"))
       bridge.awaitIdleForTests()
 
       // Terminal delivery refreshes before sending; the stale failure then forces a second
@@ -644,8 +442,7 @@ class WearProxyBridgeTest {
       var failSecondWatch = true
       var resolutions = 0
       val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
+        backgroundScope.testBridge(
           sender =
             WearMessageSender { nodeId, path, _ ->
               if (path != WearProtocol.EVENT_PATH) return@WearMessageSender
@@ -661,7 +458,6 @@ class WearProxyBridgeTest {
               setOf("watch-1", "watch-2")
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
       bridge.handleMessage("watch-1", WearProtocolCodec.encode(request("req-1")))
       bridge.handleMessage("watch-2", WearProtocolCodec.encode(request("req-2")))
@@ -677,13 +473,10 @@ class WearProxyBridgeTest {
   @Test
   fun queueOverflowRetainsTerminalEventAndEmitsResync() =
     withActorScope { actorScope ->
-      val sent = mutableListOf<SentWearMessage>()
       val requestStarted = CompletableDeferred<Unit>()
       val finishRequest = CompletableDeferred<Unit>()
-      val bridge =
-        WearProxyBridge(
-          scope = actorScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
+      val (bridge, sent) =
+        actorScope.recordingBridge(
           monotonicMillis = { 1_000L },
           handleRequest = { _, request ->
             requestStarted.complete(Unit)
@@ -695,27 +488,9 @@ class WearProxyBridgeTest {
       requestStarted.await()
 
       repeat(40) { index ->
-        bridge.publishChat(
-          buildJsonObject {
-            put("sessionKey", "main")
-            put("state", "delta")
-            put("deltaText", "$index")
-          },
-        )
+        bridge.publishChat(chatPayload(state = "delta", text = "$index"))
       }
-      bridge.publishChat(
-        buildJsonObject {
-          put("sessionKey", "main")
-          put("state", "final")
-          put(
-            "message",
-            buildJsonObject {
-              put("role", "assistant")
-              put("content", "done")
-            },
-          )
-        },
-      )
+      bridge.publishChat(chatPayload(state = "final", message = "done"))
       finishRequest.complete(Unit)
       requestJob.await()
       bridge.awaitIdleForTests()
@@ -723,7 +498,7 @@ class WearProxyBridgeTest {
       val events =
         sent
           .filter { it.path == WearProtocol.EVENT_PATH }
-          .map { (WearProtocolCodec.decode(it.data) as WearDecodeResult.Success).message as WearMessage.Event }
+          .map { it.event() }
       assertTrue(events.any { it.event == WearEventType.Resync })
       assertTrue(
         events.any { event ->
@@ -742,13 +517,7 @@ class WearProxyBridgeTest {
   @Test
   fun eventQueuePreservesSequenceAndBoundsPeers() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender = WearMessageSender { nodeId, path, data -> sent += SentWearMessage(nodeId, path, data) },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
-        )
+      val (bridge, sent) = backgroundScope.recordingBridge()
       repeat(9) { index -> bridge.handleMessage("watch-$index", WearProtocolCodec.encode(request("req-$index"))) }
       sent.clear()
 
@@ -769,7 +538,7 @@ class WearProxyBridgeTest {
       val events =
         sent
           .filter { it.path == WearProtocol.EVENT_PATH }
-          .map { (WearProtocolCodec.decode(it.data) as WearDecodeResult.Success).message as WearMessage.Event }
+          .map { it.event() }
       assertEquals(16, events.size)
       assertEquals(setOf(1L, 2L), events.map { it.sequence }.toSet())
       assertEquals(setOf(WearEventType.Connection, WearEventType.Chat), events.map { it.event }.toSet())
@@ -786,18 +555,9 @@ class WearProxyBridgeTest {
   @Test
   fun connectionEventsCarrySemanticFailureReasons() =
     runTest {
-      val sent = mutableListOf<SentWearMessage>()
-      val bridge =
-        WearProxyBridge(
-          scope = backgroundScope,
-          sender =
-            WearMessageSender { nodeId, path, data ->
-              sent += SentWearMessage(nodeId, path, data)
-            },
+      val (bridge, sent) =
+        backgroundScope.recordingBridge(
           peerResolver = WearPeerResolver { setOf("watch-1") },
-          handleRequest = { _, request ->
-            WearMessage.Response(requestId = request.requestId, ok = true)
-          },
         )
       bridge.handleMessage("watch-1", WearProtocolCodec.encode(request("req-1")))
       sent.clear()
@@ -812,8 +572,7 @@ class WearProxyBridgeTest {
       val event =
         sent
           .single { it.path == WearProtocol.EVENT_PATH }
-          .let { (WearProtocolCodec.decode(it.data) as WearDecodeResult.Success).message }
-          as WearMessage.Event
+          .event()
       val payload = checkNotNull(event.payload).jsonObject
       assertEquals(false, payload.getValue("connected").jsonPrimitive.boolean)
       assertEquals("incompatible", payload.getValue("failure").jsonPrimitive.content)
@@ -866,8 +625,7 @@ class WearProxyBridgeTest {
       val sent = mutableListOf<SentWearMessage>()
       var failFirstEvent = true
       val bridge =
-        WearProxyBridge(
-          scope = actorScope,
+        actorScope.testBridge(
           sender =
             WearMessageSender { nodeId, path, data ->
               if (path == WearProtocol.EVENT_PATH && failFirstEvent) {
@@ -879,7 +637,6 @@ class WearProxyBridgeTest {
               sent += SentWearMessage(nodeId, path, data)
             },
           monotonicMillis = { 1_000L },
-          handleRequest = { _, request -> WearMessage.Response(requestId = request.requestId, ok = true) },
         )
       bridge.handleMessage("watch-1", WearProtocolCodec.encode(request("req-1")))
 
@@ -901,24 +658,16 @@ class WearProxyBridgeTest {
       assertEquals(2, sent.count { it.path == WearProtocol.EVENT_PATH })
     }
 
-  private fun assertForeignTerminalPreservesActiveStream(state: String) {
+  private fun assertForeignTerminalPreservesStream(
+    state: String,
+    runId: String? = null,
+  ) {
     val projector = WearChatStreamProjector()
-    projectStreamEvent(projector, state = "delta", runId = "active-run", text = "Hel", message = "Hel")
+    projectStreamEvent(projector, state = "delta", runId = runId, text = "Hel", message = "Hel")
 
     projectStreamEvent(projector, state = state, runId = "older-run")
 
-    val continued = projectStreamEvent(projector, state = "delta", runId = "active-run", text = "lo")
-    assertEquals("Hello", continued.getValue("streamText").jsonPrimitive.content)
-    assertEquals("true", continued.getValue("streamTextComplete").jsonPrimitive.content)
-  }
-
-  private fun assertForeignTerminalPreservesAnonymousStream(state: String) {
-    val projector = WearChatStreamProjector()
-    projectStreamEvent(projector, state = "delta", text = "Hel", message = "Hel")
-
-    projectStreamEvent(projector, state = state, runId = "older-run")
-
-    val continued = projectStreamEvent(projector, state = "delta", text = "lo")
+    val continued = projectStreamEvent(projector, state = "delta", runId = runId, text = "lo")
     assertEquals("Hello", continued.getValue("streamText").jsonPrimitive.content)
     assertEquals("true", continued.getValue("streamTextComplete").jsonPrimitive.content)
   }
@@ -931,25 +680,70 @@ class WearProxyBridgeTest {
     message: String? = null,
   ): JsonObject =
     checkNotNull(
-      projector.project(
-        buildJsonObject {
-          put("sessionKey", "main")
-          runId?.let { put("runId", it) }
-          put("state", state)
-          text?.let { put("deltaText", it) }
-          message?.let { fullText ->
-            put(
-              "message",
-              buildJsonObject {
-                put("role", "assistant")
-                put("content", fullText)
-              },
-            )
-          }
-        },
-      ),
+      projector.project(chatPayload(state = state, runId = runId, text = text, message = message)),
     )
 
+  private fun CoroutineScope.recordingBridge(
+    peerResolver: WearPeerResolver = WearPeerResolver { emptySet() },
+    monotonicMillis: (() -> Long)? = null,
+    handleRequest: suspend (String, WearMessage.Request) -> WearMessage.Response = ::successResponse,
+  ): Pair<WearProxyBridge, MutableList<SentWearMessage>> {
+    val sent = mutableListOf<SentWearMessage>()
+    val bridge =
+      if (monotonicMillis == null) {
+        testBridge(sent.recordingSender(), peerResolver, handleRequest)
+      } else {
+        testBridge(sent.recordingSender(), peerResolver, monotonicMillis, handleRequest)
+      }
+    return bridge to sent
+  }
+
+  private fun CoroutineScope.testBridge(
+    sender: WearMessageSender,
+    peerResolver: WearPeerResolver = WearPeerResolver { emptySet() },
+    handleRequest: suspend (String, WearMessage.Request) -> WearMessage.Response = ::successResponse,
+  ): WearProxyBridge = WearProxyBridge(this, sender, peerResolver, handleRequest = handleRequest)
+
+  private fun CoroutineScope.testBridge(
+    sender: WearMessageSender,
+    peerResolver: WearPeerResolver = WearPeerResolver { emptySet() },
+    monotonicMillis: () -> Long,
+    handleRequest: suspend (String, WearMessage.Request) -> WearMessage.Response = ::successResponse,
+  ): WearProxyBridge = WearProxyBridge(this, sender, peerResolver, monotonicMillis, handleRequest)
+
+  private fun successResponse(
+    sourceNodeId: String,
+    request: WearMessage.Request,
+  ): WearMessage.Response = WearMessage.Response(requestId = request.requestId, ok = true)
+
+  private fun chatPayload(
+    state: String,
+    runId: String? = null,
+    text: String? = null,
+    message: String? = null,
+  ): JsonObject =
+    buildJsonObject {
+      put("sessionKey", "main")
+      runId?.let { put("runId", it) }
+      put("state", state)
+      text?.let { put("deltaText", it) }
+      message?.let { fullText ->
+        put(
+          "message",
+          buildJsonObject {
+            put("role", "assistant")
+            put("content", fullText)
+          },
+        )
+      }
+    }
+
+  private fun MutableList<SentWearMessage>.recordingSender(): WearMessageSender = WearMessageSender { nodeId, path, data -> add(SentWearMessage(nodeId, path, data)) }
+
+  private fun SentWearMessage.response(): WearMessage.Response = (WearProtocolCodec.decode(data) as WearDecodeResult.Success).message as WearMessage.Response
+
+  private fun SentWearMessage.event(): WearMessage.Event = (WearProtocolCodec.decode(data) as WearDecodeResult.Success).message as WearMessage.Event
+
   private fun request(requestId: String): WearMessage.Request = WearMessage.Request(requestId = requestId, method = WearRpcMethod.ProxyStatus)
 }