diff --git a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt index ee02be725b51..4f83a4cccc9e 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/wear/WearProxyBridgeTest.kt +++ b/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() - 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() - 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() 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() 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() - 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() - 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() val requestStarted = CompletableDeferred() val finishRequest = CompletableDeferred() - 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() - 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() 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() 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() 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() val requestStarted = CompletableDeferred() val finishRequest = CompletableDeferred() - 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() - 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() - 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() 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> { + val sent = mutableListOf() + 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.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) }