fix(talk): harden realtime relay cancellation ownership

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",

apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt

@@ -385,6 +385,8 @@ class TalkModeManager internal constructor(
   private val pendingRealtimePlaybackMarks = LinkedHashMap<String, PendingRealtimePlaybackMark>()
 
   @Volatile private var pendingRealtimeOutputClear: CompletableDeferred<Unit>? = 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<Unit>()
       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() }

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<Unit>())
+    assertFalse(shouldAppendRealtimeCapturedFrame(manager, 4_800))
+    setPrivateField(manager, "pendingRealtimeOutputClear", null)
+    assertTrue(shouldAppendRealtimeCapturedFrame(manager, 4_800))
   }
 
   @Test

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 {

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))
         }
     }
 

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,

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 }

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()
     }
 

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)
     }

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),

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<Data, Error>.Continuation?
         let stream = AsyncThrowingStream<Data, Error> { 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<Data, Error>.Continuation?
         let stream = AsyncThrowingStream<Data, Error> { 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<Data, Error>.Continuation?
         let secondStream = AsyncThrowingStream<Data, Error> { 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 {

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<Void, Never>?
+        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()
 

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.

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
 {

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,

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);
     }

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", {

src/gateway/talk-realtime-relay-forced-consults.ts

@@ -63,12 +63,6 @@ export function buildAlreadyDeliveredToolResult(): Record<string, string> {
   };
 }
 
-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<void> {
-  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<void> => 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<void> => submission !== undefined);
+  const submissions = submitPending();
+  if (submissions.length === 0) {
+    return;
   }
+  const settle = (pending: Promise<void>[]) =>
+    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<void> {
-  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<void> => 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) {

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<void> {
   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. */

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<typeof harness.createBridge> } = {};
+  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<typeof params.provider.createBridge>[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
       ? {

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<void>; 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;

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]) =>

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=",

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.",

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<string, unknown>;
+    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([
       [

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<DelayedToolResult>();
   private readonly markAckTimers = new Set<number>();
   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(
         () => {