diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e726b65fa7..18fb44b6ae3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -212,6 +212,7 @@ Docs: https://docs.openclaw.ai - Cron: keep recovered tool warnings diagnostic for successful scheduled runs so final cron output is delivered instead of being replaced by a post-processing warning. (#84045) Thanks @abnershang. - Plugins/perf: thread explicit plugin discovery results through `loadBundledCapabilityRuntimeRegistry`, `resolveBundledPluginSources`, and `listChannelCatalogEntries` so callers that already hold a discovery result skip redundant filesystem walks. Thanks @SebTardif. - harden update restart script creation [AI]. (#84088) Thanks @pgondhi987. +- Android/Control UI Talk: split realtime voice transcript turns, queue PCM playback writes, and add opt-in OpenClaw consult routing for Gateway relay when a realtime provider skips `openclaw_agent_consult`. (#84181) Thanks @VACInc. - Docker: keep the bundled Codex plugin in official release image keep lists so the default OpenAI agent harness remains available after Docker pruning. Fixes #83613. (#83626) Thanks @YuanHanzhong. - CLI/channels: preserve the first line of `openclaw channels logs` output when the rolling tail window starts exactly on a line boundary, mirroring the already-fixed `readLogSlice` behavior in `src/logging/log-tail.ts`. - Control UI: treat terminal session status as authoritative over stale active-run flags so completed terminal runs stop showing abort/live UI. (#84057) diff --git a/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt b/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt index 6da449c6dca..e6e575c4a99 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewaySession.kt @@ -1,12 +1,14 @@ package ai.openclaw.app.gateway import android.util.Log +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch @@ -384,6 +386,22 @@ class GatewaySession( private val client: OkHttpClient = buildClient() private var socket: WebSocket? = null private val loggerTag = "OpenClawGateway" + private val incomingMessages = Channel(Channel.UNLIMITED) + private val messagePumpJob = + scope.launch(Dispatchers.IO) { + for (text in incomingMessages) { + try { + handleMessage(text) + } catch (err: CancellationException) { + throw err + } catch (err: Throwable) { + Log.w( + loggerTag, + "gateway message handling failed: ${err.message ?: err::class.java.simpleName}", + ) + } + } + } val remoteAddress: String = formatGatewayAuthority(endpoint.host, endpoint.port) @@ -475,6 +493,11 @@ class GatewaySession( fun closeQuietly() { if (isClosed.compareAndSet(false, true)) { + incomingMessages.close() + messagePumpJob.cancel() + if (!connectDeferred.isCompleted) { + connectDeferred.completeExceptionally(IllegalStateException("Gateway closed")) + } socket?.close(1000, "bye") socket = null closedDeferred.complete(Unit) @@ -519,7 +542,7 @@ class GatewaySession( webSocket: WebSocket, text: String, ) { - scope.launch { handleMessage(text) } + incomingMessages.trySend(text) } override fun onFailure( @@ -531,6 +554,7 @@ class GatewaySession( connectDeferred.completeExceptionally(t) } if (isClosed.compareAndSet(false, true)) { + incomingMessages.close() failPending() closedDeferred.complete(Unit) onDisconnected("Gateway error: ${t.message ?: t::class.java.simpleName}") @@ -546,6 +570,7 @@ class GatewaySession( connectDeferred.completeExceptionally(IllegalStateException("Gateway closed: $reason")) } if (isClosed.compareAndSet(false, true)) { + incomingMessages.close() failPending() closedDeferred.complete(Unit) onDisconnected("Gateway closed: $reason") diff --git a/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt b/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt index 33cfdfe3faa..ea6ad555b28 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt @@ -105,6 +105,8 @@ class TalkModeManager internal constructor( private const val chatFinalWaitWithoutSubscribeMs = 6_000L private const val maxCachedRunCompletions = 128 private const val maxConversationEntries = 40 + private const val realtimePlaybackBufferMs = 240 + private const val realtimeUserFinalRewriteGraceMs = 1_500L } private val mainHandler = Handler(Looper.getMainLooper()) @@ -165,9 +167,13 @@ class TalkModeManager internal constructor( private val pendingRealtimeToolCalls = LinkedHashSet() private val pendingRealtimeToolCompletions = LinkedHashMap() private var realtimeUserEntryId: String? = null + private var realtimeUserEntryAwaitingFinal = false + private var realtimeUserEntryAwaitingFinalStartedAtMs: Long? = null private var realtimeAssistantEntryId: String? = null private val realtimePlaybackLock = Any() private var realtimeAudioTrack: AudioTrack? = null + private var realtimeAudioQueue: Channel? = null + private var realtimeAudioWriterJob: Job? = null private var realtimePlaybackIdleJob: Job? = null @Volatile @@ -785,6 +791,7 @@ class TalkModeManager internal constructor( } "audio" -> { if (realtimeOutputSuppressed) return + finishRealtimeConversationEntry(VoiceConversationRole.User) val audioBase64 = obj["audioBase64"].asStringOrNull() ?: return val bytes = try { @@ -799,16 +806,20 @@ class TalkModeManager internal constructor( "mark" -> Unit "transcript" -> { val role = obj["role"].asStringOrNull() - val text = obj["text"].asStringOrNull()?.trim().orEmpty() val isFinal = obj["final"].asBooleanOrNull() == true - if (text.isNotEmpty()) { + val text = realtimeTranscriptText(obj["text"].asStringOrNull(), isFinal) + var assistantText: String? = null + if (text != null) { when (role) { "user" -> upsertRealtimeConversation(VoiceConversationRole.User, text, isFinal) - "assistant" -> upsertRealtimeConversation(VoiceConversationRole.Assistant, text, isFinal) + "assistant" -> { + finishRealtimeConversationEntry(VoiceConversationRole.User) + assistantText = upsertRealtimeConversation(VoiceConversationRole.Assistant, text, isFinal) + } } } - if (role == "assistant" && text.isNotEmpty()) { - _lastAssistantText.value = text + if (assistantText != null) { + _lastAssistantText.value = assistantText.trim() } if (isFinal && role == "user") { realtimeOutputSuppressed = false @@ -824,6 +835,7 @@ class TalkModeManager internal constructor( callId = callId, name = name, args = obj["args"], + forced = obj["forced"].asBooleanOrNull() == true, ) } "toolResult" -> Unit @@ -849,6 +861,34 @@ class TalkModeManager internal constructor( private fun playRealtimeAudio(bytes: ByteArray) { if (!playbackEnabled || realtimeOutputSuppressed || bytes.isEmpty()) return + val queue = ensureRealtimeAudioQueue() + if (!queue.trySend(bytes).isSuccess) { + Log.w(tag, "realtime audio queue full") + } + } + + private fun ensureRealtimeAudioQueue(): Channel = + synchronized(realtimePlaybackLock) { + realtimeAudioQueue + ?: Channel(Channel.UNLIMITED).also { queue -> + realtimeAudioQueue = queue + realtimeAudioWriterJob = + scope.launch(Dispatchers.IO) { + for (chunk in queue) { + if (!playbackEnabled || realtimeOutputSuppressed || realtimeSessionId == null) continue + try { + writeRealtimeAudio(chunk) + } catch (err: CancellationException) { + throw err + } catch (err: Throwable) { + Log.w(tag, "realtime audio playback failed: ${err.message ?: err::class.java.simpleName}") + } + } + } + } + } + + private fun writeRealtimeAudio(bytes: ByteArray) { synchronized(realtimePlaybackLock) { val track = realtimeAudioTrack ?: run { @@ -858,6 +898,12 @@ class TalkModeManager internal constructor( AudioFormat.CHANNEL_OUT_MONO, AudioFormat.ENCODING_PCM_16BIT, ) + val bufferSizeBytes = + maxOf( + minBuffer * 2, + realtimeSampleRateHz * 2 * realtimePlaybackBufferMs / 1000, + bytes.size * 4, + ) val created = AudioTrack .Builder() @@ -875,16 +921,27 @@ class TalkModeManager internal constructor( .setChannelMask(AudioFormat.CHANNEL_OUT_MONO) .build(), ).setTransferMode(AudioTrack.MODE_STREAM) - .setBufferSizeInBytes(maxOf(minBuffer, bytes.size * 4)) + .setBufferSizeInBytes(bufferSizeBytes) .build() - created.play() realtimeAudioTrack = created created } + var writtenBytes = 0 + while (writtenBytes < bytes.size) { + val written = track.write(bytes, writtenBytes, bytes.size - writtenBytes) + if (written <= 0) { + Log.w(tag, "realtime audio write failed: $written") + break + } + writtenBytes += written + } + if (writtenBytes <= 0) return + if (track.playState != AudioTrack.PLAYSTATE_PLAYING) { + track.play() + } _isSpeaking.value = true _statusText.value = "Speaking…" - track.write(bytes, 0, bytes.size) - val durationMs = ((bytes.size / 2.0) / realtimeSampleRateHz * 1000.0).toLong() + val durationMs = ((writtenBytes / 2.0) / realtimeSampleRateHz * 1000.0).toLong() val now = SystemClock.elapsedRealtime() realtimePlaybackEndsAtMs = maxOf(now, realtimePlaybackEndsAtMs) + durationMs scheduleRealtimePlaybackIdle() @@ -910,6 +967,12 @@ class TalkModeManager internal constructor( } private fun stopRealtimePlayback() { + val audioQueue = realtimeAudioQueue + val audioWriterJob = realtimeAudioWriterJob + realtimeAudioQueue = null + realtimeAudioWriterJob = null + audioQueue?.close() + audioWriterJob?.cancel() realtimePlaybackIdleJob?.cancel() realtimePlaybackIdleJob = null realtimePlaybackEndsAtMs = 0L @@ -953,6 +1016,8 @@ class TalkModeManager internal constructor( pendingRealtimeToolCalls.clear() pendingRealtimeToolCompletions.clear() realtimeUserEntryId = null + realtimeUserEntryAwaitingFinal = false + realtimeUserEntryAwaitingFinalStartedAtMs = null realtimeAssistantEntryId = null stopRealtimePlayback() if (preserveStatus) { @@ -981,11 +1046,15 @@ class TalkModeManager internal constructor( callId: String, name: String, args: JsonElement?, + forced: Boolean = false, ) { val relaySessionId = realtimeSessionId ?: return pendingRealtimeToolCalls.add(callId) scope.launch { try { + if (forced) { + submitRealtimeToolWorking(callId, relaySessionId) + } val params = buildJsonObject { put("sessionKey", JsonPrimitive(mainSessionKey.ifBlank { "main" })) @@ -1086,6 +1155,7 @@ class TalkModeManager internal constructor( callId: String, result: JsonObject, sessionId: String? = realtimeSessionId, + options: JsonObject? = null, ) { val activeSessionId = sessionId ?: return val params = @@ -1093,6 +1163,7 @@ class TalkModeManager internal constructor( put("sessionId", JsonPrimitive(activeSessionId)) put("callId", JsonPrimitive(callId)) put("result", result) + if (options != null) put("options", options) } try { session.request("talk.session.submitToolResult", params.toString(), timeoutMs = 15_000) @@ -1102,27 +1173,119 @@ class TalkModeManager internal constructor( } } + private suspend fun submitRealtimeToolWorking( + callId: String, + sessionId: String, + ) { + submitRealtimeToolResult( + callId = callId, + sessionId = sessionId, + result = + buildJsonObject { + put("status", JsonPrimitive("working")) + put("tool", JsonPrimitive("openclaw_agent_consult")) + put( + "message", + JsonPrimitive( + "Tell the person briefly that you are checking, then wait for the final OpenClaw result before answering with the actual result.", + ), + ) + }, + options = buildJsonObject { put("willContinue", JsonPrimitive(true)) }, + ) + } + private fun upsertRealtimeConversation( role: VoiceConversationRole, text: String, isFinal: Boolean, - ) { - val entryId = + ): String { + var entryId = when (role) { VoiceConversationRole.User -> realtimeUserEntryId VoiceConversationRole.Assistant -> realtimeAssistantEntryId } + if (role == VoiceConversationRole.Assistant) { + finishRealtimeConversationEntry(VoiceConversationRole.User) + } + val shouldStartNewUserEntry = + role == VoiceConversationRole.User && + entryId != null && + shouldStartNewRealtimeUserEntry(entryId, text, isFinal) + if ( + role == VoiceConversationRole.User && + (entryId == null || shouldStartNewUserEntry) + ) { + finishRealtimeConversationEntry(VoiceConversationRole.Assistant) + } + if (shouldStartNewUserEntry) { + finishRealtimeConversationEntry(VoiceConversationRole.User) + entryId = null + realtimeUserEntryAwaitingFinal = false + realtimeUserEntryAwaitingFinalStartedAtMs = null + } + var resolvedText: String val resolvedEntryId = if (entryId == null) { - appendConversation(role = role, text = text, isStreaming = !isFinal) + resolvedText = text.trimStart() + appendConversation(role = role, text = resolvedText, isStreaming = !isFinal) } else { - updateConversationEntry(id = entryId, text = text, isStreaming = !isFinal) + resolvedText = updateConversationEntry(id = entryId, text = text, isStreaming = !isFinal) entryId } when (role) { - VoiceConversationRole.User -> realtimeUserEntryId = if (isFinal) null else resolvedEntryId + VoiceConversationRole.User -> { + realtimeUserEntryId = if (isFinal) null else resolvedEntryId + realtimeUserEntryAwaitingFinal = false + realtimeUserEntryAwaitingFinalStartedAtMs = null + } VoiceConversationRole.Assistant -> realtimeAssistantEntryId = if (isFinal) null else resolvedEntryId } + return resolvedText + } + + private fun finishRealtimeConversationEntry(role: VoiceConversationRole) { + val entryId = + when (role) { + VoiceConversationRole.User -> realtimeUserEntryId + VoiceConversationRole.Assistant -> realtimeAssistantEntryId + } ?: return + val current = _conversation.value + val targetIndex = current.indexOfFirst { it.id == entryId } + if (targetIndex >= 0 && current[targetIndex].isStreaming) { + val updated = current.toMutableList() + updated[targetIndex] = current[targetIndex].copy(isStreaming = false) + _conversation.value = updated + if (role == VoiceConversationRole.User) { + realtimeUserEntryAwaitingFinal = true + realtimeUserEntryAwaitingFinalStartedAtMs = SystemClock.elapsedRealtime() + } + } + when (role) { + VoiceConversationRole.User -> Unit + VoiceConversationRole.Assistant -> realtimeAssistantEntryId = null + } + } + + private fun shouldStartNewRealtimeUserEntry( + entryId: String, + incoming: String, + isFinal: Boolean, + ): Boolean { + val entry = _conversation.value.firstOrNull { it.id == entryId } ?: return false + if (entry.isStreaming) return false + val existing = entry.text + if (existing.isBlank() || incoming.isBlank()) return false + if (incoming.firstOrNull()?.isWhitespace() == true) return false + if (incoming == existing || incoming.startsWith(existing) || existing.endsWith(incoming)) return false + if (isFinal && realtimeUserEntryAwaitingFinal) { + val elapsedMs = + realtimeUserEntryAwaitingFinalStartedAtMs?.let { SystemClock.elapsedRealtime() - it } ?: Long.MAX_VALUE + if (elapsedMs <= realtimeUserFinalRewriteGraceMs && looksLikeTranscriptReplacement(existing, incoming)) { + return false + } + } + return true } private fun appendConversation( @@ -1141,7 +1304,7 @@ class TalkModeManager internal constructor( id: String, text: String, isStreaming: Boolean, - ) { + ): String { val current = _conversation.value val targetIndex = when { @@ -1149,14 +1312,112 @@ class TalkModeManager internal constructor( current[current.lastIndex].id == id -> current.lastIndex else -> current.indexOfFirst { it.id == id } } - if (targetIndex < 0) return + if (targetIndex < 0) return text val entry = current[targetIndex] - if (entry.text == text && entry.isStreaming == isStreaming) return + val updatedText = mergeRealtimeTranscriptText(entry.text, text, isFinal = !isStreaming) + if (entry.text == updatedText && entry.isStreaming == isStreaming) return entry.text val updated = current.toMutableList() - updated[targetIndex] = entry.copy(text = text, isStreaming = isStreaming) + updated[targetIndex] = entry.copy(text = updatedText, isStreaming = isStreaming) _conversation.value = updated + return updatedText } + private fun realtimeTranscriptText( + rawText: String?, + isFinal: Boolean, + ): String? { + val text = rawText ?: return null + return text.takeIf { if (isFinal) it.isNotBlank() else it.isNotEmpty() } + } + + private fun mergeRealtimeTranscriptText( + existing: String, + incoming: String, + isFinal: Boolean, + ): String { + if (existing.isBlank()) return incoming.trimStart() + if (incoming.isEmpty()) return existing + if (incoming == existing || existing.endsWith(incoming)) return existing + if (incoming.startsWith(existing)) return incoming + if (incoming.firstOrNull()?.isWhitespace() == true) return existing + incoming + if (isFinal && looksLikeTranscriptReplacement(existing, incoming)) return incoming + val overlap = findTranscriptTextOverlap(existing, incoming) + val suffix = if (overlap > 0) incoming.drop(overlap) else incoming + if (suffix.isEmpty()) return existing + val separator = + if (overlap > 0 || !shouldInsertTranscriptSpace(existing, suffix)) { + "" + } else { + " " + } + return existing + separator + suffix + } + + private fun looksLikeTranscriptReplacement( + existing: String, + incoming: String, + ): Boolean { + val existingWords = transcriptWords(existing) + val incomingWords = transcriptWords(incoming) + if (existingWords.isEmpty() || incomingWords.isEmpty()) return false + if (existingWords[0] != incomingWords[0]) return false + if (existingWords.size > 1 && incomingWords.size > 1 && existingWords[1] == incomingWords[1]) return true + val existingText = normalizeTranscriptText(existing) + val incomingText = normalizeTranscriptText(incoming) + val commonPrefix = commonPrefixLength(existingText, incomingText) + val shortest = minOf(existingText.length, incomingText.length) + return commonPrefix >= 6 && commonPrefix.toDouble() / maxOf(1, shortest).toDouble() >= 0.45 + } + + private fun transcriptWords(value: String): List = + Regex("""[\p{L}\p{N}]+""") + .findAll(value.lowercase(Locale.ROOT)) + .map { it.value } + .toList() + + private fun normalizeTranscriptText(value: String): String = value.lowercase(Locale.ROOT).replace(Regex("""\s+"""), " ").trim() + + private fun commonPrefixLength( + left: String, + right: String, + ): Int { + val max = minOf(left.length, right.length) + var index = 0 + while (index < max && left[index] == right[index]) { + index += 1 + } + return index + } + + private fun findTranscriptTextOverlap( + existing: String, + incoming: String, + ): Int { + val base = existing.lowercase(Locale.ROOT) + val next = incoming.lowercase(Locale.ROOT) + val max = minOf(base.length, next.length) + for (length in max downTo 3) { + if (base.endsWith(next.take(length))) { + return length + } + } + return 0 + } + + private fun shouldInsertTranscriptSpace( + existing: String, + incoming: String, + ): Boolean { + val last = existing.lastOrNull() ?: return false + val first = incoming.firstOrNull() ?: return false + if (last.isWhitespace() || first.isWhitespace()) return false + return first.isLetterOrDigit() && + (last.isLetterOrDigit() || transcriptSpaceAfterPunctuation.contains(last)) + } + + private val transcriptSpaceAfterPunctuation = + setOf('.', '!', '?', ',', ':', ';', ')', ']', '}', '"', '\'', '’', '”') + private fun startListeningInternal(markListening: Boolean) { val r = recognizer ?: return val intent = diff --git a/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt index 1451aebe00e..ebaf1cab0fd 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/gateway/GatewaySessionInvokeTest.kt @@ -27,6 +27,7 @@ import org.junit.runner.RunWith import org.robolectric.RobolectricTestRunner import org.robolectric.RuntimeEnvironment import org.robolectric.annotation.Config +import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicReference @@ -123,6 +124,58 @@ class GatewaySessionInvokeTest { } } + @Test + fun eventsAreDispatchedInWebSocketFrameOrder() = + runBlocking { + val json = testJson() + val connected = CompletableDeferred() + val firstEventStarted = CompletableDeferred() + val releaseFirstEvent = CompletableDeferred() + val secondEventHandled = CompletableDeferred() + val events = CopyOnWriteArrayList() + val lastDisconnect = AtomicReference("") + val server = + startGatewayServer(json) { webSocket, id, method, _ -> + if (method == "connect") { + webSocket.send(connectResponseFrame(id)) + webSocket.send("""{"type":"event","event":"voice.first","payload":{}}""") + webSocket.send("""{"type":"event","event":"voice.second","payload":{}}""") + webSocket.close(1000, "done") + } + } + + val harness = + createNodeHarness( + connected = connected, + lastDisconnect = lastDisconnect, + onEvent = { event, _ -> + if (event == "voice.first") { + firstEventStarted.complete(Unit) + runBlocking { releaseFirstEvent.await() } + } + events += event + if (event == "voice.second") { + secondEventHandled.complete(Unit) + } + }, + ) { GatewaySession.InvokeResult.ok("""{"handled":true}""") } + + try { + connectNodeSession(harness.session, server.port) + awaitConnectedOrThrow(connected, lastDisconnect, server) + withTimeout(TEST_TIMEOUT_MS) { firstEventStarted.await() } + + assertNull(withTimeoutOrNull(200) { secondEventHandled.await() }) + + releaseFirstEvent.complete(Unit) + withTimeout(TEST_TIMEOUT_MS) { secondEventHandled.await() } + assertEquals(listOf("voice.first", "voice.second"), events.toList()) + } finally { + releaseFirstEvent.complete(Unit) + shutdownHarness(harness, server) + } + } + @Test fun connect_usesBootstrapTokenWhenSharedAndDeviceTokensAreAbsent() = runBlocking { @@ -632,6 +685,7 @@ class GatewaySessionInvokeTest { private fun createNodeHarness( connected: CompletableDeferred, lastDisconnect: AtomicReference, + onEvent: (event: String, payloadJson: String?) -> Unit = { _, _ -> }, onInvoke: (GatewaySession.InvokeRequest) -> GatewaySession.InvokeResult, ): NodeHarness { val app = RuntimeEnvironment.getApplication() @@ -648,7 +702,7 @@ class GatewaySessionInvokeTest { onDisconnected = { message -> lastDisconnect.set(message) }, - onEvent = { _, _ -> }, + onEvent = onEvent, onInvoke = onInvoke, ) 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 d50dcc18807..3e1756eeb7c 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 @@ -139,6 +139,194 @@ class TalkModeManagerTest { ) } + @Test + fun realtimeTranscriptDeltasAccumulateVoiceConversation() { + val manager = createManager() + + setPrivateField(manager, "realtimeSessionId", "relay-1") + + manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "The")) + manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = " answer")) + + val entry = manager.conversation.value.single() + assertEquals("The answer", entry.text) + assertTrue(entry.isStreaming) + } + + @Test + fun realtimeTranscriptFragmentsInsertWordSpacing() { + val manager = createManager() + + 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")) + + val entry = manager.conversation.value.single() + assertEquals("Turn off the lights", entry.text) + assertTrue(entry.isStreaming) + } + + @Test + fun realtimeTranscriptFragmentsInsertSpacingAfterPunctuation() { + val manager = createManager() + + setPrivateField(manager, "realtimeSessionId", "relay-1") + + manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "Ready.")) + manager.handleGatewayEvent("talk.event", realtimeTranscriptPayload(role = "assistant", text = "What next?")) + + val entry = manager.conversation.value.single() + assertEquals("Ready. What next?", entry.text) + assertTrue(entry.isStreaming) + } + + @Test + fun realtimeFinalTranscriptCanCompleteDeltaText() { + val manager = createManager() + + 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), + ) + + val entry = manager.conversation.value.single() + assertEquals("The answer", entry.text) + assertFalse(entry.isStreaming) + } + + @Test + fun realtimeAssistantOutputSeparatesNextUserBubble() { + val manager = createManager() + + 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")) + + val entries = manager.conversation.value + assertEquals(3, entries.size) + assertEquals(VoiceConversationRole.User, entries[0].role) + assertEquals("First request", entries[0].text) + assertFalse(entries[0].isStreaming) + assertEquals(VoiceConversationRole.Assistant, entries[1].role) + assertEquals("Checking", entries[1].text) + assertFalse(entries[1].isStreaming) + assertEquals(VoiceConversationRole.User, entries[2].role) + assertEquals("Second request", entries[2].text) + assertTrue(entries[2].isStreaming) + } + + @Test + fun realtimeUserTranscriptRewriteStaysInSameBubble() { + val manager = createManager() + + 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)) + + val entry = manager.conversation.value.single() + assertEquals(VoiceConversationRole.User, entry.role) + assertEquals("Can you check?", entry.text) + assertFalse(entry.isStreaming) + } + + @Test + fun realtimeLateFinalUserTranscriptRewritesBubbleAfterAssistantStarts() { + val manager = createManager() + + 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)) + + val entries = manager.conversation.value + assertEquals(2, entries.size) + assertEquals(VoiceConversationRole.User, entries[0].role) + assertEquals("Can you check?", entries[0].text) + assertFalse(entries[0].isStreaming) + assertEquals(VoiceConversationRole.Assistant, entries[1].role) + assertEquals("Checking", entries[1].text) + } + + @Test + fun realtimeFinalNextUserAfterAssistantStartsCreatesNewBubble() { + val manager = createManager() + + 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)) + + val entries = manager.conversation.value + assertEquals(3, entries.size) + assertEquals(VoiceConversationRole.User, entries[0].role) + assertEquals("First request", entries[0].text) + assertEquals(VoiceConversationRole.Assistant, entries[1].role) + assertEquals("Checking", entries[1].text) + assertEquals(VoiceConversationRole.User, entries[2].role) + assertEquals("Second request", entries[2].text) + assertFalse(entries[2].isStreaming) + } + + @Test + fun realtimeAlternatingTurnsStayInSeparateBubbles() { + val manager = createManager() + + 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, + ), + ) + + val entries = manager.conversation.value + assertEquals(6, entries.size) + assertEquals(VoiceConversationRole.User, entries[0].role) + assertEquals("Hey, what time is it?", entries[0].text) + assertEquals(VoiceConversationRole.Assistant, entries[1].role) + assertEquals("Let me look into that for you. It's currently 7:55 PM UTC.", entries[1].text) + assertEquals(VoiceConversationRole.User, entries[2].role) + assertEquals("How's it going?", entries[2].text) + assertEquals(VoiceConversationRole.Assistant, entries[3].role) + assertEquals("Great! Ready for the next task. What can I do for you?", entries[3].text) + assertEquals(VoiceConversationRole.User, entries[4].role) + assertEquals("Turn on the basement lights", entries[4].text) + assertEquals(VoiceConversationRole.Assistant, entries[5].role) + assertEquals("Got it, let me check on that.", entries[5].text) + assertTrue(entries.none { it.isStreaming }) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun realtimeStartWithoutGatewayTurnsTalkOff() = diff --git a/docs/gateway/config-agents.md b/docs/gateway/config-agents.md index ee964d7e617..601edaa026d 100644 --- a/docs/gateway/config-agents.md +++ b/docs/gateway/config-agents.md @@ -1504,6 +1504,7 @@ Defaults for Talk mode (macOS/iOS/Android). - `speechLocale` sets the BCP 47 locale id used by iOS/macOS Talk speech recognition. Leave unset to use the device default. - `silenceTimeoutMs` controls how long Talk mode waits after user silence before it sends the transcript. Unset keeps the platform default pause window (`700 ms on macOS and Android, 900 ms on iOS`). - `realtime.instructions` appends provider-facing system instructions to OpenClaw's built-in realtime prompt, so voice style can be configured without losing default `openclaw_agent_consult` guidance. +- `realtime.consultRouting` controls Gateway relay fallback when the realtime provider produces a final user transcript without `openclaw_agent_consult`: `provider-direct` preserves direct provider replies, while `force-agent-consult` routes the finalized request through OpenClaw. --- diff --git a/docs/gateway/configuration-reference.md b/docs/gateway/configuration-reference.md index 10b1e01e562..3b154d64a39 100644 --- a/docs/gateway/configuration-reference.md +++ b/docs/gateway/configuration-reference.md @@ -52,6 +52,7 @@ Moved to a dedicated page - see - `talk.consultFastMode`: one-shot fast-mode override for Control UI Talk realtime consults - `talk.speechLocale`: optional BCP 47 locale id for Talk speech recognition on iOS/macOS - `talk.silenceTimeoutMs`: when unset, Talk keeps the platform default pause window before sending the transcript (`700 ms on macOS and Android, 900 ms on iOS`) + - `talk.realtime.consultRouting`: Gateway relay fallback for finalized realtime Talk transcripts that skip `openclaw_agent_consult` ## Tools and custom providers diff --git a/docs/nodes/talk.md b/docs/nodes/talk.md index b578c9fa540..1c3109a9e26 100644 --- a/docs/nodes/talk.md +++ b/docs/nodes/talk.md @@ -111,6 +111,7 @@ Defaults: - `realtime.providers.openai.voice`: built-in OpenAI Realtime voice id. Current `gpt-realtime-2` voices are `alloy`, `ash`, `ballad`, `coral`, `echo`, `sage`, `shimmer`, `verse`, `marin`, and `cedar`; `marin` and `cedar` are recommended for best quality. - `realtime.transport`: `webrtc` and `provider-websocket` are browser realtime transports. Android uses realtime relay only when this is `gateway-relay`; otherwise Android Talk uses its native STT/TTS loop. - `realtime.brain`: `agent-consult` routes realtime tool calls through Gateway policy; `direct-tools` is legacy direct-tool compatibility behavior; `none` is for transcription or external orchestration. +- `realtime.consultRouting`: `provider-direct` preserves the provider's direct reply when it skips `openclaw_agent_consult`; `force-agent-consult` makes Gateway relay route finalized user transcripts through OpenClaw instead. - `realtime.instructions`: appends provider-facing system instructions to OpenClaw's built-in realtime prompt. Use it for voice style and tone; OpenClaw keeps the default `openclaw_agent_consult` guidance. - `talk.catalog` exposes each provider's valid modes, transports, brain strategies, realtime audio formats, and capability flags so first-party Talk clients can avoid unsupported combinations. - Streaming transcription providers are discovered through `talk.catalog.transcription`. The current Gateway relay uses the Voice Call streaming provider config until the dedicated Talk transcription config surface is added. diff --git a/src/config/schema.help.ts b/src/config/schema.help.ts index 27d2756e797..6c643fe6b92 100644 --- a/src/config/schema.help.ts +++ b/src/config/schema.help.ts @@ -171,6 +171,8 @@ export const FIELD_HELP: Record = { "Talk byte/session transport: webrtc, provider-websocket, gateway-relay, or managed-room.", "talk.realtime.brain": "Talk reasoning strategy: agent-consult for Gateway-mediated agent help, direct-tools for local tool calls, or none.", + "talk.realtime.consultRouting": + "Gateway relay fallback for final user transcripts when the realtime provider skips openclaw_agent_consult. provider-direct preserves provider replies; force-agent-consult routes through OpenClaw.", "talk.consultThinkingLevel": "Use this to override the thinking level for the regular agent run behind Talk realtime consults.", "talk.consultFastMode": diff --git a/src/config/schema.labels.ts b/src/config/schema.labels.ts index c29a7c4c9ac..433b357ea5c 100644 --- a/src/config/schema.labels.ts +++ b/src/config/schema.labels.ts @@ -963,6 +963,7 @@ export const FIELD_LABELS: Record = { "talk.realtime.mode": "Talk Realtime Mode", "talk.realtime.transport": "Talk Realtime Transport", "talk.realtime.brain": "Talk Realtime Brain", + "talk.realtime.consultRouting": "Talk Realtime Consult Routing", channels: "Channels", "channels.defaults": "Channel Defaults", "channels.defaults.groupPolicy": "Default Group Policy", diff --git a/src/config/talk.normalize.test.ts b/src/config/talk.normalize.test.ts index feea65df7da..69727cee445 100644 --- a/src/config/talk.normalize.test.ts +++ b/src/config/talk.normalize.test.ts @@ -47,6 +47,7 @@ describe("talk normalization", () => { mode: "realtime", transport: "webrtc", brain: "agent-consult", + consultRouting: "force-agent-consult", }, interruptOnSpeech: true, }); @@ -71,6 +72,7 @@ describe("talk normalization", () => { mode: "realtime", transport: "webrtc", brain: "agent-consult", + consultRouting: "force-agent-consult", }, interruptOnSpeech: true, }); diff --git a/src/config/talk.ts b/src/config/talk.ts index 643eda90b7b..ec97a451d5f 100644 --- a/src/config/talk.ts +++ b/src/config/talk.ts @@ -132,6 +132,12 @@ function normalizeTalkRealtimeConfig(value: unknown): TalkRealtimeConfig | undef ) { normalized.brain = source.brain; } + if ( + source.consultRouting === "provider-direct" || + source.consultRouting === "force-agent-consult" + ) { + normalized.consultRouting = source.consultRouting; + } return Object.keys(normalized).length > 0 ? normalized : undefined; } diff --git a/src/config/types.gateway.ts b/src/config/types.gateway.ts index ce1bd18efae..83cdde9ec48 100644 --- a/src/config/types.gateway.ts +++ b/src/config/types.gateway.ts @@ -62,6 +62,8 @@ export type TalkRealtimeConfig = { transport?: "webrtc" | "provider-websocket" | "gateway-relay" | "managed-room"; /** Tool/agent strategy for realtime sessions. */ brain?: "agent-consult" | "direct-tools" | "none"; + /** How Gateway relay handles final user transcripts when the provider skips a consult. */ + consultRouting?: "provider-direct" | "force-agent-consult"; }; export type ResolvedTalkConfig = { diff --git a/src/config/zod-schema.talk.test.ts b/src/config/zod-schema.talk.test.ts index d0191de799a..799ee79f249 100644 --- a/src/config/zod-schema.talk.test.ts +++ b/src/config/zod-schema.talk.test.ts @@ -37,12 +37,25 @@ describe("OpenClawSchema talk validation", () => { }, }, instructions: "Speak with crisp diction.", + consultRouting: "force-agent-consult", }, }, }), ).not.toThrow(); }); + it("rejects invalid realtime Talk consult routing", () => { + expect(() => + OpenClawSchema.parse({ + talk: { + realtime: { + consultRouting: "always", + }, + }, + }), + ).toThrow(/consultRouting/i); + }); + it.each([ ["boolean", true], ["string", "1500"], diff --git a/src/config/zod-schema.ts b/src/config/zod-schema.ts index e08843784a4..84df1760b36 100644 --- a/src/config/zod-schema.ts +++ b/src/config/zod-schema.ts @@ -279,6 +279,7 @@ const TalkRealtimeSchema = z mode: z.enum(["realtime", "stt-tts", "transcription"]).optional(), transport: z.enum(["webrtc", "provider-websocket", "gateway-relay", "managed-room"]).optional(), brain: z.enum(["agent-consult", "direct-tools", "none"]).optional(), + consultRouting: z.enum(["provider-direct", "force-agent-consult"]).optional(), }) .strict() .superRefine((realtime, ctx) => { diff --git a/src/gateway/server-methods/talk-client.ts b/src/gateway/server-methods/talk-client.ts index 795875fa17c..9c2f796e8c9 100644 --- a/src/gateway/server-methods/talk-client.ts +++ b/src/gateway/server-methods/talk-client.ts @@ -1,5 +1,3 @@ -import { randomUUID } from "node:crypto"; -import { normalizeTalkSection } from "../../config/talk.js"; import { normalizeOptionalLowercaseString, normalizeOptionalString, @@ -7,21 +5,17 @@ import { import { REALTIME_VOICE_AGENT_CONSULT_TOOL, REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, - buildRealtimeVoiceAgentConsultChatMessage, } from "../../talk/agent-consult-tool.js"; import { resolveConfiguredRealtimeVoiceProvider } from "../../talk/provider-resolver.js"; import { ErrorCodes, errorShape, formatValidationErrors, - type ErrorShape, validateTalkClientCreateParams, validateTalkClientToolCallParams, } from "../protocol/index.js"; -import { registerTalkRealtimeRelayAgentRun } from "../talk-realtime-relay.js"; +import { startTalkRealtimeAgentConsult } from "../talk-agent-consult.js"; import { formatForLog } from "../ws-log.js"; -import { chatHandlers } from "./chat.js"; -import { asRecord } from "./record-shared.js"; import { buildRealtimeInstructions, buildRealtimeVoiceLaunchOptions, @@ -30,74 +24,6 @@ import { } from "./talk-shared.js"; import type { GatewayRequestHandlers } from "./types.js"; -async function startRealtimeToolCallAgentConsult(params: { - sessionKey: string; - callId: string; - args: unknown; - relaySessionId?: string; - connId?: string; - request: Parameters[0]; -}): Promise< - { ok: true; runId: string; idempotencyKey: string } | { ok: false; error: ErrorShape } -> { - let message: string; - try { - message = buildRealtimeVoiceAgentConsultChatMessage(params.args); - } catch (err) { - return { ok: false, error: errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)) }; - } - const idempotencyKey = `talk-${params.callId}-${randomUUID()}`; - const normalizedTalk = normalizeTalkSection(params.request.context.getRuntimeConfig().talk); - let chatResponse: { ok: true; result: unknown } | { ok: false; error: ErrorShape } | undefined; - await chatHandlers["chat.send"]({ - ...params.request, - req: { - type: "req", - id: `${params.request.req.id}:talk-tool-call`, - method: "chat.send", - }, - params: { - sessionKey: params.sessionKey, - message, - idempotencyKey, - ...(normalizedTalk?.consultThinkingLevel - ? { thinking: normalizedTalk.consultThinkingLevel } - : {}), - ...(typeof normalizedTalk?.consultFastMode === "boolean" - ? { fastMode: normalizedTalk.consultFastMode } - : {}), - }, - respond: (ok: boolean, result?: unknown, error?: ErrorShape) => { - chatResponse = ok - ? { ok: true, result } - : { - ok: false, - error: error ?? errorShape(ErrorCodes.UNAVAILABLE, "chat.send failed without error"), - }; - }, - } as never); - - if (!chatResponse) { - return { - ok: false, - error: errorShape(ErrorCodes.UNAVAILABLE, "chat.send did not return a realtime tool result"), - }; - } - if (!chatResponse.ok) { - return { ok: false, error: chatResponse.error }; - } - const runId = normalizeOptionalString(asRecord(chatResponse.result)?.runId) ?? idempotencyKey; - if (params.relaySessionId && params.connId) { - registerTalkRealtimeRelayAgentRun({ - relaySessionId: params.relaySessionId, - connId: params.connId, - sessionKey: params.sessionKey, - runId, - }); - } - return { ok: true, runId, idempotencyKey }; -} - export const talkClientHandlers: GatewayRequestHandlers = { "talk.client.create": async ({ params, respond, context }) => { if (!validateTalkClientCreateParams(params)) { @@ -250,13 +176,16 @@ export const talkClientHandlers: GatewayRequestHandlers = { return; } - const result = await startRealtimeToolCallAgentConsult({ + const result = await startTalkRealtimeAgentConsult({ + context: request.context, + client: request.client, + isWebchatConnect: request.isWebchatConnect, + requestId: request.req.id, sessionKey: params.sessionKey, callId: params.callId, args: params.args ?? {}, relaySessionId: normalizeOptionalString(params.relaySessionId), connId: normalizeOptionalString(request.client?.connId), - request, }); if (!result.ok) { respond(false, undefined, result.error); diff --git a/src/gateway/server-methods/talk-session.ts b/src/gateway/server-methods/talk-session.ts index bfdb6a9308f..bcc8c754278 100644 --- a/src/gateway/server-methods/talk-session.ts +++ b/src/gateway/server-methods/talk-session.ts @@ -273,6 +273,8 @@ export const talkSessionHandlers: GatewayRequestHandlers = { tools: [REALTIME_VOICE_AGENT_CONSULT_TOOL], model: launchOptions.model, voice: launchOptions.voice, + forceAgentConsultOnFinalTranscript: + realtimeConfig.consultRouting === "force-agent-consult", }); rememberUnifiedTalkSession(session.relaySessionId, { kind: "realtime-relay", diff --git a/src/gateway/server-methods/talk-shared.ts b/src/gateway/server-methods/talk-shared.ts index 91f0f0466c5..a6441df95c9 100644 --- a/src/gateway/server-methods/talk-shared.ts +++ b/src/gateway/server-methods/talk-shared.ts @@ -131,6 +131,7 @@ export function buildTalkRealtimeConfig(config: OpenClawConfig, requestedProvide mode: normalizeOptionalLowercaseString(talkRealtime?.mode), transport: normalizeOptionalLowercaseString(talkRealtime?.transport), brain: normalizeOptionalLowercaseString(talkRealtime?.brain), + consultRouting: normalizeOptionalLowercaseString(talkRealtime?.consultRouting), }; } @@ -211,7 +212,12 @@ export function resolveConfiguredRealtimeTranscriptionProvider(params: { throw new Error("No realtime transcription provider registered"); } -const DEFAULT_REALTIME_INSTRUCTIONS = `You are OpenClaw's realtime voice interface. Keep spoken replies concise. If the user asks for code, repository state, tools, files, current OpenClaw context, or deeper reasoning, call ${REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME} and then summarize the result naturally.`; +const DEFAULT_REALTIME_INSTRUCTIONS = [ + "You are OpenClaw's realtime voice interface. Keep spoken replies concise.", + `If the user asks for code, repository state, files, current OpenClaw context, tool-backed actions, or deeper reasoning, call ${REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME} and then summarize the result naturally.`, + `Do not claim you cannot use tools, perform actions, or reach OpenClaw unless ${REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME} returns that failure.`, + `When ${REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME} is in progress, speak one brief acknowledgement such as "Let me check that for you", then wait for the final OpenClaw result before answering with the actual result.`, +].join(" "); export function buildRealtimeInstructions(configuredInstructions?: string): string { const extra = normalizeOptionalString(configuredInstructions); diff --git a/src/gateway/server-methods/talk.test.ts b/src/gateway/server-methods/talk.test.ts index 09421e48330..f2978e9604e 100644 --- a/src/gateway/server-methods/talk.test.ts +++ b/src/gateway/server-methods/talk.test.ts @@ -518,6 +518,7 @@ describe("talk.session unified handlers", () => { provider: "openai", providers: { openai: { apiKey: "openai-key" } }, instructions: "Speak warmly.", + consultRouting: "force-agent-consult", }, }, }) as OpenClawConfig, @@ -537,6 +538,9 @@ describe("talk.session unified handlers", () => { expect(relayCreateInput.instructions).toContain( "Additional realtime instructions:\nSpeak warmly.", ); + expect(relayCreateInput.forceAgentConsultOnFinalTranscript).toBe(true); + expect(relayCreateInput.instructions).toContain("tool-backed actions"); + expect(relayCreateInput.instructions).toContain("Let me check that for you"); expectRespondOk(createRespond, { sessionId: "relay-unified-1", relaySessionId: "relay-unified-1", @@ -1256,6 +1260,8 @@ describe("talk.client.create handler", () => { reasoningEffort: "low", }); expect(createInput.instructions).toContain("Additional realtime instructions:\nSpeak warmly."); + expect(createInput.instructions).toContain("tool-backed actions"); + expect(createInput.instructions).toContain("Let me check that for you"); expect(createInput).not.toHaveProperty("provider"); expect(createInput).not.toHaveProperty("providers"); expect(createInput).not.toHaveProperty("transport"); diff --git a/src/gateway/talk-agent-consult.ts b/src/gateway/talk-agent-consult.ts new file mode 100644 index 00000000000..55d5b541678 --- /dev/null +++ b/src/gateway/talk-agent-consult.ts @@ -0,0 +1,91 @@ +import { randomUUID } from "node:crypto"; +import { normalizeTalkSection } from "../config/talk.js"; +import { buildRealtimeVoiceAgentConsultChatMessage } from "../talk/agent-consult-tool.js"; +import { ErrorCodes, errorShape, type ConnectParams, type ErrorShape } from "./protocol/index.js"; +import { chatHandlers } from "./server-methods/chat.js"; +import type { + GatewayClient, + GatewayRequestContext, + GatewayRequestHandlers, +} from "./server-methods/shared-types.js"; +import { registerTalkRealtimeRelayAgentRun } from "./talk-realtime-relay.js"; +import { formatForLog } from "./ws-log.js"; + +export async function startTalkRealtimeAgentConsult(params: { + context: GatewayRequestContext; + client: GatewayClient | null; + isWebchatConnect: (params: ConnectParams | null | undefined) => boolean; + requestId: string; + sessionKey: string; + callId: string; + args: unknown; + relaySessionId?: string; + connId?: string; +}): Promise< + { ok: true; runId: string; idempotencyKey: string } | { ok: false; error: ErrorShape } +> { + let message: string; + try { + message = buildRealtimeVoiceAgentConsultChatMessage(params.args); + } catch (err) { + return { ok: false, error: errorShape(ErrorCodes.INVALID_REQUEST, formatForLog(err)) }; + } + const idempotencyKey = `talk-${params.callId}-${randomUUID()}`; + const normalizedTalk = normalizeTalkSection(params.context.getRuntimeConfig().talk); + let chatResponse: { ok: true; result: unknown } | { ok: false; error: ErrorShape } | undefined; + await chatHandlers["chat.send"]({ + req: { + type: "req", + id: `${params.requestId}:talk-tool-call`, + method: "chat.send", + }, + client: params.client, + isWebchatConnect: params.isWebchatConnect, + context: params.context, + params: { + sessionKey: params.sessionKey, + message, + idempotencyKey, + ...(normalizedTalk?.consultThinkingLevel + ? { thinking: normalizedTalk.consultThinkingLevel } + : {}), + ...(typeof normalizedTalk?.consultFastMode === "boolean" + ? { fastMode: normalizedTalk.consultFastMode } + : {}), + }, + respond: (ok: boolean, result?: unknown, error?: ErrorShape) => { + chatResponse = ok + ? { ok: true, result } + : { + ok: false, + error: error ?? errorShape(ErrorCodes.UNAVAILABLE, "chat.send failed without error"), + }; + }, + } as Parameters[0]); + + if (!chatResponse) { + return { + ok: false, + error: errorShape(ErrorCodes.UNAVAILABLE, "chat.send did not return a realtime tool result"), + }; + } + if (!chatResponse.ok) { + return { ok: false, error: chatResponse.error }; + } + const result = chatResponse.result; + const runId = + result && typeof result === "object" && !Array.isArray(result) + ? typeof (result as Record).runId === "string" + ? (result as Record).runId + : idempotencyKey + : idempotencyKey; + if (params.relaySessionId && params.connId) { + registerTalkRealtimeRelayAgentRun({ + relaySessionId: params.relaySessionId, + connId: params.connId, + sessionKey: params.sessionKey, + runId, + }); + } + return { ok: true, runId, idempotencyKey }; +} diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index 6d1a69ce9aa..16ef7bc2f8f 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -14,6 +14,7 @@ import { describe("talk realtime gateway relay", () => { afterEach(() => { clearTalkRealtimeRelaySessionsForTest(); + vi.useRealTimers(); }); function createIdleRelayProvider(): RealtimeVoiceProviderPlugin { @@ -172,7 +173,7 @@ describe("talk realtime gateway relay", () => { itemId: "item-1", callId: "call-1", name: "openclaw_agent_consult", - args: { question: "what now" }, + args: { question: "hello" }, }); }), sendAudio: vi.fn(), @@ -295,7 +296,7 @@ describe("talk realtime gateway relay", () => { itemId: "item-1", callId: "call-1", name: "openclaw_agent_consult", - args: { question: "what now" }, + args: { question: "hello" }, }); expectRecordFields(toolCallPayload.talkEvent, { type: "tool.call", @@ -342,12 +343,23 @@ describe("talk realtime gateway relay", () => { expect(bridge.submitToolResult).toHaveBeenNthCalledWith( 1, "call-1", + { + status: "working", + tool: "openclaw_agent_consult", + message: + "Tell the person briefly that you are checking, then wait for the final OpenClaw result before answering with the actual result.", + }, + { willContinue: true }, + ); + expect(bridge.submitToolResult).toHaveBeenNthCalledWith( + 2, + "call-1", { status: "working" }, { willContinue: true }, ); - expect(bridge.submitToolResult).toHaveBeenNthCalledWith(2, "call-1", { ok: true }, undefined); + expect(bridge.submitToolResult).toHaveBeenNthCalledWith(3, "call-1", { ok: true }, undefined); expect(bridge.submitToolResult).toHaveBeenNthCalledWith( - 3, + 4, "call-2", { status: "already_delivered" }, { suppressResponse: true }, @@ -386,16 +398,16 @@ describe("talk realtime gateway relay", () => { (payload as Record).type === "toolResult" && (payload as Record).callId === "call-1", ); - expect(toolResultPayloads).toHaveLength(2); + expect(toolResultPayloads).toHaveLength(3); expectRecordFields(toolResultPayloads[0], { relaySessionId: session.relaySessionId, type: "toolResult", callId: "call-1", }); expectRecordFields(toolResultPayloads[0]?.talkEvent, { - type: "tool.result", + type: "tool.progress", callId: "call-1", - final: false, + payload: { name: "openclaw_agent_consult", status: "working" }, }); expectRecordFields(toolResultPayloads[1], { relaySessionId: session.relaySessionId, @@ -403,6 +415,16 @@ describe("talk realtime gateway relay", () => { callId: "call-1", }); expectRecordFields(toolResultPayloads[1]?.talkEvent, { + type: "tool.result", + callId: "call-1", + final: false, + }); + expectRecordFields(toolResultPayloads[2], { + relaySessionId: session.relaySessionId, + type: "toolResult", + callId: "call-1", + }); + expectRecordFields(toolResultPayloads[2]?.talkEvent, { type: "tool.result", callId: "call-1", final: true, @@ -417,6 +439,376 @@ describe("talk realtime gateway relay", () => { expectRecordFields(closePayload.talkEvent, { type: "session.closed", final: true }); }); + it("preserves provider-direct replies unless forced consult routing is configured", async () => { + vi.useFakeTimers(); + + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const bridge = { + supportsToolResultContinuation: true, + connect: vi.fn(async () => undefined), + sendAudio: vi.fn(), + setMediaTimestamp: vi.fn(), + sendUserMessage: vi.fn(), + triggerGreeting: vi.fn(), + handleBargeIn: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + isConnected: vi.fn(() => true), + }; + const provider: RealtimeVoiceProviderPlugin = { + id: "relay-test", + label: "Relay Test", + isConfigured: () => true, + createBridge: (req) => { + bridgeRequest = req; + return bridge; + }, + }; + const events: Array<{ event: string; payload: unknown; connIds: string[] }> = []; + const context = { + broadcastToConnIds: (event: string, payload: unknown, connIds: ReadonlySet) => { + events.push({ event, payload, connIds: [...connIds] }); + }, + } as never; + + const session = createTalkRealtimeRelaySession({ + context, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "be brief", + tools: [], + }); + await Promise.resolve(); + + bridgeRequest?.onTranscript?.("user", "Can you answer directly?", true); + expect(bridge.sendUserMessage).toHaveBeenLastCalledWith("Can you answer directly?"); + expect( + events.some((entry) => { + const payload = entry.payload; + return ( + typeof payload === "object" && + payload !== null && + (payload as Record).type === "toolCall" && + (payload as Record).forced === true + ); + }), + ).toBe(false); + + stopTalkRealtimeRelaySession({ relaySessionId: session.relaySessionId, connId: "conn-1" }); + }); + + it("forces an agent consult when configured and realtime transcript finalizes without a provider tool call", async () => { + vi.useFakeTimers(); + + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const bridge = { + supportsToolResultContinuation: true, + connect: vi.fn(async () => undefined), + sendAudio: vi.fn(), + setMediaTimestamp: vi.fn(), + sendUserMessage: vi.fn(), + triggerGreeting: vi.fn(), + handleBargeIn: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + isConnected: vi.fn(() => true), + }; + const provider: RealtimeVoiceProviderPlugin = { + id: "relay-test", + label: "Relay Test", + isConfigured: () => true, + createBridge: (req) => { + bridgeRequest = req; + return bridge; + }, + }; + const events: Array<{ event: string; payload: unknown; connIds: string[] }> = []; + const context = { + broadcastToConnIds: (event: string, payload: unknown, connIds: ReadonlySet) => { + events.push({ event, payload, connIds: [...connIds] }); + }, + } as never; + + const session = createTalkRealtimeRelaySession({ + context, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "be brief", + tools: [], + forceAgentConsultOnFinalTranscript: true, + }); + await Promise.resolve(); + + bridgeRequest?.onTranscript?.("user", "Can you check this?", true); + expect(bridge.sendUserMessage).not.toHaveBeenCalledWith("Can you check this?"); + + await vi.advanceTimersByTimeAsync(250); + + const forcedToolCall = findEventPayload( + events, + (payload) => payload.type === "toolCall" && payload.forced === true, + ); + expectRecordFields(forcedToolCall, { + relaySessionId: session.relaySessionId, + type: "toolCall", + name: "openclaw_agent_consult", + forced: true, + }); + expectRecordFields(forcedToolCall.args, { + question: "Can you check this?", + responseStyle: "Reply in a concise spoken tone.", + }); + expectRecordFields(forcedToolCall.talkEvent, { type: "tool.call" }); + expectRecordFields((forcedToolCall.talkEvent as Record).payload, { + forced: true, + }); + expect(bridge.handleBargeIn).toHaveBeenCalledWith({ + audioPlaybackActive: true, + force: true, + }); + + const callId = String(forcedToolCall.callId); + submitTalkRealtimeRelayToolResult({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + callId, + result: { status: "working" }, + options: { willContinue: true }, + }); + expect(bridge.sendUserMessage).toHaveBeenLastCalledWith( + "Briefly tell the person that you are checking with OpenClaw. Do not answer the request yet. Wait for the OpenClaw result before giving the actual answer.", + ); + + bridgeRequest?.onToolCall?.({ + itemId: "native-item", + callId: "native-call", + name: "openclaw_agent_consult", + args: { question: "Can you check this?" }, + }); + expect(bridge.submitToolResult).toHaveBeenLastCalledWith( + "native-call", + { + status: "working", + tool: "openclaw_agent_consult", + message: + "Tell the person briefly that you are checking, then wait for the final OpenClaw result before answering with the actual result.", + }, + { willContinue: true }, + ); + + submitTalkRealtimeRelayToolResult({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + callId, + result: { result: "Here is the checked answer." }, + }); + expect(bridge.submitToolResult).toHaveBeenLastCalledWith( + "native-call", + { + status: "already_delivered", + message: "OpenClaw already delivered this consult result internally. Do not repeat it.", + }, + { suppressResponse: true }, + ); + expect(bridge.sendUserMessage).toHaveBeenLastCalledWith( + [ + "OpenClaw finished checking. Speak this result naturally and concisely.", + "Do not mention tool calls, JSON, or internal routing.", + "", + "Here is the checked answer.", + ].join("\n"), + ); + expect( + bridge.submitToolResult.mock.invocationCallOrder[ + bridge.submitToolResult.mock.invocationCallOrder.length - 1 + ], + ).toBeLessThan( + bridge.sendUserMessage.mock.invocationCallOrder[ + bridge.sendUserMessage.mock.invocationCallOrder.length - 1 + ] ?? 0, + ); + expect( + events.some((entry) => { + const payload = entry.payload; + return ( + typeof payload === "object" && + payload !== null && + (payload as Record).type === "toolCall" && + (payload as Record).callId === "native-call" + ); + }), + ).toBe(false); + + bridgeRequest?.onToolCall?.({ + itemId: "native-other-item", + callId: "native-other-call", + name: "openclaw_agent_consult", + args: { question: "Can you check something else?" }, + }); + expect(bridge.submitToolResult).toHaveBeenLastCalledWith( + "native-other-call", + { + status: "working", + tool: "openclaw_agent_consult", + message: + "Tell the person briefly that you are checking, then wait for the final OpenClaw result before answering with the actual result.", + }, + { willContinue: true }, + ); + const nativeOtherToolCall = findEventPayload( + events, + (payload) => payload.type === "toolCall" && payload.callId === "native-other-call", + ); + expectRecordFields(nativeOtherToolCall, { + relaySessionId: session.relaySessionId, + type: "toolCall", + callId: "native-other-call", + name: "openclaw_agent_consult", + args: { question: "Can you check something else?" }, + }); + stopTalkRealtimeRelaySession({ relaySessionId: session.relaySessionId, connId: "conn-1" }); + }); + + it("does not force a duplicate consult after native consult or cancellation", async () => { + vi.useFakeTimers(); + + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const bridge = { + supportsToolResultContinuation: true, + connect: vi.fn(async () => undefined), + sendAudio: vi.fn(), + setMediaTimestamp: vi.fn(), + sendUserMessage: vi.fn(), + triggerGreeting: vi.fn(), + handleBargeIn: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + isConnected: vi.fn(() => true), + }; + const provider: RealtimeVoiceProviderPlugin = { + id: "relay-test", + label: "Relay Test", + isConfigured: () => true, + createBridge: (req) => { + bridgeRequest = req; + return bridge; + }, + }; + const events: Array<{ event: string; payload: unknown; connIds: string[] }> = []; + const context = { + broadcastToConnIds: (event: string, payload: unknown, connIds: ReadonlySet) => { + events.push({ event, payload, connIds: [...connIds] }); + }, + } as never; + + const nativeSession = createTalkRealtimeRelaySession({ + context, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "be brief", + tools: [], + forceAgentConsultOnFinalTranscript: true, + }); + await Promise.resolve(); + bridgeRequest?.onTranscript?.("user", "Can you check this?", true); + bridgeRequest?.onToolCall?.({ + itemId: "native-item", + callId: "native-call", + name: "openclaw_agent_consult", + args: { question: "Can you check this for me?" }, + }); + await vi.advanceTimersByTimeAsync(250); + + expect( + events.some((entry) => { + const payload = entry.payload; + return ( + typeof payload === "object" && + payload !== null && + (payload as Record).type === "toolCall" && + (payload as Record).forced === true + ); + }), + ).toBe(false); + stopTalkRealtimeRelaySession({ + relaySessionId: nativeSession.relaySessionId, + connId: "conn-1", + }); + + const unicodeSession = createTalkRealtimeRelaySession({ + context, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "be brief", + tools: [], + forceAgentConsultOnFinalTranscript: true, + }); + await Promise.resolve(); + bridgeRequest?.onTranscript?.("user", "проверь статус", true); + bridgeRequest?.onToolCall?.({ + itemId: "unicode-native-item", + callId: "unicode-native-call", + name: "openclaw_agent_consult", + args: { question: "проверь статус" }, + }); + await vi.advanceTimersByTimeAsync(250); + expect( + events.some((entry) => { + const payload = entry.payload; + return ( + typeof payload === "object" && + payload !== null && + (payload as Record).type === "toolCall" && + (payload as Record).forced === true + ); + }), + ).toBe(false); + stopTalkRealtimeRelaySession({ + relaySessionId: unicodeSession.relaySessionId, + connId: "conn-1", + }); + + const cancelledSession = createTalkRealtimeRelaySession({ + context, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "be brief", + tools: [], + forceAgentConsultOnFinalTranscript: true, + }); + await Promise.resolve(); + bridgeRequest?.onTranscript?.("user", "Cancel this consult", true); + cancelTalkRealtimeRelayTurn({ + relaySessionId: cancelledSession.relaySessionId, + connId: "conn-1", + reason: "barge-in", + }); + await vi.advanceTimersByTimeAsync(250); + expect( + events.some((entry) => { + const payload = entry.payload; + return ( + typeof payload === "object" && + payload !== null && + (payload as Record).type === "toolCall" && + (payload as Record).forced === true + ); + }), + ).toBe(false); + stopTalkRealtimeRelaySession({ + relaySessionId: cancelledSession.relaySessionId, + connId: "conn-1", + }); + }); + it("rejects relay control from a different connection", () => { const provider: RealtimeVoiceProviderPlugin = { id: "relay-test", diff --git a/src/gateway/talk-realtime-relay.ts b/src/gateway/talk-realtime-relay.ts index 5b5f97904ec..e363b007faa 100644 --- a/src/gateway/talk-realtime-relay.ts +++ b/src/gateway/talk-realtime-relay.ts @@ -1,6 +1,10 @@ import { randomUUID } from "node:crypto"; import type { OpenClawConfig } from "../config/types.js"; import type { RealtimeVoiceProviderPlugin } from "../plugins/types.js"; +import { + REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, + buildRealtimeVoiceAgentConsultWorkingResponse, +} from "../talk/agent-consult-tool.js"; import { recordTalkObservabilityEvent } from "../talk/observability.js"; import { REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, @@ -28,6 +32,35 @@ const MAX_AUDIO_BASE64_BYTES = 512 * 1024; const MAX_RELAY_SESSIONS_PER_CONN = 2; const MAX_RELAY_SESSIONS_GLOBAL = 64; const RELAY_EVENT = "talk.event"; +const FORCED_CONSULT_FALLBACK_DELAY_MS = 200; +const FORCED_CONSULT_NATIVE_DEDUPE_MS = 2_000; +const FORCED_CONSULT_RESULT_MAX_CHARS = 1_800; +const CONSULT_QUESTION_STOPWORDS = new Set([ + "a", + "an", + "and", + "are", + "can", + "check", + "could", + "for", + "in", + "is", + "it", + "look", + "me", + "of", + "on", + "or", + "please", + "see", + "that", + "the", + "this", + "to", + "would", + "you", +]); type TalkRealtimeRelayEventPayload = | { relaySessionId: string; type: "ready" } @@ -49,6 +82,7 @@ type TalkRealtimeRelayEventPayload = callId: string; name: string; args: unknown; + forced?: boolean; } | { relaySessionId: string; type: "toolResult"; callId: string } | { relaySessionId: string; type: "error"; message: string } @@ -56,6 +90,14 @@ type TalkRealtimeRelayEventPayload = type TalkRealtimeRelayEvent = TalkRealtimeRelayEventPayload & { talkEvent?: TalkEvent }; +type ForcedConsultState = { + question: string; + nativeCallIds: Set; + cancelled?: boolean; + completedAtMs?: number; + cleanupTimer?: ReturnType; +}; + type RelaySession = { id: string; connId: string; @@ -65,6 +107,10 @@ type RelaySession = { expiresAtMs: number; cleanupTimer: ReturnType; activeAgentRuns: Map; + forcedConsultTimer?: ReturnType; + pendingForcedConsultQuestion?: string; + lastProviderConsult?: { atMs: number; question?: string }; + forcedConsultCalls: Map; }; type CreateTalkRealtimeRelaySessionParams = { @@ -77,6 +123,7 @@ type CreateTalkRealtimeRelaySessionParams = { tools: RealtimeVoiceTool[]; model?: string; voice?: string; + forceAgentConsultOnFinalTranscript?: boolean; }; type TalkRealtimeRelaySessionResult = { @@ -95,6 +142,187 @@ function formatError(error: unknown): string { return error instanceof Error ? error.message : String(error); } +function readSpeakableToolResultText(result: unknown): string | undefined { + if (!result || typeof result !== "object" || Array.isArray(result)) { + return undefined; + } + const record = result as Record; + const value = + typeof record.text === "string" + ? record.text + : typeof record.result === "string" + ? record.result + : record.error; + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + if (!trimmed) { + return undefined; + } + return trimmed.length <= FORCED_CONSULT_RESULT_MAX_CHARS + ? trimmed + : `${trimmed.slice(0, FORCED_CONSULT_RESULT_MAX_CHARS - 16).trimEnd()} [truncated]`; +} + +function isWorkingToolResult(result: unknown): boolean { + return ( + Boolean(result) && + typeof result === "object" && + !Array.isArray(result) && + (result as Record).status === "working" + ); +} + +function readConsultQuestion(args: unknown): string | undefined { + if (!args || typeof args !== "object" || Array.isArray(args)) { + return undefined; + } + const record = args as Record; + for (const key of ["question", "prompt", "query", "task"]) { + const value = record[key]; + if (typeof value === "string" && value.trim()) { + return value.trim(); + } + } + return undefined; +} + +function normalizeConsultQuestion(value: string | undefined): string | undefined { + return ( + value + ?.toLowerCase() + .replace(/[^\p{L}\p{N}]+/gu, " ") + .replace(/\s+/gu, " ") + .trim() || undefined + ); +} + +function questionTokens(value: string | undefined): Set { + const normalized = normalizeConsultQuestion(value); + if (!normalized) { + return new Set(); + } + return new Set( + normalized + .split(/[^\p{L}\p{N}]+/gu) + .map((token) => token.trim()) + .filter((token) => token.length >= 2 && !CONSULT_QUESTION_STOPWORDS.has(token)), + ); +} + +function consultQuestionsMatch(left: string | undefined, right: string | undefined): boolean { + const normalizedLeft = normalizeConsultQuestion(left); + const normalizedRight = normalizeConsultQuestion(right); + if (!normalizedLeft || !normalizedRight) { + return false; + } + if ( + normalizedLeft === normalizedRight || + normalizedLeft.includes(normalizedRight) || + normalizedRight.includes(normalizedLeft) + ) { + return true; + } + const leftTokens = questionTokens(normalizedLeft); + const rightTokens = questionTokens(normalizedRight); + if (leftTokens.size === 0 || rightTokens.size === 0) { + return false; + } + let overlap = 0; + for (const token of leftTokens) { + if (rightTokens.has(token)) { + overlap += 1; + } + } + return overlap / Math.min(leftTokens.size, rightTokens.size) >= 0.6; +} + +function buildForcedConsultCheckingPrompt(): string { + return [ + "Briefly tell the person that you are checking with OpenClaw.", + "Do not answer the request yet. Wait for the OpenClaw result before giving the actual answer.", + ].join(" "); +} + +function buildForcedConsultSpeechPrompt(text: string): string { + return [ + "OpenClaw finished checking. Speak this result naturally and concisely.", + "Do not mention tool calls, JSON, or internal routing.", + "", + text, + ].join("\n"); +} + +function buildAlreadyDeliveredToolResult(): Record { + return { + status: "already_delivered", + message: "OpenClaw already delivered this consult result internally. Do not repeat it.", + }; +} + +function clearPendingForcedConsult(session: RelaySession): void { + if (!session.forcedConsultTimer) { + return; + } + clearTimeout(session.forcedConsultTimer); + session.forcedConsultTimer = undefined; + session.pendingForcedConsultQuestion = undefined; +} + +function clearForcedConsultStates(session: RelaySession): void { + for (const state of session.forcedConsultCalls.values()) { + if (state.cleanupTimer) { + clearTimeout(state.cleanupTimer); + } + } + session.forcedConsultCalls.clear(); +} + +function cancelForcedConsults(session: RelaySession): void { + for (const state of session.forcedConsultCalls.values()) { + state.cancelled = true; + } +} + +function scheduleForcedConsultCleanup( + session: RelaySession, + callId: string, + state: ForcedConsultState, +): void { + if (state.cleanupTimer) { + clearTimeout(state.cleanupTimer); + } + state.cleanupTimer = setTimeout(() => { + if (session.forcedConsultCalls.get(callId) === state) { + session.forcedConsultCalls.delete(callId); + } + }, FORCED_CONSULT_NATIVE_DEDUPE_MS); + state.cleanupTimer.unref?.(); +} + +function findActiveForcedConsult( + session: RelaySession, + nativeArgs: unknown, +): ForcedConsultState | undefined { + const nativeQuestion = readConsultQuestion(nativeArgs); + if (!nativeQuestion) { + return undefined; + } + const now = Date.now(); + for (const state of session.forcedConsultCalls.values()) { + if (state.cancelled) { + continue; + } + const active = + !state.completedAtMs || now - state.completedAtMs < FORCED_CONSULT_NATIVE_DEDUPE_MS; + if (active && consultQuestionsMatch(state.question, nativeQuestion)) { + return state; + } + } + return undefined; +} + function broadcastToOwner( context: GatewayRequestContext, connId: string, @@ -115,6 +343,8 @@ function abortRelayAgentRuns(session: RelaySession, reason: string): void { } function closeRelaySession(session: RelaySession, reason: "completed" | "error"): void { + clearPendingForcedConsult(session); + clearForcedConsultStates(session); relaySessions.delete(session.id); forgetUnifiedTalkSession(session.id); clearTimeout(session.cleanupTimer); @@ -177,7 +407,6 @@ export function createTalkRealtimeRelaySession( { onEvent: recordTalkObservabilityEvent }, ); let relay: RelaySession | undefined; - let bridgeSession: RealtimeVoiceBridgeSession | undefined; const emit = (event: TalkRealtimeRelayEventPayload, talkEvent?: TalkEventInput) => broadcastToOwner(params.context, params.connId, { ...event, @@ -256,11 +485,37 @@ export function createTalkRealtimeRelaySession( }, ); if (role === "user" && final && text.trim()) { - bridgeSession?.sendUserMessage(text.trim()); + const question = text.trim(); + if (params.forceAgentConsultOnFinalTranscript) { + scheduleForcedAgentConsult(relay, question); + } else { + relay?.bridge.sendUserMessage(question); + } } }, onToolCall: (toolCall) => { const turnId = relay ? ensureRelayTurn(relay) : undefined; + if (relay && toolCall.name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) { + const nativeQuestion = readConsultQuestion(toolCall.args); + relay.lastProviderConsult = { + atMs: Date.now(), + question: nativeQuestion, + }; + if (consultQuestionsMatch(relay.pendingForcedConsultQuestion, nativeQuestion)) { + clearPendingForcedConsult(relay); + } + const forcedConsult = findActiveForcedConsult(relay, toolCall.args); + if (forcedConsult) { + forcedConsult.nativeCallIds.add(toolCall.callId); + if (forcedConsult.completedAtMs) { + submitAlreadyDeliveredToolResult(relay, toolCall.callId, turnId); + } else { + submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId, turnId); + } + return; + } + submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId, turnId); + } emit( { relaySessionId, @@ -291,6 +546,8 @@ export function createTalkRealtimeRelaySession( if (!active) { return; } + clearPendingForcedConsult(active); + clearForcedConsultStates(active); relaySessions.delete(relaySessionId); forgetUnifiedTalkSession(relaySessionId); clearTimeout(active.cleanupTimer); @@ -301,7 +558,6 @@ export function createTalkRealtimeRelaySession( ); }, }); - bridgeSession = bridge; relay = { id: relaySessionId, connId: params.connId, @@ -316,6 +572,7 @@ export function createTalkRealtimeRelaySession( } }, RELAY_SESSION_TTL_MS), activeAgentRuns: new Map(), + forcedConsultCalls: new Map(), }; relay.cleanupTimer.unref?.(); relaySessions.set(relaySessionId, relay); @@ -343,6 +600,105 @@ export function createTalkRealtimeRelaySession( }; } +function scheduleForcedAgentConsult(session: RelaySession | undefined, question: string): void { + if (!session || !question.trim()) { + return; + } + clearPendingForcedConsult(session); + session.pendingForcedConsultQuestion = question; + session.forcedConsultTimer = setTimeout(() => { + session.forcedConsultTimer = undefined; + session.pendingForcedConsultQuestion = undefined; + if (!relaySessions.has(session.id)) { + return; + } + const providerConsult = session.lastProviderConsult; + if ( + providerConsult && + Date.now() - providerConsult.atMs < FORCED_CONSULT_NATIVE_DEDUPE_MS && + consultQuestionsMatch(providerConsult.question, question) + ) { + return; + } + const turnId = ensureRelayTurn(session); + const callId = `forced-consult-${randomUUID()}`; + const itemId = `forced-consult-item-${randomUUID()}`; + session.forcedConsultCalls.set(callId, { question, nativeCallIds: new Set() }); + session.bridge.handleBargeIn({ audioPlaybackActive: true, force: true }); + broadcastToOwner(session.context, session.connId, { + relaySessionId: session.id, + type: "toolCall", + itemId, + callId, + name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, + forced: true, + args: { + question, + context: + "The realtime provider produced a final user transcript without invoking openclaw_agent_consult, so OpenClaw is forcing the consult for realtime Talk.", + responseStyle: "Reply in a concise spoken tone.", + }, + talkEvent: session.talk.emit({ + type: "tool.call", + itemId, + callId, + turnId, + payload: { + name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, + args: { question }, + forced: true, + }, + }), + }); + }, FORCED_CONSULT_FALLBACK_DELAY_MS); + session.forcedConsultTimer.unref?.(); +} + +function submitAlreadyDeliveredToolResult( + session: RelaySession, + callId: string, + turnId = ensureRelayTurn(session), +): void { + const result = buildAlreadyDeliveredToolResult(); + session.bridge.submitToolResult(callId, result, { suppressResponse: true }); + broadcastToOwner(session.context, session.connId, { + relaySessionId: session.id, + type: "toolResult", + callId, + talkEvent: session.talk.emit({ + type: "tool.result", + callId, + turnId, + payload: { name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, result }, + final: true, + }), + }); +} + +function submitRealtimeAgentConsultWorkingResponse( + session: RelaySession, + callId: string, + turnId = ensureRelayTurn(session), +): void { + if (!session.bridge.bridge.supportsToolResultContinuation) { + return; + } + session.bridge.submitToolResult(callId, buildRealtimeVoiceAgentConsultWorkingResponse("person"), { + willContinue: true, + }); + broadcastToOwner(session.context, session.connId, { + relaySessionId: session.id, + type: "toolResult", + callId, + talkEvent: session.talk.emit({ + type: "tool.progress", + callId, + turnId, + payload: { name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, status: "working" }, + }), + }); +} + function ensureRelayTurn(session: RelaySession): string { const turn = session.talk.ensureTurn(); if (turn.event) { @@ -403,6 +759,42 @@ export function submitTalkRealtimeRelayToolResult(params: { options?: RealtimeVoiceToolResultOptions; }): void { const session = getRelaySession(params.relaySessionId, params.connId); + const forcedConsult = session.forcedConsultCalls.get(params.callId); + if (forcedConsult) { + const turnId = ensureRelayTurn(session); + if (forcedConsult.cancelled) { + forcedConsult.completedAtMs = Date.now(); + } else if (isWorkingToolResult(params.result)) { + session.bridge.sendUserMessage(buildForcedConsultCheckingPrompt()); + } else { + forcedConsult.completedAtMs = Date.now(); + const text = readSpeakableToolResultText(params.result); + for (const nativeCallId of forcedConsult.nativeCallIds) { + submitAlreadyDeliveredToolResult(session, nativeCallId, turnId); + } + if (text) { + session.bridge.sendUserMessage(buildForcedConsultSpeechPrompt(text)); + } + scheduleForcedConsultCleanup(session, params.callId, forcedConsult); + } + const final = params.options?.willContinue !== true; + if (forcedConsult.cancelled && final) { + scheduleForcedConsultCleanup(session, params.callId, forcedConsult); + } + broadcastToOwner(session.context, session.connId, { + relaySessionId: session.id, + type: "toolResult", + callId: params.callId, + talkEvent: session.talk.emit({ + type: "tool.result", + callId: params.callId, + turnId, + payload: { result: params.result, forced: true }, + final, + }), + }); + return; + } session.bridge.submitToolResult(params.callId, params.result, params.options); const turnId = ensureRelayTurn(session); const final = params.options?.willContinue !== true; @@ -438,6 +830,8 @@ export function cancelTalkRealtimeRelayTurn(params: { const session = getRelaySession(params.relaySessionId, params.connId); const turnId = ensureRelayTurn(session); const reason = params.reason ?? "client-cancelled"; + clearPendingForcedConsult(session); + cancelForcedConsults(session); session.bridge.handleBargeIn({ audioPlaybackActive: true }); abortRelayAgentRuns(session, reason); const cancelled = session.talk.cancelTurn({ @@ -461,6 +855,8 @@ export function stopTalkRealtimeRelaySession(params: { export function clearTalkRealtimeRelaySessionsForTest(): void { for (const session of relaySessions.values()) { + clearPendingForcedConsult(session); + clearForcedConsultStates(session); clearTimeout(session.cleanupTimer); forgetUnifiedTalkSession(session.id); session.bridge.close(); diff --git a/src/talk/agent-consult-tool.ts b/src/talk/agent-consult-tool.ts index 6e2861d6b2a..7efe14cefd0 100644 --- a/src/talk/agent-consult-tool.ts +++ b/src/talk/agent-consult-tool.ts @@ -26,7 +26,7 @@ export const REALTIME_VOICE_AGENT_CONSULT_TOOL: RealtimeVoiceTool = { type: "function", name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, description: - "Delegate the caller's request to the configured OpenClaw agent for normal tool-backed work, context, memory, or reasoning before speaking.", + "Delegate the caller's request to the configured OpenClaw agent for normal tool-backed work, actions, context, memory, or reasoning before speaking.", parameters: { type: "object", properties: { diff --git a/ui/src/styles/chat/layout.css b/ui/src/styles/chat/layout.css index d8f2c2e16f1..3b73ac86233 100644 --- a/ui/src/styles/chat/layout.css +++ b/ui/src/styles/chat/layout.css @@ -701,6 +701,62 @@ color: var(--text); } +.agent-chat__voice-turns { + display: flex; + flex-direction: column; + gap: 8px; + padding: 0 4px 8px; + background: transparent; +} + +.agent-chat__voice-turn { + display: grid; + grid-template-columns: minmax(56px, max-content) minmax(0, 1fr) 12px; + gap: 8px; + align-items: baseline; + max-width: min(100%, 780px); + padding: 8px 10px; + border: 1px solid color-mix(in srgb, var(--border) 55%, transparent); + border-radius: 8px; + background: color-mix(in srgb, var(--bg) 82%, transparent); +} + +.agent-chat__voice-turn--user { + align-self: flex-end; + background: color-mix(in srgb, var(--accent) 10%, var(--bg)); +} + +.agent-chat__voice-turn--assistant { + align-self: flex-start; +} + +.agent-chat__voice-turn-speaker { + min-width: 0; + overflow: hidden; + color: var(--muted); + font-size: 12px; + font-weight: 700; + text-overflow: ellipsis; + white-space: nowrap; +} + +.agent-chat__voice-turn-text { + min-width: 0; + overflow-wrap: anywhere; + color: var(--text); + font-size: 13px; + line-height: 1.45; + white-space: pre-wrap; +} + +.agent-chat__voice-turn-stream { + width: 6px; + height: 6px; + border-radius: 999px; + background: var(--accent); + opacity: 0.7; +} + .agent-chat__talk-options { display: flex; align-items: flex-start; diff --git a/ui/src/styles/chat/layout.test.ts b/ui/src/styles/chat/layout.test.ts index 34bf71c93f8..052b7eb454a 100644 --- a/ui/src/styles/chat/layout.test.ts +++ b/ui/src/styles/chat/layout.test.ts @@ -87,4 +87,12 @@ describe("chat layout styles", () => { expect(css).toContain(".chat-loading-skeleton .chat-bubble"); expect(css).toContain("width: 100%;"); }); + + it("lets realtime Talk turns flow in the chat thread", () => { + const css = readLayoutCss(); + + expect(css).toContain(".agent-chat__voice-turns"); + expect(css).toContain("background: transparent;"); + expect(css).not.toContain("max-height: min(28vh, 220px);"); + }); }); diff --git a/ui/src/ui/app-lifecycle-connect.node.test.ts b/ui/src/ui/app-lifecycle-connect.node.test.ts index bfce1934167..399f62af583 100644 --- a/ui/src/ui/app-lifecycle-connect.node.test.ts +++ b/ui/src/ui/app-lifecycle-connect.node.test.ts @@ -39,10 +39,12 @@ vi.mock("./app-scroll.ts", () => ({ scheduleLogsScroll: vi.fn(), })); -import { handleConnected } from "./app-lifecycle.ts"; +import { handleConnected, handleUpdated } from "./app-lifecycle.ts"; import { startNodesPolling } from "./app-polling.ts"; +import { scheduleChatScroll } from "./app-scroll.ts"; const startNodesPollingMock = vi.mocked(startNodesPolling); +const scheduleChatScrollMock = vi.mocked(scheduleChatScroll); function createDeferred() { let resolve: (() => void) | undefined; @@ -71,7 +73,7 @@ function createHost() { chatLoading: false, chatMessages: [], chatToolMessages: [], - chatStream: "", + chatStream: "" as string | null, logsAutoFollow: false, logsAtBottom: true, logsEntries: [], @@ -86,6 +88,7 @@ describe("handleConnected", () => { connectGatewayMock.mockReset(); loadBootstrapMock.mockReset(); startNodesPollingMock.mockReset(); + scheduleChatScrollMock.mockReset(); vi.stubGlobal("window", { addEventListener: vi.fn(), }); @@ -146,4 +149,16 @@ describe("handleConnected", () => { handleConnected(nodesHost as never); expect(startNodesPollingMock).toHaveBeenCalledWith(nodesHost); }); + + it("keeps realtime Talk turns pinned in the chat flow", () => { + const host = createHost(); + host.chatStream = null; + + handleUpdated( + host as unknown as Parameters[0], + new Map([["realtimeTalkConversation", []]]), + ); + + expect(scheduleChatScrollMock).toHaveBeenCalledWith(host, true); + }); }); diff --git a/ui/src/ui/app-lifecycle.ts b/ui/src/ui/app-lifecycle.ts index b36179acb2a..e947eed8804 100644 --- a/ui/src/ui/app-lifecycle.ts +++ b/ui/src/ui/app-lifecycle.ts @@ -42,6 +42,8 @@ type LifecycleHost = { realtimeTalkStatus?: string; realtimeTalkDetail?: string | null; realtimeTalkTranscript?: string | null; + realtimeTalkConversation?: unknown[]; + resetRealtimeTalkConversation?: () => void; chatLoading: boolean; chatMessages: unknown[]; chatToolMessages: unknown[]; @@ -132,6 +134,7 @@ export function handleDisconnected(host: LifecycleHost) { host.realtimeTalkStatus = "idle"; host.realtimeTalkDetail = null; host.realtimeTalkTranscript = null; + host.resetRealtimeTalkConversation?.(); host.client?.stop(); host.client = null; host.connected = false; @@ -152,6 +155,7 @@ export function handleUpdated(host: LifecycleHost, changed: Map void; updateRealtimeTalkOptions: (next: Partial) => void; chatManualRefreshInFlight: boolean; chatHeaderControlsHidden: boolean; diff --git a/ui/src/ui/app.talk.test.ts b/ui/src/ui/app.talk.test.ts index 4fb77342a61..cb9cb29c597 100644 --- a/ui/src/ui/app.talk.test.ts +++ b/ui/src/ui/app.talk.test.ts @@ -34,6 +34,7 @@ describe("OpenClawApp Talk controls", () => { lastError: string | null; realtimeTalkActive: boolean; realtimeTalkDetail: string | null; + realtimeTalkConversation: Array<{ id: string; role: string; text: string }>; realtimeTalkStatus: string; realtimeTalkSession: { stop(): void } | null; realtimeTalkTranscript: string | null; @@ -45,6 +46,7 @@ describe("OpenClawApp Talk controls", () => { connected: { value: true, writable: true }, lastError: { value: null, writable: true }, realtimeTalkActive: { value: true, writable: true }, + realtimeTalkConversation: { value: [], writable: true }, realtimeTalkDetail: { value: null, writable: true }, realtimeTalkSession: { value: { stop: staleStop }, writable: true }, realtimeTalkStatus: { value: "error", writable: true }, @@ -63,4 +65,54 @@ describe("OpenClawApp Talk controls", () => { expect(session?.start).toBe(startMock); expect(session?.stop).toBe(stopMock); }); + + it("accumulates Talk transcripts as ordered conversation turns", async () => { + const { OpenClawApp } = await import("./app.ts"); + const app = Object.create(OpenClawApp.prototype) as { + client: unknown; + connected: boolean; + lastError: string | null; + realtimeTalkActive: boolean; + realtimeTalkConversation: Array<{ role: string; text: string; isStreaming: boolean }>; + realtimeTalkDetail: string | null; + realtimeTalkStatus: string; + realtimeTalkSession: { stop(): void } | null; + realtimeTalkTranscript: string | null; + sessionKey: string; + }; + Object.defineProperties(app, { + client: { value: { request: vi.fn() }, writable: true }, + connected: { value: true, writable: true }, + lastError: { value: null, writable: true }, + realtimeTalkActive: { value: false, writable: true }, + realtimeTalkConversation: { value: [], writable: true }, + realtimeTalkDetail: { value: null, writable: true }, + realtimeTalkSession: { value: null, writable: true }, + realtimeTalkStatus: { value: "idle", writable: true }, + realtimeTalkTranscript: { value: null, writable: true }, + sessionKey: { value: "main", writable: true }, + }); + + await OpenClawApp.prototype.toggleRealtimeTalk.call(app as never); + const callbacks = realtimeTalkCtor.mock.calls[0]?.[2] as + | { + onTranscript?: (entry: { + role: "user" | "assistant"; + text: string; + final: boolean; + }) => void; + } + | undefined; + + callbacks?.onTranscript?.({ role: "user", text: "Turn off", final: false }); + callbacks?.onTranscript?.({ role: "user", text: "the lights", final: false }); + callbacks?.onTranscript?.({ role: "assistant", text: "Checking", final: false }); + callbacks?.onTranscript?.({ role: "user", text: "Second request", final: true }); + + expect(app.realtimeTalkConversation).toMatchObject([ + { role: "user", text: "Turn off the lights", isStreaming: false }, + { role: "assistant", text: "Checking", isStreaming: false }, + { role: "user", text: "Second request", isStreaming: false }, + ]); + }); }); diff --git a/ui/src/ui/app.ts b/ui/src/ui/app.ts index 06f18b48668..b5fd3b09bcc 100644 --- a/ui/src/ui/app.ts +++ b/ui/src/ui/app.ts @@ -67,6 +67,12 @@ import { import type { AppViewState } from "./app-view-state.ts"; import { normalizeAssistantIdentity } from "./assistant-identity.ts"; import { exportChatMarkdown } from "./chat/export.ts"; +import { + createRealtimeTalkConversationState, + updateRealtimeTalkConversation, + type RealtimeTalkConversationEntry, + type RealtimeTalkConversationState, +} from "./chat/realtime-talk-conversation.ts"; import { RealtimeTalkSession, type RealtimeTalkLaunchOptions, @@ -245,6 +251,7 @@ export class OpenClawApp extends LitElement { @state() realtimeTalkStatus: RealtimeTalkStatus = "idle"; @state() realtimeTalkDetail: string | null = null; @state() realtimeTalkTranscript: string | null = null; + @state() realtimeTalkConversation: RealtimeTalkConversationEntry[] = []; @state() realtimeTalkOptionsOpen = false; @state() realtimeTalkOptions = { provider: "", @@ -257,6 +264,8 @@ export class OpenClawApp extends LitElement { reasoningEffort: "", }; private realtimeTalkSession: RealtimeTalkSession | null = null; + private realtimeTalkConversationState: RealtimeTalkConversationState = + createRealtimeTalkConversationState(); private nativeBridgeCleanup: (() => void) | null = null; @state() chatManualRefreshInFlight = false; @state() chatHeaderControlsHidden = false; @@ -1067,6 +1076,7 @@ export class OpenClawApp extends LitElement { this.realtimeTalkStatus = "idle"; this.realtimeTalkDetail = null; this.realtimeTalkTranscript = null; + this.resetRealtimeTalkConversation(); return; } } @@ -1078,6 +1088,7 @@ export class OpenClawApp extends LitElement { this.realtimeTalkStatus = "connecting"; this.realtimeTalkDetail = null; this.realtimeTalkTranscript = null; + this.resetRealtimeTalkConversation(); const session = new RealtimeTalkSession( this.client, this.sessionKey, @@ -1091,6 +1102,11 @@ export class OpenClawApp extends LitElement { }, onTranscript: (entry) => { this.realtimeTalkTranscript = `${entry.role === "user" ? "You" : "OpenClaw"}: ${entry.text}`; + this.realtimeTalkConversationState = updateRealtimeTalkConversation( + this.realtimeTalkConversationState, + entry, + ); + this.realtimeTalkConversation = this.realtimeTalkConversationState.entries; }, }, this.buildRealtimeTalkLaunchOptions(), @@ -1110,6 +1126,11 @@ export class OpenClawApp extends LitElement { } } + resetRealtimeTalkConversation() { + this.realtimeTalkConversationState = createRealtimeTalkConversationState(); + this.realtimeTalkConversation = []; + } + async steerQueuedChatMessage(id: string) { await steerQueuedChatMessageInternal( this as unknown as Parameters[0], diff --git a/ui/src/ui/chat/realtime-talk-conversation.test.ts b/ui/src/ui/chat/realtime-talk-conversation.test.ts new file mode 100644 index 00000000000..4915d37c702 --- /dev/null +++ b/ui/src/ui/chat/realtime-talk-conversation.test.ts @@ -0,0 +1,149 @@ +import { describe, expect, it } from "vitest"; +import { + createRealtimeTalkConversationState, + finishRealtimeConversationEntry, + updateRealtimeTalkConversation, +} from "./realtime-talk-conversation.ts"; + +describe("realtime Talk conversation", () => { + it("inserts spacing between adjacent transcript fragments", () => { + let state = createRealtimeTalkConversationState(); + + state = updateRealtimeTalkConversation(state, { + role: "user", + text: "Turn off", + final: false, + nowMs: 1, + }); + state = updateRealtimeTalkConversation(state, { + role: "user", + text: "the lights", + final: false, + nowMs: 2, + }); + + expect(state.entries).toMatchObject([ + { role: "user", text: "Turn off the lights", isStreaming: true }, + ]); + }); + + it("inserts spacing after punctuation-ended transcript fragments", () => { + let state = createRealtimeTalkConversationState(); + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: "Ready.", + final: false, + nowMs: 1, + }); + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: "What next?", + final: false, + nowMs: 2, + }); + + expect(state.entries).toMatchObject([ + { role: "assistant", text: "Ready. What next?", isStreaming: true }, + ]); + }); + + it("keeps a late final rewrite in the original user bubble", () => { + let state = createRealtimeTalkConversationState(); + + state = updateRealtimeTalkConversation(state, { + role: "user", + text: "Can you tack", + final: false, + nowMs: 1, + }); + state = finishRealtimeConversationEntry(state, "user", 2); + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: "Checking", + final: false, + nowMs: 3, + }); + state = updateRealtimeTalkConversation(state, { + role: "user", + text: "Can you check?", + final: true, + nowMs: 4, + }); + + expect(state.entries).toMatchObject([ + { role: "user", text: "Can you check?", isStreaming: false }, + { role: "assistant", text: "Checking", isStreaming: true }, + ]); + }); + + it("creates a new bubble for the next final user turn after assistant output starts", () => { + let state = createRealtimeTalkConversationState(); + + state = updateRealtimeTalkConversation(state, { + role: "user", + text: "First request", + final: false, + nowMs: 1, + }); + state = finishRealtimeConversationEntry(state, "user", 2); + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: "Checking", + final: false, + nowMs: 3, + }); + state = updateRealtimeTalkConversation(state, { + role: "user", + text: "Second request", + final: true, + nowMs: 4, + }); + + expect(state.entries).toMatchObject([ + { role: "user", text: "First request", isStreaming: false }, + { role: "assistant", text: "Checking", isStreaming: false }, + { role: "user", text: "Second request", isStreaming: false }, + ]); + }); + + it("keeps alternating realtime turns as separate bubbles", () => { + let state = createRealtimeTalkConversationState(); + + for (const update of [ + { role: "user" as const, text: "Hey, what time is it?", final: true }, + { + role: "assistant" as const, + text: "Let me look into that for you. It's currently 7:55 PM UTC.", + final: true, + }, + { role: "user" as const, text: "How's it going?", final: true }, + { + role: "assistant" as const, + text: "Great! Ready for the next task. What can I do for you?", + final: true, + }, + { role: "user" as const, text: "Turn on the basement lights", final: true }, + { role: "assistant" as const, text: "Got it, let me check on that.", final: true }, + ]) { + state = updateRealtimeTalkConversation(state, update); + } + + expect(state.entries).toMatchObject([ + { role: "user", text: "Hey, what time is it?", isStreaming: false }, + { + role: "assistant", + text: "Let me look into that for you. It's currently 7:55 PM UTC.", + isStreaming: false, + }, + { role: "user", text: "How's it going?", isStreaming: false }, + { + role: "assistant", + text: "Great! Ready for the next task. What can I do for you?", + isStreaming: false, + }, + { role: "user", text: "Turn on the basement lights", isStreaming: false }, + { role: "assistant", text: "Got it, let me check on that.", isStreaming: false }, + ]); + }); +}); diff --git a/ui/src/ui/chat/realtime-talk-conversation.ts b/ui/src/ui/chat/realtime-talk-conversation.ts new file mode 100644 index 00000000000..36b71af4f39 --- /dev/null +++ b/ui/src/ui/chat/realtime-talk-conversation.ts @@ -0,0 +1,294 @@ +export type RealtimeTalkConversationRole = "user" | "assistant"; + +export type RealtimeTalkConversationEntry = { + id: string; + role: RealtimeTalkConversationRole; + text: string; + isStreaming: boolean; +}; + +export type RealtimeTalkConversationState = { + entries: RealtimeTalkConversationEntry[]; + nextEntryId: number; + userEntryId: string | null; + userEntryAwaitingFinal: boolean; + userEntryAwaitingFinalStartedAtMs: number | null; + assistantEntryId: string | null; +}; + +export type RealtimeTalkTranscriptUpdate = { + role: RealtimeTalkConversationRole; + text: string; + final: boolean; + nowMs?: number; +}; + +const MAX_CONVERSATION_ENTRIES = 60; +const USER_FINAL_REWRITE_GRACE_MS = 1_500; + +export function createRealtimeTalkConversationState(): RealtimeTalkConversationState { + return { + entries: [], + nextEntryId: 1, + userEntryId: null, + userEntryAwaitingFinal: false, + userEntryAwaitingFinalStartedAtMs: null, + assistantEntryId: null, + }; +} + +export function updateRealtimeTalkConversation( + state: RealtimeTalkConversationState, + update: RealtimeTalkTranscriptUpdate, +): RealtimeTalkConversationState { + const text = update.text; + if (update.final ? text.trim() === "" : text === "") { + return state; + } + const nowMs = update.nowMs ?? Date.now(); + if (update.role === "assistant") { + const preparedState = finishRealtimeConversationEntry(state, "user", nowMs); + return upsertRealtimeConversationEntry( + preparedState, + update.role, + preparedState.assistantEntryId, + text, + update.final, + nowMs, + ); + } + const entryId = state.userEntryId; + const shouldStartNewUserEntry = + entryId !== null && shouldStartNewRealtimeUserEntry(state, entryId, text, update.final, nowMs); + const assistantClosedState = + entryId === null || shouldStartNewUserEntry + ? finishRealtimeConversationEntry(state, "assistant", nowMs) + : state; + const nextState = + shouldStartNewUserEntry && entryId !== null + ? { + ...finishRealtimeConversationEntry(assistantClosedState, "user", nowMs), + userEntryId: null, + userEntryAwaitingFinal: false, + userEntryAwaitingFinalStartedAtMs: null, + } + : assistantClosedState; + return upsertRealtimeConversationEntry( + nextState, + update.role, + shouldStartNewUserEntry ? null : entryId, + text, + update.final, + nowMs, + ); +} + +function upsertRealtimeConversationEntry( + state: RealtimeTalkConversationState, + role: RealtimeTalkConversationRole, + entryId: string | null, + text: string, + isFinal: boolean, + nowMs: number, +): RealtimeTalkConversationState { + if (entryId === null) { + const id = `rt-${state.nextEntryId}`; + const entries = [ + ...state.entries, + { id, role, text: text.trimStart(), isStreaming: !isFinal }, + ].slice(-MAX_CONVERSATION_ENTRIES); + return rememberRealtimeConversationEntry( + { ...state, entries, nextEntryId: state.nextEntryId + 1 }, + role, + id, + isFinal, + nowMs, + ); + } + + const targetIndex = state.entries.findIndex((entry) => entry.id === entryId); + if (targetIndex === -1) { + return upsertRealtimeConversationEntry(state, role, null, text, isFinal, nowMs); + } + const entry = state.entries[targetIndex]; + const updatedText = mergeRealtimeTranscriptText(entry.text, text, isFinal); + const entries = + entry.text === updatedText && entry.isStreaming === !isFinal + ? state.entries + : state.entries.map((candidate, index) => + index === targetIndex + ? { ...candidate, text: updatedText, isStreaming: !isFinal } + : candidate, + ); + return rememberRealtimeConversationEntry({ ...state, entries }, role, entryId, isFinal, nowMs); +} + +function rememberRealtimeConversationEntry( + state: RealtimeTalkConversationState, + role: RealtimeTalkConversationRole, + entryId: string, + isFinal: boolean, + _nowMs: number, +): RealtimeTalkConversationState { + if (role === "user") { + return { + ...state, + userEntryId: isFinal ? null : entryId, + userEntryAwaitingFinal: false, + userEntryAwaitingFinalStartedAtMs: null, + }; + } + return { ...state, assistantEntryId: isFinal ? null : entryId }; +} + +export function finishRealtimeConversationEntry( + state: RealtimeTalkConversationState, + role: RealtimeTalkConversationRole, + nowMs: number = Date.now(), +): RealtimeTalkConversationState { + const entryId = role === "user" ? state.userEntryId : state.assistantEntryId; + if (entryId === null) { + return state; + } + const entries = state.entries.map((entry) => + entry.id === entryId && entry.isStreaming ? { ...entry, isStreaming: false } : entry, + ); + if (role === "user") { + return { + ...state, + entries, + userEntryAwaitingFinal: true, + userEntryAwaitingFinalStartedAtMs: nowMs, + }; + } + return { ...state, entries, assistantEntryId: null }; +} + +function shouldStartNewRealtimeUserEntry( + state: RealtimeTalkConversationState, + entryId: string, + incoming: string, + isFinal: boolean, + nowMs: number, +): boolean { + const entry = state.entries.find((candidate) => candidate.id === entryId); + if (!entry || entry.isStreaming) { + return false; + } + const existing = entry.text; + if (existing.trim() === "" || incoming.trim() === "") { + return false; + } + if (incoming[0] && /\s/.test(incoming[0])) { + return false; + } + if (incoming === existing || incoming.startsWith(existing) || existing.endsWith(incoming)) { + return false; + } + if (isFinal && state.userEntryAwaitingFinal) { + const elapsed = + state.userEntryAwaitingFinalStartedAtMs === null + ? Number.POSITIVE_INFINITY + : nowMs - state.userEntryAwaitingFinalStartedAtMs; + if ( + elapsed <= USER_FINAL_REWRITE_GRACE_MS && + looksLikeTranscriptReplacement(existing, incoming) + ) { + return false; + } + } + return true; +} + +export function mergeRealtimeTranscriptText( + existing: string, + incoming: string, + isFinal: boolean, +): string { + if (existing.trim() === "") { + return incoming.trimStart(); + } + if (incoming === "") { + return existing; + } + if (incoming === existing || existing.endsWith(incoming)) { + return existing; + } + if (incoming.startsWith(existing)) { + return incoming; + } + if (incoming[0] && /\s/.test(incoming[0])) { + return `${existing}${incoming}`; + } + if (isFinal && looksLikeTranscriptReplacement(existing, incoming)) { + return incoming; + } + const overlap = findTextOverlap(existing, incoming); + const suffix = overlap > 0 ? incoming.slice(overlap) : incoming; + if (suffix === "") { + return existing; + } + const separator = overlap > 0 || !shouldInsertTranscriptSpace(existing, suffix) ? "" : " "; + return `${existing}${separator}${suffix}`; +} + +function looksLikeTranscriptReplacement(existing: string, incoming: string): boolean { + const existingWords = transcriptWords(existing); + const incomingWords = transcriptWords(incoming); + if (existingWords.length === 0 || incomingWords.length === 0) { + return false; + } + if (existingWords[0] !== incomingWords[0]) { + return false; + } + if ( + existingWords.length > 1 && + incomingWords.length > 1 && + existingWords[1] === incomingWords[1] + ) { + return true; + } + const existingText = normalizeTranscriptText(existing); + const incomingText = normalizeTranscriptText(incoming); + const commonPrefix = commonPrefixLength(existingText, incomingText); + const shortest = Math.min(existingText.length, incomingText.length); + return commonPrefix >= 6 && commonPrefix / Math.max(1, shortest) >= 0.45; +} + +function transcriptWords(value: string): string[] { + return [...value.toLowerCase().matchAll(/[\p{L}\p{N}]+/gu)].map((match) => match[0]); +} + +function normalizeTranscriptText(value: string): string { + return value.toLowerCase().replace(/\s+/g, " ").trim(); +} + +function commonPrefixLength(left: string, right: string): number { + const max = Math.min(left.length, right.length); + let index = 0; + while (index < max && left[index] === right[index]) { + index += 1; + } + return index; +} + +function findTextOverlap(base: string, next: string): number { + const normalizedBase = base.toLowerCase(); + const normalizedNext = next.toLowerCase(); + const max = Math.min(normalizedBase.length, normalizedNext.length); + for (let length = max; length >= 3; length -= 1) { + if (normalizedBase.endsWith(normalizedNext.slice(0, length))) { + return length; + } + } + return 0; +} + +function shouldInsertTranscriptSpace(base: string, next: string): boolean { + const last = base.at(-1); + const first = next[0]; + if (!last || !first || /\s/.test(last) || /\s/.test(first)) { + return false; + } + return /[\p{L}\p{N}.!?,:;)\]}"'’”]/u.test(last) && /[\p{L}\p{N}]/u.test(first); +} diff --git a/ui/src/ui/chat/realtime-talk-gateway-relay.ts b/ui/src/ui/chat/realtime-talk-gateway-relay.ts index 5a4aab0a4c0..d4a6442caaa 100644 --- a/ui/src/ui/chat/realtime-talk-gateway-relay.ts +++ b/ui/src/ui/chat/realtime-talk-gateway-relay.ts @@ -28,6 +28,7 @@ type GatewayRelayEvent = { callId?: string; name?: string; args?: unknown; + forced?: boolean; } | { type?: "error"; message?: string } | { type?: "close"; reason?: string } @@ -243,6 +244,18 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport const abortController = new AbortController(); this.consultAbortControllers.add(abortController); try { + if (event.forced) { + this.submitToolResult( + callId, + { + status: "working", + tool: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, + message: + "Tell the person briefly that you are checking, then wait for the final OpenClaw result before answering with the actual result.", + }, + { willContinue: true }, + ); + } await submitRealtimeTalkConsult({ ctx: this.ctx, callId, @@ -256,11 +269,16 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport } } - private submitToolResult(callId: string, result: unknown): void { + private submitToolResult( + callId: string, + result: unknown, + options?: { suppressResponse?: boolean; willContinue?: boolean }, + ): void { void this.ctx.client.request("talk.session.submitToolResult", { sessionId: this.session.relaySessionId, callId, result, + ...(options ? { options } : {}), }); } diff --git a/ui/src/ui/realtime-talk-gateway-relay.test.ts b/ui/src/ui/realtime-talk-gateway-relay.test.ts index ead93c961bb..cbf1b249511 100644 --- a/ui/src/ui/realtime-talk-gateway-relay.test.ts +++ b/ui/src/ui/realtime-talk-gateway-relay.test.ts @@ -413,6 +413,49 @@ describe("GatewayRelayRealtimeTalkTransport", () => { transport.stop(); }); + it("submits an interim working result for forced consult tool calls", async () => { + const client = createClient(); + vi.mocked(client.request).mockImplementation(async (method) => { + if (method === "talk.client.toolCall") { + return { runId: "run-1" }; + } + return {}; + }); + const transport = new GatewayRelayRealtimeTalkTransport(createSession(), { + callbacks: {}, + client, + sessionKey: "main", + }); + + await transport.start(); + emitGatewayFrame({ + event: "talk.event", + payload: { + relaySessionId: "relay-1", + type: "toolCall", + callId: "call-1", + name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, + forced: true, + args: { question: "status?" }, + }, + }); + + await vi.waitFor(() => + expect(client.request).toHaveBeenCalledWith("talk.session.submitToolResult", { + sessionId: "relay-1", + callId: "call-1", + result: { + status: "working", + tool: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, + message: + "Tell the person briefly that you are checking, then wait for the final OpenClaw result before answering with the actual result.", + }, + options: { willContinue: true }, + }), + ); + transport.stop(); + }); + it("aborts in-flight consults when the relay transport stops", async () => { const client = createClient(); vi.mocked(client.request).mockImplementation(async (method, params) => { diff --git a/ui/src/ui/views/chat.test.ts b/ui/src/ui/views/chat.test.ts index f5c931d69d9..7cea94781f1 100644 --- a/ui/src/ui/views/chat.test.ts +++ b/ui/src/ui/views/chat.test.ts @@ -546,6 +546,32 @@ afterEach(() => { }); describe("chat loading skeleton", () => { + it("renders realtime Talk transcript as ordered voice turns", () => { + const container = renderChatView({ + realtimeTalkActive: true, + realtimeTalkConversation: [ + { id: "u1", role: "user", text: "Turn off the lights", isStreaming: false }, + { id: "a1", role: "assistant", text: "Checking", isStreaming: true }, + { id: "u2", role: "user", text: "Second request", isStreaming: false }, + ], + }); + + const turns = [...container.querySelectorAll(".agent-chat__voice-turn")]; + expect(turns.map((turn) => turn.getAttribute("data-role"))).toEqual([ + "user", + "assistant", + "user", + ]); + expect(turns.map((turn) => turn.textContent?.replace(/\s+/g, " ").trim())).toEqual([ + "You Turn off the lights", + "Val Checking", + "You Second request", + ]); + expect(container.querySelector(".chat-thread-inner .agent-chat__voice-turns")).not.toBeNull(); + expect(container.querySelector(".agent-chat__input .agent-chat__voice-turns")).toBeNull(); + expect(container.querySelector(".agent-chat__welcome")).toBeNull(); + }); + it("shows the skeleton while the initial history load has no rendered content", () => { const container = renderChatView({ loading: true }); diff --git a/ui/src/ui/views/chat.ts b/ui/src/ui/views/chat.ts index a6764c63ca3..b4bf910bbc0 100644 --- a/ui/src/ui/views/chat.ts +++ b/ui/src/ui/views/chat.ts @@ -28,6 +28,7 @@ import { import type { ChatInputHistoryKeyInput, ChatInputHistoryKeyResult } from "../chat/input-history.ts"; import { PinnedMessages } from "../chat/pinned-messages.ts"; import { getPinnedMessageSummary } from "../chat/pinned-summary.ts"; +import type { RealtimeTalkConversationEntry } from "../chat/realtime-talk-conversation.ts"; import type { RealtimeTalkStatus } from "../chat/realtime-talk.ts"; import { renderChatRunControls } from "../chat/run-controls.ts"; import type { ChatRunUiStatus } from "../chat/run-lifecycle.ts"; @@ -100,6 +101,7 @@ export type ChatProps = { realtimeTalkStatus?: RealtimeTalkStatus; realtimeTalkDetail?: string | null; realtimeTalkTranscript?: string | null; + realtimeTalkConversation?: RealtimeTalkConversationEntry[]; realtimeTalkOptionsOpen?: boolean; realtimeTalkOptions?: { provider: string; @@ -329,6 +331,40 @@ function renderRealtimeTalkOptions(props: ChatProps) { `; } +function renderRealtimeTalkConversation(props: ChatProps) { + const entries = props.realtimeTalkConversation ?? []; + if (entries.length === 0) { + return nothing; + } + return html` +
+ ${repeat( + entries, + (entry) => entry.id, + (entry) => { + const label = + entry.role === "user" ? props.userName?.trim() || "You" : props.assistantName; + return html` +
+ ${label} + ${entry.text} + ${entry.isStreaming + ? html`` + : nothing} +
+ `; + }, + )} +
+ `; +} + interface ChatEphemeralState { slashMenuOpen: boolean; slashMenuItems: SlashCommandDef[]; @@ -1042,7 +1078,8 @@ export function renderChat(props: ChatProps) { expandedToolCards.set(toolCardId, !expandedToolCards.get(toolCardId)); requestUpdate(); }; - const isEmpty = chatItems.length === 0 && !props.loading; + const hasRealtimeTalkConversation = (props.realtimeTalkConversation?.length ?? 0) > 0; + const isEmpty = chatItems.length === 0 && !props.loading && !hasRealtimeTalkConversation; const showLoadingSkeleton = props.loading && chatItems.length === 0; const thread = html` @@ -1191,6 +1228,7 @@ export function renderChat(props: ChatProps) { return nothing; }, )} + ${renderRealtimeTalkConversation(props)} `; @@ -1447,7 +1485,9 @@ export function renderChat(props: ChatProps) { ? html`
${props.realtimeTalkDetail ?? - props.realtimeTalkTranscript ?? + ((props.realtimeTalkConversation?.length ?? 0) === 0 + ? props.realtimeTalkTranscript + : null) ?? (props.realtimeTalkStatus === "thinking" ? "Asking OpenClaw..." : props.realtimeTalkStatus === "connecting"