diff --git a/apps/android/app/src/test/java/ai/openclaw/app/voice/RealtimeAgentCoordinatorTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/voice/RealtimeAgentCoordinatorTest.kt index aae5cd256f7f..18cf2287a8e0 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/voice/RealtimeAgentCoordinatorTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/voice/RealtimeAgentCoordinatorTest.kt @@ -17,47 +17,26 @@ import org.robolectric.RobolectricTestRunner @OptIn(ExperimentalCoroutinesApi::class) @RunWith(RobolectricTestRunner::class) class RealtimeAgentCoordinatorTest { + private lateinit var calls: MutableList + @Test fun `consult correlates the active run and submits its final text`() = runTest { - val calls = mutableListOf() val working = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") """{"runId":"run-1"}""" else "{}" }, onWorking = working::add, ) val session = RealtimeAgentSession("relay-1", "session-1") coordinator.beginSession(session) - assertTrue( - coordinator.handleToolCall( - callId = "call-1", - name = "openclaw_agent_consult", - args = null, - forced = false, - ), - ) + assertTrue(coordinator.consult("call-1")) runCurrent() assertEquals(listOf(session), working) - assertFalse( - coordinator.handleChatEvent( - sessionKey = "other-session", - runId = "run-1", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"wrong"}"""), - ), - ) - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-1", - runId = "run-1", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"done"}"""), - ), - ) + assertFalse(coordinator.complete("other-session", "run-1", "wrong")) + assertTrue(coordinator.complete("session-1", "run-1", "done")) runCurrent() val consult = calls.single { it.method == "talk.client.toolCall" } @@ -72,25 +51,16 @@ class RealtimeAgentCoordinatorTest { @Test fun `early completion waits for run metadata`() = runTest { - val calls = mutableListOf() val response = CompletableDeferred() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") response.await() else "{}" }, ) coordinator.beginSession(RealtimeAgentSession("relay-1", "session-1")) - coordinator.handleToolCall("call-1", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-1") runCurrent() - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-1", - runId = "run-early", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"early"}"""), - ), - ) + assertTrue(coordinator.complete("session-1", "run-early", "early")) response.complete("""{"runId":"run-early"}""") runCurrent() @@ -105,10 +75,8 @@ class RealtimeAgentCoordinatorTest { @Test fun `validates tool names and dispatches control without a consult`() = runTest { - val calls = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> when (method) { "talk.session.steer" -> """{"status":"steered"}""" @@ -143,11 +111,9 @@ class RealtimeAgentCoordinatorTest { @Test fun `forced consult reports working then returns gateway errors`() = runTest { - val calls = mutableListOf() val errors = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") error("gateway offline") else "{}" }, @@ -155,12 +121,7 @@ class RealtimeAgentCoordinatorTest { ) coordinator.beginSession(RealtimeAgentSession("relay-1", "session-1")) - coordinator.handleToolCall( - callId = "call-1", - name = "openclaw_agent_consult", - args = null, - forced = true, - ) + coordinator.consult("call-1", forced = true) runCurrent() val results = calls.filter { it.method == "talk.session.submitToolResult" } @@ -174,13 +135,11 @@ class RealtimeAgentCoordinatorTest { @Test fun `session replacement quarantines the old run while a call id is reused`() = runTest { - val calls = mutableListOf() val oldResponse = CompletableDeferred() val newResponse = CompletableDeferred() var requestCount = 0 val coordinator = coordinator( - calls = calls, responses = { method -> if (method != "talk.client.toolCall") { "{}" @@ -192,21 +151,14 @@ class RealtimeAgentCoordinatorTest { }, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-shared", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-shared") runCurrent() coordinator.beginSession(RealtimeAgentSession("relay-new", "session-main")) - coordinator.handleToolCall("call-shared", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-shared") runCurrent() - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "run-new", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"fresh"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "run-new", "fresh")) newResponse.complete("""{"runId":"run-new"}""") runCurrent() @@ -216,41 +168,25 @@ class RealtimeAgentCoordinatorTest { oldResponse.complete("""{"runId":"run-old"}""") runCurrent() - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "run-old", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"stale"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "run-old", "stale")) assertEquals(1, calls.count { it.method == "talk.session.submitToolResult" }) } @Test fun `transport reset cancels an old consult before a new gateway session`() = runTest { - val calls = mutableListOf() val oldResponse = CompletableDeferred() val unhandled = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") oldResponse.await() else "{}" }, onUnhandledCompletion = unhandled::add, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) - coordinator.handleToolCall("call-old-2", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") + coordinator.consult("call-old-2") runCurrent() - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "cached-old-run", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"stale cached"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "cached-old-run", "stale cached")) coordinator.resetTransport() coordinator.beginSession(RealtimeAgentSession("relay-new", "session-main")) @@ -259,42 +195,26 @@ class RealtimeAgentCoordinatorTest { assertTrue(calls.none { it.method == "talk.session.submitToolResult" }) assertTrue(unhandled.isEmpty()) - assertFalse( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "run-old", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"stale"}"""), - ), - ) + assertFalse(coordinator.complete("session-main", "run-old", "stale")) } @Test fun `transport reset rejects a reused retired run id without stranding the call`() = runTest { - val calls = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") """{"runId":"shared-run"}""" else "{}" }, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") runCurrent() coordinator.resetTransport() coordinator.beginSession(RealtimeAgentSession("relay-new", "session-main")) - coordinator.handleToolCall("call-new", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-new") runCurrent() - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "shared-run", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"late"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "shared-run", "late")) runCurrent() val result = calls.single { it.method == "talk.session.submitToolResult" } @@ -306,27 +226,18 @@ class RealtimeAgentCoordinatorTest { @Test fun `old session request does not buffer a new session completion`() = runTest { - val calls = mutableListOf() val oldResponse = CompletableDeferred() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") oldResponse.await() else "{}" }, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-old")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") runCurrent() coordinator.beginSession(RealtimeAgentSession("relay-new", "session-new")) - assertFalse( - coordinator.handleChatEvent( - sessionKey = "session-new", - runId = "ordinary-run", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"ordinary"}"""), - ), - ) + assertFalse(coordinator.complete("session-new", "ordinary-run", "ordinary")) oldResponse.complete("""{"runId":"run-old"}""") runCurrent() assertTrue(calls.none { it.method == "talk.session.submitToolResult" }) @@ -335,26 +246,17 @@ class RealtimeAgentCoordinatorTest { @Test fun `session replacement consumes a known old run with the same session key`() = runTest { - val calls = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") """{"runId":"run-old"}""" else "{}" }, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") runCurrent() coordinator.beginSession(RealtimeAgentSession("relay-new", "session-main")) - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "run-old", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"stale"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "run-old", "stale")) runCurrent() assertTrue(calls.none { it.method == "talk.session.submitToolResult" }) @@ -363,26 +265,17 @@ class RealtimeAgentCoordinatorTest { @Test fun `session end retains unresolved correlation and quarantines its final`() = runTest { - val calls = mutableListOf() val response = CompletableDeferred() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") response.await() else "{}" }, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") runCurrent() coordinator.endSession("relay-old") - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "run-old", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"stale"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "run-old", "stale")) response.complete("""{"runId":"run-old"}""") runCurrent() @@ -392,17 +285,15 @@ class RealtimeAgentCoordinatorTest { @Test fun `session end suppresses a late gateway failure`() = runTest { - val calls = mutableListOf() val response = CompletableDeferred() val errors = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") response.await() else "{}" }, onError = errors::add, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") runCurrent() coordinator.endSession("relay-old") @@ -416,26 +307,17 @@ class RealtimeAgentCoordinatorTest { @Test fun `session cleanup releases uncertain early completions`() = runTest { - val calls = mutableListOf() val response = CompletableDeferred() val unhandled = mutableListOf() val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") response.await() else "{}" }, onUnhandledCompletion = unhandled::add, ) coordinator.beginSession(RealtimeAgentSession("relay-old", "session-main")) - coordinator.handleToolCall("call-old", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-old") runCurrent() - assertTrue( - coordinator.handleChatEvent( - sessionKey = "session-main", - runId = "uncertain-run", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"ordinary"}"""), - ), - ) + assertTrue(coordinator.complete("session-main", "uncertain-run", "ordinary")) coordinator.beginSession(RealtimeAgentSession("relay-new", "session-main")) response.complete("""{"runId":"old-tool-run"}""") @@ -449,13 +331,11 @@ class RealtimeAgentCoordinatorTest { @Test fun `early completion overflow fails pending calls instead of stranding them`() = runTest { - val calls = mutableListOf() val unhandled = mutableListOf() val responses = List(3) { CompletableDeferred() } var responseIndex = 0 val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") responses[responseIndex++].await() else "{}" }, @@ -464,16 +344,11 @@ class RealtimeAgentCoordinatorTest { ) coordinator.beginSession(RealtimeAgentSession("relay-1", "session-1")) repeat(3) { index -> - coordinator.handleToolCall("call-${index + 1}", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-${index + 1}") } runCurrent() repeat(3) { index -> - coordinator.handleChatEvent( - sessionKey = "session-1", - runId = "run-${index + 1}", - state = "final", - message = Json.parseToJsonElement("""{"role":"assistant","content":"result-${index + 1}"}"""), - ) + coordinator.complete("session-1", "run-${index + 1}", "result-${index + 1}") } runCurrent() responses.take(2).forEachIndexed { index, response -> response.complete("""{"runId":"run-${index + 1}"}""") } @@ -496,11 +371,9 @@ class RealtimeAgentCoordinatorTest { @Test fun `registered runs count against the concurrent call limit`() = runTest { - val calls = mutableListOf() var runIndex = 0 val coordinator = coordinator( - calls = calls, responses = { method -> if (method == "talk.client.toolCall") """{"runId":"run-${++runIndex}"}""" else "{}" }, @@ -509,7 +382,7 @@ class RealtimeAgentCoordinatorTest { coordinator.beginSession(RealtimeAgentSession("relay-1", "session-1")) repeat(3) { index -> - coordinator.handleToolCall("call-${index + 1}", "openclaw_agent_consult", null, forced = false) + coordinator.consult("call-${index + 1}") runCurrent() } @@ -520,23 +393,36 @@ class RealtimeAgentCoordinatorTest { } private fun kotlinx.coroutines.test.TestScope.coordinator( - calls: MutableList, responses: suspend (String) -> String, onWorking: (RealtimeAgentSession) -> Unit = {}, onError: (String) -> Unit = {}, onUnhandledCompletion: (RealtimeAgentUnhandledCompletion) -> Unit = {}, maxCachedCompletions: Int = 128, - ) = RealtimeAgentCoordinator( - parentScope = backgroundScope, - requestGateway = { method, params, timeoutMs -> - calls += GatewayCall(method, params.orEmpty(), timeoutMs) - responses(method) - }, - onWorking = onWorking, - onError = { _, message -> onError(message) }, - onUnhandledCompletion = onUnhandledCompletion, - maxCachedCompletions = maxCachedCompletions, - ) + ): RealtimeAgentCoordinator { + calls = mutableListOf() + return RealtimeAgentCoordinator( + parentScope = backgroundScope, + requestGateway = { method, params, timeoutMs -> + calls += GatewayCall(method, params.orEmpty(), timeoutMs) + responses(method) + }, + onWorking = onWorking, + onError = { _, message -> onError(message) }, + onUnhandledCompletion = onUnhandledCompletion, + maxCachedCompletions = maxCachedCompletions, + ) + } + + private fun RealtimeAgentCoordinator.consult( + callId: String, + forced: Boolean = false, + ): Boolean = handleToolCall(callId, "openclaw_agent_consult", null, forced) + + private fun RealtimeAgentCoordinator.complete( + sessionKey: String, + runId: String, + text: String, + ): Boolean = handleChatEvent(sessionKey, runId, "final", Json.parseToJsonElement("""{"role":"assistant","content":"$text"}""")) private data class GatewayCall( val method: String, diff --git a/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt index ecd2a1c667f2..3831d54ff30a 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/voice/TalkModeManagerTest.kt @@ -1,7 +1,7 @@ package ai.openclaw.app.voice -import ai.openclaw.app.gateway.DeviceAuthEntry -import ai.openclaw.app.gateway.DeviceAuthTokenStore +import ai.openclaw.app.SecurePrefs +import ai.openclaw.app.gateway.DeviceAuthStore import ai.openclaw.app.gateway.GatewayRequestRejected import ai.openclaw.app.gateway.GatewaySession import ai.openclaw.app.gateway.testDeviceIdentityStore @@ -29,6 +29,7 @@ import kotlinx.coroutines.async import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.launch import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.currentTime @@ -49,6 +50,7 @@ import org.robolectric.annotation.Config import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicLong +@OptIn(ExperimentalCoroutinesApi::class) @RunWith(RobolectricTestRunner::class) @Config(sdk = [34]) class TalkModeManagerTest { @@ -150,18 +152,11 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun beginPushToTalkRejectsInvalidatedCaptureBeforeStarting() = runTest { - val app = RuntimeEnvironment.getApplication() - shadowOf(app).grantPermissions(Manifest.permission.RECORD_AUDIO) - val packageManager = shadowOf(app.packageManager) - val speechService = ComponentName(app, "TestSpeechRecognitionService") - packageManager.addServiceIfNotPresent(speechService) - packageManager.addIntentFilterForService(speechService, IntentFilter(RecognitionService.SERVICE_INTERFACE)) + installSpeechRecognitionService() val manager = createManager() - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain { val error = runCatching { manager.beginPushToTalk( @@ -173,8 +168,6 @@ class TalkModeManagerTest { assertEquals("NODE_BACKGROUND_UNAVAILABLE: command requires foreground", error?.message) assertNull(readPrivateField(manager, "activePttCaptureId")) assertFalse(manager.isListening.value) - } finally { - Dispatchers.resetMain() } } @@ -198,22 +191,18 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun staleCancellationDoesNotStopNewerPushToTalkCapture() = runTest { val manager = createManager() val completion = CompletableDeferred() setPrivateField(manager, "activePttCaptureId", "capture-new") setPrivateField(manager, "pttCompletion", completion) - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain { val payload = manager.cancelPushToTalk("capture-old") assertEquals("idle", payload.status) assertEquals("capture-new", readPrivateField(manager, "activePttCaptureId")) assertFalse(completion.isCompleted) - } finally { - Dispatchers.resetMain() } } @@ -235,7 +224,6 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun cancelledOneShotWaitCleansItsCapture() = runTest { val manager = createManager() @@ -244,8 +232,7 @@ class TalkModeManagerTest { setPrivateField(manager, "pttCompletion", completion) setMutableStateFlow(manager, "_isListening", true) val start = TalkPttOnceStart.Started(captureId = "capture-1", completion = completion) - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain { val wait = launch { manager.awaitPushToTalkOnce(start) } advanceUntilIdle() wait.cancel() @@ -256,13 +243,10 @@ class TalkModeManagerTest { assertNull(readPrivateField(manager, "pttCompletion")) assertFalse(manager.isListening.value) assertTrue(completion.isCompleted) - } finally { - Dispatchers.resetMain() } } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun staleStopDoesNotSubmitNewerPushToTalkCapture() = runTest { val manager = createManager() @@ -270,16 +254,13 @@ class TalkModeManagerTest { setPrivateField(manager, "activePttCaptureId", "capture-new") setPrivateField(manager, "pttCompletion", completion) setPrivateField(manager, "lastTranscript", "new partial transcript") - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain { val payload = manager.endPushToTalk("capture-old") assertEquals("idle", payload.status) assertEquals("capture-new", readPrivateField(manager, "activePttCaptureId")) assertEquals("new partial transcript", readPrivateField(manager, "lastTranscript")) assertFalse(completion.isCompleted) - } finally { - Dispatchers.resetMain() } } @@ -306,7 +287,6 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun releaseKeepsWaitingPastOldGraceForLateTerminalSegment() = runTest { val manager = createManager(isConnected = { false }) @@ -317,8 +297,7 @@ class TalkModeManagerTest { @Suppress("UNCHECKED_CAST") (readPrivateField(manager, "pttFinalSegments") as MutableList) += "early segment" val listener = recognitionListener(manager, "capture-1") - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain { val ending = async { manager.endPushToTalk("capture-1") } runCurrent() @@ -334,29 +313,20 @@ class TalkModeManagerTest { advanceUntilIdle() assertEquals("early segment. late segment", ending.await().transcript) - } finally { - Dispatchers.resetMain() } } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun cancelledEndPushToTalkClearsPendingReleaseBeforeNextBegin() = runTest { - val app = RuntimeEnvironment.getApplication() - shadowOf(app).grantPermissions(Manifest.permission.RECORD_AUDIO) - val packageManager = shadowOf(app.packageManager) - val speechService = ComponentName(app, "TestSpeechRecognitionService") - packageManager.addServiceIfNotPresent(speechService) - packageManager.addIntentFilterForService(speechService, IntentFilter(RecognitionService.SERVICE_INTERFACE)) + installSpeechRecognitionService() val manager = createManager() setPrivateField(manager, "activePttCaptureId", "capture-a") setPrivateField(manager, "pttReleaseCompletion", CompletableDeferred()) setPrivateField(manager, "pttRecognitionRung", silenceSegmentedRung()) @Suppress("UNCHECKED_CAST") (readPrivateField(manager, "pttFinalSegments") as MutableList) += "capture a" - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain(cleanup = manager::stopAllCapture) { val ending = async { manager.endPushToTalk("capture-a") } runCurrent() ending.cancel() @@ -372,22 +342,13 @@ class TalkModeManagerTest { assertEquals(started.captureId, readPrivateField(manager, "activePttCaptureId")) assertEquals(emptyList(), readPrivateField(manager, "pttFinalSegments")) - } finally { - manager.stopAllCapture() - Dispatchers.resetMain() } } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun replacementBeginDrainsPendingReleaseBeforeStartingNewCapture() = runTest { - val app = RuntimeEnvironment.getApplication() - shadowOf(app).grantPermissions(Manifest.permission.RECORD_AUDIO) - val packageManager = shadowOf(app.packageManager) - val speechService = ComponentName(app, "TestSpeechRecognitionService") - packageManager.addServiceIfNotPresent(speechService) - packageManager.addIntentFilterForService(speechService, IntentFilter(RecognitionService.SERVICE_INTERFACE)) + installSpeechRecognitionService() var connectionChecks = 0 val manager = createManager( @@ -402,8 +363,7 @@ class TalkModeManagerTest { setPrivateField(manager, "pttRecognitionRung", silenceSegmentedRung()) @Suppress("UNCHECKED_CAST") (readPrivateField(manager, "pttFinalSegments") as MutableList) += "first segment" - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) - try { + withMain(cleanup = manager::stopAllCapture) { val ending = async { manager.endPushToTalk("capture-a") } runCurrent() val starting = async { manager.beginPushToTalk(allowNewCapture = true) } @@ -418,9 +378,6 @@ class TalkModeManagerTest { assertEquals("first segment", ended.transcript) assertEquals(started.captureId, readPrivateField(manager, "activePttCaptureId")) assertEquals(emptyList(), readPrivateField(manager, "pttFinalSegments")) - } finally { - manager.stopAllCapture() - Dispatchers.resetMain() } } @@ -470,10 +427,7 @@ class TalkModeManagerTest { setPrivateField(manager, "realtimeSessionId", "relay-1") setMutableStateFlow(manager, "_isEnabled", true) - manager.handleGatewayEvent( - "talk.event", - """{"relaySessionId":"relay-1","type":"close","reason":"error"}""", - ) + manager.realtimeEvent("""{"relaySessionId":"relay-1","type":"close","reason":"error"}""") assertFalse(manager.isEnabled.value) assertTrue(stoppedByRelay) @@ -491,16 +445,12 @@ class TalkModeManagerTest { setMutableStateFlow(manager, "_isEnabled", true) setTalkFailure(manager, verbatimText("Échec de Talk : session refusée.")) - manager.handleGatewayEvent( - "talk.event", - """{"relaySessionId":"relay-1","type":"close","reason":"error"}""", - ) + manager.realtimeEvent("""{"relaySessionId":"relay-1","type":"close","reason":"error"}""") assertEquals("Échec de Talk : session refusée.", manager.statusText.value) } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun localizedOffStatusDoesNotBecomeRealtimeStartFailure() = runTest { val manager = createManager(scope = this) @@ -524,25 +474,20 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun realtimePlaybackMarkAcknowledgesAfterQueuedAudioBarrier() = runTest { val acknowledgements = mutableListOf>() - val dispatcher = StandardTestDispatcher(testScheduler) val manager = createManager( scope = this, - realtimePlaybackDispatcher = dispatcher, + realtimePlaybackDispatcher = StandardTestDispatcher(testScheduler), realtimeMarkAcknowledger = { sessionId, markName -> acknowledgements += sessionId to markName }, ) setPrivateField(manager, "realtimeSessionId", "relay-1") - manager.handleGatewayEvent( - "talk.event", - """{"relaySessionId":"relay-1","type":"mark","markName":"audio-1"}""", - ) + manager.realtimeEvent("""{"relaySessionId":"relay-1","type":"mark","markName":"audio-1"}""") runCurrent() assertEquals(listOf("relay-1" to "audio-1"), acknowledgements) @@ -550,14 +495,12 @@ class TalkModeManagerTest { @Test fun realtimeTranscriptsPopulateVoiceConversation() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "hello")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "hello world", final = true)) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "hi")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "hi there", final = true)) + manager.transcript("user", "hello") + manager.transcript("user", "hello world", final = true) + manager.transcript("assistant", "hi") + manager.transcript("assistant", "hi there", final = true) assertEquals( listOf( @@ -578,41 +521,35 @@ class TalkModeManagerTest { @Test fun realtimeUserTranscriptsDriveSpeechActive() { - val manager = createManager() - - setPrivateField(manager, "realtimeSessionId", "relay-1") + val manager = createRealtimeManager() assertFalse(manager.speechActive.value) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "hello")) + manager.transcript("user", "hello") assertTrue(manager.speechActive.value) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "hello world", final = true)) + manager.transcript("user", "hello world", final = true) assertFalse(manager.speechActive.value) } @Test fun finalUserTranscriptMarksAwaitingAgentUntilStatusMovesOn() { - val manager = createManager() - - setPrivateField(manager, "realtimeSessionId", "relay-1") + val manager = createRealtimeManager() assertFalse(manager.awaitingAgent.value) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "hello", final = true)) + manager.transcript("user", "hello", final = true) assertTrue(manager.awaitingAgent.value) // Any later status transition clears the typed flag; forgetting it at a // new setStatus site fails safe instead of showing a stale Thinking wave. - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "hi there", final = true)) + manager.transcript("assistant", "hi there", final = true) manager.stopAllCapture() assertFalse(manager.awaitingAgent.value) } @Test fun realtimeTranscriptDeltasAccumulateVoiceConversation() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "The")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = " answer")) + manager.transcript("assistant", "The") + manager.transcript("assistant", " answer") val entry = manager.conversation.value.single() assertEquals("The answer", entry.text) @@ -621,12 +558,10 @@ class TalkModeManagerTest { @Test fun realtimeTranscriptFragmentsInsertWordSpacing() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Turn off")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "the lights")) + manager.transcript("user", "Turn off") + manager.transcript("user", "the lights") val entry = manager.conversation.value.single() assertEquals("Turn off the lights", entry.text) @@ -635,12 +570,10 @@ class TalkModeManagerTest { @Test fun realtimeTranscriptFragmentsInsertSpacingAfterPunctuation() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "Ready.")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "What next?")) + manager.transcript("assistant", "Ready.") + manager.transcript("assistant", "What next?") val entry = manager.conversation.value.single() assertEquals("Ready. What next?", entry.text) @@ -649,15 +582,10 @@ class TalkModeManagerTest { @Test fun realtimeFinalTranscriptCanCompleteDeltaText() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "The")) - manager.handleGatewayEvent( - "talk.event", - realtimeTranscriptPayload(role = "assistant", text = " answer", final = true), - ) + manager.transcript("assistant", "The") + manager.transcript("assistant", " answer", final = true) val entry = manager.conversation.value.single() assertEquals("The answer", entry.text) @@ -666,13 +594,11 @@ class TalkModeManagerTest { @Test fun realtimeAssistantOutputSeparatesNextUserBubble() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "First request")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "Checking")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Second request")) + manager.transcript("user", "First request") + manager.transcript("assistant", "Checking") + manager.transcript("user", "Second request") val entries = manager.conversation.value assertEquals(3, entries.size) @@ -689,12 +615,10 @@ class TalkModeManagerTest { @Test fun realtimeUserTranscriptRewriteStaysInSameBubble() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Can you tack")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Can you check?", final = true)) + manager.transcript("user", "Can you tack") + manager.transcript("user", "Can you check?", final = true) val entry = manager.conversation.value.single() assertEquals(VoiceConversationRole.User, entry.role) @@ -704,13 +628,11 @@ class TalkModeManagerTest { @Test fun realtimeLateFinalUserTranscriptRewritesBubbleAfterAssistantStarts() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Can you tack")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "Checking")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Can you check?", final = true)) + manager.transcript("user", "Can you tack") + manager.transcript("assistant", "Checking") + manager.transcript("user", "Can you check?", final = true) val entries = manager.conversation.value assertEquals(2, entries.size) @@ -723,13 +645,11 @@ class TalkModeManagerTest { @Test fun realtimeFinalNextUserAfterAssistantStartsCreatesNewBubble() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "First request")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "Checking")) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Second request", final = true)) + manager.transcript("user", "First request") + manager.transcript("assistant", "Checking") + manager.transcript("user", "Second request", final = true) val entries = manager.conversation.value assertEquals(3, entries.size) @@ -744,37 +664,14 @@ class TalkModeManagerTest { @Test fun realtimeAlternatingTurnsStayInSeparateBubbles() { - val manager = createManager() + val manager = createRealtimeManager() - setPrivateField(manager, "realtimeSessionId", "relay-1") - - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Hey, what time is it?", final = true)) - manager.handleGatewayEvent( - "talk.event", - realtimeTranscriptPayload( - role = "assistant", - text = "Let me look into that for you. It's currently 7:55 PM UTC.", - final = true, - ), - ) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "How's it going?", final = true)) - manager.handleGatewayEvent( - "talk.event", - realtimeTranscriptPayload( - role = "assistant", - text = "Great! Ready for the next task. What can I do for you?", - final = true, - ), - ) - manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "user", text = "Turn on the basement lights", final = true)) - manager.handleGatewayEvent( - "talk.event", - realtimeTranscriptPayload( - role = "assistant", - text = "Got it, let me check on that.", - final = true, - ), - ) + manager.transcript("user", "Hey, what time is it?", final = true) + manager.transcript("assistant", "Let me look into that for you. It's currently 7:55 PM UTC.", final = true) + manager.transcript("user", "How's it going?", final = true) + manager.transcript("assistant", "Great! Ready for the next task. What can I do for you?", final = true) + manager.transcript("user", "Turn on the basement lights", final = true) + manager.transcript("assistant", "Got it, let me check on that.", final = true) val entries = manager.conversation.value assertEquals(6, entries.size) @@ -816,7 +713,6 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun realtimeStartWithoutGatewayTurnsTalkOff() = runTest { val stoppedByRelay = AtomicBoolean(false) @@ -838,21 +734,14 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun browserOnlyRealtimeConfigStartsNativeTalkInsteadOfRelay() = runTest { - val app = RuntimeEnvironment.getApplication() - shadowOf(app).grantPermissions(Manifest.permission.RECORD_AUDIO) - val packageManager = shadowOf(app.packageManager) - val speechService = ComponentName(app, "TestSpeechRecognitionService") - packageManager.addServiceIfNotPresent(speechService) - packageManager.addIntentFilterForService(speechService, IntentFilter(RecognitionService.SERVICE_INTERFACE)) - Dispatchers.setMain(StandardTestDispatcher(testScheduler)) + installSpeechRecognitionService() val manager = createManager( scope = this, ) - try { + withMain(cleanup = { manager.setEnabled(false) }) { setPrivateField(manager, "configLoaded", true) setPrivateField(manager, "realtimeRelayModelSupported", false) manager.setEnabled(true) @@ -862,9 +751,6 @@ class TalkModeManagerTest { assertTrue(manager.isListening.value) assertNull(readPrivateField(manager, "realtimeSessionId")) assertEquals("Listening", manager.statusText.value) - } finally { - manager.setEnabled(false) - Dispatchers.resetMain() } } @@ -1014,17 +900,11 @@ class TalkModeManagerTest { assertEquals("Listening", manager.statusText.value) assertTrue(readPrivateField(manager, "realtimeOutputSuppressed") as Boolean) - manager.handleGatewayEvent( - "talk.event", - """{"relaySessionId":"relay-1","type":"transcript","role":"user","text":"stale","final":true}""", - ) + manager.realtimeEvent("""{"relaySessionId":"relay-1","type":"transcript","role":"user","text":"stale","final":true}""") assertTrue(readPrivateField(manager, "realtimeOutputSuppressed") as Boolean) - manager.handleGatewayEvent( - "talk.event", - """{"relaySessionId":"relay-1","type":"inputAudio","byteLength":4800}""", - ) + manager.realtimeEvent("""{"relaySessionId":"relay-1","type":"inputAudio","byteLength":4800}""") assertFalse(readPrivateField(manager, "realtimeOutputSuppressed") as Boolean) manager.stopAllCapture() @@ -1102,7 +982,6 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun cancelledQueuedFinalizerResumesOnlyItsRealtimeCaptureOnMain() = runTest { val finalizerDispatcher = StandardTestDispatcher() @@ -1110,8 +989,7 @@ class TalkModeManagerTest { createManager( scope = CoroutineScope(SupervisorJob() + finalizerDispatcher), ) - Dispatchers.setMain(Dispatchers.Unconfined) - try { + withMain(dispatcher = Dispatchers.Unconfined, cleanup = manager::stopAllCapture) { setMutableStateFlow(manager, "_isEnabled", true) manager.pauseRealtimeCaptureForPushToTalk("capture-1") setPrivateField(manager, "activePttCaptureId", "capture-1") @@ -1132,9 +1010,6 @@ class TalkModeManagerTest { assertNull(manager.finishingPushToTalkCaptureId) assertNull(readPrivateField(manager, "realtimeCapturePause")) assertNull(readPrivateField(manager, "activePttCaptureId")) - } finally { - manager.stopAllCapture() - Dispatchers.resetMain() } } @@ -1146,10 +1021,7 @@ class TalkModeManagerTest { setPrivateField(manager, "realtimeSessionId", "relay-1") setPrivateField(manager, "finishingPttCaptureId", "capture-1") - manager.handleGatewayEvent( - "talk.event", - """{"relaySessionId":"relay-1","type":"close","reason":"completed"}""", - ) + manager.realtimeEvent("""{"relaySessionId":"relay-1","type":"close","reason":"completed"}""") assertNull(readPrivateField(manager, "realtimeCapturePause")) assertEquals("capture-1", manager.finishingPushToTalkCaptureId) @@ -1185,7 +1057,6 @@ class TalkModeManagerTest { } @Test - @OptIn(ExperimentalCoroutinesApi::class) fun chatFinalWaitUsesGatewayEventTimeout() = runTest { val manager = createManager(scope = this) @@ -1208,12 +1079,11 @@ class TalkModeManagerTest { realtimeMarkAcknowledger: (suspend (String, String) -> Unit)? = null, ): TalkModeManager { val app = RuntimeEnvironment.getApplication() - val sessionJob = SupervisorJob() val session = GatewaySession( - scope = CoroutineScope(sessionJob + Dispatchers.Default), + scope = CoroutineScope(SupervisorJob() + Dispatchers.Default), identityStore = testDeviceIdentityStore(app), - deviceAuthStore = InMemoryDeviceAuthStore(), + deviceAuthStore = DeviceAuthStore(SecurePrefs(app, app.getSharedPreferences("talk-mode-test-${System.nanoTime()}", 0))), onConnected = {}, onDisconnected = {}, onEvent = { _, _ -> }, @@ -1232,6 +1102,40 @@ class TalkModeManagerTest { ) } + private fun createRealtimeManager(): TalkModeManager = createManager().also { setPrivateField(it, "realtimeSessionId", "relay-1") } + + private suspend fun TestScope.withMain( + dispatcher: CoroutineDispatcher = StandardTestDispatcher(testScheduler), + cleanup: () -> Unit = {}, + block: suspend () -> Unit, + ) { + Dispatchers.setMain(dispatcher) + try { + block() + } finally { + cleanup() + Dispatchers.resetMain() + } + } + + private fun TalkModeManager.transcript( + role: String, + text: String, + final: Boolean = false, + ) = handleGatewayEvent("talk.event", realtimeTranscriptPayload(role, text, final)) + + private fun TalkModeManager.realtimeEvent(payload: String) = handleGatewayEvent("talk.event", payload) + + private fun installSpeechRecognitionService() { + val app = RuntimeEnvironment.getApplication() + shadowOf(app).grantPermissions(Manifest.permission.RECORD_AUDIO) + val speechService = ComponentName(app, "TestSpeechRecognitionService") + shadowOf(app.packageManager).apply { + addServiceIfNotPresent(speechService) + addIntentFilterForService(speechService, IntentFilter(RecognitionService.SERVICE_INTERFACE)) + } + } + @Suppress("UNCHECKED_CAST") private fun playbackGeneration(manager: TalkModeManager) = readPrivateField(manager, "playbackGeneration") as AtomicLong @@ -1303,35 +1207,13 @@ class TalkModeManagerTest { runId: String, text: String, role: String = "assistant", - ): String = - """ - { - "runId": "$runId", - "sessionKey": "main", - "state": "final", - "message": { - "role": "$role", - "content": [ - { "type": "text", "text": "$text" } - ] - } - } - """.trimIndent() + ): String = """{"runId":"$runId","sessionKey":"main","state":"final","message":{"role":"$role","content":[{"type":"text","text":"$text"}]}}""" private fun realtimeTranscriptPayload( role: String, text: String, final: Boolean = false, - ): String = - """ - { - "relaySessionId": "relay-1", - "type": "transcript", - "role": "$role", - "text": "$text", - "final": $final - } - """.trimIndent() + ): String = """{"relaySessionId":"relay-1","type":"transcript","role":"$role","text":"$text","final":$final}""" } private class FakeTalkSpeechSynthesizer : TalkSpeechSynthesizing { @@ -1361,25 +1243,3 @@ private class FakeTalkAudioPlayer : TalkAudioPlaying { stopped = true } } - -private class InMemoryDeviceAuthStore : DeviceAuthTokenStore { - override fun loadEntry( - gatewayId: String, - deviceId: String, - role: String, - ): DeviceAuthEntry? = null - - override fun saveToken( - gatewayId: String, - deviceId: String, - role: String, - token: String, - scopes: List, - ) = Unit - - override fun clearToken( - gatewayId: String, - deviceId: String, - role: String, - ) = Unit -}