ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-23 02:45:38 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(talk): bound Apple relay audio lifecycle

Browse Source 
Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>
This commit is contained in:
Vincent Koc
2026-08-21 06:08:24 -07:00
parent 697b2967f6
commit e9055a8aef
9 changed files with 3056 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/.i18n/native-source.json
Generated
+136
View File
@@ -29897,6 +29897,17 @@
}
]
},
{
"id": "native.apple.024d68fa654acf2c",
"source": "Gateway connection was replaced before realtime startup finished",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.165da8a03dd68ee1",
"source": "Gateway data changed while you edited. Save will validate the current revision.",
@@ -29919,6 +29930,17 @@
}
]
},
{
"id": "native.apple.da84239bbdff6838",
"source": "Gateway did not return a realtime relay session",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.8104be3229e9088b",
"source": "Gateway is not connected",
@@ -29927,6 +29949,10 @@
{
"kind": "ui-localized-call",
"path": "apps/ios/Sources/Design/SettingsSystemAgentChat.swift"
},
{
"kind": "ui-localized-call",
"path": "apps/ios/Sources/Voice/TalkModeManager.swift"
}
]
},
@@ -39326,6 +39352,83 @@
}
]
},
{
"id": "native.apple.3459811c080ed880",
"source": "Realtime audio failed: %@",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.4f81777177325153",
"source": "Realtime audio input fell behind. Reconnecting…",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.1d3037a7bc3ec7cb",
"source": "Realtime audio playback failed. Reconnecting…",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.90f64f78a77015d7",
"source": "Realtime audio playback fell behind. Reconnecting…",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.f40c2764425389ef",
"source": "Realtime closed before it became ready.",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.59d22abf27122e23",
"source": "Realtime connection ended before it became ready.",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.b236c658fc54cbe5",
"source": "Realtime did not become ready in time.",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.afd0d1396586ab0b",
"source": "Realtime disconnected",
@@ -39337,6 +39440,17 @@
}
]
},
{
"id": "native.apple.ca3a762269827f37",
"source": "Realtime failed",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.f350f10af66f4a87",
"source": "Realtime failed before connecting",
@@ -39348,6 +39462,28 @@
}
]
},
{
"id": "native.apple.96ece1b431ac3f96",
"source": "Realtime output cancellation failed: %@",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.83f84908b64f015c",
"source": "Realtime tool call did not return a run id",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift"
}
]
},
{
"id": "native.apple.651a2b218e29ac62",
"source": "Realtime unavailable",
apps/ios/Sources/Voice/RealtimeTalkRelaySession.swift
+2 -1
View File
@@ -36,7 +36,8 @@ final class IOSRealtimeTalkAudioCapture: RealtimeTalkAudioCapturing {
func start(
targetSampleRate: Double,
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void) throws
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void,
onFailure _: @escaping @MainActor (String) -> Void) throws
{
self.stop()
let input = self.audioEngine.inputNode
apps/ios/Sources/Voice/TalkModeGatewayConfig.swift
+2
View File
@@ -9,6 +9,8 @@ enum TalkModeExecutionMode: Equatable {
struct TalkRuntimeIssue: Equatable {
enum Code: String {
case audioInputUnavailable = "audio_input_unavailable"
case realtimeOutputCancelFailed = "realtime_output_cancel_failed"
case realtimeUnavailable = "realtime_unavailable"
}
apps/ios/Sources/Voice/TalkModeManager.swift
+36 -16
View File
@@ -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 {
@@ -2353,26 +2356,24 @@ final class TalkModeManager: NSObject {
model: self.realtimeModelId,
voice: self.realtimeVoiceId),
audioCapture: IOSRealtimeTalkAudioCapture(),
pcmPlayer: self.pcmPlayer,
pcmPlayer: RealtimePCMStreamingAudioPlayer(),
onStatus: { [weak self] status in
guard let self, self.realtimeRelayGeneration == relayGeneration else { return }
self.handleRealtimeRelayStatus(status)
},
onIssue: { [weak self] relayIssue in
guard let self, self.realtimeRelayGeneration == relayGeneration else { return }
let issue = TalkRuntimeIssue(
code: .realtimeUnavailable,
message: relayIssue.message,
provider: relayIssue.provider,
model: relayIssue.model,
transport: relayIssue.transport,
phase: relayIssue.phase)
let issue = Self.runtimeIssue(from: relayIssue)
self.realtimeRelayStartIssue = issue
self.pendingRealtimeIssue = issue
self.gatewayTalkLastIssueText = issue.diagnosticSummary
self.gatewayTalkActiveModeTitle = String(localized: "Realtime unavailable")
self.gatewayTalkActiveModeSubtitle = issue.displayMessage
},
onTermination: { [weak self] termination in
guard let self, self.realtimeRelayGeneration == relayGeneration else { return }
self.handleRealtimeRelayTermination(termination)
},
onSpeakingChanged: { [weak self] speaking in
guard let self, self.realtimeRelayGeneration == relayGeneration else { return }
self.isSpeaking = speaking
@@ -4159,17 +4160,13 @@ extension TalkModeManager {
let phase = Self.phase(forRealtimeStatus: status)
if status == "Listening (Realtime)" {
// Ready can be followed by a buffered close before start() resumes. Commit continuous
// state here so the close still enters bounded recovery.
// state here so the typed terminal callback still enters bounded recovery.
self.markRealtimeSessionReady()
} else {
self.setStatus(
Self.presentationText(forRealtimeStatus: status),
phase: phase,
watchPresentation: Self.watchPresentation(forRealtimeStatus: status))
if status == "Ready" {
self.realtimeRelaySession = nil
self.handleRealtimeSessionFinish()
}
}
self.isListening = phase == .listening
if phase == .thinking || phase == .connecting {
@@ -4179,6 +4176,13 @@ extension TalkModeManager {
}
}
private func handleRealtimeRelayTermination(_ termination: RealtimeTalkRelayTermination) {
GatewayDiagnostics.log("talk realtime relay terminated reason=\(String(describing: termination))")
self.realtimeRelaySession = nil
guard self.captureMode != .pushToTalk else { return }
self.handleRealtimeSessionFinish()
}
private func prepareRealtimeRelayStart() {
self.realtimeRelayStartIssue = nil
self.pendingRealtimeIssue = nil
@@ -4210,6 +4214,16 @@ extension TalkModeManager {
phase: phase)
}
static func runtimeIssue(from issue: RealtimeTalkRelayIssue) -> TalkRuntimeIssue {
TalkRuntimeIssue(
code: TalkRuntimeIssue.Code(rawValue: issue.code) ?? .realtimeUnavailable,
message: issue.message,
provider: issue.provider,
model: issue.model,
transport: issue.transport,
phase: issue.phase)
}
private func realtimeIssue(from error: Error, phase: String) -> TalkRuntimeIssue {
if let gatewayError = error as? GatewayResponseError,
let issue = Self.talkRuntimeIssue(
@@ -5097,6 +5111,12 @@ extension TalkModeManager {
self.handleRealtimeRelayStatus(status)
}
func _test_handleRealtimeRelayTermination(
_ termination: RealtimeTalkRelayTermination = .remoteClose(reason: "completed"))
{
self.handleRealtimeRelayTermination(termination)
}
func _test_prepareEnabledRealtimeSessionForClose() {
self.isEnabled = true
self.gatewayConnected = true
apps/ios/Tests/TalkModeConfigParsingTests.swift
+12
View File
@@ -403,6 +403,16 @@ struct TalkModeManagerTests {
#expect(issue.technicalDetails.contains("code: realtime_unavailable"))
}
@Test func `relay issue preserves known code and falls back for unknown code`() {
let codes = ["audio_input_unavailable", "realtime_output_cancel_failed", "future_code"].map { rawCode in
TalkModeManager.runtimeIssue(from: RealtimeTalkRelayIssue(
code: rawCode,
message: "failed")).code
}
#expect(codes == [.audioInputUnavailable, .realtimeOutputCancelFailed, .realtimeUnavailable])
}
@Test func `native fallback keeps realtime issue visible`() {
let manager = TalkModeManager(allowSimulatorCapture: true)
let issue = TalkRuntimeIssue(
@@ -480,6 +490,7 @@ struct TalkModeManagerTests {
#expect(manager._test_gatewayTalkActiveModeTitle() != "Not active")
manager._test_handleRealtimeRelayStatus("Ready")
manager._test_handleRealtimeRelayTermination()
#expect(manager.statusText == "Ready")
#expect(manager._test_gatewayTalkActiveModeTitle() == "Not active")
@@ -524,6 +535,7 @@ struct TalkModeManagerTests {
manager._test_handleRealtimeRelayStatus("Listening (Realtime)")
manager._test_handleRealtimeRelayStatus("Ready")
manager._test_handleRealtimeRelayTermination()
#expect(manager.statusText == "Reconnecting")
#expect(manager._test_rapidRealtimeRestartCount() == 1)
apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimePCMStreamingAudioPlayer.swift
+237
View File
@@ -0,0 +1,237 @@
#if Talk && canImport(ElevenLabsKit) && (os(iOS) || os(macOS))
import AVFAudio
import Foundation
@MainActor
public final class RealtimePCMStreamingAudioPlayer: PCMStreamingAudioPlaying {
static let frameDurationSeconds = 0.020
static let maxScheduledBuffers = 3
typealias Completion = @Sendable () -> Void
private let preparePlayback: (Double) throws -> Void
private let scheduleFrame: (Data, Double, @escaping Completion) throws -> Void
private let stopPlayback: () -> Void
private let playbackTime: () -> Double?
private var generation: UInt64 = 0
private var nextBufferID: UInt64 = 0
private var scheduledBufferIDs: Set<UInt64> = []
private var slotWaiters: [CheckedContinuation<Bool, Never>] = []
private var playbackContinuation: CheckedContinuation<StreamingPlaybackResult, Never>?
private var inputTask: Task<Void, Never>?
private var inputFinished = false
public convenience init() {
let engine = AVAudioEngine()
let node = AVAudioPlayerNode()
engine.attach(node)
var format: AVAudioFormat?
self.init(
preparePlayback: { sampleRate in
node.stop()
engine.stop()
engine.disconnectNodeOutput(node)
guard let nextFormat = AVAudioFormat(
commonFormat: .pcmFormatInt16,
sampleRate: sampleRate,
channels: 1,
interleaved: false)
else {
throw NSError(domain: "RealtimePCMStreamingAudioPlayer", code: 1)
}
format = nextFormat
engine.connect(node, to: engine.mainMixerNode, format: nextFormat)
engine.prepare()
try engine.start()
node.play()
},
scheduleFrame: { data, _, completion in
guard let format else {
throw NSError(domain: "RealtimePCMStreamingAudioPlayer", code: 2)
}
let frames = AVAudioFrameCount(data.count / MemoryLayout<Int16>.size)
guard let buffer = AVAudioPCMBuffer(pcmFormat: format, frameCapacity: frames),
let channel = buffer.int16ChannelData?[0]
else {
throw NSError(domain: "RealtimePCMStreamingAudioPlayer", code: 3)
}
buffer.frameLength = frames
data.copyBytes(
to: UnsafeMutableRawBufferPointer(
start: channel,
count: data.count))
node.scheduleBuffer(
buffer,
completionCallbackType: .dataPlayedBack)
{ _ in completion() }
},
stopPlayback: {
node.stop()
engine.stop()
},
playbackTime: {
guard let renderTime = node.lastRenderTime,
let playerTime = node.playerTime(forNodeTime: renderTime)
else { return nil }
return Double(playerTime.sampleTime) / playerTime.sampleRate
})
}
init(
preparePlayback: @escaping (Double) throws -> Void,
scheduleFrame: @escaping (Data, Double, @escaping Completion) throws -> Void,
stopPlayback: @escaping () -> Void,
playbackTime: @escaping () -> Double?)
{
self.preparePlayback = preparePlayback
self.scheduleFrame = scheduleFrame
self.stopPlayback = stopPlayback
self.playbackTime = playbackTime
}
public func play(
stream: AsyncThrowingStream<Data, Error>,
sampleRate: Double) async -> StreamingPlaybackResult
{
_ = self.stop()
guard sampleRate > 0 else {
return StreamingPlaybackResult(finished: false, interruptedAt: nil)
}
self.generation &+= 1
let generation = self.generation
do {
try self.preparePlayback(sampleRate)
} catch {
return StreamingPlaybackResult(finished: false, interruptedAt: nil)
}
return await withCheckedContinuation { continuation in
self.playbackContinuation = continuation
self.inputTask = Task { @MainActor [weak self] in
await self?.consume(stream: stream, sampleRate: sampleRate, generation: generation)
}
}
}
public func stop() -> Double? {
let interruptedAt = self.playbackTime()
self.generation &+= 1
self.inputTask?.cancel()
self.inputTask = nil
self.inputFinished = false
self.scheduledBufferIDs.removeAll()
let waiters = self.slotWaiters
self.slotWaiters.removeAll()
for waiter in waiters {
waiter.resume(returning: false)
}
let continuation = self.playbackContinuation
self.playbackContinuation = nil
self.stopPlayback()
continuation?.resume(returning: StreamingPlaybackResult(
finished: false,
interruptedAt: interruptedAt))
return interruptedAt
}
private func consume(
stream: AsyncThrowingStream<Data, Error>,
sampleRate: Double,
generation: UInt64) async
{
let frameBytes = max(
MemoryLayout<Int16>.size,
Int((sampleRate * Self.frameDurationSeconds).rounded()) * MemoryLayout<Int16>.size)
var pending = Data()
do {
for try await chunk in stream {
try Task.checkCancellation()
pending.append(chunk)
while pending.count >= frameBytes {
let frame = Data(pending.prefix(frameBytes))
pending.removeFirst(frameBytes)
guard await self.schedule(
frame: frame,
sampleRate: sampleRate,
generation: generation)
else { return }
}
}
if !pending.isEmpty {
pending.append(Data(repeating: 0, count: frameBytes - pending.count))
guard await self.schedule(
frame: pending,
sampleRate: sampleRate,
generation: generation)
else { return }
}
guard self.generation == generation else { return }
self.inputFinished = true
self.finishIfDrained(generation: generation)
} catch {
self.finish(generation: generation, finished: false)
}
}
private func schedule(frame: Data, sampleRate: Double, generation: UInt64) async -> Bool {
while self.generation == generation,
self.scheduledBufferIDs.count >= Self.maxScheduledBuffers
{
let admitted = await withCheckedContinuation { continuation in
self.slotWaiters.append(continuation)
}
guard admitted else { return false }
}
guard self.generation == generation, !Task.isCancelled else { return false }
self.nextBufferID &+= 1
let bufferID = self.nextBufferID
self.scheduledBufferIDs.insert(bufferID)
do {
try self.scheduleFrame(frame, sampleRate) { [weak self] in
Task { @MainActor in
self?.completed(bufferID: bufferID, generation: generation)
}
}
return true
} catch {
self.scheduledBufferIDs.remove(bufferID)
self.finish(generation: generation, finished: false)
return false
}
}
private func completed(bufferID: UInt64, generation: UInt64) {
guard self.generation == generation,
self.scheduledBufferIDs.remove(bufferID) != nil
else { return }
if !self.slotWaiters.isEmpty {
self.slotWaiters.removeFirst().resume(returning: true)
}
self.finishIfDrained(generation: generation)
}
private func finishIfDrained(generation: UInt64) {
guard self.inputFinished, self.scheduledBufferIDs.isEmpty else { return }
self.finish(generation: generation, finished: true)
}
private func finish(generation: UInt64, finished: Bool) {
guard self.generation == generation else { return }
let interruptedAt = finished ? nil : self.playbackTime()
self.generation &+= 1
self.scheduledBufferIDs.removeAll()
let waiters = self.slotWaiters
self.slotWaiters.removeAll()
for waiter in waiters {
waiter.resume(returning: false)
}
self.inputTask = nil
self.inputFinished = false
let continuation = self.playbackContinuation
self.playbackContinuation = nil
self.stopPlayback()
continuation?.resume(returning: StreamingPlaybackResult(
finished: finished,
interruptedAt: interruptedAt))
}
}
#endif
apps/shared/OpenClawKit/Sources/OpenClawKit/RealtimeTalkRelaySession.swift
+503 -112
View File
@@ -55,7 +55,8 @@ public protocol RealtimeTalkAudioCapturing: AnyObject {
func start(
targetSampleRate: Double,
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void) throws
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void,
onFailure: @escaping @MainActor (String) -> Void) throws
func stop()
}
@@ -113,6 +114,18 @@ public struct RealtimeTalkTranscript: Equatable, Sendable {
}
}
public enum RealtimeTalkRelayTermination: Equatable, Sendable {
case remoteClose(reason: String?)
case eventStreamEnded
case audioInputFailed(message: String)
case outputCancellationFailed
case outputPlaybackOverflow
}
private enum RealtimeAudioSendOutcome {
case sent, inactive, saturated, failed(String)
}
private actor RealtimeAudioSender {
private let request: @Sendable (String, [String: AnyCodable]?, Double) async throws -> Data
private var relaySessionId: String?
@@ -131,25 +144,27 @@ private actor RealtimeAudioSender {
self.relaySessionId = nil
}
func send(_ data: Data, timestampMs: Double) async -> String? {
guard !Task.isCancelled else { return nil }
guard let relaySessionId else { return nil }
guard self.pendingSends < self.maxPendingSends else { return nil }
func send(_ data: Data, timestampMs: Double) async -> RealtimeAudioSendOutcome {
guard !Task.isCancelled, let relaySessionId else { return .inactive }
guard self.pendingSends < self.maxPendingSends else { return .saturated }
self.pendingSends += 1
defer { self.pendingSends -= 1 }
// The Gateway carries this straight into the provider's media timeline, and OpenAI rejects
// a `conversation.item.truncate` whose `audio_end_ms` is not an integer -- a fractional
// timestamp here kills the session on the first barge-in.
let payload: [String: AnyCodable] = [
"sessionId": AnyCodable(relaySessionId),
"audioBase64": AnyCodable(data.base64EncodedString()),
"timestamp": AnyCodable(timestampMs),
"timestamp": AnyCodable(timestampMs.rounded()),
]
do {
try Task.checkCancellation()
let response = try await self.request("talk.session.appendAudio", payload, 8000)
try Task.checkCancellation()
_ = try JSONDecoder().decode(TalkSessionOkResult.self, from: response)
return nil
return .sent
} catch {
return error.localizedDescription
return Task.isCancelled ? .inactive : .failed(error.localizedDescription)
}
}
}
@@ -194,6 +209,15 @@ public final class RealtimeTalkRelaySession {
case cancelled
}
/// Startup abandons for two very different reasons and callers must not treat them alike.
/// Local cancellation is caller-initiated and stays silent; a lost Gateway route is an
/// external failure the runtime has to see, or Talk reports listening with no relay behind it.
private enum LifecycleStatus {
case current
case cancelledLocally
case routeLost
}
private nonisolated static let expectedInputEncoding = "pcm16"
private nonisolated static let expectedOutputEncoding = "pcm16"
private nonisolated static let defaultSampleRateHz = 24000
@@ -201,6 +225,9 @@ public final class RealtimeTalkRelaySession {
private nonisolated static let bargeInCooldownMs: Double = 900
private nonisolated static let minOutputBeforeBargeInMs: Double = 250
private nonisolated static let startupReadyTimeoutSeconds = 12
/// At the protocol's 20 ms cadence this bounds queued relay audio to 640 ms / 30,720 bytes.
/// Overflow terminates the session so recovery replaces a lagging playback path.
private nonisolated static let maxBufferedOutputChunks = 32
private let transport: RealtimeTalkRelayTransport
private let audioCapture: any RealtimeTalkAudioCapturing
@@ -209,6 +236,7 @@ public final class RealtimeTalkRelaySession {
private let logger = Logger(subsystem: "ai.openclawfoundation.app", category: "RealtimeTalkRelay")
private let onStatus: (String) -> Void
private let onIssue: (RealtimeTalkRelayIssue) -> Void
private let onTermination: (RealtimeTalkRelayTermination) -> Void
private let onSpeakingChanged: (Bool) -> Void
private let onInputLevel: (Double) -> Void
private let onOutputLevel: (Double?) -> Void
@@ -230,17 +258,25 @@ public final class RealtimeTalkRelaySession {
private var audioSendTasks: [UUID: Task<Void, Never>] = [:]
private var outputTask: Task<Void, Never>?
private var outputContinuation: AsyncThrowingStream<Data, Error>.Continuation?
/// Provider deltas may span any number of frames; retain only the partial tail so the
/// AsyncStream's 32 slots always contain bounded 20 ms PCM chunks.
private var pendingOutputAudio = Data()
private var outputIdleTask: Task<Void, Never>?
private var outputSessionId = 0
private var pendingOutputChunks: [Data] = []
private var pendingOutputDone = false
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 cancelledOutputTurnId: String?
private var outputCancellationTask: Task<Void, Never>?
private var outputStartedAtMs: Double?
private var outputPlaybackExpectedEndMs: Double = 0
private var lastBargeInAtMs: Double = 0
@@ -262,6 +298,7 @@ public final class RealtimeTalkRelaySession {
pcmPlayer: PCMStreamingAudioPlaying,
onStatus: @escaping (String) -> Void,
onIssue: @escaping (RealtimeTalkRelayIssue) -> Void = { _ in },
onTermination: @escaping (RealtimeTalkRelayTermination) -> Void = { _ in },
onSpeakingChanged: @escaping (Bool) -> Void,
onInputLevel: @escaping (Double) -> Void = { _ in },
onOutputLevel: @escaping (Double?) -> Void = { _ in },
@@ -273,6 +310,7 @@ public final class RealtimeTalkRelaySession {
self.pcmPlayer = pcmPlayer
self.onStatus = onStatus
self.onIssue = onIssue
self.onTermination = onTermination
self.onSpeakingChanged = onSpeakingChanged
self.onInputLevel = onInputLevel
self.onOutputLevel = onOutputLevel
@@ -290,11 +328,16 @@ public final class RealtimeTalkRelaySession {
self.pendingPreRelayEvents.removeAll()
self.onStatus("Connecting realtime…")
let eventStream = await self.transport.subscribeServerEvents(200)
guard await self.isCurrentLifecycle(lifecycleGeneration) else { return }
switch await self.lifecycleStatus(lifecycleGeneration) {
case .current: break
case .cancelledLocally: return
case .routeLost: throw Self.gatewayRouteLostError()
}
self.startEventPump(stream: eventStream, lifecycleGeneration: lifecycleGeneration)
do {
let result = try await self.createRelaySession()
guard await self.isCurrentLifecycle(lifecycleGeneration) else {
let statusAfterCreate = await self.lifecycleStatus(lifecycleGeneration)
if statusAfterCreate != .current {
if let relaySessionId = result.relaysessionid?.trimmingCharacters(in: .whitespacesAndNewlines),
!relaySessionId.isEmpty
{
@@ -302,13 +345,27 @@ public final class RealtimeTalkRelaySession {
transport: self.transport,
relaySessionId: relaySessionId)
}
if statusAfterCreate == .routeLost {
throw Self.gatewayRouteLostError()
}
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 {
throw NSError(domain: "RealtimeTalkRelay", code: 1, userInfo: [
NSLocalizedDescriptionKey: "Gateway did not return a realtime relay session",
NSLocalizedDescriptionKey: String(
localized: "Gateway did not return a realtime relay session"),
])
}
self.relaySessionId = relaySessionId
@@ -319,7 +376,11 @@ public final class RealtimeTalkRelaySession {
try self.startMicrophonePump(lifecycleGeneration: lifecycleGeneration)
self.onStatus("Waiting for realtime…")
await self.drainPendingPreRelayEvents(lifecycleGeneration: lifecycleGeneration)
guard await self.isCurrentLifecycle(lifecycleGeneration) else { return }
switch await self.lifecycleStatus(lifecycleGeneration) {
case .current: break
case .cancelledLocally: return
case .routeLost: throw Self.gatewayRouteLostError()
}
switch await self.waitForStartupResult(
timeoutSeconds: Self.startupReadyTimeoutSeconds,
lifecycleGeneration: lifecycleGeneration)
@@ -327,7 +388,6 @@ public final class RealtimeTalkRelaySession {
case .ready:
return
case let .failed(issue):
self.close(sendClose: true)
throw NSError(domain: "RealtimeTalkRelay", code: 6, userInfo: [
NSLocalizedDescriptionKey: issue.message,
])
@@ -335,7 +395,9 @@ public final class RealtimeTalkRelaySession {
return
}
} catch {
guard await self.isCurrentLifecycle(lifecycleGeneration) else { return }
// A lost route must still surface: swallowing here would discard both the original
// failure and the route loss, leaving the runtime with nothing to fall back from.
if await self.lifecycleStatus(lifecycleGeneration) == .cancelledLocally { return }
let createdRelaySessionId = self.relaySessionId
self.close(sendClose: false)
if let createdRelaySessionId {
@@ -362,14 +424,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.cancelledOutputTurnId = nil
self.isOutputPaused = false
self.stopOutputPlayback()
if sendClose, let relaySessionId = self.relaySessionId {
Task { [transport] in
@@ -380,6 +441,21 @@ public final class RealtimeTalkRelaySession {
self.onSpeakingChanged(false)
}
/// Deliberately not a `CancellationError`: the runtime treats those as caller-initiated and
/// returns silently, while any other error routes Talk to its native fallback.
private nonisolated static func gatewayRouteLostError() -> NSError {
NSError(domain: "RealtimeTalkRelay", code: 7, userInfo: [
NSLocalizedDescriptionKey: String(
localized: "Gateway connection was replaced before realtime startup finished"),
])
}
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
@@ -388,18 +464,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
@@ -416,6 +480,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),
@@ -457,8 +529,35 @@ public final class RealtimeTalkRelaySession {
if Task.isCancelled { return }
await self?.handleGatewayEvent(event, lifecycleGeneration: lifecycleGeneration)
}
guard !Task.isCancelled else { return }
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")
guard self.hasReceivedReady else {
guard !self.hasReceivedFailure else { return }
let issue = RealtimeTalkRelayIssue(
message: String(localized: "Realtime connection ended before it became ready."),
provider: self.options.provider,
model: self.options.model,
transport: "gateway-relay",
phase: "connect")
self.hasReceivedFailure = true
self.startupIssue = issue
self.onIssue(issue)
self.onStatus(issue.message)
self.finishStartupWait(.failed(issue))
return
}
self.onStatus("Ready")
self.close(sendClose: false)
self.onTermination(.eventStreamEnded)
}
private func handleGatewayEvent(_ event: EventFrame, lifecycleGeneration: UInt64) async {
guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return }
@@ -482,25 +581,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()
self.handleOutputAudioDone(payload)
case "clear":
let marks = self.takePendingPlaybackMarks()
self.stopOutputPlayback()
self.acknowledgePlaybackMarks(marks)
self.handleOutputClear(payload)
case "mark":
self.handlePlaybackMark(payload)
case "transcript":
@@ -508,7 +593,7 @@ public final class RealtimeTalkRelaySession {
case "toolCall":
self.startToolCall(payload, lifecycleGeneration: lifecycleGeneration)
case "error":
let message = payload["message"]?.stringValue ?? "Realtime failed"
let message = payload["message"]?.stringValue ?? String(localized: "Realtime failed")
let issue = Self.issue(
payload: payload,
fallbackMessage: message,
@@ -524,9 +609,13 @@ public final class RealtimeTalkRelaySession {
self.logger.debug("talk realtime: close")
if self.hasReceivedReady {
self.onStatus("Ready")
let reason = self.nonEmpty(payload["reason"]?.stringValue)
self.close(sendClose: false)
self.onTermination(.remoteClose(reason: reason))
return
} else if !self.hasReceivedFailure {
let issue = RealtimeTalkRelayIssue(
message: "Realtime closed before it became ready.",
message: String(localized: "Realtime closed before it became ready."),
provider: self.options.provider,
model: self.options.model,
transport: "gateway-relay",
@@ -536,12 +625,39 @@ public final class RealtimeTalkRelaySession {
self.finishStartupWait(.failed(issue))
self.onStatus("Realtime failed before connecting")
}
self.close(sendClose: false)
default:
return
}
}
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 {
self.awaitingOutputClear = false
if self.outputCancellationTask == nil { 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
@@ -587,7 +703,7 @@ public final class RealtimeTalkRelaySession {
return
}
let issue = RealtimeTalkRelayIssue(
message: "Realtime did not become ready in time.",
message: String(localized: "Realtime did not become ready in time."),
provider: self.options.provider,
model: self.options.model,
transport: "gateway-relay",
@@ -698,7 +814,8 @@ public final class RealtimeTalkRelaySession {
lifecycleGeneration: lifecycleGeneration)
guard let runId = startResponse.runId ?? startResponse.idempotencyKey else {
throw NSError(domain: "RealtimeTalkRelay", code: 3, userInfo: [
NSLocalizedDescriptionKey: "Realtime tool call did not return a run id",
NSLocalizedDescriptionKey: String(
localized: "Realtime tool call did not return a run id"),
])
}
let completion = await self.waitForChatCompletion(
@@ -844,11 +961,15 @@ public final class RealtimeTalkRelaySession {
return try JSONDecoder().decode(type, from: response)
}
private func isCurrentLifecycle(_ lifecycleGeneration: UInt64) async -> Bool {
guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return false }
private func lifecycleStatus(_ lifecycleGeneration: UInt64) async -> LifecycleStatus {
guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return .cancelledLocally }
let routeIsCurrent = await self.transport.isCurrent()
return self.isCurrentLifecycleLocally(lifecycleGeneration) &&
routeIsCurrent
guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return .cancelledLocally }
return routeIsCurrent ? .current : .routeLost
}
private func isCurrentLifecycle(_ lifecycleGeneration: UInt64) async -> Bool {
await self.lifecycleStatus(lifecycleGeneration) == .current
}
private func isCurrentLifecycleLocally(_ lifecycleGeneration: UInt64) -> Bool {
@@ -869,57 +990,41 @@ public final class RealtimeTalkRelaySession {
}
envelope.begin(sampleRate: self.outputSampleRateHz)
self.outputEnvelope = envelope
let stream = AsyncThrowingStream<Data, Error> { continuation in
self.outputContinuation = continuation
}
let stream = AsyncThrowingStream<Data, Error>(
bufferingPolicy: .bufferingOldest(Self.maxBufferedOutputChunks))
{ continuation in self.outputContinuation = continuation }
self.outputTask = Task { [weak self] in
guard let self else { return }
guard self.outputSessionId == sessionId, !self.isClosed, !Task.isCancelled else { return }
let result = await self.pcmPlayer.play(stream: stream, sampleRate: self.outputSampleRateHz)
await MainActor.run {
guard self.outputSessionId == sessionId else { return }
self.outputTask = nil
self.outputContinuation = nil
if !result.finished, let interruptedAt = result.interruptedAt {
self.logger.info("realtime output interrupted at \(interruptedAt, privacy: .public)s")
if !result.finished {
if let interruptedAt = result.interruptedAt {
self.logger.info("realtime output interrupted at \(interruptedAt, privacy: .public)s")
}
self.handleOutputPlaybackFailure(
String(localized: "Realtime audio playback failed. Reconnecting…"))
return
}
self.markOutputPlaybackFinished()
self.startPendingOutputPlaybackIfNeeded()
}
}
}
private func finishOutputPlaybackStream() {
guard let continuation = self.outputContinuation else {
if self.outputTask != nil, !self.pendingOutputChunks.isEmpty {
self.pendingOutputDone = true
}
return
guard let continuation = self.outputContinuation else { return }
if !self.pendingOutputAudio.isEmpty {
let trailingFrame = self.pendingOutputAudio
self.pendingOutputAudio.removeAll(keepingCapacity: true)
guard self.yieldOutputAudioFrame(trailingFrame) else { return }
}
continuation.finish()
self.outputContinuation = nil
}
private func startPendingOutputPlaybackIfNeeded() {
guard !self.pendingOutputChunks.isEmpty else {
self.pendingOutputDone = false
return
}
let chunks = self.pendingOutputChunks
let shouldFinish = self.pendingOutputDone
self.pendingOutputChunks = []
self.pendingOutputDone = false
self.ensureOutputPlaybackStarted()
for chunk in chunks {
self.markOutputAudioStarted(byteCount: chunk.count, nowMs: ProcessInfo.processInfo.systemUptime * 1000)
self.onSpeakingChanged(true)
self.outputEnvelope?.append(chunk)
self.outputContinuation?.yield(chunk)
}
if shouldFinish {
self.finishOutputPlaybackStream()
}
}
private func scheduleOutputPlaybackIdle(expectedEndMs: Double) {
self.outputIdleTask?.cancel()
let nowMs = ProcessInfo.processInfo.systemUptime * 1000
@@ -947,6 +1052,7 @@ public final class RealtimeTalkRelaySession {
self.outputIdleTask = nil
}
self.isOutputPlaying = false
self.outputIdentity = nil
self.outputStartedAtMs = nil
self.outputPlaybackExpectedEndMs = 0
self.outputEnvelope?.cancel()
@@ -984,8 +1090,9 @@ public final class RealtimeTalkRelaySession {
do {
_ = try await transport.request("talk.session.acknowledgeMark", payload, 8000)
} catch {
let message = Self.safeLogMessage(error.localizedDescription)
logger.warning(
"talk realtime: mark acknowledgement failed=\(Self.safeLogMessage(error.localizedDescription), privacy: .public)")
"talk realtime: mark acknowledgement failed=\(message, privacy: .public)")
}
}
}
@@ -995,14 +1102,14 @@ public final class RealtimeTalkRelaySession {
self.outputSessionId += 1
self.outputContinuation?.finish()
self.outputContinuation = nil
self.pendingOutputAudio.removeAll(keepingCapacity: true)
self.outputTask?.cancel()
self.outputTask = nil
self.outputIdleTask?.cancel()
self.outputIdleTask = nil
self.pendingOutputChunks = []
self.pendingOutputDone = false
_ = self.pcmPlayer.stop()
self.isOutputPlaying = false
self.outputIdentity = nil
self.outputStartedAtMs = nil
self.outputPlaybackExpectedEndMs = 0
self.outputEnvelope?.cancel()
@@ -1050,22 +1157,260 @@ public final class RealtimeTalkRelaySession {
}
}
extension RealtimeTalkRelaySession {
private struct OutputIdentity {
enum Relation {
case same
case different
case unknown
}
let turnId: String?
init(_ payload: [String: AnyCodable]) {
let turnId = payload["talkEvent"]?.dictionaryValue?["turnId"]?.stringValue?
.trimmingCharacters(in: .whitespacesAndNewlines)
self.turnId = turnId?.isEmpty == false ? turnId : nil
}
func isEmpty() -> Bool {
self.turnId == nil
}
func relation(to other: OutputIdentity) -> Relation {
if self.isEmpty() || other.isEmpty() {
return self.isEmpty() == other.isEmpty() ? .same : .different
}
if let turnId, let otherTurnId = other.turnId {
return turnId == otherTurnId ? .same : .different
}
return .unknown
}
}
@discardableResult
public func cancelOutput(reason: String = "user") -> Bool {
guard let relaySessionId,
let outputIdentity = self.outputIdentity,
let turnId = outputIdentity.turnId
else { return false }
self.outputCancellationGeneration &+= 1
let cancellationGeneration = self.outputCancellationGeneration
self.outputCancellationTask?.cancel()
self.suppressedOutputIdentity = outputIdentity
self.cancelledOutputTurnId = outputIdentity.turnId
self.awaitingOutputClear = true
self.stopOutputPlayback()
self.outputCancellationTask = Task { [weak self, transport] in
let payload: [String: AnyCodable] = [
"sessionId": AnyCodable(relaySessionId),
"reason": AnyCodable(reason),
"turnId": AnyCodable(turnId),
]
do {
let response = try await transport.request("talk.session.cancelOutput", payload, 8000)
let result = try JSONDecoder().decode(TalkSessionCancelOutputResult.self, from: response)
guard result.ok else { throw URLError(.badServerResponse) }
guard let self, self.isCurrentOutputCancellation(cancellationGeneration) else { return }
switch result.status?.stringValue {
case "stale", "idle":
self.retireOutputCancellation()
case nil, "applied":
guard result.turnid == nil || result.turnid == turnId else {
throw URLError(.badServerResponse)
}
if self.awaitingOutputClear {
self.outputCancellationTask = nil
} else {
self.retireOutputCancellation()
}
default:
throw URLError(.badServerResponse)
}
} catch {
guard let self, self.isCurrentOutputCancellation(cancellationGeneration) else { return }
let issue = RealtimeTalkRelayIssue(
code: "realtime_output_cancel_failed",
message: String(
format: String(localized: "Realtime output cancellation failed: %@"),
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)
}
}
return true
}
private func isCurrentOutputCancellation(_ generation: UInt64) -> Bool {
generation == self.outputCancellationGeneration && !self.isClosed
}
private func handleOutputAudio(_ payload: [String: AnyCodable]) {
guard !self.isOutputPaused else { return }
let incomingIdentity = OutputIdentity(payload)
guard let incomingTurnId = incomingIdentity.turnId else {
self.handleOutputPlaybackOverflow()
return
}
guard !self.awaitingOutputClear else { return }
if let cancelledOutputTurnId {
if incomingTurnId == cancelledOutputTurnId {
return
}
}
guard let base64 = payload["audioBase64"]?.stringValue else { return }
guard let data = Data(base64Encoded: base64) else {
self.handleOutputPlaybackOverflow()
return
}
if let currentIdentity = self.outputIdentity,
currentIdentity.relation(to: incomingIdentity) == .different
{
self.stopOutputPlayback()
} else if self.outputContinuation == nil, self.outputTask != nil {
self.stopOutputPlayback()
}
self.outputIdentity = incomingIdentity
self.recordOutputAudioChunk(byteCount: data.count)
self.markOutputAudioStarted(byteCount: data.count, nowMs: ProcessInfo.processInfo.systemUptime * 1000)
self.onSpeakingChanged(true)
self.ensureOutputPlaybackStarted()
self.bufferOutputAudio(data)
}
private func bufferOutputAudio(_ data: Data) {
let frameByteCount = max(2, Int((self.outputSampleRateHz * 0.02).rounded()) * 2)
var offset = data.startIndex
if !self.pendingOutputAudio.isEmpty {
let fillCount = min(frameByteCount - self.pendingOutputAudio.count, data.count)
let fillEnd = data.index(offset, offsetBy: fillCount)
self.pendingOutputAudio.append(data[offset..<fillEnd])
offset = fillEnd
if self.pendingOutputAudio.count == frameByteCount {
let frame = self.pendingOutputAudio
self.pendingOutputAudio.removeAll(keepingCapacity: true)
guard self.yieldOutputAudioFrame(frame) else { return }
}
}
while data.distance(from: offset, to: data.endIndex) >= frameByteCount {
let frameEnd = data.index(offset, offsetBy: frameByteCount)
let frame = Data(data[offset..<frameEnd])
offset = frameEnd
guard self.yieldOutputAudioFrame(frame) else { return }
}
if offset < data.endIndex {
self.pendingOutputAudio.append(data[offset...])
}
}
private func yieldOutputAudioFrame(_ data: Data) -> Bool {
guard let continuation = self.outputContinuation else { return false }
switch continuation.yield(data) {
case .enqueued:
self.outputEnvelope?.append(data)
return true
case .dropped:
self.handleOutputPlaybackOverflow()
return false
case .terminated:
return false
@unknown default:
self.handleOutputPlaybackOverflow()
return false
}
}
private func handleOutputAudioDone(_ payload: [String: AnyCodable]) {
let incomingIdentity = OutputIdentity(payload)
if !incomingIdentity.isEmpty(),
let outputIdentity,
outputIdentity.relation(to: incomingIdentity) != .same
{
return
}
self.finishOutputPlaybackStream()
}
private func handleOutputPlaybackOverflow() {
self.handleOutputPlaybackFailure(
String(localized: "Realtime audio playback fell behind. Reconnecting…"))
}
private func handleOutputPlaybackFailure(_ message: String) {
guard !self.isClosed else { return }
let issue = RealtimeTalkRelayIssue(
message: message,
provider: self.options.provider,
model: self.options.model,
transport: "gateway-relay",
phase: "output-playback")
self.onIssue(issue)
self.onStatus(message)
self.close(sendClose: true)
self.onTermination(.outputPlaybackOverflow)
}
private func retireOutputCancellation() {
self.outputCancellationGeneration &+= 1
self.outputCancellationTask?.cancel()
self.outputCancellationTask = nil
self.suppressedOutputIdentity = nil
self.awaitingOutputClear = false
}
}
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) { [weak self] frame in
Task { @MainActor [weak self] in
_ = self?.enqueueMicrophoneFrame(
frame.data,
timestampMs: frame.timestampMs,
rms: frame.rms,
try self.audioCapture.start(
targetSampleRate: self.inputSampleRateHz,
onAudio: { [weak self] frame in
Task { @MainActor [weak self] in
_ = self?.enqueueMicrophoneFrame(
frame.data,
timestampMs: frame.timestampMs,
rms: frame.rms,
lifecycleGeneration: lifecycleGeneration,
audioCaptureGeneration: audioCaptureGeneration)
}
},
onFailure: { [weak self] message in
self?.handleAudioInputFailure(
message,
lifecycleGeneration: lifecycleGeneration,
audioCaptureGeneration: audioCaptureGeneration)
}
}
})
}
private func handleAudioInputFailure(
_ message: String,
lifecycleGeneration: UInt64,
audioCaptureGeneration: UInt64)
{
guard !self.isClosed, self.isCurrentLifecycleLocally(lifecycleGeneration),
self.audioCaptureGeneration == audioCaptureGeneration
else { return }
let issue = RealtimeTalkRelayIssue(
code: "audio_input_unavailable",
message: message,
provider: self.options.provider,
model: self.options.model,
transport: "gateway-relay",
phase: "audio-input")
self.logger.error("talk realtime microphone failed: \(Self.safeLogMessage(message), privacy: .public)")
self.onIssue(issue)
self.onStatus(message)
self.close(sendClose: true)
self.onTermination(.audioInputFailed(message: message))
}
@discardableResult
@@ -1078,7 +1423,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)
@@ -1100,10 +1445,24 @@ 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 let message = await audioSender.send(encoded, timestampMs: timestampMs) else { return }
guard self.isCurrentLifecycleLocally(lifecycleGeneration) else { return }
self.onStatus("Realtime audio failed: \(message)")
guard self.isCurrentLifecycleLocally(lifecycleGeneration),
self.audioCaptureGeneration == audioCaptureGeneration,
!self.isInputPaused, self.suppressedOutputIdentity == nil
else { return }
switch await audioSender.send(encoded, timestampMs: timestampMs) {
case .sent, .inactive:
return
case .saturated:
self.handleAudioInputFailure(
String(localized: "Realtime audio input fell behind. Reconnecting…"),
lifecycleGeneration: lifecycleGeneration,
audioCaptureGeneration: audioCaptureGeneration)
case let .failed(message):
self.handleAudioInputFailure(
String(format: String(localized: "Realtime audio failed: %@"), message),
lifecycleGeneration: lifecycleGeneration,
audioCaptureGeneration: audioCaptureGeneration)
}
}
self.audioSendTasks[taskID] = task
return task
@@ -1132,8 +1491,10 @@ extension RealtimeTalkRelaySession {
guard timestampMs - self.lastSuppressedEchoLogAtMs >= 1000 else { return }
self.lastSuppressedEchoLogAtMs = timestampMs
let maxRms = String(format: "%.4f", Double(self.suppressedEchoMaxRms))
let frames = self.suppressedEchoFrameCount
let bytes = self.suppressedEchoByteCount
self.logger.debug(
"talk realtime mic suppressed during output: buffers=\(self.suppressedEchoFrameCount) bytes=\(self.suppressedEchoByteCount) maxRms=\(maxRms)")
"talk realtime mic suppressed during output: buffers=\(frames) bytes=\(bytes) maxRms=\(maxRms)")
self.suppressedEchoFrameCount = 0
self.suppressedEchoByteCount = 0
self.suppressedEchoMaxRms = 0
@@ -1141,19 +1502,32 @@ extension RealtimeTalkRelaySession {
private func stopMicrophonePump() {
self.audioCaptureGeneration &+= 1
for task in self.audioSendTasks.values {
task.cancel()
}
self.audioSendTasks.removeAll()
self.audioCapture.stop()
}
}
#if DEBUG
extension RealtimeTalkRelaySession {
// periphery:ignore - package tests drive a relay session without a live gateway handshake.
func _test_setRelaySessionId(_ relaySessionId: String) {
self.relaySessionId = relaySessionId
}
// periphery:ignore - package tests inject gateway events without a live socket.
func _test_handleGatewayEvent(_ event: EventFrame) async {
await self.handleGatewayEvent(event, lifecycleGeneration: self.lifecycleGeneration)
}
// periphery:ignore - package tests end the event stream deterministically.
func _test_handleEventStreamEnded() async {
await self.handleEventStreamEnded(lifecycleGeneration: self.lifecycleGeneration)
}
// periphery:ignore - package tests observe startup cancellation without waiting out the timeout.
func _test_waitForStartupCancelled(timeoutSeconds: Int) async -> Bool {
if case .cancelled = await self.waitForStartupResult(
timeoutSeconds: timeoutSeconds,
@@ -1164,6 +1538,7 @@ extension RealtimeTalkRelaySession {
return false
}
// periphery:ignore - package tests await in-flight tool calls before asserting.
func _test_waitForToolCalls() async {
let tasks = self.toolCallTasks.values
for task in tasks {
@@ -1171,26 +1546,32 @@ extension RealtimeTalkRelaySession {
}
}
func _test_startupReadyTimeoutSeconds() -> Int {
Self.startupReadyTimeoutSeconds
// periphery:ignore - package tests capture the exact owned cancellation before replacement or stop.
func _test_outputCancellationTask() -> Task<Void, Never>? {
self.outputCancellationTask
}
// periphery:ignore - package tests start output playback without decoding real audio.
func _test_markOutputAudioStarted(nowMs: Double) {
self.markOutputAudioStarted(byteCount: 4800, nowMs: nowMs)
}
// periphery:ignore - package tests finish playback without a real player callback.
func _test_markOutputPlaybackFinished() {
self.markOutputPlaybackFinished()
}
// periphery:ignore - package tests observe barge-in timing state.
func _test_outputStartedAtMs() -> Double? {
self.outputStartedAtMs
}
// periphery:ignore - package tests observe playback state without exposing it publicly.
func _test_isOutputPlaying() -> Bool {
self.isOutputPlaying
}
// periphery:ignore - package tests exercise the audio sender without a started session.
func _test_prepareAudioSender(relaySessionId: String) {
self.isClosed = false
self.audioSender = RealtimeAudioSender(
@@ -1198,13 +1579,23 @@ extension RealtimeTalkRelaySession {
request: self.transport.request)
}
func _test_enqueueMicrophoneFrame(_ data: Data) -> Task<Void, Never>? {
// periphery:ignore - package tests enqueue frames without a live capture device.
func _test_enqueueMicrophoneFrame(
_ data: Data,
timestampMs: Double = 1) -> Task<Void, Never>?
{
self.enqueueMicrophoneFrame(
data,
timestampMs: 1,
timestampMs: timestampMs,
rms: 0.01,
lifecycleGeneration: self.lifecycleGeneration,
audioCaptureGeneration: self.audioCaptureGeneration)
}
// periphery:ignore - package tests start the pump to observe capture failure handling.
func _test_startMicrophonePump() throws {
try self.startMicrophonePump(lifecycleGeneration: self.lifecycleGeneration)
}
}
#endif
#endif
apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimePCMStreamingAudioPlayerTests.swift
+340
View File
@@ -0,0 +1,340 @@
#if Talk && canImport(ElevenLabsKit) && (os(iOS) || os(macOS))
import Foundation
import Testing
@testable import OpenClawKit
private struct RealtimePCMPlaybackWaitTimeout: Error {
let label: String
}
private struct RealtimePCMPlaybackFailure: Error {}
private let realtimePCMPlaybackWaitTimeoutSeconds = 15.0
@MainActor
private final class RealtimePCMPlaybackBackend {
private struct Waiter {
let count: Int
let continuation: CheckedContinuation<Void, any Error>
}
private(set) var scheduledFrames: [Data] = []
private(set) var completions: [@Sendable () -> Void] = []
private(set) var activeCount = 0
private(set) var maxActiveCount = 0
private var completedCallbacks = 0
private var scheduledWaiters: [UUID: Waiter] = [:]
private var completionWaiters: [UUID: Waiter] = [:]
func prepare(sampleRate _: Double) throws {}
func schedule(
data: Data,
sampleRate _: Double,
completion: @escaping @Sendable () -> Void) throws
{
self.scheduledFrames.append(data)
self.activeCount += 1
self.maxActiveCount = max(self.maxActiveCount, self.activeCount)
self.resumeScheduledWaiters()
self.completions.append { [weak self] in
Task { @MainActor in
self?.activeCount -= 1
self?.completedCallbacks += 1
self?.resumeCompletionWaiters()
completion()
}
}
}
func stop() {
self.activeCount = 0
}
func complete(at index: Int = 0) {
self.completions.remove(at: index)()
}
func takeCompletion(at index: Int = 0) -> @Sendable () -> Void {
self.completions.remove(at: index)
}
func waitForScheduledFrames(_ count: Int) async throws {
if self.scheduledFrames.count >= count { return }
try await AsyncTimeout.withTimeout(
seconds: realtimePCMPlaybackWaitTimeoutSeconds,
onTimeout: { RealtimePCMPlaybackWaitTimeout(label: "scheduled frames \(count)") },
operation: { try await self.waitForScheduledFramesWithoutDeadline(count) })
}
func waitForCompletionCallbacks(_ count: Int) async throws {
if self.completedCallbacks >= count { return }
try await AsyncTimeout.withTimeout(
seconds: realtimePCMPlaybackWaitTimeoutSeconds,
onTimeout: { RealtimePCMPlaybackWaitTimeout(label: "completion callbacks \(count)") },
operation: { try await self.waitForCompletionCallbacksWithoutDeadline(count) })
}
private func waitForScheduledFramesWithoutDeadline(_ count: Int) async throws {
let id = UUID()
try await withTaskCancellationHandler {
try await withCheckedThrowingContinuation { continuation in
if self.scheduledFrames.count >= count {
continuation.resume()
} else {
self.scheduledWaiters[id] = Waiter(count: count, continuation: continuation)
}
}
} onCancel: {
Task { @MainActor in self.cancelScheduledWaiter(id) }
}
}
private func waitForCompletionCallbacksWithoutDeadline(_ count: Int) async throws {
let id = UUID()
try await withTaskCancellationHandler {
try await withCheckedThrowingContinuation { continuation in
if self.completedCallbacks >= count {
continuation.resume()
} else {
self.completionWaiters[id] = Waiter(count: count, continuation: continuation)
}
}
} onCancel: {
Task { @MainActor in self.cancelCompletionWaiter(id) }
}
}
private func cancelScheduledWaiter(_ id: UUID) {
self.scheduledWaiters.removeValue(forKey: id)?.continuation.resume(throwing: CancellationError())
}
private func cancelCompletionWaiter(_ id: UUID) {
self.completionWaiters.removeValue(forKey: id)?.continuation.resume(throwing: CancellationError())
}
private func resumeScheduledWaiters() {
let ready = self.scheduledWaiters.filter { self.scheduledFrames.count >= $0.value.count }
for (id, waiter) in ready {
self.scheduledWaiters.removeValue(forKey: id)
waiter.continuation.resume()
}
}
private func resumeCompletionWaiters() {
let ready = self.completionWaiters.filter { self.completedCallbacks >= $0.value.count }
for (id, waiter) in ready {
self.completionWaiters.removeValue(forKey: id)
waiter.continuation.resume()
}
}
}
@MainActor
private final class RealtimePCMPlaybackResultProbe {
private(set) var results: [StreamingPlaybackResult] = []
func record(_ result: StreamingPlaybackResult) {
self.results.append(result)
}
}
@MainActor
private func makeRealtimePCMPlayer(
backend: RealtimePCMPlaybackBackend) -> RealtimePCMStreamingAudioPlayer
{
RealtimePCMStreamingAudioPlayer(
preparePlayback: backend.prepare,
scheduleFrame: backend.schedule,
stopPlayback: backend.stop,
playbackTime: { nil })
}
private func waitForPlayback(_ task: Task<Void, Never>, label: String) async throws {
try await AsyncTimeout.withTimeout(
seconds: realtimePCMPlaybackWaitTimeoutSeconds,
onTimeout: { RealtimePCMPlaybackWaitTimeout(label: label) },
operation: { await task.value })
}
@MainActor
struct RealtimePCMStreamingAudioPlayerTests {
private let sampleRate = 8000.0
private var frameBytes: Int {
Int(self.sampleRate * RealtimePCMStreamingAudioPlayer.frameDurationSeconds) * 2
}
@Test func `prepare failure returns unfinished playback`() async {
let player = RealtimePCMStreamingAudioPlayer(
preparePlayback: { _ in throw RealtimePCMPlaybackFailure() },
scheduleFrame: { _, _, _ in },
stopPlayback: {},
playbackTime: { nil })
let (stream, continuation) = AsyncThrowingStream<Data, Error>.makeStream()
continuation.finish()
let result = await player.play(stream: stream, sampleRate: self.sampleRate)
#expect(!result.finished)
#expect(result.interruptedAt == nil)
}
@Test func `schedule failure returns unfinished playback`() async throws {
let player = RealtimePCMStreamingAudioPlayer(
preparePlayback: { _ in },
scheduleFrame: { _, _, _ in throw RealtimePCMPlaybackFailure() },
stopPlayback: {},
playbackTime: { nil })
let (stream, continuation) = AsyncThrowingStream<Data, Error>.makeStream()
let playback = Task {
await player.play(stream: stream, sampleRate: self.sampleRate)
}
continuation.yield(Data(repeating: 1, count: self.frameBytes))
continuation.finish()
let probe = RealtimePCMPlaybackResultProbe()
let observed = Task {
await probe.record(playback.value)
}
try await waitForPlayback(observed, label: "schedule failure")
#expect(probe.results.count == 1)
#expect(probe.results.first?.finished == false)
#expect(probe.results.first?.interruptedAt == nil)
}
@Test func `withheld completions cap scheduling and one completion admits one frame`() async throws {
let backend = RealtimePCMPlaybackBackend()
let player = makeRealtimePCMPlayer(backend: backend)
let probe = RealtimePCMPlaybackResultProbe()
let (stream, continuation) = AsyncThrowingStream<Data, Error>.makeStream()
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()
try await backend.waitForScheduledFrames(3)
#expect(backend.scheduledFrames.count == 3)
#expect(backend.maxActiveCount == 3)
#expect(probe.results.isEmpty)
backend.complete()
try await backend.waitForScheduledFrames(4)
#expect(backend.scheduledFrames.count == 4)
#expect(backend.maxActiveCount == 3)
#expect(probe.results.isEmpty)
backend.complete()
try await backend.waitForScheduledFrames(5)
for _ in 0..<3 {
backend.complete()
}
try await waitForPlayback(playback, label: "five-frame playback")
#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 throws {
let backend = RealtimePCMPlaybackBackend()
let player = makeRealtimePCMPlayer(backend: backend)
let probe = RealtimePCMPlaybackResultProbe()
let (stream, continuation) = AsyncThrowingStream<Data, Error>.makeStream()
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()
try await backend.waitForScheduledFrames(2)
#expect(backend.completions.count == 2)
#expect(probe.results.isEmpty)
backend.complete()
#expect(backend.completions.count == 1)
#expect(probe.results.isEmpty)
backend.complete()
try await waitForPlayback(playback, label: "completed input playback")
#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 throws {
let backend = RealtimePCMPlaybackBackend()
let player = makeRealtimePCMPlayer(backend: backend)
let (firstStream, firstContinuation) = AsyncThrowingStream<Data, Error>.makeStream()
let firstProbe = RealtimePCMPlaybackResultProbe()
let firstPlayback = Task {
let result = await player.play(stream: firstStream, sampleRate: self.sampleRate)
firstProbe.record(result)
}
firstContinuation.yield(Data(repeating: 1, count: self.frameBytes))
try await backend.waitForScheduledFrames(1)
let staleCompletion = backend.takeCompletion()
_ = player.stop()
try await waitForPlayback(firstPlayback, label: "stopped A playback")
#expect(firstProbe.results.map(\.finished) == [false])
let (secondStream, secondContinuation) = AsyncThrowingStream<Data, Error>.makeStream()
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()
try await backend.waitForScheduledFrames(4)
#expect(backend.activeCount == 3)
#expect(probe.results.isEmpty)
staleCompletion()
try await backend.waitForCompletionCallbacks(1)
#expect(backend.scheduledFrames.count == 4)
#expect(backend.completions.count == 3)
#expect(firstProbe.results.map(\.finished) == [false])
#expect(probe.results.isEmpty)
backend.complete()
try await backend.waitForScheduledFrames(5)
#expect(backend.scheduledFrames.count == 5)
#expect(probe.results.isEmpty)
backend.complete()
try await backend.waitForScheduledFrames(6)
#expect(backend.scheduledFrames.count == 6)
#expect(probe.results.isEmpty)
for _ in 0..<3 {
backend.complete()
}
try await waitForPlayback(secondPlayback, label: "replacement B playback")
#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 throws {
let backend = RealtimePCMPlaybackBackend()
let player = makeRealtimePCMPlayer(backend: backend)
let probe = RealtimePCMPlaybackResultProbe()
let (stream, continuation) = AsyncThrowingStream<Data, Error>.makeStream()
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))
try await backend.waitForScheduledFrames(3)
_ = player.stop()
_ = player.stop()
try await waitForPlayback(playback, label: "stopped active playback")
#expect(probe.results.count == 1)
#expect(probe.results.first?.finished == false)
}
}
#endif
apps/shared/OpenClawKit/Tests/OpenClawKitTests/RealtimeTalkRelaySessionTests.swift
+1788 -35
View File
File diff suppressed because it is too large Load Diff