diff --git a/apps/ios/Sources/Voice/RealtimeTalkRelaySession.swift b/apps/ios/Sources/Voice/RealtimeTalkRelaySession.swift index 3fb47fa86278..98155b590c81 100644 --- a/apps/ios/Sources/Voice/RealtimeTalkRelaySession.swift +++ b/apps/ios/Sources/Voice/RealtimeTalkRelaySession.swift @@ -80,6 +80,11 @@ extension RealtimeTalkRelayTransport { guard await gateway.currentRoute() == route else { throw CancellationError() } return response }, - isCurrent: { await gateway.currentRoute() == route }) + isCurrent: { await gateway.currentRoute() == route }, + supportsOutputGeneration: { + await gateway.supportsServerCapability( + .talkOutputGeneration, + ifCurrentRoute: route) == true + }) } } diff --git a/apps/macos/Sources/OpenClaw/GatewayConnection+RealtimeTalk.swift b/apps/macos/Sources/OpenClaw/GatewayConnection+RealtimeTalk.swift index eb65cb0b38aa..37509e9ced19 100644 --- a/apps/macos/Sources/OpenClaw/GatewayConnection+RealtimeTalk.swift +++ b/apps/macos/Sources/OpenClaw/GatewayConnection+RealtimeTalk.swift @@ -38,6 +38,11 @@ extension GatewayConnection { }, isCurrent: { await self.isCurrentServerLease(lease) + }, + supportsOutputGeneration: { + await self.supportsServerCapability( + .talkOutputGeneration, + ifCurrentServerLease: lease) == true }) } diff --git a/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift b/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift index b0cfdd766a31..559a26aa20d7 100644 --- a/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift +++ b/apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift @@ -206,6 +206,26 @@ extension TalkModeRuntime { relayGeneration: relayGeneration) } + func setRealtimeInputPaused( + _ paused: Bool, + session: RealtimeTalkRelaySession, + relayGeneration: UInt64) async -> Bool + { + do { + try await MainActor.run { + try session.setInputPaused(paused) + } + return true + } catch { + self.logger.error( + "talk realtime pause transition failed: \(error.localizedDescription, privacy: .public)") + await self.handleRealtimeInputRestartFailure( + error.localizedDescription, + relayGeneration: relayGeneration) + return false + } + } + private func handleRealtimeTermination( _ termination: RealtimeTalkRelayTermination, relayGeneration: UInt64) async @@ -258,10 +278,11 @@ extension TalkModeRuntime { } } - private func handleRealtimeSpeakingChanged(_ speaking: Bool, relayGeneration: UInt64) async { + func handleRealtimeSpeakingChanged(_ speaking: Bool, relayGeneration: UInt64) async { guard self.realtimeRelayGeneration == relayGeneration, self.realtimeSession != nil, - self.isEnabled + self.isEnabled, + !self.isPaused else { return } if speaking { self.phase = .speaking @@ -272,29 +293,32 @@ extension TalkModeRuntime { } } - private func handleRealtimeInputLevel(_ level: Double, relayGeneration: UInt64) async { + func handleRealtimeInputLevel(_ level: Double, relayGeneration: UInt64) async { guard self.realtimeRelayGeneration == relayGeneration, self.realtimeSession != nil, - self.isEnabled + self.isEnabled, + !self.isPaused else { return } await MainActor.run { TalkModeController.shared.updateLevel(level) } } - private func handleRealtimeOutputLevel(_ level: Double?, relayGeneration: UInt64) async { + func handleRealtimeOutputLevel(_ level: Double?, relayGeneration: UInt64) async { guard self.realtimeRelayGeneration == relayGeneration, self.realtimeSession != nil, - self.isEnabled + self.isEnabled, + !self.isPaused else { return } await MainActor.run { TalkModeController.shared.updateSpeakingLevel(level) } } - private func handleRealtimeTranscript( + func handleRealtimeTranscript( _ transcript: RealtimeTalkTranscript, relayGeneration: UInt64) async { guard self.realtimeRelayGeneration == relayGeneration, self.realtimeSession != nil, - self.isEnabled + self.isEnabled, + !self.isPaused else { return } let text = transcript.text.trimmingCharacters(in: .whitespacesAndNewlines) guard !text.isEmpty else { return } diff --git a/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift b/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift index 7c588dcdc85e..3a85bf2f3187 100644 --- a/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift +++ b/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift @@ -104,6 +104,7 @@ actor TalkModeRuntime { var bypassRealtimeOnNextStart = false var realtimeRelayGeneration: UInt64 = 0 var realtimeRelayStartGeneration: UInt64? + private var pendingRealtimeRelayStartLifecycleGeneration: Int? var realtimeRestartGeneration: UInt64 = 0 var realtimeRestartTask: Task? var startDependencies: StartDependencies? @@ -157,29 +158,61 @@ actor TalkModeRuntime { guard self.isEnabled else { return } + if paused { + self.pendingRealtimeRelayStartLifecycleGeneration = nil + if self.realtimeRelayStartGeneration != nil { + self.realtimeRelayGeneration &+= 1 + } + } else if self.realtimeRelayStartGeneration != nil, self.shouldAttemptRealtimeRelay() { + self.pendingRealtimeRelayStartLifecycleGeneration = self.lifecycleGeneration + return + } + if paused, self.realtimeSession == nil { self.cancelScheduledRealtimeRecovery() } if let realtimeSession { let relayGeneration = self.realtimeRelayGeneration - do { - try await MainActor.run { - try realtimeSession.setInputPaused(paused) + if paused { + await MainActor.run { realtimeSession.setOutputPaused(true) } + guard self.isPaused, + self.realtimeRelayGeneration == relayGeneration, + self.realtimeSession === realtimeSession, + await self.setRealtimeInputPaused( + true, + session: realtimeSession, + relayGeneration: relayGeneration) + else { return } + self.lastTranscript = "" + self.lastHeard = nil + self.lastSpeechEnergyAt = nil + self.phase = .idle + await MainActor.run { + TalkModeController.shared.updateLevel(0) + TalkModeController.shared.updateSpeakingLevel(nil) + TalkModeController.shared.updatePartialTranscript("") + TalkModeController.shared.updatePhase(.idle) } - if !paused { - self.phase = .listening - await MainActor.run { TalkModeController.shared.updatePhase(.listening) } + } else { + guard await self.setRealtimeInputPaused( + false, + session: realtimeSession, + relayGeneration: relayGeneration), + !self.isPaused, + self.realtimeRelayGeneration == relayGeneration, + self.realtimeSession === realtimeSession + else { return } + await MainActor.run { realtimeSession.setOutputPaused(false) } + guard !self.isPaused, + self.realtimeRelayGeneration == relayGeneration, + self.realtimeSession === realtimeSession + else { + await MainActor.run { realtimeSession.setOutputPaused(true) } + return } - } catch { - // Only the unpause branch throws, and the session restores its own input-paused - // state on the way out. Without this the runtime would report Talk as running - // with no microphone, so route it through the typed close-and-recover path. - self.logger.error( - "talk realtime pause transition failed: \(error.localizedDescription, privacy: .public)") - await self.handleRealtimeInputRestartFailure( - error.localizedDescription, - relayGeneration: relayGeneration) + self.phase = .listening + await MainActor.run { TalkModeController.shared.updatePhase(.listening) } } return } @@ -264,8 +297,12 @@ actor TalkModeRuntime { } return } catch is CancellationError { + if self.consumePendingRealtimeRelayStart() { + await self.start() + } return } catch { + self.pendingRealtimeRelayStartLifecycleGeneration = nil guard self.canOwnNativeFallback(fallbackOwner) else { return } self.logger.error( "talk realtime unavailable; using native fallback: " + @@ -291,6 +328,7 @@ actor TalkModeRuntime { } private func stop() async { + self.pendingRealtimeRelayStartLifecycleGeneration = nil self.resetRealtimeRecoveryState() self.realtimeRelayGeneration &+= 1 self.realtimeRelayStartGeneration = nil @@ -321,6 +359,14 @@ actor TalkModeRuntime { } } + func consumePendingRealtimeRelayStart() -> Bool { + guard let generation = self.pendingRealtimeRelayStartLifecycleGeneration else { return false } + self.pendingRealtimeRelayStartLifecycleGeneration = nil + return generation == self.lifecycleGeneration && self.isEnabled && !self.isPaused && + self.realtimeSession == nil && self.realtimeRelayStartGeneration == nil && + self.shouldAttemptRealtimeRelay() + } + #if DEBUG func _test_enableRealtimeRelaySelection() { (self.macOSRealtimeRelayOptIn, self.hasGatewayRealtimeRelayTuple) = (true, true) diff --git a/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift b/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift index 87d39fe29c43..e369368c521c 100644 --- a/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift +++ b/apps/macos/Tests/OpenClawIPCTests/TalkModeRuntimeSpeechTests.swift @@ -13,12 +13,14 @@ private enum RuntimeTestAudioCaptureError: Error { private final class RuntimeTestAudioCapture: RealtimeTalkAudioCapturing { let suppressesInputDuringOutput = false var startError: Error? + private(set) var startCount = 0 func start( targetSampleRate: Double, onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void, onFailure: @escaping @MainActor (String) -> Void) throws { + self.startCount += 1 if let startError = self.startError { throw startError } @@ -148,6 +150,12 @@ private enum RuntimeRelayStartError: Error { case failed } +enum RuntimeRelayStartupPauseOutcome: Equatable { + case resume + case remainPaused + case disable +} + @MainActor private func makeRuntimeTestRealtimeSession( player: RuntimeTestPCMPlayer) -> RealtimeTalkRelaySession @@ -335,6 +343,84 @@ struct TalkModeRuntimeSpeechTests { session.stop() } + @Test @MainActor func `pausing realtime resets visible state and ignores late callbacks`() async { + let runtime = TalkModeRuntime() + let player = RuntimeTestPCMPlayer() + let session = makeRuntimeTestRealtimeSession(player: player) + let relayGeneration = await runtime._test_prepareEnabledRealtimeSessionForClose(session) + TalkModeController.shared.updatePhase(.speaking) + TalkModeController.shared.updateLevel(0.8) + TalkModeController.shared.updatePartialTranscript("stale") + + await runtime.setPaused(true) + await runtime.handleRealtimeSpeakingChanged(true, relayGeneration: relayGeneration) + await runtime.handleRealtimeInputLevel(0.9, relayGeneration: relayGeneration) + await runtime.handleRealtimeOutputLevel(0.8, relayGeneration: relayGeneration) + await runtime.handleRealtimeTranscript( + .init(role: "user", text: "late transcript", isFinal: false), + relayGeneration: relayGeneration) + + #expect(await runtime._test_phase() == .idle) + #expect(TalkModeController.shared.phase == .idle) + #expect(TalkModeController.shared.level == 0) + #expect(TalkModeController.shared.partialTranscript.isEmpty) + #expect(player.stopCount == 0) + + await runtime._test_cancelRealtimeRecovery() + session.stop() + } + + @Test @MainActor func `resuming realtime restarts input and reuses the relay`() async throws { + let runtime = TalkModeRuntime() + let audioCapture = RuntimeTestAudioCapture() + let player = RuntimeTestPCMPlayer() + let eventChannel = AsyncStream.makeStream() + let result = TalkSessionCreateResult( + sessionid: "talk-session", + mode: AnyCodable("realtime"), + transport: AnyCodable("gateway-relay"), + brain: AnyCodable("agent-consult"), + relaysessionid: "relay-1") + let resultData = try JSONEncoder().encode(result) + let session = RealtimeTalkRelaySession( + transport: RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in eventChannel.stream }, + request: { method, _, _ in + if method == "talk.session.create" { + eventChannel.continuation.yield(EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "ready", + ]), + seq: nil, + stateversion: nil)) + return resultData + } + return Data("{\"ok\":true}".utf8) + }), + options: .init(sessionKey: "main", provider: "openai", model: "gpt-realtime-2", voice: nil), + audioCapture: audioCapture, + pcmPlayer: player, + onStatus: { _ in }, + onSpeakingChanged: { _ in }) + try await session.start() + let relayGeneration = await runtime._test_prepareEnabledRealtimeSessionForClose(session) + + await runtime.setPaused(true) + await runtime.setPaused(false) + + #expect(audioCapture.startCount == 2) + #expect(await runtime._test_realtimeSessionIs(session)) + await runtime.handleRealtimeSpeakingChanged(true, relayGeneration: relayGeneration) + #expect(await runtime._test_phase() == .speaking) + + await runtime._test_cancelRealtimeRecovery() + session.stop() + eventChannel.continuation.finish() + } + @Test @MainActor func `disabling during relay startup stops the published session`() async { let runtime = TalkModeRuntime() let lifecycleGeneration = await runtime._test_prepareEnabledLifecycle() @@ -368,6 +454,58 @@ struct TalkModeRuntimeSpeechTests { #expect(player.stopCount == 1) } + @Test(arguments: [ + RuntimeRelayStartupPauseOutcome.resume, + .remainPaused, + .disable, + ]) + @MainActor + func `relay startup pause retries only a matching resume`( + outcome: RuntimeRelayStartupPauseOutcome) async + { + let runtime = TalkModeRuntime() + let lifecycleGeneration = await runtime._test_prepareEnabledLifecycle() + await runtime._test_enableRealtimeRelaySelection() + let barrier = RuntimeContinuationBarrier() + let probe = RuntimeCommitProbe() + let player = RuntimeTestPCMPlayer() + let session = makeRuntimeTestRealtimeSession(player: player) + let attempt = Task { @MainActor in + do { + try await runtime._test_startRealtimeRelay( + lifecycleGeneration: lifecycleGeneration, + makeSession: { session }, + start: { _ in + probe.record("start") + await barrier.wait() + }) + return true + } catch { + return false + } + } + + await barrier.waitUntilEntered() + #expect(await runtime._test_realtimeSessionIs(session)) + await runtime.setPaused(true) + if outcome != .remainPaused { + await runtime.setPaused(false) + } + if outcome == .disable { + await runtime.setEnabled(false) + } + await barrier.release() + + #expect(await attempt.value == false) + #expect(await runtime._test_realtimeSessionIsActive() == false) + if await runtime.consumePendingRealtimeRelayStart() { probe.record("retry") } + if await runtime.consumePendingRealtimeRelayStart() { probe.record("retry") } + #expect(probe.values() == (outcome == .resume ? ["start", "retry"] : ["start"])) + #expect(player.stopCount == 1) + + await runtime.setEnabled(false) + } + @Test func `processed recognition start failure retries a fresh raw capture`() { let probe = RuntimeCommitProbe() diff --git a/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift b/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift index d58db3b1e0c0..4c55231b092b 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift @@ -65,15 +65,18 @@ public struct RealtimeTalkRelayTransport: Sendable { public let subscribeServerEvents: @Sendable (Int) async -> AsyncStream public let request: @Sendable (String, [String: AnyCodable]?, Double) async throws -> Data public let isCurrent: @Sendable () async -> Bool + public let supportsOutputGeneration: @Sendable () async -> Bool public init( subscribeServerEvents: @escaping @Sendable (Int) async -> AsyncStream, request: @escaping @Sendable (String, [String: AnyCodable]?, Double) async throws -> Data, - isCurrent: @escaping @Sendable () async -> Bool = { true }) + isCurrent: @escaping @Sendable () async -> Bool = { true }, + supportsOutputGeneration: @escaping @Sendable () async -> Bool = { false }) { self.subscribeServerEvents = subscribeServerEvents self.request = request self.isCurrent = isCurrent + self.supportsOutputGeneration = supportsOutputGeneration } } @@ -118,6 +121,7 @@ public enum RealtimeTalkRelayTermination: Equatable, Sendable { case remoteClose(reason: String?) case eventStreamEnded case audioCaptureFailed(message: String) + case outputCancellationFailed } private actor RealtimeAudioSender { @@ -257,10 +261,17 @@ public final class RealtimeTalkRelaySession { private var pendingPlaybackMarks: [String] = [] private var audioSender: RealtimeAudioSender? private var isInputPaused = false + private var isOutputPaused = false private var audioCaptureGeneration: UInt64 = 0 private var isClosed = false private var lifecycleGeneration: UInt64 = 0 + private var outputCancellationGeneration: UInt64 = 0 private var isOutputPlaying = false + private var outputIdentity: OutputIdentity? + private var suppressedOutputIdentity: OutputIdentity? + private var awaitingOutputClear = false + private var cancelledOutputGenerationWatermark: Int? + private var outputCancellationTask: Task? private var outputStartedAtMs: Double? private var outputPlaybackExpectedEndMs: Double = 0 private var lastBargeInAtMs: Double = 0 @@ -334,6 +345,16 @@ public final class RealtimeTalkRelaySession { } return } + if let startupIssue { + if let relaySessionId = result.relaysessionid?.trimmingCharacters(in: .whitespacesAndNewlines), + !relaySessionId.isEmpty + { + await Self.closeRelaySession( + transport: self.transport, + relaySessionId: relaySessionId) + } + throw Self.startupFailureError(startupIssue) + } guard let relaySessionId = result.relaysessionid?.trimmingCharacters(in: .whitespacesAndNewlines), !relaySessionId.isEmpty else { @@ -397,14 +418,13 @@ public final class RealtimeTalkRelaySession { for task in self.toolCallTasks.values { task.cancel() } - for task in self.audioSendTasks.values { - task.cancel() - } - self.audioSendTasks.removeAll() self.pendingPlaybackMarks.removeAll() let audioSender = self.audioSender self.audioSender = nil Task { await audioSender?.close() } + self.retireOutputCancellation() + self.cancelledOutputGenerationWatermark = nil + self.isOutputPaused = false self.stopOutputPlayback() if sendClose, let relaySessionId = self.relaySessionId { Task { [transport] in @@ -423,6 +443,12 @@ public final class RealtimeTalkRelaySession { ]) } + private nonisolated static func startupFailureError(_ issue: RealtimeTalkRelayIssue) -> NSError { + NSError(domain: "RealtimeTalkRelay", code: 6, userInfo: [ + NSLocalizedDescriptionKey: issue.message, + ]) + } + private nonisolated static func closeRelaySession( transport: RealtimeTalkRelayTransport, relaySessionId: String) async @@ -431,18 +457,6 @@ public final class RealtimeTalkRelaySession { _ = try? await transport.request("talk.session.close", payload, 8000) } - public func cancelOutput(reason: String = "user") { - self.stopOutputPlayback() - guard let relaySessionId else { return } - Task { [transport] in - let payload: [String: AnyCodable] = [ - "sessionId": AnyCodable(relaySessionId), - "reason": AnyCodable(reason), - ] - _ = try? await transport.request("talk.session.cancelOutput", payload, 8000) - } - } - public func setInputPaused(_ paused: Bool) throws { guard self.isInputPaused != paused else { return } self.isInputPaused = paused @@ -459,6 +473,14 @@ public final class RealtimeTalkRelaySession { } } + public func setOutputPaused(_ paused: Bool) { + guard self.isOutputPaused != paused else { return } + self.isOutputPaused = paused + if paused, self.isOutputPlaying { + self.cancelOutput(reason: "pause") + } + } + private func createRelaySession() async throws -> TalkSessionCreateResult { var payload: [String: AnyCodable] = [ "sessionKey": AnyCodable(self.options.sessionKey), @@ -504,7 +526,9 @@ public final class RealtimeTalkRelaySession { await self?.handleEventStreamEnded(lifecycleGeneration: lifecycleGeneration) } } +} +extension RealtimeTalkRelaySession { private func handleEventStreamEnded(lifecycleGeneration: UInt64) async { guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return } self.logger.debug("talk realtime: event stream ended") @@ -550,25 +574,11 @@ public final class RealtimeTalkRelaySession { self.finishStartupWait(.ready) self.onStatus("Listening (Realtime)") case "audio": - guard let base64 = payload["audioBase64"]?.stringValue, - let data = Data(base64Encoded: base64) - else { return } - self.recordOutputAudioChunk(byteCount: data.count) - self.markOutputAudioStarted(byteCount: data.count, nowMs: ProcessInfo.processInfo.systemUptime * 1000) - self.onSpeakingChanged(true) - if self.outputContinuation == nil, self.outputTask != nil { - self.pendingOutputChunks.append(data) - return - } - self.ensureOutputPlaybackStarted() - self.outputEnvelope?.append(data) - self.outputContinuation?.yield(data) + self.handleOutputAudio(payload) case "audioDone": self.finishOutputPlaybackStream() case "clear": - let marks = self.takePendingPlaybackMarks() - self.stopOutputPlayback() - self.acknowledgePlaybackMarks(marks) + self.handleOutputClear(payload) case "mark": self.handlePlaybackMark(payload) case "transcript": @@ -613,6 +623,38 @@ public final class RealtimeTalkRelaySession { } } + private func handleOutputClear(_ payload: [String: AnyCodable]) { + let clearIdentity = OutputIdentity(payload) + if self.awaitingOutputClear, + let suppressed = self.suppressedOutputIdentity + { + let clearsSuppressed = + clearIdentity.isEmpty() + ? suppressed.isEmpty() + : suppressed.isEmpty() || suppressed.relation(to: clearIdentity) == .same + if clearsSuppressed { + if let generation = clearIdentity.outputGeneration ?? suppressed.outputGeneration { + self.cancelledOutputGenerationWatermark = max( + self.cancelledOutputGenerationWatermark ?? 0, + generation) + } + self.retireOutputCancellation() + } + } + let currentMatches = + clearIdentity.isEmpty() + ? self.outputIdentity == nil + : self.outputIdentity?.relation(to: clearIdentity) == .same + guard currentMatches else { return } + let marks = self.takePendingPlaybackMarks() + // Cancellation already published the stopped state. A later clear with no + // active output only retires the fence; it must not emit a duplicate callback. + if self.isOutputPlaying || self.outputIdentity != nil { + self.stopOutputPlayback() + } + self.acknowledgePlaybackMarks(marks) + } + private func waitForStartupResult( timeoutSeconds: Int, lifecycleGeneration: UInt64) async -> StartupWaitResult @@ -975,7 +1017,7 @@ public final class RealtimeTalkRelaySession { } private func startPendingOutputPlaybackIfNeeded() { - guard !self.pendingOutputChunks.isEmpty else { + guard !self.isOutputPaused, !self.pendingOutputChunks.isEmpty else { self.pendingOutputDone = false return } @@ -1022,6 +1064,7 @@ public final class RealtimeTalkRelaySession { self.outputIdleTask = nil } self.isOutputPlaying = false + self.outputIdentity = nil self.outputStartedAtMs = nil self.outputPlaybackExpectedEndMs = 0 self.outputEnvelope?.cancel() @@ -1079,6 +1122,7 @@ public final class RealtimeTalkRelaySession { self.pendingOutputDone = false _ = self.pcmPlayer.stop() self.isOutputPlaying = false + self.outputIdentity = nil self.outputStartedAtMs = nil self.outputPlaybackExpectedEndMs = 0 self.outputEnvelope?.cancel() @@ -1126,11 +1170,143 @@ public final class RealtimeTalkRelaySession { } } +extension RealtimeTalkRelaySession { + private struct OutputIdentity { + enum Relation { + case same + case different + case unknown + } + + let turnId: String? + let outputGeneration: Int? + + init(_ payload: [String: AnyCodable]) { + let turnId = payload["talkEvent"]?.dictionaryValue?["turnId"]?.stringValue? + .trimmingCharacters(in: .whitespacesAndNewlines) + self.turnId = turnId?.isEmpty == false ? turnId : nil + self.outputGeneration = RealtimeTalkRelaySession.outputGeneration(payload["outputGeneration"]) + } + + func isEmpty() -> Bool { + self.turnId == nil && self.outputGeneration == nil + } + + func relation(to other: OutputIdentity) -> Relation { + if self.isEmpty() || other.isEmpty() { + return self.isEmpty() == other.isEmpty() ? .same : .different + } + if let outputGeneration, let otherGeneration = other.outputGeneration { + return outputGeneration == otherGeneration ? .same : .different + } + if let turnId, let otherTurnId = other.turnId { + return turnId == otherTurnId ? .same : .different + } + return .unknown + } + } + + @discardableResult + public func cancelOutputIfPlaying(reason: String = "user") -> Bool { + guard self.isOutputPlaying else { return false } + self.cancelOutput(reason: reason) + return true + } + + public func cancelOutput(reason: String = "user") { + guard let relaySessionId else { return } + let outputIdentity = self.outputIdentity ?? OutputIdentity([:]) + self.outputCancellationGeneration &+= 1 + let cancellationGeneration = self.outputCancellationGeneration + self.outputCancellationTask?.cancel() + self.suppressedOutputIdentity = outputIdentity + self.awaitingOutputClear = true + self.stopOutputPlayback() + self.outputCancellationTask = Task { [weak self, transport] in + var payload: [String: AnyCodable] = [ + "sessionId": AnyCodable(relaySessionId), + "reason": AnyCodable(reason), + ] + if let turnId = outputIdentity.turnId { + payload["turnId"] = AnyCodable(turnId) + } + if await transport.supportsOutputGeneration(), + let outputGeneration = outputIdentity.outputGeneration + { + payload["outputGeneration"] = AnyCodable(outputGeneration) + } + do { + _ = try await transport.request("talk.session.cancelOutput", payload, 8000) + } catch { + guard let self, self.isCurrentOutputCancellation(cancellationGeneration) else { return } + let issue = RealtimeTalkRelayIssue( + code: "realtime_output_cancel_failed", + message: error.localizedDescription, + provider: self.options.provider, + model: self.options.model, + transport: "gateway-relay", + phase: "output-cancel") + self.onIssue(issue) + self.onStatus(issue.message) + // A failed current cancellation leaves remote output ownership unknown. + // Keep the fence until terminal teardown makes late audio impossible. + self.close(sendClose: true) + self.onTermination(.outputCancellationFailed) + } + } + } + + private func isCurrentOutputCancellation(_ generation: UInt64) -> Bool { + generation == self.outputCancellationGeneration && !self.isClosed + } + + private func handleOutputAudio(_ payload: [String: AnyCodable]) { + guard !self.isOutputPaused else { return } + guard let base64 = payload["audioBase64"]?.stringValue, + let data = Data(base64Encoded: base64) + else { return } + let incomingIdentity = OutputIdentity(payload) + guard !self.awaitingOutputClear else { return } + if let watermark = self.cancelledOutputGenerationWatermark { + guard let generation = incomingIdentity.outputGeneration, generation > watermark else { return } + } + if !incomingIdentity.isEmpty() { + self.outputIdentity = incomingIdentity + } + self.recordOutputAudioChunk(byteCount: data.count) + self.markOutputAudioStarted(byteCount: data.count, nowMs: ProcessInfo.processInfo.systemUptime * 1000) + self.onSpeakingChanged(true) + if self.outputContinuation == nil, self.outputTask != nil { + self.pendingOutputChunks.append(data) + return + } + self.ensureOutputPlaybackStarted() + self.outputEnvelope?.append(data) + self.outputContinuation?.yield(data) + } + + private func retireOutputCancellation() { + self.outputCancellationGeneration &+= 1 + self.outputCancellationTask?.cancel() + self.outputCancellationTask = nil + self.suppressedOutputIdentity = nil + self.awaitingOutputClear = false + } + + private nonisolated static func outputGeneration(_ value: AnyCodable?) -> Int? { + guard let raw = value?.doubleValue, + raw > 0, + raw <= Double(Int.max), + raw.rounded(.towardZero) == raw + else { return nil } + return Int(raw) + } +} + extension RealtimeTalkRelaySession { private func startMicrophonePump(lifecycleGeneration: UInt64) throws { self.stopMicrophonePump() guard !self.isInputPaused else { return } - self.audioCaptureGeneration &+= 1 let audioCaptureGeneration = self.audioCaptureGeneration try self.audioCapture.start( targetSampleRate: self.inputSampleRateHz, @@ -1206,9 +1382,15 @@ extension RealtimeTalkRelaySession { let task = Task { @MainActor [weak self, audioSender] in guard let self else { return } defer { self.audioSendTasks.removeValue(forKey: taskID) } - guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return } + guard self.isCurrentLifecycleLocally(lifecycleGeneration), + self.audioCaptureGeneration == audioCaptureGeneration, + !self.isInputPaused + else { return } guard let message = await audioSender.send(encoded, timestampMs: timestampMs) else { return } - guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return } + guard self.isCurrentLifecycleLocally(lifecycleGeneration), + self.audioCaptureGeneration == audioCaptureGeneration, + !self.isInputPaused + else { return } self.onStatus("Realtime audio failed: \(message)") } self.audioSendTasks[taskID] = task @@ -1249,6 +1431,10 @@ extension RealtimeTalkRelaySession { private func stopMicrophonePump() { self.audioCaptureGeneration &+= 1 + for task in self.audioSendTasks.values { + task.cancel() + } + self.audioSendTasks.removeAll() self.audioCapture.stop() } } diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift index 13a91de833f4..860252b1e02f 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift @@ -14,6 +14,20 @@ private final class UnusedPCMStreamingAudioPlayer: PCMStreamingAudioPlaying { } } +@MainActor +private final class DrainingPCMStreamingAudioPlayer: PCMStreamingAudioPlaying { + func play(stream: AsyncThrowingStream, sampleRate _: Double) async -> StreamingPlaybackResult { + do { + for try await _ in stream {} + } catch {} + return StreamingPlaybackResult(finished: true, interruptedAt: nil) + } + + func stop() -> Double? { + nil + } +} + @MainActor private final class TestRealtimeTalkAudioCapture: RealtimeTalkAudioCapturing { var suppressesInputDuringOutput = false @@ -97,14 +111,81 @@ private actor RealtimeRelayRouteFlag { } } +private actor RealtimeRelayEventSource { + private var continuation: AsyncStream.Continuation? + + func stream() -> AsyncStream { + AsyncStream { self.continuation = $0 } + } + + func finish() { + self.continuation?.finish() + } +} + private func unusedRealtimeRelayTransport() -> RealtimeTalkRelayTransport { RealtimeTalkRelayTransport( subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, request: { _, _, _ in throw CancellationError() }) } +private func outputAudioEvent(generation: Int) -> EventFrame { + EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "audio", + "audioBase64": Data([0x01]).base64EncodedString(), + "outputGeneration": generation, + ]), + seq: nil, + stateversion: nil) +} + @MainActor struct RealtimeTalkRelaySessionTests { + enum CancellationRetirement { + case clear + case close + } + + private func makeIdleCancellationSession( + _ onSpeakingChanged: @escaping (Bool) -> Void) -> RealtimeTalkRelaySession + { + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { _, _, _ in Data("{\"ok\":true}".utf8) }) + let session = RealtimeTalkRelaySession( + transport: transport, + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: onSpeakingChanged) + session._test_setRelaySessionId("relay-1") + return session + } + + private func makeAudioSendSession() -> (RealtimeTalkRelaySession, RealtimeRelayStartupRequestLog) { + let requests = RealtimeRelayStartupRequestLog() + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + return Data("{\"ok\":true}".utf8) + }) + let session = RealtimeTalkRelaySession( + transport: transport, + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: UnusedPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { _ in }) + session._test_prepareAudioSender(relaySessionId: "relay-1") + return (session, requests) + } + @Test func `transcript callback carries typed partial and final values`() async { var transcripts: [RealtimeTalkTranscript] = [] let session = RealtimeTalkRelaySession( @@ -226,6 +307,356 @@ struct RealtimeTalkRelaySessionTests { #expect(request.params?["sessionId"]?.stringValue == "relay-1") #expect(request.params?["markName"]?.stringValue == "audio-1") } +} + +extension RealtimeTalkRelaySessionTests { + @Test func `output cancellation fences delayed audio and preserves exact identity`() async throws { + let requests = RealtimeRelayStartupRequestLog() + var speakingStates: [Bool] = [] + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + return Data("{\"ok\":true}".utf8) + }, + supportsOutputGeneration: { true }) + let session = RealtimeTalkRelaySession( + transport: transport, + 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") + let audio: (Int, String) -> EventFrame = { generation, turnId in + EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "audio", + "audioBase64": Data([0x01]).base64EncodedString(), + "outputGeneration": generation, + "talkEvent": ["turnId": turnId], + ]), + seq: nil, + stateversion: nil) + } + let clear: (Int) -> EventFrame = { generation in + EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "clear", + "outputGeneration": generation, + ]), + seq: nil, + stateversion: nil) + } + + await session._test_handleGatewayEvent(audio(7, "turn-7")) + #expect(session.cancelOutputIfPlaying(reason: "barge-in")) + for _ in 0..<10 { + if await !requests.snapshot().isEmpty { break } + await Task.yield() + } + let request = try #require(await requests.snapshot().first) + #expect(request.method == "talk.session.cancelOutput") + #expect(request.params?["sessionId"]?.stringValue == "relay-1") + #expect(request.params?["turnId"]?.stringValue == "turn-7") + #expect(request.params?["outputGeneration"]?.doubleValue == 7) + #expect(request.params?["reason"]?.stringValue == "barge-in") + + await session._test_handleGatewayEvent(audio(7, "turn-7")) + await session._test_handleGatewayEvent(audio(8, "turn-8")) + #expect(speakingStates.first == true) + #expect(!speakingStates.dropFirst().contains(true)) + await session._test_handleGatewayEvent(clear(7)) + await session._test_handleGatewayEvent(audio(7, "turn-7")) + #expect(speakingStates == [true, false]) + await session._test_handleGatewayEvent(audio(8, "turn-8")) + #expect(speakingStates == [true, false, true]) + await session._test_handleGatewayEvent(clear(7)) + #expect(speakingStates == [true, false, true]) + await session._test_handleGatewayEvent(clear(8)) + #expect(speakingStates == [true, false, true, false]) + } + + @Test func `legacy cancellation keeps turn identity but omits generation`() async throws { + let requests = RealtimeRelayStartupRequestLog() + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + return Data("{\"ok\":true}".utf8) + }) + let session = RealtimeTalkRelaySession( + transport: transport, + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { _ in }) + session._test_setRelaySessionId("relay-1") + await session._test_handleGatewayEvent(EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "audio", + "audioBase64": Data([0x01]).base64EncodedString(), + "outputGeneration": 7, + "talkEvent": ["turnId": "turn-7"], + ]), + seq: nil, + stateversion: nil)) + + #expect(session.cancelOutputIfPlaying(reason: "barge-in")) + for _ in 0..<10 { + if await !requests.snapshot().isEmpty { break } + await Task.yield() + } + let request = try #require(await requests.snapshot().first) + #expect(request.params?["turnId"]?.stringValue == "turn-7") + #expect(request.params?["outputGeneration"] == nil) + } + + @Test func `idle cancellation and pause retain the relay without false interruption`() async { + let requests = RealtimeRelayStartupRequestLog() + var speakingStates: [Bool] = [] + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + return Data("{\"ok\":true}".utf8) + }, + supportsOutputGeneration: { true }) + let session = RealtimeTalkRelaySession( + transport: transport, + 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") + + #expect(!session.cancelOutputIfPlaying()) + session.setOutputPaused(true) + await session._test_handleGatewayEvent(outputAudioEvent(generation: 1)) + #expect(speakingStates.isEmpty) + #expect(await requests.snapshot().isEmpty) + session.setOutputPaused(false) + await session._test_handleGatewayEvent(outputAudioEvent(generation: 2)) + #expect(speakingStates == [true]) + #expect(await requests.snapshot().isEmpty) + } + + @Test func `idle cancellation waits for clear while cancellation without relay stays unfenced`() async { + var speakingStates: [Bool] = [] + let session = self.makeIdleCancellationSession { speakingStates.append($0) } + session.cancelOutput(reason: "barge-in") + await session._test_handleGatewayEvent(outputAudioEvent(generation: 1)) + #expect(speakingStates == [false]) + await session._test_handleGatewayEvent(EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "clear", + ]), + seq: nil, + stateversion: nil)) + await session._test_handleGatewayEvent(outputAudioEvent(generation: 1)) + #expect(speakingStates == [false, true]) + + var unfencedStates: [Bool] = [] + let unfenced = RealtimeTalkRelaySession( + transport: unusedRealtimeRelayTransport(), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { unfencedStates.append($0) }) + unfenced.cancelOutput() + unfenced._test_setRelaySessionId("relay-1") + await unfenced._test_handleGatewayEvent(outputAudioEvent(generation: 1)) + #expect(unfencedStates == [true]) + } + + @Test func `active output pause cancels the exact generation`() async throws { + 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("{\"ok\":true}".utf8) + }, + supportsOutputGeneration: { true }), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onSpeakingChanged: { _ in }) + session._test_setRelaySessionId("relay-1") + await session._test_handleGatewayEvent(EventFrame( + type: "event", + event: "talk.event", + payload: AnyCodable([ + "relaySessionId": "relay-1", + "type": "audio", + "audioBase64": Data([0x01]).base64EncodedString(), + "outputGeneration": 7, + "talkEvent": ["turnId": "turn-7"], + ]), + seq: nil, + stateversion: nil)) + + session.setOutputPaused(true) + for _ in 0..<10 { + if await !requests.snapshot().isEmpty { break } + await Task.yield() + } + let request = try #require(await requests.snapshot().first) + #expect(request.params?["turnId"]?.stringValue == "turn-7") + #expect(request.params?["outputGeneration"]?.doubleValue == 7) + #expect(request.params?["reason"]?.stringValue == "pause") + } + + @Test func `current cancellation failure terminates and rejects late audio`() async { + let requests = RealtimeRelayStartupRequestLog() + var issues: [RealtimeTalkRelayIssue] = [] + var terminations: [RealtimeTalkRelayTermination] = [] + var speakingStates: [Bool] = [] + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + if method == "talk.session.cancelOutput" { + throw URLError(.cannotConnectToHost) + } + return Data("{\"ok\":true}".utf8) + }) + let session = RealtimeTalkRelaySession( + transport: transport, + options: .init(sessionKey: "main", provider: "openai", model: "gpt-realtime-2", voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onIssue: { issues.append($0) }, + onTermination: { terminations.append($0) }, + onSpeakingChanged: { speakingStates.append($0) }) + session._test_setRelaySessionId("relay-1") + await session._test_handleGatewayEvent(outputAudioEvent(generation: 1)) + + session.cancelOutput() + for _ in 0..<50 { + if !issues.isEmpty { break } + await Task.yield() + } + await session._test_handleGatewayEvent(outputAudioEvent(generation: 2)) + for _ in 0..<50 { + if await requests.snapshot().contains(where: { $0.method == "talk.session.close" }) { break } + await Task.yield() + } + + #expect(issues.map(\.code) == ["realtime_output_cancel_failed"]) + #expect(issues.map(\.phase) == ["output-cancel"]) + #expect(terminations == [.outputCancellationFailed]) + #expect(await requests.snapshot().map(\.method) == [ + "talk.session.cancelOutput", + "talk.session.close", + ]) + #expect(speakingStates.first == true) + #expect(!speakingStates.dropFirst().contains(true)) + } + + @Test func `superseded cancellation failure leaves the active fence intact`() async { + let barrier = RealtimeRelayStartupBarrier() + let requests = RealtimeRelayStartupRequestLog() + var issues: [RealtimeTalkRelayIssue] = [] + var speakingStates: [Bool] = [] + let transport = RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, params, _ in + await requests.record(method: method, params: params) + if await requests.snapshot().count == 1 { + await barrier.suspend() + throw URLError(.cancelled) + } + return Data("{\"ok\":true}".utf8) + }) + let session = RealtimeTalkRelaySession( + transport: transport, + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onIssue: { issues.append($0) }, + onSpeakingChanged: { speakingStates.append($0) }) + session._test_setRelaySessionId("relay-1") + + session.cancelOutput() + await barrier.waitUntilEntered() + session.cancelOutput() + await barrier.release() + for _ in 0..<10 { + if await requests.snapshot().count == 2 { break } + await Task.yield() + } + await session._test_handleGatewayEvent(outputAudioEvent(generation: 1)) + + #expect(issues.isEmpty) + #expect(await requests.snapshot().count == 2) + #expect(!speakingStates.contains(true)) + } + + @Test(arguments: [CancellationRetirement.clear, .close]) + func `clear and close retire in flight cancellation failures`( + retirement: CancellationRetirement) async + { + let barrier = RealtimeRelayStartupBarrier() + var issues: [RealtimeTalkRelayIssue] = [] + let session = RealtimeTalkRelaySession( + transport: RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, + request: { method, _, _ in + if method == "talk.session.cancelOutput" { + await barrier.suspend() + throw URLError(.cancelled) + } + return Data("{\"ok\":true}".utf8) + }), + options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), + audioCapture: TestRealtimeTalkAudioCapture(), + pcmPlayer: DrainingPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onIssue: { issues.append($0) }, + onSpeakingChanged: { _ in }) + session._test_setRelaySessionId("relay-1") + + 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", + ]), + seq: nil, + stateversion: nil)) + case .close: + session.stop() + } + await barrier.release() + await Task.yield() + + #expect(issues.isEmpty) + } @Test func `close after classified error does not replace issue`() async { var issues: [RealtimeTalkRelayIssue] = [] @@ -376,6 +807,59 @@ struct RealtimeTalkRelaySessionTests { #expect(!audioCapture.isStarted) } + @Test func `event stream ending during relay creation closes the late relay`() async throws { + let barrier = RealtimeRelayStartupBarrier() + let events = RealtimeRelayEventSource() + let requests = RealtimeRelayStartupRequestLog() + let audioCapture = TestRealtimeTalkAudioCapture() + var issues: [RealtimeTalkRelayIssue] = [] + let result = TalkSessionCreateResult( + sessionid: "talk-session", + mode: AnyCodable("realtime"), + transport: AnyCodable("gateway-relay"), + brain: AnyCodable("agent-consult"), + relaysessionid: "relay-1") + let resultData = try JSONEncoder().encode(result) + let session = RealtimeTalkRelaySession( + transport: RealtimeTalkRelayTransport( + subscribeServerEvents: { _ in await events.stream() }, + request: { method, params, _ in + await requests.record(method: method, params: params) + if method == "talk.session.create" { + await barrier.suspend() + return resultData + } + return Data("{\"ok\":true}".utf8) + }), + options: .init(sessionKey: "main", provider: "openai", model: "gpt-realtime-2", voice: nil), + audioCapture: audioCapture, + pcmPlayer: UnusedPCMStreamingAudioPlayer(), + onStatus: { _ in }, + onIssue: { issues.append($0) }, + onSpeakingChanged: { _ in }) + let start = Task { @MainActor in + do { + try await session.start() + return nil as String? + } catch { + return error.localizedDescription + } + } + await barrier.waitUntilEntered() + + await events.finish() + while issues.isEmpty { + await Task.yield() + } + await barrier.release() + + #expect(await start.value == "Realtime connection ended before it became ready.") + #expect(audioCapture.startCount == 0) + let recorded = await requests.snapshot() + #expect(recorded.map(\.method) == ["talk.session.create", "talk.session.close"]) + #expect(recorded.last?.params?["sessionId"]?.stringValue == "relay-1") + } + @Test func `microphone failure terminates relay and reports typed issue`() async throws { let requests = RealtimeRelayStartupRequestLog() let transport = RealtimeTalkRelayTransport( @@ -636,28 +1120,23 @@ struct RealtimeTalkRelaySessionTests { #expect(statuses == ["Thinking…"]) } - @Test func `stop cancels buffered microphone audio before dispatch`() async throws { - let requests = RealtimeRelayStartupRequestLog() - let transport = RealtimeTalkRelayTransport( - subscribeServerEvents: { _ in AsyncStream { $0.finish() } }, - request: { method, params, _ in - await requests.record(method: method, params: params) - return Data("{\"ok\":true}".utf8) - }) - let session = RealtimeTalkRelaySession( - transport: transport, - options: .init(sessionKey: "main", provider: "openai", model: nil, voice: nil), - audioCapture: TestRealtimeTalkAudioCapture(), - pcmPlayer: UnusedPCMStreamingAudioPlayer(), - onStatus: { _ in }, - onSpeakingChanged: { _ in }) - session._test_prepareAudioSender(relaySessionId: "relay-1") - let send = try #require(session._test_enqueueMicrophoneFrame(Data([0x01, 0x02]))) + @Test func `stop and pause discard buffered microphone audio before dispatch`() async throws { + let (stoppedSession, stoppedRequests) = self.makeAudioSendSession() + let stoppedSend = try #require(stoppedSession._test_enqueueMicrophoneFrame(Data([0x01]))) - session.stop() - await send.value + stoppedSession.stop() + await stoppedSend.value + #expect(await stoppedRequests.snapshot().isEmpty) - #expect(await requests.snapshot().isEmpty) + let (session, requests) = self.makeAudioSendSession() + _ = try #require(session._test_enqueueMicrophoneFrame(Data([0x01]))) + try session.setInputPaused(true) + try session.setInputPaused(false) + try await (#require(session._test_enqueueMicrophoneFrame(Data([0x02])))).value + + #expect(await requests.snapshot().compactMap { $0.params?["audioBase64"]?.stringValue } == [ + Data([0x02]).base64EncodedString(), + ]) } @Test func `gateway route lost during startup fails instead of reporting ready`() async throws {