diff --git a/apps/.i18n/native-source.json b/apps/.i18n/native-source.json index 8baf031c8438..b84684c5adc3 100644 --- a/apps/.i18n/native-source.json +++ b/apps/.i18n/native-source.json @@ -29953,6 +29953,10 @@ { "kind": "ui-localized-call", "path": "apps/ios/Sources/Design/SettingsSystemAgentChat.swift" + }, + { + "kind": "ui-localized-call", + "path": "apps/ios/Sources/Voice/TalkModeManager.swift" } ] }, @@ -39426,6 +39430,28 @@ } ] }, + { + "id": "native.apple.48e407d88c565c02", + "source": "Realtime disconnected repeatedly — using native speech", + "surface": "apple", + "sites": [ + { + "kind": "ui-localized-call", + "path": "apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift" + } + ] + }, + { + "id": "native.apple.ab73f15b8147eb41", + "source": "Realtime disconnected — reconnecting…", + "surface": "apple", + "sites": [ + { + "kind": "ui-localized-call", + "path": "apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift" + } + ] + }, { "id": "native.apple.ca3a762269827f37", "source": "Realtime failed", @@ -39452,6 +39478,17 @@ } ] }, + { + "id": "native.apple.b5b843b6a0a96cc5", + "source": "Realtime microphone became unavailable: %@", + "surface": "apple", + "sites": [ + { + "kind": "ui-localized-call", + "path": "apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift" + } + ] + }, { "id": "native.apple.83f84908b64f015c", "source": "Realtime tool call did not return a run id", 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 cdc9de6649b8..a099c4252376 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 @@ -385,6 +385,8 @@ class TalkModeManager internal constructor( private val pendingRealtimePlaybackMarks = LinkedHashMap() @Volatile private var pendingRealtimeOutputClear: CompletableDeferred? = null + + @Volatile private var realtimeOutputTurnId: String? = null private val realtimeOutputCancellationMutex = Mutex() @Volatile @@ -1305,7 +1307,7 @@ class TalkModeManager internal constructor( } } - private fun shouldAppendRealtimeCapturedFrame(length: Int): Boolean = !isRealtimePlaybackActive() && length > 0 + private fun shouldAppendRealtimeCapturedFrame(length: Int): Boolean = pendingRealtimeOutputClear == null && !isRealtimePlaybackActive() && length > 0 private fun isRealtimePlaybackActive(): Boolean = _isSpeaking.value || SystemClock.elapsedRealtime() < realtimePlaybackEndsAtMs @@ -1338,6 +1340,9 @@ class TalkModeManager internal constructor( } "audio" -> { if (realtimeOutputSuppressed) return + val turnId = obj["talkEvent"].asObjectOrNull()?.get("turnId").asStringOrNull() ?: return + if (turnId.isBlank()) return + realtimeOutputTurnId = turnId finishRealtimeConversationEntry(VoiceConversationRole.User) val audioBase64 = obj["audioBase64"].asStringOrNull() ?: return val bytes = @@ -1350,8 +1355,11 @@ class TalkModeManager internal constructor( playRealtimeAudio(bytes) } "clear" -> { + val turnId = obj["talkEvent"].asObjectOrNull()?.get("turnId").asStringOrNull() + if (!turnId.isNullOrBlank() && turnId != realtimeOutputTurnId) return val marks = takePendingRealtimePlaybackMarks() stopRealtimePlayback() + realtimeOutputTurnId = null acknowledgeRealtimePlaybackMarks(marks) pendingRealtimeOutputClear?.complete(Unit) } @@ -1684,6 +1692,7 @@ class TalkModeManager internal constructor( currentSessionId to currentCaptureJobs } realtimeOutputSuppressed = false + realtimeOutputTurnId = null pendingRealtimeOutputClear?.cancel() pendingRealtimeOutputClear = null if (cancelCapture) { @@ -2915,6 +2924,7 @@ class TalkModeManager internal constructor( private suspend fun cancelRealtimeOutput(reason: String): Boolean = realtimeOutputCancellationMutex.withLock { val sessionId = realtimeSessionId ?: return@withLock true + val turnId = realtimeOutputTurnId val clear = CompletableDeferred() pendingRealtimeOutputClear = clear try { @@ -2922,8 +2932,11 @@ class TalkModeManager internal constructor( buildJsonObject { put("sessionId", JsonPrimitive(sessionId)) put("reason", JsonPrimitive(reason)) + if (turnId != null) put("turnId", JsonPrimitive(turnId)) } - requestGateway("talk.session.cancelOutput", params.toString(), timeoutMs = 5_000) + val result = + json.parseToJsonElement(requestGateway("talk.session.cancelOutput", params.toString(), timeoutMs = 5_000)).asObjectOrNull() + if (result?.get("status").asStringOrNull() != "applied") clear.complete(Unit) // The response confirms provider cancellation; clear confirms that the // old playback boundary reached Android before capture can resume. withTimeout(2_000) { clear.await() } 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 3831d54ff30a..5ce7d595136a 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 @@ -789,7 +789,7 @@ class TalkModeManagerTest { } @Test - fun realtimeAudioFramesStreamUntilPlaybackStarts() { + fun realtimeAudioFramesStreamUntilPlaybackOrCancellationStarts() { val manager = createManager() assertFalse(shouldAppendRealtimeCapturedFrame(manager, 0)) @@ -803,6 +803,10 @@ class TalkModeManagerTest { setPrivateField(manager, "realtimePlaybackEndsAtMs", SystemClock.elapsedRealtime() - 1) assertTrue(shouldAppendRealtimeCapturedFrame(manager, 4_800)) + setPrivateField(manager, "pendingRealtimeOutputClear", CompletableDeferred()) + assertFalse(shouldAppendRealtimeCapturedFrame(manager, 4_800)) + setPrivateField(manager, "pendingRealtimeOutputClear", null) + assertTrue(shouldAppendRealtimeCapturedFrame(manager, 4_800)) } @Test diff --git a/apps/ios/Sources/Voice/TalkModeManager.swift b/apps/ios/Sources/Voice/TalkModeManager.swift index abf0a9e36d45..34765869c303 100644 --- a/apps/ios/Sources/Voice/TalkModeManager.swift +++ b/apps/ios/Sources/Voice/TalkModeManager.swift @@ -2179,7 +2179,8 @@ final class TalkModeManager: NSObject { return .started } guard let gateway else { - return .unavailable(realtimeIssue(message: "Gateway not connected", phase: "start")) + return .unavailable( + realtimeIssue(message: String(localized: "Gateway is not connected"), phase: "start")) } let startedAt = Self.nowSeconds() if self.prefetchedRealtimeSession == nil, let prefetchTask = realtimePrefetchTask { @@ -2310,7 +2311,8 @@ final class TalkModeManager: NSObject { private func startRealtimeRelayIfAvailable(attemptID: Int) async -> RealtimeStartResult { guard let gateway else { - return .unavailable(realtimeIssue(message: "Gateway not connected", phase: "start")) + return .unavailable( + realtimeIssue(message: String(localized: "Gateway is not connected"), phase: "start")) } guard self.foregroundAudioCaptureAllowed else { self.setStatus( @@ -2322,7 +2324,8 @@ final class TalkModeManager: NSObject { } guard self.isCurrentStartAttempt(attemptID) else { return .ignored } guard let gatewayRoute = await gateway.currentRoute() else { - return .unavailable(realtimeIssue(message: "Gateway not connected", phase: "start")) + return .unavailable( + realtimeIssue(message: String(localized: "Gateway is not connected"), phase: "start")) } guard self.isCurrentStartAttempt(attemptID) else { return .ignored } if self.realtimeRelaySession != nil { diff --git a/apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift b/apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift index 1f1059c6272a..ac0a04212bad 100644 --- a/apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift +++ b/apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift @@ -265,7 +265,9 @@ final class MacRealtimeTalkAudioCapture: RealtimeTalkAudioCapturing { "realtime input restart failed: \(error.localizedDescription, privacy: .public)") let onFailure = self.onFailure self.stop() - onFailure?("Realtime microphone became unavailable: \(error.localizedDescription)") + onFailure?(String( + format: String(localized: "Realtime microphone became unavailable: %@"), + error.localizedDescription)) } } diff --git a/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift b/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift index 9887a2f8b311..de4243174abf 100644 --- a/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift +++ b/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift @@ -40,6 +40,7 @@ extension TalkModeRuntime { lifecycleGeneration: generation, relayGeneration: relayGeneration, start: { session in try await session.start() }) + self.realtimeSessionReadyAt = Date() self.phase = .listening await MainActor.run { TalkModeController.shared.updatePartialTranscript("") @@ -172,9 +173,6 @@ extension TalkModeRuntime { self.realtimeSession != nil else { return } self.logger.debug("talk realtime status=\(status, privacy: .public)") - if status == "Listening (Realtime)", self.realtimeSessionReadyAt == nil { - self.realtimeSessionReadyAt = Date() - } } private func handleRealtimeIssue(_ issue: RealtimeTalkRelayIssue, relayGeneration: UInt64) async { @@ -259,7 +257,8 @@ extension TalkModeRuntime { let lifecycleGeneration = self.lifecycleGeneration if let delay = Self.realtimeRestartDelayNanoseconds(attempt: attempt) { await MainActor.run { - TalkModeController.shared.updatePartialTranscript("Realtime disconnected — reconnecting…") + TalkModeController.shared.updatePartialTranscript( + String(localized: "Realtime disconnected — reconnecting…")) } self.scheduleRealtimeRecovery( after: delay, @@ -269,7 +268,7 @@ extension TalkModeRuntime { self.bypassRealtimeOnNextStart = true await MainActor.run { TalkModeController.shared.updatePartialTranscript( - "Realtime disconnected repeatedly — using native speech") + String(localized: "Realtime disconnected repeatedly — using native speech")) } self.scheduleRealtimeRecovery( after: nil, diff --git a/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift b/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift index 041f1ac82430..a45bc7d455f3 100644 --- a/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift +++ b/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift @@ -22,12 +22,6 @@ actor TalkModeRuntime { case fallback } - struct NativeFallbackOwner { - let lifecycleGeneration: Int - let recognitionGeneration: Int - let realtimeRelayGeneration: UInt64 - } - let logger = Logger(subsystem: "ai.openclaw", category: "talk.runtime") let ttsLogger = Logger(subsystem: "ai.openclaw", category: "talk.tts") static let defaultModelIdFallback = "eleven_v3" @@ -246,26 +240,6 @@ actor TalkModeRuntime { generation == self.lifecycleGeneration && self.isEnabled } - private func canOwnNativeFallback(_ owner: NativeFallbackOwner) -> Bool { - self.isCurrent(owner.lifecycleGeneration) && - !self.isPaused && - self.recognitionGeneration == owner.recognitionGeneration && - self.realtimeRelayGeneration == owner.realtimeRelayGeneration && - self.realtimeRelayStartGeneration == nil && - self.realtimeSession == nil - } - - func transitionToNativeFallback( - owner: NativeFallbackOwner, - projectFailure: @Sendable () async -> Void) async -> Bool - { - guard self.canOwnNativeFallback(owner) else { return false } - await projectFailure() - // Projection crosses actors. Revalidate before native capture can replace - // a newer realtime or recognition owner. - return self.canOwnNativeFallback(owner) - } - func start() async { let gen = self.lifecycleGeneration guard voiceWakeSupported else { return } @@ -286,11 +260,10 @@ actor TalkModeRuntime { } let bypassRealtime = self.bypassRealtimeOnNextStart self.bypassRealtimeOnNextStart = false + var nativeFallbackStatus: String? if self.shouldAttemptRealtimeRelay(), !bypassRealtime { - let fallbackOwner = NativeFallbackOwner( - lifecycleGeneration: gen, - recognitionGeneration: self.recognitionGeneration, - realtimeRelayGeneration: self.realtimeRelayGeneration &+ 1) + let fallbackRecognitionGeneration = self.recognitionGeneration + let fallbackRealtimeRelayGeneration = self.realtimeRelayGeneration &+ 1 do { try await self.startRealtimeRelay(generation: gen) return @@ -301,22 +274,19 @@ actor TalkModeRuntime { return } catch { self.pendingRealtimeRelayStartLifecycleGeneration = nil - guard self.canOwnNativeFallback(fallbackOwner) else { return } + guard self.isCurrent(gen), !self.isPaused, + self.recognitionGeneration == fallbackRecognitionGeneration, + self.realtimeRelayGeneration == fallbackRealtimeRelayGeneration, + self.realtimeRelayStartGeneration == nil, + self.realtimeSession == nil + else { return } self.logger.error( "talk realtime unavailable; using native fallback: " + "\(error.localizedDescription, privacy: .public)") - guard await self.transitionToNativeFallback( - owner: fallbackOwner, - projectFailure: { - await MainActor.run { - TalkModeController.shared.updatePartialTranscript( - String(localized: "Realtime unavailable — using native speech")) - } - }) - else { return } + nativeFallbackStatus = String(localized: "Realtime unavailable — using native speech") } } - await self.startNativeFallback(generation: gen) + await self.startNativeFallback(generation: gen, status: nativeFallbackStatus) } private func stop() async { @@ -369,15 +339,41 @@ actor TalkModeRuntime { } #endif - private func startNativeFallback(generation: Int) async { + private func startNativeFallback(generation: Int, status: String? = nil) async { + let relayGeneration = self.realtimeRelayGeneration guard await self.startRecognition(lifecycleGeneration: generation), - self.isCurrent(generation), !self.isPaused else { return } + await self.commitNativeFallback( + lifecycleGeneration: generation, + recognitionGeneration: self.recognitionGeneration, + relayGeneration: relayGeneration, + status: status) + else { return } self.startAudioInputObserver() - self.phase = .listening - await MainActor.run { TalkModeController.shared.updatePhase(.listening) } self.startSilenceMonitor() } + func commitNativeFallback( + lifecycleGeneration: Int, + recognitionGeneration: Int, + relayGeneration: UInt64, + status: String?) async -> Bool + { + let ownsFallback = { + self.canCommitRecognitionStart( + lifecycleGeneration: lifecycleGeneration, + recognitionAttempt: recognitionGeneration) && + self.realtimeRelayGeneration == relayGeneration && + self.realtimeRelayStartGeneration == nil && self.realtimeSession == nil + } + guard ownsFallback() else { return false } + self.phase = .listening + await MainActor.run { + if let status { TalkModeController.shared.updatePartialTranscript(status) } + TalkModeController.shared.updatePhase(.listening) + } + return ownsFallback() + } + func inputDeviceSelectionDidChange() async { if let realtimeSession { guard self.isEnabled, !self.isPaused else { return } diff --git a/apps/macos/Tests/OpenClawIPCTests/GatewayConnectionControlTests.swift b/apps/macos/Tests/OpenClawIPCTests/GatewayConnectionControlTests.swift index ddcd02b8f850..c39dc1c9fffb 100644 --- a/apps/macos/Tests/OpenClawIPCTests/GatewayConnectionControlTests.swift +++ b/apps/macos/Tests/OpenClawIPCTests/GatewayConnectionControlTests.swift @@ -339,9 +339,18 @@ private func assertConfigLookupCannotRecreateRoute( socketGeneration: socketGeneration) } - var iterator = events.makeAsyncIterator() - #expect(await iterator.next() != nil) - #expect(await iterator.next() == nil) + let terminalRead = Task { + var iterator = events.makeAsyncIterator() + let first = await iterator.next() + let terminal = await iterator.next() + return (first, terminal) + } + let (first, terminal) = try await AsyncTimeout.withTimeout( + seconds: 1, + onTimeout: { CancellationError() }, + operation: { await terminalRead.value }) + #expect(first != nil) + #expect(terminal == nil) await connection.shutdown() } diff --git a/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift b/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift index 888003908d20..9607db5d9b4f 100644 --- a/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift +++ b/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift @@ -605,81 +605,42 @@ struct TalkModeRuntimeSpeechTests { sessionB.stop() } - @Test @MainActor func `current relay failure owner can transition to native fallback`() async { + @Test @MainActor func `current relay failure owner can transition to native fallback`() async throws { let runtime = TalkModeRuntime() let lifecycleGeneration = await runtime._test_prepareEnabledLifecycle() - let fallbackProbe = RuntimeCommitProbe() - let player = RuntimeTestPCMPlayer() - let session = makeRuntimeTestRealtimeSession(player: player) - do { - try await runtime._test_startRealtimeRelay( - lifecycleGeneration: lifecycleGeneration, - makeSession: { session }, - start: { _ in throw RuntimeRelayStartError.failed }) - Issue.record("Expected relay start failure") - } catch {} - let owner = await TalkModeRuntime.NativeFallbackOwner( + let recognitionGeneration = try #require(await runtime._test_beginRecognitionAttempt( + lifecycleGeneration: lifecycleGeneration)) + let relayGeneration = await runtime.realtimeRelayGeneration + + #expect(await runtime.commitNativeFallback( lifecycleGeneration: lifecycleGeneration, - recognitionGeneration: runtime.recognitionGeneration, - realtimeRelayGeneration: runtime.realtimeRelayGeneration) - - if await runtime.transitionToNativeFallback(owner: owner, projectFailure: {}) { - fallbackProbe.record("native") - } - - #expect(fallbackProbe.values() == ["native"]) - #expect(await runtime._test_realtimeSessionIsActive() == false) - #expect(player.stopCount == 1) - #expect(await runtime._test_beginRecognitionAttempt( - lifecycleGeneration: lifecycleGeneration) != nil) + recognitionGeneration: recognitionGeneration, + relayGeneration: relayGeneration, + status: "native")) + #expect(TalkModeController.shared.partialTranscript == "native") await runtime.setEnabled(false) } - @Test @MainActor func `stale relay fallback cannot replace successor recognition owner`() async { + @Test @MainActor func `stale relay fallback cannot replace successor recognition owner`() async throws { let runtime = TalkModeRuntime() let lifecycleGeneration = await runtime._test_prepareEnabledLifecycle() - let fallbackProjectionBarrier = RuntimeContinuationBarrier() - let fallbackProbe = RuntimeCommitProbe() - let player = RuntimeTestPCMPlayer() - let session = makeRuntimeTestRealtimeSession(player: player) - do { - try await runtime._test_startRealtimeRelay( - lifecycleGeneration: lifecycleGeneration, - makeSession: { session }, - start: { _ in throw RuntimeRelayStartError.failed }) - Issue.record("Expected relay start failure") - } catch {} - let owner = await TalkModeRuntime.NativeFallbackOwner( - lifecycleGeneration: lifecycleGeneration, - recognitionGeneration: runtime.recognitionGeneration, - realtimeRelayGeneration: runtime.realtimeRelayGeneration) - - let transition = Task { - let accepted = await runtime.transitionToNativeFallback( - owner: owner, - projectFailure: { - await MainActor.run { - TalkModeController.shared.updatePartialTranscript("stale fallback") - } - await fallbackProjectionBarrier.wait() - }) - if accepted { - fallbackProbe.record("native") - } - } - await fallbackProjectionBarrier.waitUntilEntered() - let successorRecognition = await runtime._test_beginRecognitionAttempt( - lifecycleGeneration: lifecycleGeneration) + let staleRecognition = try #require(await runtime._test_beginRecognitionAttempt( + lifecycleGeneration: lifecycleGeneration)) + let relayGeneration = await runtime.realtimeRelayGeneration + let successorRecognition = try #require(await runtime._test_beginRecognitionAttempt( + lifecycleGeneration: lifecycleGeneration)) TalkModeController.shared.updatePartialTranscript("successor") - await fallbackProjectionBarrier.release() - await transition.value - #expect(successorRecognition != nil) + let accepted = await runtime.commitNativeFallback( + lifecycleGeneration: lifecycleGeneration, + recognitionGeneration: staleRecognition, + relayGeneration: relayGeneration, + status: "stale fallback") + #expect(await runtime.recognitionGeneration == successorRecognition) #expect(TalkModeController.shared.partialTranscript == "successor") - #expect(fallbackProbe.values().isEmpty) - #expect(player.stopCount == 1) + #expect(!accepted) await runtime.setEnabled(false) } diff --git a/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift b/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift index a85487dcbe22..5197129ef9ea 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift @@ -636,7 +636,8 @@ extension RealtimeTalkRelaySession { ? suppressed.isEmpty() : suppressed.isEmpty() || suppressed.relation(to: clearIdentity) == .same if clearsSuppressed { - self.retireOutputCancellation() + self.awaitingOutputClear = false + if self.outputCancellationTask == nil { self.retireOutputCancellation() } } } let currentMatches = @@ -1190,7 +1191,14 @@ extension RealtimeTalkRelaySession { "turnId": AnyCodable(turnId), ] do { - _ = try await transport.request("talk.session.cancelOutput", payload, 8000) + let response = try await transport.request("talk.session.cancelOutput", payload, 8000) + let status = try? JSONDecoder().decode([String: String].self, from: response)["status"] + guard let self, self.isCurrentOutputCancellation(cancellationGeneration) else { return } + if status == "stale" || status == "idle" || !self.awaitingOutputClear { + self.retireOutputCancellation() + } else { + self.outputCancellationTask = nil + } } catch { guard let self, self.isCurrentOutputCancellation(cancellationGeneration) else { return } let issue = RealtimeTalkRelayIssue( @@ -1354,7 +1362,7 @@ extension RealtimeTalkRelaySession { { guard self.isCurrentLifecycleLocally(lifecycleGeneration), self.audioCaptureGeneration == audioCaptureGeneration, - !self.isInputPaused, + !self.isInputPaused, self.suppressedOutputIdentity == nil, let audioSender = self.audioSender else { return nil } self.recordMicrophoneFrame(byteCount: encoded.count, rms: rms, timestampMs: timestampMs) @@ -1378,7 +1386,7 @@ extension RealtimeTalkRelaySession { defer { self.audioSendTasks.removeValue(forKey: taskID) } guard self.isCurrentLifecycleLocally(lifecycleGeneration), self.audioCaptureGeneration == audioCaptureGeneration, - !self.isInputPaused + !self.isInputPaused, self.suppressedOutputIdentity == nil else { return } guard let message = await audioSender.send(encoded, timestampMs: timestampMs) else { return } guard self.isCurrentLifecycleLocally(lifecycleGeneration), diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimePCMStreamingAudioPlayerTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimePCMStreamingAudioPlayerTests.swift index ad758abe3af4..b701c4db90bd 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimePCMStreamingAudioPlayerTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimePCMStreamingAudioPlayerTests.swift @@ -80,50 +80,65 @@ struct RealtimePCMStreamingAudioPlayerTests { @Test func `withheld completions cap scheduling and one completion admits one frame`() async { let backend = RealtimePCMPlaybackBackend() let player = makeRealtimePCMPlayer(backend: backend) + let probe = RealtimePCMPlaybackResultProbe() var continuation: AsyncThrowingStream.Continuation? let stream = AsyncThrowingStream { continuation = $0 } - let playback = Task { await player.play(stream: stream, sampleRate: self.sampleRate) } + let playback = Task { + let result = await player.play(stream: stream, sampleRate: self.sampleRate) + probe.record(result) + } continuation?.yield(Data(repeating: 1, count: self.frameBytes * 5)) + continuation?.finish() await waitUntil { backend.scheduledFrames.count == 3 } #expect(backend.scheduledFrames.count == 3) #expect(backend.maxActiveCount == 3) + #expect(probe.results.isEmpty) backend.complete() await waitUntil { backend.scheduledFrames.count == 4 } #expect(backend.scheduledFrames.count == 4) #expect(backend.maxActiveCount == 3) - backend.complete() - await waitUntil { backend.scheduledFrames.count == 5 } - #expect(backend.scheduledFrames.count == 5) - #expect(backend.maxActiveCount == 3) + #expect(probe.results.isEmpty) - continuation?.finish() while !backend.completions.isEmpty { backend.complete() await Task.yield() } - #expect(await (playback.value).finished) + await playback.value + #expect(probe.results.count == 1) + #expect(probe.results.first?.finished == true) + #expect(probe.results.first?.interruptedAt == nil) + #expect(backend.scheduledFrames.count == 5) #expect(backend.scheduledFrames.allSatisfy { $0.count == self.frameBytes }) } @Test func `playback finishes only after input and every scheduled frame complete`() async { let backend = RealtimePCMPlaybackBackend() let player = makeRealtimePCMPlayer(backend: backend) + let probe = RealtimePCMPlaybackResultProbe() var continuation: AsyncThrowingStream.Continuation? let stream = AsyncThrowingStream { continuation = $0 } - let playback = Task { await player.play(stream: stream, sampleRate: self.sampleRate) } + let playback = Task { + let result = await player.play(stream: stream, sampleRate: self.sampleRate) + probe.record(result) + } continuation?.yield(Data(repeating: 1, count: self.frameBytes * 2)) continuation?.finish() await waitUntil { backend.scheduledFrames.count == 2 } #expect(backend.completions.count == 2) + #expect(probe.results.isEmpty) backend.complete() await Task.yield() #expect(backend.completions.count == 1) + #expect(probe.results.isEmpty) backend.complete() - #expect(await (playback.value).finished) + await playback.value + #expect(probe.results.count == 1) + #expect(probe.results.first?.finished == true) + #expect(probe.results.first?.interruptedAt == nil) } @Test func `stop restart ignores stale buffer completions`() async { @@ -141,16 +156,32 @@ struct RealtimePCMStreamingAudioPlayerTests { var secondContinuation: AsyncThrowingStream.Continuation? let secondStream = AsyncThrowingStream { secondContinuation = $0 } - let secondPlayback = Task { await player.play(stream: secondStream, sampleRate: self.sampleRate) } - secondContinuation?.yield(Data(repeating: 2, count: self.frameBytes)) + let probe = RealtimePCMPlaybackResultProbe() + let secondPlayback = Task { + let result = await player.play(stream: secondStream, sampleRate: self.sampleRate) + probe.record(result) + } + secondContinuation?.yield(Data(repeating: 2, count: self.frameBytes * 5)) secondContinuation?.finish() - await waitUntil { backend.completions.count == 1 } + await waitUntil { backend.completions.count == 3 } staleCompletion() await Task.yield() - #expect(backend.completions.count == 1) + #expect(backend.scheduledFrames.count == 4) + #expect(backend.completions.count == 3) + #expect(probe.results.isEmpty) backend.complete() - #expect(await (secondPlayback.value).finished) + await waitUntil { backend.scheduledFrames.count == 5 } + #expect(backend.scheduledFrames.count == 5) + #expect(probe.results.isEmpty) + while !backend.completions.isEmpty { + backend.complete() + await Task.yield() + } + await secondPlayback.value + #expect(probe.results.count == 1) + #expect(probe.results.first?.finished == true) + #expect(probe.results.first?.interruptedAt == nil) } @Test func `stop resumes the active playback exactly once`() async { diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift index 46521db9f303..a2543774d7f0 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift @@ -152,12 +152,16 @@ private func unusedRealtimeRelayTransport() -> RealtimeTalkRelayTransport { request: { _, _, _ in throw CancellationError() }) } -private func outputAudioEvent(turnId: String, data: Data = Data([0x01])) -> EventFrame { +private func outputAudioEvent( + turnId: String, + data: Data = Data([0x01]), + relaySessionId: String = "relay-1") -> EventFrame +{ EventFrame( type: "event", event: "talk.event", payload: AnyCodable([ - "relaySessionId": "relay-1", + "relaySessionId": relaySessionId, "type": "audio", "audioBase64": data.base64EncodedString(), "talkEvent": ["turnId": turnId], @@ -168,11 +172,6 @@ private func outputAudioEvent(turnId: String, data: Data = Data([0x01])) -> Even @MainActor struct RealtimeTalkRelaySessionTests { - enum CancellationRetirement { - case clear - case close - } - private func makeIdleCancellationSession( _ onSpeakingChanged: @escaping (Bool) -> Void) -> RealtimeTalkRelaySession { @@ -429,6 +428,37 @@ extension RealtimeTalkRelaySessionTests { #expect(!session._test_isOutputPlaying()) } + @Test func `stale relay session cannot clear successor playback`() async { + var speakingStates: [Bool] = [] + let session = RealtimeTalkRelaySession( + transport: unusedRealtimeRelayTransport(), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: StalledPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { speakingStates.append($0) }) + session._test_setRelaySessionId("relay-1") + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-a")) + session._test_setRelaySessionId("relay-2") + await session._test_handleGatewayEvent( + outputAudioEvent(turnId: "turn-b", relaySessionId: "relay-2")) + + await session._test_handleGatewayEvent(EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "clear", + "talkEvent": ["turnId": "turn-a"], + ]), + seq: nil, + stateversion: nil)) + + #expect(session._test_isOutputPlaying()) + #expect(speakingStates == [true, false, true]) + session.stop() + } + @Test func `output cancellation fences delayed audio and preserves exact identity`() async throws { let requests = RealtimeRelayStartupRequestLog() var speakingStates: [Bool] = [] @@ -564,6 +594,40 @@ extension RealtimeTalkRelaySessionTests { #expect(await requests.snapshot().isEmpty) } + @Test(arguments: ["stale", "idle"]) + func `non applied cancellation retires the wait without reopening the old turn`( + status: String) async + { + var speakingStates: [Bool] = [] + let requests = RealtimeRelayStartupRequestLog() + let session = RealtimeTalkRelaySession( + transport: RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + return Data("{\"status\":\"\(status)\"}".utf8) + }), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { speakingStates.append($0) }) + session._test_setRelaySessionId("relay-1") + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-1")) + + #expect(session.cancelOutput()) + while await requests.snapshot().isEmpty { + await Task.yield() + } + for _ in 0..<5 { + await Task.yield() + } + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-1")) + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-2")) + + #expect(speakingStates == [true, false, true]) + } + @Test func `cancellation without active identified output is a no-op`() async { var speakingStates: [Bool] = [] let session = self.makeIdleCancellationSession { speakingStates.append($0) } @@ -654,6 +718,29 @@ extension RealtimeTalkRelaySessionTests { #expect(terminations == [.outputPlaybackOverflow]) } + @Test func `exact maximum output audio frame is accepted`() async { + var issues: [RealtimeTalkRelayIssue] = [] + var terminations: [RealtimeTalkRelayTermination] = [] + let session = RealtimeTalkRelaySession( + transport: unusedRealtimeRelayTransport(), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: StalledPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onIssue: { issues.append($0) }, + onTermination: { terminations.append($0) }, + onSpeakingChanged: { _ in }) + session._test_setRelaySessionId("relay-1") + + await session._test_handleGatewayEvent( + outputAudioEvent(turnId: "turn-1", data: Data(repeating: 1, count: 960))) + + #expect(issues.isEmpty) + #expect(terminations.isEmpty) + #expect(session._test_isOutputPlaying()) + session.stop() + } + @Test func `active output pause cancels the exact turn`() async throws { let requests = RealtimeRelayStartupRequestLog() let session = RealtimeTalkRelaySession( @@ -791,10 +878,57 @@ extension RealtimeTalkRelaySessionTests { #expect(speakingStates == [true, false, true, false]) } - @Test(arguments: [CancellationRetirement.clear, .close]) - func `clear and close retire in flight cancellation failures`( - retirement: CancellationRetirement) async - { + @Test func `clear keeps microphone fenced until cancellation response`() async throws { + let barrier = RealtimeRelayStartupBarrier() + let requests = RealtimeRelayStartupRequestLog() + let session = RealtimeTalkRelaySession( + transport: RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + if method == "talk.session.cancelOutput" { + await barrier.suspend() + return Data("{\"status\":\"applied\",\"turnId\":\"turn-1\"}".utf8) + } + return Data("{\"ok\":true}".utf8) + }), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { _ in }) + session._test_setRelaySessionId("relay-1") + session._test_prepareAudioSender(relaySessionId: "relay-1") + await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-1")) + #expect(session.cancelOutput()) + await barrier.waitUntilEntered() + await session._test_handleGatewayEvent(EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "clear", + "talkEvent": ["turnId": "turn-1"], + ]), + seq: nil, + stateversion: nil)) + + #expect(session._test_enqueueMicrophoneFrame(Data([0x01])) == nil) + await barrier.release() + var admitted: Task? + for _ in 0..<10 where admitted == nil { + await Task.yield() + admitted = session._test_enqueueMicrophoneFrame(Data([0x02])) + } + let admittedTask = try #require(admitted) + await admittedTask.value + #expect(await requests.snapshot().map(\.method) == [ + "talk.session.cancelOutput", + "talk.session.appendAudio", + ]) + } + + @Test func `close retires in flight cancellation failure`() async { let barrier = RealtimeRelayStartupBarrier() var issues: [RealtimeTalkRelayIssue] = [] let session = RealtimeTalkRelaySession( @@ -818,21 +952,7 @@ extension RealtimeTalkRelaySessionTests { await session._test_handleGatewayEvent(outputAudioEvent(turnId: "turn-1")) #expect(session.cancelOutput()) await barrier.waitUntilEntered() - switch retirement { - case .clear: - await session._test_handleGatewayEvent(EventFrame( - type: "event", - event: "talk.event", - payload: AnyCodable([ - "relaySessionId": "relay-1", - "type": "clear", - "talkEvent": ["turnId": "turn-1"], - ]), - seq: nil, - stateversion: nil)) - case .close: - session.stop() - } + session.stop() await barrier.release() await Task.yield() diff --git a/docs/gateway/protocol.md b/docs/gateway/protocol.md index 48c63682dbc6..22e193bbb7ce 100644 --- a/docs/gateway/protocol.md +++ b/docs/gateway/protocol.md @@ -588,7 +588,7 @@ methods. Treat this as feature discovery, not a full enumeration of - `talk.config` returns the effective Talk config payload; `includeSecrets` requires `operator.talk.secrets` (or `operator.admin`). - `talk.session.create` (`operator.talk`) creates a gateway-owned Talk session for `realtime/gateway-relay`, `transcription/gateway-relay`, or `stt-tts/managed-room`. For `stt-tts/managed-room`, non-admin callers that pass `sessionKey` must also pass `spawnedBy` for scoped session-key visibility; unscoped `sessionKey` creation and `brain: "direct-tools"` require `operator.admin`. - `talk.session.appendAudio` appends base64 PCM input audio to gateway-owned realtime relay and transcription sessions. - - `talk.session.cancelOutput` stops assistant audio output, primarily for VAD-gated barge-in in gateway relay sessions. + - `talk.session.cancelOutput` stops assistant audio output, primarily for VAD-gated barge-in in gateway relay sessions. Send the current `talk.event.turnId`; the result is `applied`, `stale`, or `idle`. - `talk.session.submitToolResult` completes a provider tool call emitted by a gateway-owned realtime relay session. The request waits for any asynchronous completion signal exposed by the provider bridge; failed submissions keep the linked run active and do not emit a successful tool-result event. Pass `options: { willContinue: true }` for interim tool output or `options: { suppressResponse: true }` when the provider bridge advertises suppression support and the result should not start another response. - `talk.session.steer` sends active-run voice control into a gateway-owned agent-backed Talk session: `{ sessionId, text, mode? }`, where `mode` is `status`, `steer`, `cancel`, or `followup`; omitted mode is classified from the spoken text. - `talk.session.close` closes a gateway-owned relay, transcription, or managed-room session and emits terminal Talk events. diff --git a/docs/nodes/talk.md b/docs/nodes/talk.md index c5033dc7a37c..de30a6138c73 100644 --- a/docs/nodes/talk.md +++ b/docs/nodes/talk.md @@ -92,7 +92,9 @@ then 2 s). If those attempts are exhausted, the overlay reports Losing the microphone mid-session closes the relay and takes the same route. Relay output cancellation is turn-scoped. Clients copy the current `turnId` from the -`talk.event` audio envelope; missing, empty, or stale turn ids are ignored: +`talk.event` audio envelope. Matching ids return `applied`, stale ids return `stale`, and +sessions without an active turn return `idle`. Older clients that omit `turnId` still cancel +the current turn: ```json { diff --git a/docs/plugins/sdk-migration.md b/docs/plugins/sdk-migration.md index f6d182712b4e..2eb2347b9cb0 100644 --- a/docs/plugins/sdk-migration.md +++ b/docs/plugins/sdk-migration.md @@ -981,7 +981,7 @@ await gateway.request("talk.session.create", { sessionKey: "main", }); await gateway.request("talk.session.appendAudio", { sessionId, audioBase64 }); -await gateway.request("talk.session.cancelOutput", { sessionId, reason: "barge-in" }); +await gateway.request("talk.session.cancelOutput", { sessionId, turnId, reason: "barge-in" }); await gateway.request("talk.session.submitToolResult", { sessionId, callId, diff --git a/src/gateway/server-methods/talk-session.ts b/src/gateway/server-methods/talk-session.ts index 43f5f5b9b3b8..59a76f44fc60 100644 --- a/src/gateway/server-methods/talk-session.ts +++ b/src/gateway/server-methods/talk-session.ts @@ -381,7 +381,7 @@ export const talkSessionHandlers: GatewayRequestHandlers = { const session = getUnifiedTalkSession(params.sessionId); if (session.kind === "realtime-relay") { const connId = requireUnifiedTalkSessionConn(session, client?.connId); - sendTalkRealtimeRelayAudio({ + await sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId, audioBase64: params.audioBase64, @@ -426,13 +426,13 @@ export const talkSessionHandlers: GatewayRequestHandlers = { return; } const connId = requireUnifiedTalkSessionConn(session, client?.connId); - cancelTalkRealtimeRelayTurn({ + const result = await cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId, reason: normalizeOptionalString(params.reason) ?? "output-cancelled", turnId: normalizeOptionalString(params.turnId), }); - respondOk(respond); + respondOk(respond, result); } catch (err) { respondUnavailable(respond, err); } diff --git a/src/gateway/server-methods/talk.test.ts b/src/gateway/server-methods/talk.test.ts index 21c4faada356..d56cfa3c6b55 100644 --- a/src/gateway/server-methods/talk.test.ts +++ b/src/gateway/server-methods/talk.test.ts @@ -1732,6 +1732,10 @@ describe("talk.session unified handlers", () => { }); const cancelRespond = vi.fn(); + mocks.cancelTalkRealtimeRelayTurn.mockReturnValueOnce({ + status: "applied", + turnId: "turn-7", + }); await callTalkHandler("talk.session.cancelOutput", { params: { sessionId: "relay-unified-1", reason: "barge-in", turnId: "turn-7" }, id: "3", @@ -1744,6 +1748,7 @@ describe("talk.session unified handlers", () => { reason: "barge-in", turnId: "turn-7", }); + expectRespondOk(cancelRespond, { status: "applied", turnId: "turn-7" }); const markRespond = vi.fn(); await callTalkHandler("talk.session.acknowledgeMark", { diff --git a/src/gateway/talk-realtime-relay-forced-consults.ts b/src/gateway/talk-realtime-relay-forced-consults.ts index ee8be0aeeb91..8fe4f88c0d92 100644 --- a/src/gateway/talk-realtime-relay-forced-consults.ts +++ b/src/gateway/talk-realtime-relay-forced-consults.ts @@ -63,12 +63,6 @@ export function buildAlreadyDeliveredToolResult(): Record { }; } -export function cancelForcedConsults(session: RelaySession): void { - for (const handle of session.harness.forcedConsults.handles()) { - session.harness.forcedConsults.markCancelled(handle); - } -} - export function submitRelayAgentControlProviderResults( session: RelaySession, result: RealtimeVoiceAgentControlResult, @@ -228,28 +222,34 @@ function drainForcedTerminalProviderResults( handle: RealtimeVoiceForcedConsultHandle, terminal: ForcedTerminalProviderResult, ): void | Promise { - if (session.forcedTerminalProviderResults.get(handle.id) !== terminal) { + const isCurrent = () => + relaySessions.get(session.id) === session && + session.toolResultEpoch === terminal.epoch && + session.forcedTerminalProviderResults.get(handle.id) === terminal; + if (!isCurrent()) { return; } - const submissions = session.harness.forcedConsults - .nativeCallIds(handle) - .map((callId) => - submitForcedConsultProviderResult(session, callId, terminal.result, terminal.options), - ); - const pending = submissions.filter( - (submission): submission is Promise => submission !== undefined, - ); - if (pending.length > 0) { - return Promise.all(pending).then(() => - drainForcedTerminalProviderResults(session, handle, terminal), - ); - } - const hasUnsubmittedCall = session.harness.forcedConsults - .nativeCallIds(handle) - .some((callId) => !session.toolCalls.isProviderCompleted(callId)); - if (hasUnsubmittedCall) { - return drainForcedTerminalProviderResults(session, handle, terminal); + const callIds = () => + terminal.nativeCallIds ?? session.harness.forcedConsults.nativeCallIds(handle); + const submitPending = () => + callIds() + .filter((callId) => !session.toolCalls.isProviderCompleted(callId)) + .map((callId) => + submitForcedConsultProviderResult(session, callId, terminal.result, terminal.options), + ) + .filter((submission): submission is Promise => submission !== undefined); + const submissions = submitPending(); + if (submissions.length === 0) { + return; } + const settle = (pending: Promise[]) => + terminal.nativeCallIds ? Promise.allSettled(pending) : Promise.all(pending); + return settle(submissions).then(async () => { + if (!isCurrent()) { + return; + } + await settle(submitPending()); + }); } function drainForcedTerminalProviderResultsAfterPending( @@ -257,16 +257,17 @@ function drainForcedTerminalProviderResultsAfterPending( handle: RealtimeVoiceForcedConsultHandle, terminal: ForcedTerminalProviderResult, ): void | Promise { - const pending = session.harness.forcedConsults - .nativeCallIds(handle) + const pending = (terminal.nativeCallIds ?? session.harness.forcedConsults.nativeCallIds(handle)) .map((callId) => session.pendingProviderToolResults.get(callId)) .filter((submission): submission is Promise => submission !== undefined); if (pending.length === 0) { return drainForcedTerminalProviderResults(session, handle, terminal); } - return Promise.allSettled(pending).then(() => - drainForcedTerminalProviderResults(session, handle, terminal), - ); + return Promise.allSettled(pending).then(() => { + if (relaySessions.get(session.id) === session && session.toolResultEpoch === terminal.epoch) { + return drainForcedTerminalProviderResults(session, handle, terminal); + } + }); } export function submitRealtimeAgentConsultWorkingResponse( @@ -322,12 +323,16 @@ export function submitForcedTalkRealtimeRelayToolResult( const providerResult = buildRealtimeVoiceAgentCancelProviderResult( "OpenClaw cancelled this consult before completion. Do not restart it.", ); - const terminal: ForcedTerminalProviderResult = { - result: providerResult, - options: suppressedToolResultOptions(session), - turnId, - epoch: session.toolResultEpoch, - }; + const existing = session.forcedTerminalProviderResults.get(forcedConsult.id); + const terminal: ForcedTerminalProviderResult = + existing?.epoch === session.toolResultEpoch + ? existing + : { + result: providerResult, + options: suppressedToolResultOptions(session), + turnId, + epoch: session.toolResultEpoch, + }; session.forcedTerminalProviderResults.set(forcedConsult.id, terminal); const clearTerminal = () => { if (session.forcedTerminalProviderResults.get(forcedConsult.id) === terminal) { diff --git a/src/gateway/talk-realtime-relay-operations.ts b/src/gateway/talk-realtime-relay-operations.ts index a62c46f0e13f..243384596029 100644 --- a/src/gateway/talk-realtime-relay-operations.ts +++ b/src/gateway/talk-realtime-relay-operations.ts @@ -1,4 +1,5 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; +import { createDeferredCore } from "../shared/deferred.js"; import { buildRealtimeVoiceAgentCancelProviderResult } from "../talk/agent-run-control-shared.js"; import { controlRealtimeVoiceAgentRun, @@ -13,7 +14,6 @@ import type { TalkEvent } from "../talk/talk-session-controller.js"; import { abortChatRunById } from "./chat-abort.js"; import { formatError } from "./server-utils.js"; import { - cancelForcedConsults, submitForcedTalkRealtimeRelayToolResult, submitRelayAgentControlProviderResults, } from "./talk-realtime-relay-forced-consults.js"; @@ -53,6 +53,8 @@ import { } from "./talk-relay-session-lifecycle.js"; import { forgetUnifiedTalkSession } from "./talk-session-registry.js"; +const TURN_BOUND_CANCELLATION_DRAIN_MS = 1_000; + /** Ensure a gateway-relay call has its durable record before transcript-free RPCs. */ export function ensureTalkRealtimeRelayVoiceSession(params: { relaySessionId: string; @@ -109,6 +111,7 @@ export function closeRelaySession( ): void { const disposition = options?.disposition ?? "abort"; session.harness.close(); + session.outputOwnership.drain?.resolve(); relaySessions.delete(session.id); forgetUnifiedTalkSession(session.id); clearTimeout(session.cleanupTimer); @@ -199,11 +202,14 @@ export function sendTalkRealtimeRelayAudio(params: { connId: string; audioBase64: string; timestamp?: number; -}): void { +}): void | Promise { if (params.audioBase64.length > MAX_AUDIO_BASE64_BYTES) { throw new Error("Realtime relay audio frame is too large"); } const session = getRelaySession(params.relaySessionId, params.connId); + if (session.outputOwnership.phase === "cancelling") { + return session.outputOwnership.drain!.promise.then(() => sendTalkRealtimeRelayAudio(params)); + } const audio = decodeTalkRelayAudioBase64(params.audioBase64, "Realtime relay"); const turnId = ensureRelayTurn(session); session.bridge.sendAudio(audio); @@ -244,6 +250,12 @@ export function submitTalkRealtimeRelayToolResult(params: { if (session.toolCalls.isAgentCompleted(params.callId)) { return; } + if ( + session.outputOwnership.phase === "cancelling" && + !session.toolCalls.hasCancelled(params.callId) + ) { + return; + } if (!session.toolCalls.tryAdmit([params.callId])) { return; } @@ -514,40 +526,58 @@ export async function steerTalkRealtimeRelayAgentRun(params: { } /** Cancels the active relay turn, aborts agent work, and clears provider audio. */ -export function cancelTalkRealtimeRelayTurn(params: { +export async function cancelTalkRealtimeRelayTurn(params: { relaySessionId: string; connId: string; reason?: string; turnId?: string; -}): void { +}) { const session = getRelaySession(params.relaySessionId, params.connId); - const requestedTurnId = normalizeOptionalString(params.turnId); - if (!requestedTurnId || session.harness.talk.activeTurnId !== requestedTurnId) { - return; + const turnId = session.harness.talk.activeTurnId; + if (!turnId) { + return { status: "idle" as const }; } - const turnId = requestedTurnId; - session.toolResultEpoch += 1; + const requestedTurnId = normalizeOptionalString(params.turnId); + if (requestedTurnId && turnId !== requestedTurnId) { + return { status: "stale" as const }; + } + if (session.outputOwnership.phase === "owned" && session.outputOwnership.turnId !== turnId) { + return { status: "stale" as const }; + } + const forcedConsults = session.harness.forcedConsults.handles().map((handle) => ({ + handle, + nativeCallIds: session.harness.forcedConsults.nativeCallIds(handle), + })); + const rootCallIds = new Set([ + ...session.activeAgentToolCalls.keys(), + ...forcedConsults.map(({ handle }) => handle.id), + ]); + const terminalEpoch = ++session.toolResultEpoch; session.forcedTerminalProviderResults.clear(); const reason = params.reason ?? "client-cancelled"; - cancelForcedConsults(session); - for (const callId of session.activeAgentToolCalls.keys()) { - if (!session.toolCalls.markCancelled([callId], turnId)) { - return; - } + if ( + !session.toolCalls.markCancelled( + [...rootCallIds, ...forcedConsults.flatMap(({ nativeCallIds }) => nativeCallIds)], + turnId, + ) + ) { + throw new Error("Realtime relay cancellation could not record tool state"); } - for (const forcedConsult of session.harness.forcedConsults.handles()) { - if (session.harness.forcedConsults.isCancelled(forcedConsult)) { - if ( - !session.toolCalls.markCancelled( - [forcedConsult.id, ...session.harness.forcedConsults.nativeCallIds(forcedConsult)], - turnId, - ) - ) { - return; - } - } + for (const { handle, nativeCallIds } of forcedConsults) { + session.harness.forcedConsults.markCancelled(handle); + session.forcedTerminalProviderResults.set(handle.id, { + result: buildRealtimeVoiceAgentCancelProviderResult( + "OpenClaw cancelled this consult before completion. Do not restart it.", + ), + options: suppressedToolResultOptions(session), + turnId, + epoch: terminalEpoch, + nativeCallIds, + }); } - session.harness.handleBargeIn({ audioPlaybackActive: true }, noFallbackRelayOutputFlush); + session.outputOwnership.phase = "cancelling"; + session.outputOwnership.turnId = turnId; + const cancellationDrained = (session.outputOwnership.drain = createDeferredCore()); abortRelayAgentRuns(session, reason); const cancelled = session.harness.talk.cancelTurn({ turnId, @@ -558,6 +588,37 @@ export function cancelTalkRealtimeRelayTurn(params: { type: "clear", talkEvent: cancelled.ok ? cancelled.event : undefined, }); + const closeAfterCancellation = () => { + if ( + relaySessions.get(session.id) === session && + session.toolResultEpoch === terminalEpoch && + session.outputOwnership.phase === "cancelling" + ) { + session.outputOwnership.drain?.resolve(); + closeRelaySession(session, "completed"); + } + }; + setTimeout(closeAfterCancellation, TURN_BOUND_CANCELLATION_DRAIN_MS).unref?.(); + const terminalDrain = Promise.allSettled( + [...rootCallIds].map(async (callId) => { + await submitTalkRealtimeRelayToolResult({ + relaySessionId: session.id, + connId: session.connId, + callId, + result: { status: "cancelled" }, + }); + }), + ); + if (session.outputOwnership.mode === "exact-response") { + try { + session.bridge.handleBargeIn({ audioPlaybackActive: true }); + } catch { + session.failSession("Realtime provider cancellation failed. Reconnecting."); + } + } else { + void terminalDrain.then(closeAfterCancellation); + } + return cancellationDrained.promise.then(() => ({ status: "applied" as const, turnId })); } /** Drops one provider generation without sending cancellation into its replacement. */ diff --git a/src/gateway/talk-realtime-relay-session-create.ts b/src/gateway/talk-realtime-relay-session-create.ts index e912fcf0570d..4d2743807c1b 100644 --- a/src/gateway/talk-realtime-relay-session-create.ts +++ b/src/gateway/talk-realtime-relay-session-create.ts @@ -46,6 +46,7 @@ import { relaySessions, type CreateTalkRealtimeRelaySessionParams, type RelaySession, + TalkRealtimeRelayOutputOwnership, type TalkRealtimeRelayEventPayload, type TalkRealtimeRelaySessionResult, } from "./talk-realtime-relay-state.js"; @@ -108,7 +109,6 @@ export function createTalkRealtimeRelaySession( ...(talkEvent ? { talkEvent: harness.emit(talkEvent) } : {}), }); let currentOutputItemId: string | undefined; - let currentOutputResponseId: string | undefined; let ready = false; let continuityResetActive = false; let failureEmitted = false; @@ -122,6 +122,16 @@ export function createTalkRealtimeRelaySession( return relay && relaySessions.get(relay.id) === relay ? relay : undefined; }; const bridgeRef: { current?: ReturnType } = {}; + const outputOwnership = new TalkRealtimeRelayOutputOwnership( + () => harness.talk.activeTurnId, + (message) => { + const relay = getActiveRelay(); + relay?.failSession(message); + if (!relay) { + constructionTerminal.current ??= { kind: "error", error: new Error(message) }; + } + }, + ); const relaySessionKey = params.sessionKey?.trim(); const relayAgentId = relaySessionKey ? resolveTalkSessionAgentId(params.cfg ?? params.context.getRuntimeConfig(), relaySessionKey) @@ -192,16 +202,7 @@ export function createTalkRealtimeRelaySession( ); }, }); - // The generic harness should stay transport-neutral. Wrap only this relay provider - // invocation so provider-owned delegations cannot acquire the host runner elsewhere. - const relayProvider = { - ...params.provider, - createBridge: (request: Parameters[0]) => - params.provider.createBridge({ - ...request, - runAgentConsult, - }), - }; + const relayProvider = outputOwnership.bind(params.provider, runAgentConsult); const bridge = harness.createBridge({ provider: relayProvider, cfg: params.cfg, @@ -227,7 +228,13 @@ export function createTalkRealtimeRelaySession( ); return; } - const turnId = ensureRelayTurn(relay); + if (outputOwnership.phase === "cancelling") { + return; + } + const outputTurnId = outputOwnership.resolve(true); + if (!outputTurnId) { + return; + } for (let offset = 0; offset < audio.byteLength; offset += RELAY_OUTPUT_AUDIO_FRAME_BYTES) { const frame = audio.subarray( offset, @@ -239,11 +246,11 @@ export function createTalkRealtimeRelaySession( type: "audio", audioBase64: frame.toString("base64"), ...(currentOutputItemId ? { itemId: currentOutputItemId } : {}), - ...(currentOutputResponseId ? { responseId: currentOutputResponseId } : {}), + ...(outputOwnership.responseId ? { responseId: outputOwnership.responseId } : {}), }, { type: "output.audio.delta", - turnId, + turnId: outputTurnId, payload: { byteLength: frame.byteLength }, }, ); @@ -254,12 +261,15 @@ export function createTalkRealtimeRelaySession( if (!relay) { return; } - const turnId = ensureRelayTurn(relay); + const outputTurnId = outputOwnership.resolve(false); + if (!outputTurnId) { + return; + } emit( { relaySessionId, type: "clear", ...(reason ? { reason } : {}) }, { type: "output.audio.done", - turnId, + turnId: outputTurnId, payload: { reason: reason ?? "clear" }, final: true, }, @@ -270,12 +280,18 @@ export function createTalkRealtimeRelaySession( if (!relay) { return; } - const turnId = ensureRelayTurn(relay); + const outputTurnId = outputOwnership.resolve(false); + if (!outputTurnId) { + if (outputOwnership.phase !== "owned") { + bridgeRef.current?.acknowledgeMark(markName); + } + return; + } emit( { relaySessionId, type: "mark", markName }, { type: "output.audio.done", - turnId, + turnId: outputTurnId, payload: { markName }, final: true, }, @@ -294,7 +310,10 @@ export function createTalkRealtimeRelaySession( continuityResetActive = true; ready = false; currentOutputItemId = undefined; - currentOutputResponseId = undefined; + outputOwnership.outputGeneration += 1; + outputOwnership.drain?.resolve(); + outputOwnership.phase = "unowned"; + outputOwnership.turnId = outputOwnership.responseId = undefined; const talkEvent = resetTalkRealtimeRelayContinuity(relay, event.type); if (!getActiveRelay()) { return; @@ -312,6 +331,13 @@ export function createTalkRealtimeRelaySession( if (event.type === "session.created") { continuityResetActive = false; } + if ( + (event.type === "response.done" || event.type === "response.cancelled") && + outputOwnership.finish(event.responseId, true) === "cancelled" + ) { + currentOutputItemId = undefined; + return; + } if (event.type === "tool.call.cancelled" && event.itemId) { const relayCallId = cancelTalkRealtimeRelayProviderToolCall(relay, event.itemId); if (relayCallId) { @@ -330,7 +356,7 @@ export function createTalkRealtimeRelaySession( event.type === "response.output_audio.delta" ) { currentOutputItemId = event.itemId ?? currentOutputItemId; - currentOutputResponseId = event.responseId ?? currentOutputResponseId; + outputOwnership.responseId = event.responseId ?? outputOwnership.responseId; } }, onResponseDone: (outcome) => { @@ -338,21 +364,27 @@ export function createTalkRealtimeRelaySession( if (!relay) { return; } + const responseId = outcome.responseId ?? outputOwnership.responseId; + const disposition = outputOwnership.finish(responseId); + if (disposition === "ignore") { + return; + } + if (disposition === "cancelled") { + currentOutputItemId = undefined; + return; + } const terminalTalkEvent = harness.talk.recentEvents.at(-1); broadcastToOwner(params.context, params.connId, { relaySessionId, type: "audioDone", ...(currentOutputItemId ? { itemId: currentOutputItemId } : {}), - ...((outcome.responseId ?? currentOutputResponseId) - ? { responseId: outcome.responseId ?? currentOutputResponseId } - : {}), + ...(responseId ? { responseId } : {}), ...(terminalTalkEvent && (terminalTalkEvent.type === "turn.ended" || terminalTalkEvent.type === "turn.cancelled") ? { talkEvent: terminalTalkEvent } : {}), }); currentOutputItemId = undefined; - currentOutputResponseId = undefined; if (outcome.status === "failed" || outcome.status === "incomplete") { const issue = realtimeRelayIssue({ message: outcome.message, @@ -374,10 +406,17 @@ export function createTalkRealtimeRelaySession( if (!relay) { return; } + if (role === "assistant" && outputOwnership.phase === "cancelling") { + return; + } if (final && !enqueueRelayVoiceTranscript(relay, role, text)) { return; } - const turnId = ensureRelayTurn(relay); + const outputTurnId = role === "assistant" ? outputOwnership.resolve(true) : undefined; + if (role === "assistant" && !outputTurnId) { + return; + } + const turnId = outputTurnId ?? ensureRelayTurn(relay); const eventType = role === "assistant" ? final @@ -414,6 +453,13 @@ export function createTalkRealtimeRelaySession( if (!relay) { return; } + if (outputOwnership.phase === "cancelling") { + return; + } + const outputTurnId = outputOwnership.resolve(true); + if (!outputTurnId) { + return; + } const providerCallId = toolCall.callId; const relayCallId = adoptRelayProviderToolCallId(relay, providerCallId); if (!relayCallId) { @@ -446,7 +492,6 @@ export function createTalkRealtimeRelaySession( } shouldSubmitWorkingResult = true; } - const turnId = ensureRelayTurn(relay); emit( { relaySessionId, @@ -460,12 +505,12 @@ export function createTalkRealtimeRelaySession( type: "tool.call", itemId: toolCall.itemId, callId: relayCallId, - turnId, + turnId: outputTurnId, payload: { name: toolCall.name, args: toolCall.args }, }, ); if (shouldSubmitWorkingResult) { - return submitRealtimeAgentConsultWorkingResponse(relay, relayCallId, turnId); + return submitRealtimeAgentConsultWorkingResponse(relay, relayCallId, outputTurnId); } }, onReady: () => { @@ -573,6 +618,7 @@ export function createTalkRealtimeRelaySession( context: params.context, bridge, harness, + outputOwnership, sessionKey: initialSessionKey, ...(initialSessionKey ? { diff --git a/src/gateway/talk-realtime-relay-state.ts b/src/gateway/talk-realtime-relay-state.ts index ae16e1439cab..c762a8c9717f 100644 --- a/src/gateway/talk-realtime-relay-state.ts +++ b/src/gateway/talk-realtime-relay-state.ts @@ -6,6 +6,7 @@ import type { RealtimeVoiceAgentControlResult } from "../talk/agent-run-control. import type { RealtimeVoiceBrowserAudioContract, RealtimeVoiceAudioClearReason, + RealtimeVoiceAgentConsultRunner, RealtimeVoiceProviderConfig, RealtimeVoiceTool, RealtimeVoiceToolResultOptions, @@ -77,6 +78,7 @@ export type ForcedTerminalProviderResult = { options?: RealtimeVoiceToolResultOptions; turnId: string; epoch: number; + nativeCallIds?: readonly string[]; }; export type RelayAgentControlProviderSubmission = { @@ -84,12 +86,101 @@ export type RelayAgentControlProviderSubmission = { providerResponseStarted: boolean; }; +type RelayProvider = RealtimeVoiceProviderPlugin; +export class TalkRealtimeRelayOutputOwnership { + mode: "turn-bound" | "exact-response" = "turn-bound"; + phase: "unowned" | "owned" | "cancelling" = "unowned"; + outputGeneration = 0; + turnId?: string; + responseId?: string; + drain?: { promise: Promise; resolve: () => void }; + + constructor( + private readonly activeTurnId: () => string | undefined, + private readonly fail: (message: string) => void, + ) {} + + responseCreated(responseId: string | undefined): boolean { + const normalizedResponseId = responseId?.trim(); + const turnId = this.activeTurnId(); + if ( + !normalizedResponseId || + !turnId || + (this.phase !== "unowned" && normalizedResponseId !== this.responseId) + ) { + this.fail("Realtime provider output has no live response owner."); + return false; + } + this.mode = "exact-response"; + if (this.phase === "unowned") { + Object.assign(this, { phase: "owned" as const, turnId, responseId: normalizedResponseId }); + } + return normalizedResponseId === this.responseId; + } + + resolve(claim: boolean): string | undefined { + const activeTurnId = this.activeTurnId(); + if ( + this.phase !== "cancelling" && + activeTurnId && + this.mode === "turn-bound" && + claim && + this.phase === "unowned" + ) { + Object.assign(this, { phase: "owned" as const, turnId: activeTurnId }); + } + const turnId = + this.phase === "owned" && this.turnId === activeTurnId ? activeTurnId : undefined; + if (!turnId && (claim || this.phase === "owned")) { + this.fail("Realtime provider output has no live response owner."); + } + return turnId; + } + + finish(responseId: string | undefined, cancellationEvent = false) { + const cancelled = this.phase === "cancelling"; + if ( + (cancelled && this.mode === "turn-bound") || + (cancellationEvent && !cancelled) || + (this.mode === "exact-response" && + (this.phase === "unowned" || this.responseId !== responseId)) + ) { + return "ignore"; + } + this.drain?.resolve(); + Object.assign(this, { phase: "unowned" as const, turnId: undefined, responseId: undefined }); + return cancelled ? "cancelled" : "completed"; + } + + bind(provider: RelayProvider, runAgentConsult: RealtimeVoiceAgentConsultRunner): RelayProvider { + return { + ...provider, + createBridge: (request) => + provider.createBridge({ + ...request, + onEvent: (event) => { + if ( + event.direction === "server" && + event.type === "response.created" && + !this.responseCreated(event.responseId) + ) { + return; + } + request.onEvent?.(event); + }, + runAgentConsult, + }), + }; + } +} + export type RelaySession = { id: string; connId: string; context: GatewayRequestContext; bridge: RealtimeVoiceBridgeSession; harness: RealtimeVoiceSessionHarness; + outputOwnership: TalkRealtimeRelayOutputOwnership; sessionKey?: string; agentId?: string; expiresAtMs: number; diff --git a/src/gateway/talk-realtime-relay.test.ts b/src/gateway/talk-realtime-relay.test.ts index 7e55826131cd..db3ab6e06ae3 100644 --- a/src/gateway/talk-realtime-relay.test.ts +++ b/src/gateway/talk-realtime-relay.test.ts @@ -136,7 +136,7 @@ describe("talk realtime gateway relay", () => { throw new Error("expected realtime bridge request"); } - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-1", audioBase64: Buffer.from("first").toString("base64"), @@ -166,7 +166,7 @@ describe("talk realtime gateway relay", () => { expect(relaySessions.has(session.relaySessionId)).toBe(true); expect(close).not.toHaveBeenCalled(); - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-1", audioBase64: Buffer.from("later").toString("base64"), @@ -177,7 +177,7 @@ describe("talk realtime gateway relay", () => { type: "response.created", responseId: "response-2", }); - bridgeRequest.onResponseDone?.({ status: "completed", responseId: "response-2" }); + bridgeRequest.onResponseDone?.({ status: "completed" }); bridgeRequest.onEvent?.({ direction: "server", responseId: "response-2", @@ -348,22 +348,24 @@ describe("talk realtime gateway relay", () => { await Promise.resolve(); expect(broadcastToConnIds).toHaveBeenCalledTimes(eventCountAfterClose); expect(bridgeToolResults[0]).not.toHaveBeenCalled(); - expect(() => - sendTalkRealtimeRelayAudio({ - relaySessionId: firstOwned.relaySessionId, - connId: "conn-owner", - audioBase64: "AQI=", - }), + expect( + () => + void sendTalkRealtimeRelayAudio({ + relaySessionId: firstOwned.relaySessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), ).toThrow("Unknown realtime relay session"); - expect(() => - sendTalkRealtimeRelayAudio({ - relaySessionId: secondOwned.relaySessionId, - connId: "conn-owner", - audioBase64: "AQI=", - }), + expect( + () => + void sendTalkRealtimeRelayAudio({ + relaySessionId: secondOwned.relaySessionId, + connId: "conn-owner", + audioBase64: "AQI=", + }), ).toThrow("Unknown realtime relay session"); - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: unrelated.relaySessionId, connId: "conn-other", audioBase64: "AQI=", @@ -1026,6 +1028,7 @@ describe("talk realtime gateway relay", () => { instructions: "brief", tools: [], }); + ensureActiveRelayTurnId(session.relaySessionId); if (options.register !== false) { registerTalkRealtimeRelayAgentRun({ @@ -1351,6 +1354,12 @@ describe("talk realtime gateway relay", () => { } const request = bridgeRequest; request.onReady?.(); + relay.harness.talk.startTurn({ turnId: "turn-old" }); + request.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "old-response", + }); request.onEvent?.({ direction: "server", type: "response.audio.delta", @@ -1407,6 +1416,12 @@ describe("talk realtime gateway relay", () => { expect(submitToolResult).toHaveBeenCalledTimes(1); request.onEvent?.({ direction: "server", type: "session.created" }); + relay.harness.talk.startTurn({ turnId: "turn-new" }); + request.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "fresh-response", + }); request.onAudio(Buffer.from("fresh audio")); const freshAudio = fixture.broadcastToConnIds.mock.calls .map(([, payload]) => payload) @@ -1421,7 +1436,7 @@ describe("talk realtime gateway relay", () => { audioBase64: Buffer.from("fresh audio").toString("base64"), }); expect(freshAudio).not.toHaveProperty("itemId"); - expect(freshAudio).not.toHaveProperty("responseId"); + expect(freshAudio).toHaveProperty("responseId", "fresh-response"); request.onToolCall?.({ itemId: "fresh-item", @@ -1481,9 +1496,14 @@ describe("talk realtime gateway relay", () => { supportsToolResultContinuation: true, connect: vi.fn(async () => { bridgeRequest?.onReady?.(); + bridgeRequest?.onTranscript?.("user", "hel", false); + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "response-1", + }); bridgeRequest?.onAudio(Buffer.from("audio-out")); bridgeRequest?.onMark?.("mark-1"); - bridgeRequest?.onTranscript?.("user", "hel", false); bridgeRequest?.onTranscript?.("user", "hello", true); bridgeRequest?.onTranscript?.("assistant", "hi there", true); bridgeRequest?.onToolCall?.({ @@ -1650,7 +1670,7 @@ describe("talk realtime gateway relay", () => { }); expectDelivery(toolCallPayload, false); - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-1", audioBase64: Buffer.from("audio-in").toString("base64"), @@ -1681,7 +1701,7 @@ describe("talk realtime gateway relay", () => { connId: "conn-1", markName: "mark-1", }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -2606,7 +2626,7 @@ describe("talk realtime gateway relay", () => { }); await Promise.resolve(); bridgeRequest?.onTranscript?.("user", "Cancel this consult", true); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: cancelledSession.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -2624,10 +2644,7 @@ describe("talk realtime gateway relay", () => { ); }), ).toBe(false); - stopTalkRealtimeRelaySession({ - relaySessionId: cancelledSession.relaySessionId, - connId: "conn-1", - }); + expect(relaySessions.has(cancelledSession.relaySessionId)).toBe(false); }); it("rejects relay control from a different connection", () => { @@ -2646,12 +2663,13 @@ describe("talk realtime gateway relay", () => { tools: [], }); - expect(() => - sendTalkRealtimeRelayAudio({ - relaySessionId: session.relaySessionId, - connId: "conn-2", - audioBase64: Buffer.from("audio").toString("base64"), - }), + expect( + () => + void sendTalkRealtimeRelayAudio({ + relaySessionId: session.relaySessionId, + connId: "conn-2", + audioBase64: Buffer.from("audio").toString("base64"), + }), ).toThrow("Unknown realtime relay session"); }); @@ -2681,7 +2699,7 @@ describe("talk realtime gateway relay", () => { tools: [], }); - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-1", audioBase64: Buffer.from("audio").toString("base64"), @@ -2690,6 +2708,11 @@ describe("talk realtime gateway relay", () => { for (let index = 0; index < audio.length; index += 1) { audio[index] = index % 251; } + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "response-1", + }); bridgeRequest?.onEvent?.({ direction: "server", type: "response.output_audio.delta", @@ -2759,11 +2782,33 @@ describe("talk realtime gateway relay", () => { expect(events.some((entry) => entry.payload.type === "error")).toBe(true); }); + it("ignores provider clear with no live response owner", () => { + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + bridgeRequest = request; + return makeRelayTransport(); + }; + const broadcastToConnIds = vi.fn(); + createTalkRealtimeRelaySession({ + context: { broadcastToConnIds } as never, + connId: "conn-1", + provider, + providerConfig: {}, + instructions: "brief", + tools: [], + }); + + bridgeRequest?.onClearAudio("barge-in"); + + expect(broadcastToConnIds).not.toHaveBeenCalled(); + }); + it("aborts linked agent consult runs when the relay turn is cancelled", () => { const { abortController, broadcast, nodeSendToSession, removeChatRun, chatRunState, session } = createAbortableRelayRunFixture(); relaySessions.get(session.relaySessionId)?.harness.talk.startTurn({ turnId: "turn-1" }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -2791,7 +2836,7 @@ describe("talk realtime gateway relay", () => { }; relay?.forcedTerminalProviderResults.set("call-1", forcedResult); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -2805,33 +2850,194 @@ describe("talk realtime gateway relay", () => { expect(broadcast).not.toHaveBeenCalled(); }); - it("ignores missing and empty turn cancellation without mutating relay state", () => { - const { abortController, broadcast, session } = createAbortableRelayRunFixture(); - const relay = relaySessions.get(session.relaySessionId); - expect(relay).toBeDefined(); - relay?.harness.talk.startTurn({ turnId: "turn-b" }); - const epoch = relay?.toolResultEpoch; - const forcedResult = { - result: { status: "cancelled" }, - turnId: "turn-b", - epoch: epoch ?? 0, - }; - relay?.forcedTerminalProviderResults.set("call-1", forcedResult); + it.each([undefined, "", " "])( + "preserves legacy current-turn cancellation for turn id %j", + async (turnId) => { + const { abortController, broadcast, session } = createAbortableRelayRunFixture(); + const relay = relaySessions.get(session.relaySessionId); + expect(relay).toBeDefined(); + relay?.harness.talk.startTurn({ turnId: "turn-b" }); + expect( + await cancelTalkRealtimeRelayTurn({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + reason: "barge-in", + turnId, + }), + ).toEqual({ status: "applied", turnId: "turn-b" }); - for (const turnId of [undefined, "", " "]) { - cancelTalkRealtimeRelayTurn({ + expect(relay?.harness.talk.activeTurnId).toBeUndefined(); + expect(abortController.signal.aborted).toBe(true); + expect(broadcast).toHaveBeenCalled(); + }, + ); + + it("recovers an exact-response relay only after matching provider cancellation", async () => { + vi.useFakeTimers(); + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const handleBargeIn = vi.fn(); + const close = vi.fn(); + const sendAudio = vi.fn(); + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + bridgeRequest = request; + return makeRelayTransport({ + handleBargeIn, + close, + sendAudio, + submitToolResult: vi.fn(() => { + throw new Error("provider rejected cancellation"); + }), + }); + }; + const { session } = createAbortableRelayRunFixture(provider); + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "response-1", + }); + + let cancellationSettled = false; + const cancellation = cancelTalkRealtimeRelayTurn({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + turnId: ensureActiveRelayTurnId(session.relaySessionId), + }); + void cancellation.then(() => (cancellationSettled = true)); + await Promise.resolve(); + expect(handleBargeIn).toHaveBeenCalledWith({ audioPlaybackActive: true }); + expect(relaySessions.has(session.relaySessionId)).toBe(true); + expect(cancellationSettled).toBe(false); + + const resumedAudio = sendTalkRealtimeRelayAudio({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + audioBase64: "AQI=", + }); + bridgeRequest?.onResponseDone?.({ status: "completed", responseId: "response-other" }); + await Promise.resolve(); + expect(cancellationSettled).toBe(false); + expect(sendAudio).not.toHaveBeenCalled(); + + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.cancelled", + responseId: "response-1", + }); + await expect(Promise.all([cancellation, resumedAudio])).resolves.toEqual([ + { status: "applied", turnId: expect.any(String) }, + undefined, + ]); + expect(sendAudio).toHaveBeenCalledWith(Buffer.from([1, 2])); + await vi.advanceTimersByTimeAsync(1_000); + expect(relaySessions.has(session.relaySessionId)).toBe(true); + expect(close).not.toHaveBeenCalled(); + }); + + it("closes an exact-response relay when cancellation is never confirmed", async () => { + vi.useFakeTimers(); + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const close = vi.fn(); + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + bridgeRequest = request; + return makeRelayTransport({ close }); + }; + const { session } = createAbortableRelayRunFixture(provider); + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "response-1", + }); + + let cancellationSettled = false; + const cancellation = cancelTalkRealtimeRelayTurn({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + turnId: ensureActiveRelayTurnId(session.relaySessionId), + }); + void cancellation.then(() => (cancellationSettled = true)); + await vi.advanceTimersByTimeAsync(999); + expect(relaySessions.has(session.relaySessionId)).toBe(true); + expect(cancellationSettled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); + expect(relaySessions.has(session.relaySessionId)).toBe(false); + expect(close).toHaveBeenCalledOnce(); + }); + + it("fails visibly when a replacement response starts before cancellation confirms", async () => { + let bridgeRequest: RealtimeVoiceBridgeCreateRequest | undefined; + const close = vi.fn(); + const provider = createIdleRelayProvider(); + provider.createBridge = (request) => { + bridgeRequest = request; + return makeRelayTransport({ close }); + }; + const { broadcastToConnIds, session } = createAbortableRelayRunFixture(provider); + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "response-1", + }); + const cancellation = cancelTalkRealtimeRelayTurn({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + turnId: ensureActiveRelayTurnId(session.relaySessionId), + }); + + bridgeRequest?.onEvent?.({ + direction: "server", + type: "response.created", + responseId: "response-2", + }); + + expect(close).toHaveBeenCalledOnce(); + expect(relaySessions.has(session.relaySessionId)).toBe(false); + await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); + expect( + broadcastToConnIds.mock.calls.some((call) => (call[1] as { type?: string }).type === "error"), + ).toBe(true); + }); + + it("closes a stalled turn-bound cancellation after its drain deadline", async () => { + vi.useFakeTimers(); + const pending = createDeferred(); + const close = vi.fn(); + const provider = createIdleRelayProvider(); + provider.createBridge = () => + makeRelayTransport({ close, submitToolResult: vi.fn(() => pending.promise) }); + const { session } = createAbortableRelayRunFixture(provider); + + let cancellationSettled = false; + const cancellation = cancelTalkRealtimeRelayTurn({ + relaySessionId: session.relaySessionId, + connId: "conn-1", + turnId: ensureActiveRelayTurnId(session.relaySessionId), + }); + void cancellation.then(() => (cancellationSettled = true)); + const pendingAudio = Promise.resolve( + sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-1", - reason: "barge-in", - turnId, - }); - } - - expect(relay?.harness.talk.activeTurnId).toBe("turn-b"); - expect(relay?.toolResultEpoch).toBe(epoch); - expect(relay?.forcedTerminalProviderResults.get("call-1")).toBe(forcedResult); - expect(abortController.signal.aborted).toBe(false); - expect(broadcast).not.toHaveBeenCalled(); + audioBase64: "AQI=", + }), + ); + let audioSettled = false; + void pendingAudio.then( + () => (audioSettled = true), + () => (audioSettled = true), + ); + await vi.advanceTimersByTimeAsync(999); + expect(relaySessions.has(session.relaySessionId)).toBe(true); + expect(cancellationSettled).toBe(false); + expect(audioSettled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await expect(cancellation).resolves.toEqual({ status: "applied", turnId: expect.any(String) }); + await expect(pendingAudio).rejects.toThrow("Unknown realtime relay session"); + expect(relaySessions.has(session.relaySessionId)).toBe(false); + expect(close).toHaveBeenCalledOnce(); + pending.resolve(); }); it("terminally satisfies a late normal result after turn cancellation without a new turn", async () => { @@ -2842,7 +3048,7 @@ describe("talk realtime gateway relay", () => { submitToolResult, }); const { broadcastToConnIds, session } = createAbortableRelayRunFixture(provider); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -2888,7 +3094,7 @@ describe("talk realtime gateway relay", () => { callId: "call-1", result: { ok: true }, }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -2935,6 +3141,7 @@ describe("talk realtime gateway relay", () => { instructions: "brief", tools: [], }); + ensureActiveRelayTurnId(session.relaySessionId); bridgeRequest?.onToolCall?.({ itemId: "item-1", callId: "call-1", @@ -3103,7 +3310,7 @@ describe("talk realtime gateway relay", () => { callId: "call-1", result: { answer: "stale" }, }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3162,14 +3369,16 @@ describe("talk realtime gateway relay", () => { result: { answer: "stale" }, }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", turnId: ensureActiveRelayTurnId(session.relaySessionId), }); + expect(relaySessions.has(session.relaySessionId)).toBe(true); workingAccepted.resolve(); await Promise.all([working, final]); + await vi.waitFor(() => expect(relaySessions.has(session.relaySessionId)).toBe(false)); expect(submitToolResult.mock.calls.map((call) => call[1])).toEqual([ { status: "working" }, @@ -3187,12 +3396,6 @@ describe("talk realtime gateway relay", () => { ), ).toBe(false); - await submitTalkRealtimeRelayToolResult({ - relaySessionId: session.relaySessionId, - connId: "conn-1", - callId: "call-1", - result: { error: "late abort" }, - }); expect(submitToolResult).toHaveBeenCalledTimes(2); }); @@ -3217,7 +3420,7 @@ describe("talk realtime gateway relay", () => { callId: "call-1", result: { answer: "stale" }, }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3306,7 +3509,7 @@ describe("talk realtime gateway relay", () => { ).toBeUndefined(); expect(bridge.submitToolResult).toHaveBeenCalledTimes(1); expect(toolResultEvents()).toHaveLength(1); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3349,7 +3552,7 @@ describe("talk realtime gateway relay", () => { ), ).toBe(false); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3489,7 +3692,7 @@ describe("talk realtime gateway relay", () => { text: "cancel that", mode: "cancel", }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3733,7 +3936,7 @@ describe("talk realtime gateway relay", () => { it("terminally cancels late forced working even when willContinue is set", async () => { const fixture = await createSuppressionUnsupportedForcedConsultFixture(["native-1"]); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: fixture.session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3761,10 +3964,7 @@ describe("talk realtime gateway relay", () => { }); expect(startedTurns()).toBe(beforeLateCalls); - expect(fixture.submitToolResult.mock.calls.map((call) => call[0])).toEqual([ - "native-2", - "native-1", - ]); + expect(fixture.submitToolResult.mock.calls.map((call) => call[0])).toEqual(["native-1"]); for (const call of fixture.submitToolResult.mock.calls) { expect(call[1]).toEqual({ status: "cancelled", @@ -3808,7 +4008,7 @@ describe("talk realtime gateway relay", () => { result: { answer: "stale" }, }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: fixture.session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -3831,7 +4031,7 @@ describe("talk realtime gateway relay", () => { (entry.payload as { type?: string; callId?: string }).type === "toolResult" && (entry.payload as { callId?: string }).callId === fixture.callId, ), - ).toBe(false); + ).toBe(true); }); it("supersedes a rejected forced final with canonical cancellation", async () => { @@ -3847,7 +4047,7 @@ describe("talk realtime gateway relay", () => { callId: fixture.callId, result: { answer: "stale" }, }); - cancelTalkRealtimeRelayTurn({ + void cancelTalkRealtimeRelayTurn({ relaySessionId: fixture.session.relaySessionId, connId: "conn-1", reason: "barge-in", @@ -4048,7 +4248,7 @@ describe("talk realtime gateway relay", () => { throw new Error("expected active relay"); } relay.expiresAtMs = Date.now() - 1; - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-1", audioBase64: "AQI=", @@ -4058,7 +4258,7 @@ describe("talk realtime gateway relay", () => { { name: "fails connection ownership", close: (session: { relaySessionId: string }) => { - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn-other", audioBase64: "AQI=", @@ -4178,6 +4378,7 @@ describe("talk realtime gateway relay", () => { }); const retainedSession = relaySessions.get(session.relaySessionId); expect(retainedSession).toBeDefined(); + ensureActiveRelayTurnId(session.relaySessionId); const toolCallEventCount = () => broadcastToConnIds.mock.calls.filter( ([, payload]) => diff --git a/src/gateway/talk-relay-audio-base64.test.ts b/src/gateway/talk-relay-audio-base64.test.ts index d58aa79e3b5f..94baa9b7caa3 100644 --- a/src/gateway/talk-relay-audio-base64.test.ts +++ b/src/gateway/talk-relay-audio-base64.test.ts @@ -102,12 +102,13 @@ describe("Talk relay audio base64", () => { realtime.set(session.relaySessionId, "conn"); await Promise.resolve(); events.length = 0; - expect(() => - sendTalkRealtimeRelayAudio({ - relaySessionId: session.relaySessionId, - connId: "conn", - audioBase64: "AB", - }), + expect( + () => + void sendTalkRealtimeRelayAudio({ + relaySessionId: session.relaySessionId, + connId: "conn", + audioBase64: "AB", + }), ).toThrow("Realtime relay audio frame is invalid base64"); expect(sendAudio).not.toHaveBeenCalled(); expect(events).toEqual([]); @@ -127,7 +128,7 @@ describe("Talk relay audio base64", () => { realtime.set(session.relaySessionId, "conn"); await Promise.resolve(); events.length = 0; - sendTalkRealtimeRelayAudio({ + void sendTalkRealtimeRelayAudio({ relaySessionId: session.relaySessionId, connId: "conn", audioBase64: "YXVkaW8taW4=", diff --git a/ui/src/i18n/locales/en.ts b/ui/src/i18n/locales/en.ts index 5d3f839fc15b..7388c95560e8 100644 --- a/ui/src/i18n/locales/en.ts +++ b/ui/src/i18n/locales/en.ts @@ -5948,6 +5948,7 @@ export const en: TranslationMap = { dictationProviderUnavailable: "No transcription provider is configured for dictation.", dictationRecording: "Recording {elapsed}", dictationReleaseToInsert: "Release to insert dictation", + realtimeTalkMissingTurnIdentity: "Realtime output is missing its turn identity.", realtimeTalkRequiresMicrophone: "Realtime voice input requires browser microphone access.", selectedMicrophoneUnavailable: "The selected microphone is unavailable. Choose another input or System default.", diff --git a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts index 88ecbfcfbb1b..3c78b85301be 100644 --- a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts +++ b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts @@ -142,7 +142,14 @@ function emitGatewayFrame(frame: GatewayFrame): void { } function emitTalkEvent(payload: unknown): void { - emitGatewayFrame({ event: "talk.event", payload }); + let eventPayload = payload; + if (payload && typeof payload === "object" && !Array.isArray(payload)) { + const event = payload as Record; + if ((event.type === "audio" || event.type === "clear") && event.talkEvent === undefined) { + eventPayload = { ...event, talkEvent: { turnId: "turn-1" } }; + } + } + emitGatewayFrame({ event: "talk.event", payload: eventPayload }); } function pumpMicrophone(samples: Float32Array): void { @@ -507,6 +514,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { { sessionId: "relay-1", reason: "playback-overflow", + turnId: "turn-1", }, ], ]), @@ -551,6 +559,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { { sessionId: "relay-1", reason: "playback-overflow", + turnId: "turn-1", }, ], ]), @@ -899,6 +908,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { { sessionId: "relay-1", reason: "barge-in", + turnId: "turn-1", }, ], ]); @@ -1257,9 +1267,17 @@ describe("GatewayRelayRealtimeTalkTransport", () => { { sessionId: "relay-1", reason: "barge-in", + turnId: "turn-1", }, ], ]); + const appendCountBeforeClear = requestCallsFor(client, "talk.session.appendAudio").length; + emitTalkEvent({ relaySessionId: "relay-1", type: "clear" }); + pumpMicrophone(speech); + await Promise.resolve(); + expect(requestCallsFor(client, "talk.session.appendAudio")).toHaveLength( + appendCountBeforeClear, + ); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -1269,6 +1287,12 @@ describe("GatewayRelayRealtimeTalkTransport", () => { pumpMicrophone(speech); pumpMicrophone(speech); expect(requestCallsFor(client, "talk.session.cancelOutput")).toHaveLength(2); + emitTalkEvent({ relaySessionId: "relay-1", type: "clear" }); + pumpMicrophone(speech); + await Promise.resolve(); + expect(requestCallsFor(client, "talk.session.appendAudio")).toHaveLength( + appendCountBeforeClear, + ); emitGatewayFrame({ event: "chat", payload: { runId: "run-1", state: "final", message: { text: "ready" } }, @@ -1278,10 +1302,20 @@ describe("GatewayRelayRealtimeTalkTransport", () => { resolveCancellations[0]?.(); await vi.advanceTimersByTimeAsync(0); + pumpMicrophone(speech); + await Promise.resolve(); + expect(requestCallsFor(client, "talk.session.appendAudio")).toHaveLength( + appendCountBeforeClear, + ); expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(0); resolveCancellations[1]?.(); await vi.advanceTimersByTimeAsync(0); + pumpMicrophone(speech); + await Promise.resolve(); + expect(requestCallsFor(client, "talk.session.appendAudio")).toHaveLength( + appendCountBeforeClear + 1, + ); expect(requestCallsFor(client, "talk.session.submitToolResult")).toEqual([ [ diff --git a/ui/src/pages/chat/realtime-talk-gateway-relay.ts b/ui/src/pages/chat/realtime-talk-gateway-relay.ts index 9ea6ee605252..7da9c79ab23d 100644 --- a/ui/src/pages/chat/realtime-talk-gateway-relay.ts +++ b/ui/src/pages/chat/realtime-talk-gateway-relay.ts @@ -1,3 +1,4 @@ +import { t } from "../../i18n/index.ts"; import { formatUiError } from "../../lib/format-error.ts"; import { bytesToBase64, @@ -59,6 +60,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport private readonly delayedToolResults = new Set(); private readonly markAckTimers = new Set(); private cancelRequestedForPlayback = false; + private activeOutputTurnId: string | null = null; private playbackOverflowed = false; private pendingOutputCancellations = 0; private speechFramesDuringPlayback = 0; @@ -196,6 +198,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport this.media?.getTracks().forEach((track) => track.stop()); this.media = null; this.playbackOverflowed = false; + this.activeOutputTurnId = null; this.stopOutput(); void this.inputContext?.close(); this.inputContext = null; @@ -217,7 +220,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport const abortController = this.audioAppendAbortController; // Live microphone frames become stale once the Gateway falls behind, so fail at // the ownership cap instead of silently dropping speech or growing a latency queue. - if (!abortController || abortController.signal.aborted) { + if (!abortController || abortController.signal.aborted || this.pendingOutputCancellations) { return; } if (this.pendingAudioAppends.size >= MAX_PENDING_AUDIO_APPENDS) { @@ -320,14 +323,28 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport return; case "audio": if (event.audioBase64 && !this.playbackOverflowed) { + const turnId = event.talkEvent?.turnId?.trim(); + if (!turnId) { + this.ctx.callbacks.onStatus?.( + "error", + t("chat.composer.realtimeTalkMissingTurnIdentity"), + ); + this.stop(); + return; + } + this.activeOutputTurnId = turnId; this.cancelRequestedForPlayback = false; this.speechFramesDuringPlayback = 0; this.playPcm16(event.audioBase64); } return; case "clear": + if (event.talkEvent?.turnId && event.talkEvent.turnId !== this.activeOutputTurnId) { + return; + } this.playbackOverflowed = false; this.stopOutput({ releaseDelayedToolResults: this.pendingOutputCancellations === 0 }); + this.activeOutputTurnId = null; if (event.talkEvent?.type === "turn.cancelled") { this.abortConsults(); } @@ -671,6 +688,12 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport if ((requirePlayback && !this.outputQueue.isPlaying) || this.cancelRequestedForPlayback) { return; } + const turnId = this.activeOutputTurnId; + if (!turnId) { + this.ctx.callbacks.onStatus?.("error", t("chat.composer.realtimeTalkMissingTurnIdentity")); + this.stop(); + return; + } this.cancelRequestedForPlayback = true; // Keep completed consult results until the Gateway records this cancellation. // Releasing earlier can let the provider answer from a turn the user interrupted. @@ -681,6 +704,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport .request("talk.session.cancelOutput", { sessionId: this.session.relaySessionId, reason, + turnId, }) .then( () => {