feat(talk): add macOS realtime Gateway relay (#128204)

* feat(talk): add macOS realtime relay transport

Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>

* feat(talk): integrate realtime relay with macOS Talk

Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>

* fix(talk): keep realtime bootstrap config redacted

Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>

* fix(talk): surface failed native fallback

Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>

* test(talk): inject deterministic relay capture

Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>

---------

Co-authored-by: Zhilong Zheng <zhengzhilong1115@gmail.com>
This commit is contained in:
Vincent Koc
2026-08-23 07:30:30 -07:00
committed by GitHub
parent 5283334fc6
commit 000db97c82
18 changed files with 4637 additions and 328 deletions
+110
View File
@@ -39386,6 +39386,17 @@
}
]
},
{
"id": "native.apple.11f6f4fa597f9df4",
"source": "Realtime Talk requested an invalid audio sample rate",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift"
}
]
},
{
"id": "native.apple.400963cb08a3c253",
"source": "Realtime Voice",
@@ -39504,6 +39515,28 @@
}
]
},
{
"id": "native.apple.48e407d88c565c02",
"source": "Realtime disconnected repeatedly — using native speech",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift"
}
]
},
{
"id": "native.apple.ab73f15b8147eb41",
"source": "Realtime disconnected — reconnecting…",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift"
}
]
},
{
"id": "native.apple.ca3a762269827f37",
"source": "Realtime failed",
@@ -39526,6 +39559,17 @@
}
]
},
{
"id": "native.apple.b5b843b6a0a96cc5",
"source": "Realtime microphone became unavailable: %@",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift"
}
]
},
{
"id": "native.apple.96ece1b431ac3f96",
"source": "Realtime output cancellation failed: %@",
@@ -39559,6 +39603,28 @@
}
]
},
{
"id": "native.apple.babb170df3172374",
"source": "Realtime unavailable — native speech could not start",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift"
}
]
},
{
"id": "native.apple.ddb5344f621cd627",
"source": "Realtime unavailable — using native speech",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/TalkModeRuntime+Realtime.swift"
}
]
},
{
"id": "native.apple.4c945b4b74b4de2d",
"source": "Realtime voice did not start.",
@@ -42907,6 +42973,28 @@
}
]
},
{
"id": "native.apple.d40597f2d391f8cf",
"source": "Selected audio input has no usable Float32 format",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift"
}
]
},
{
"id": "native.apple.8a4aa2d658ca95a3",
"source": "Selected input and system default are unavailable",
"surface": "apple",
"sites": [
{
"kind": "ui-localized-call",
"path": "apps/macos/Sources/OpenClaw/MacRealtimeTalkAudioCapture.swift"
}
]
},
{
"id": "native.apple.209377172e2dd2aa",
"source": "Selected reasoning is unavailable for this target.",
@@ -48328,6 +48416,28 @@
}
]
},
{
"id": "native.apple.050cddb7d6d24e52",
"source": "Use realtime Gateway relay",
"surface": "apple",
"sites": [
{
"kind": "ui-named-argument",
"path": "apps/macos/Sources/OpenClaw/VoiceWakeSettings.swift"
}
]
},
{
"id": "native.apple.d7c89897a6c2da7f",
"source": "Use the Gateway's configured realtime voice session on this Mac. Requires realtime, gateway-relay, and agent-consult in Talk settings.",
"surface": "apple",
"sites": [
{
"kind": "ui-named-argument-multiline",
"path": "apps/macos/Sources/OpenClaw/VoiceWakeSettings.swift"
}
]
},
{
"id": "native.apple.58566125ab806740",
"source": "Use the panel + Canvas",
@@ -20,6 +20,7 @@ extension AppState {
state.voiceWakeAdditionalLocaleIDs = ["en-US", "de-DE"]
state.voicePushToTalkEnabled = false
state.talkEnabled = false
state.talkRealtimeRelayEnabled = false
state.talkPhaseSoundsEnabled = true
state.talkShiftToStopEnabled = true
state.iconOverride = .system
@@ -0,0 +1,10 @@
import Foundation
extension AppState {
func persistTalkRealtimeRelayPreference(previousValue: Bool) {
guard !self.isPreview else { return }
AppDefaults.standard.set(self.talkRealtimeRelayEnabled, forKey: talkRealtimeRelayEnabledKey)
guard self.talkEnabled, self.talkRealtimeRelayEnabled != previousValue else { return }
Task { await TalkModeRuntime.shared.realtimeRelayPreferenceDidChange() }
}
}
+5 -1
View File
@@ -81,7 +81,7 @@ final class AppState {
private static let logger = Logger(subsystem: "ai.openclaw", category: "app-state")
private static let execApprovalsReadRetryAttempts = 5
private let isPreview: Bool
let isPreview: Bool
@ObservationIgnored private let execApprovalsDefaultsAsyncResolver:
@MainActor () async -> Result<ExecApprovalsResolvedDefaults, ExecApprovalsReadError>
@ObservationIgnored private let execApprovalsReadRetryDelay: Duration
@@ -260,6 +260,10 @@ final class AppState {
}
}
var talkRealtimeRelayEnabled = isTalkRealtimeRelayEnabled() {
didSet { self.persistTalkRealtimeRelayPreference(previousValue: oldValue) }
}
var talkPhaseSoundsEnabled: Bool {
didSet {
self.ifNotPreview {
@@ -31,6 +31,7 @@ let voiceWakeAdditionalLocalesKey = "openclaw.voiceWakeAdditionalLocaleIDs"
let voicePushToTalkEnabledKey = "openclaw.voicePushToTalkEnabled"
let voiceWakeTriggersTalkModeKey = "openclaw.voiceWakeTriggersTalkMode"
let talkEnabledKey = "openclaw.talkEnabled"
let talkRealtimeRelayEnabledKey = "openclaw.talkRealtimeRelayEnabled"
let talkPhaseSoundsEnabledKey = "openclaw.talkPhaseSoundsEnabled"
let talkShiftToStopEnabledKey = "openclaw.talkShiftToStopEnabled"
let iconOverrideKey = "openclaw.iconOverride"
@@ -48,6 +49,10 @@ let cookieSyncEnabledKey = "openclaw.cookieSyncEnabled"
let cookieSyncIntoProfileKey = "openclaw.cookieSyncIntoProfile"
let cookieSyncDomainsKey = "openclaw.cookieSyncDomains"
func isTalkRealtimeRelayEnabled(defaults: UserDefaults = AppDefaults.standard) -> Bool {
defaults.object(forKey: talkRealtimeRelayEnabledKey) as? Bool ?? false
}
func isComputerControlEnabled(
defaults: UserDefaults = AppDefaults.standard,
launchPlan: AppLaunchRuntimePlan = .current) -> Bool
@@ -0,0 +1,149 @@
import Foundation
import OpenClawChatUI
import OpenClawKit
import OpenClawProtocol
extension GatewayConnection {
struct RealtimeTalkBootstrap: @unchecked Sendable {
let transport: RealtimeTalkRelayTransport
let configSnapshot: ConfigSnapshot
let sessionKey: String
}
/// Freezes config and relay traffic to one physical Gateway socket.
///
/// A route replacement between config resolution and session creation must
/// fail this attempt instead of silently moving the relay to a new owner.
func acquireRealtimeTalkBootstrap() async throws -> RealtimeTalkBootstrap {
let lease = try await self.acquireServerLease()
let data = try await self.request(
method: Method.talkConfig.rawValue,
params: [:],
timeoutMs: 8000,
ifCurrentServerLease: lease)
let snapshot = try JSONDecoder().decode(ConfigSnapshot.self, from: data)
guard await self.isCurrentServerLease(lease) else {
throw OpenClawChatTransportSendError.notDispatched
}
let configuredSessionKey = snapshot.config?["session"]?.dictionaryValue?["mainKey"]?
.stringValue?.trimmingCharacters(in: .whitespacesAndNewlines)
return RealtimeTalkBootstrap(
transport: self.realtimeTalkTransport(ifCurrentServerLease: lease),
configSnapshot: snapshot,
sessionKey: configuredSessionKey?.isEmpty == false ? configuredSessionKey! : "main")
}
/// Creates a realtime Talk transport bound to one physical Gateway socket.
///
/// Gateway relay sessions are owned by the connection that created them. A
/// route-only transport could silently move follow-up audio or close calls to
/// a replacement socket after reconnecting, where that session does not exist.
func acquireRealtimeTalkTransport() async throws -> RealtimeTalkRelayTransport {
let lease = try await self.acquireServerLease()
return self.realtimeTalkTransport(ifCurrentServerLease: lease)
}
private func realtimeTalkTransport(
ifCurrentServerLease lease: ServerLease) -> RealtimeTalkRelayTransport
{
RealtimeTalkRelayTransport(
subscribeServerEvents: { bufferingNewest in
let pushes = await self.subscribe(
bufferingNewest: bufferingNewest,
ifCurrentServerLease: lease)
return AsyncStream(bufferingPolicy: .bufferingNewest(bufferingNewest)) { continuation in
let task = Task {
for await push in pushes {
guard case let .event(event) = push else { continue }
switch continuation.yield(event) {
case .enqueued:
continue
case .dropped, .terminated:
continuation.finish()
return
@unknown default:
continuation.finish()
return
}
}
continuation.finish()
}
continuation.onTermination = { @Sendable _ in
task.cancel()
}
}
},
request: { method, params, timeoutMs in
try await self.request(
method: method,
params: params,
timeoutMs: timeoutMs,
ifCurrentServerLease: lease)
},
isCurrent: {
await self.isCurrentServerLease(lease)
})
}
func subscribe(
bufferingNewest: Int,
ifCurrentServerLease lease: ServerLease) -> AsyncStream<GatewayPush>
{
let id = UUID()
let connection = self
return AsyncStream(bufferingPolicy: .bufferingNewest(bufferingNewest)) { continuation in
guard self.serverLeaseMatchesCurrentState(lease) else {
continuation.finish()
return
}
if let snapshot = self.lastSnapshot {
switch continuation.yield(.snapshot(snapshot)) {
case .enqueued:
break
case .dropped, .terminated:
continuation.finish()
return
@unknown default:
continuation.finish()
return
}
}
self.realtimeTalkSubscribers[lease.socketGeneration, default: [:]][id] = continuation
continuation.onTermination = { @Sendable _ in
Task {
await connection.removeRealtimeTalkSubscriber(
id,
socketGeneration: lease.socketGeneration)
}
}
}
}
func removeRealtimeTalkSubscriber(_ id: UUID, socketGeneration: UInt64) {
self.realtimeTalkSubscribers[socketGeneration]?[id] = nil
if self.realtimeTalkSubscribers[socketGeneration]?.isEmpty == true {
self.realtimeTalkSubscribers[socketGeneration] = nil
}
}
func finishRealtimeTalkSubscribers(socketGeneration: UInt64? = nil) {
let subscribers: [AsyncStream<GatewayPush>.Continuation]
if let socketGeneration {
if let removed = self.realtimeTalkSubscribers.removeValue(forKey: socketGeneration) {
subscribers = Array(removed.values)
} else {
subscribers = []
}
} else {
subscribers = self.realtimeTalkSubscribers.values.flatMap(\.values)
self.realtimeTalkSubscribers.removeAll()
}
subscribers.forEach { $0.finish() }
}
#if DEBUG
func _test_activeSocketGeneration() -> UInt64? {
self.activeSocketGeneration
}
#endif
}
@@ -70,7 +70,7 @@ actor GatewayConnection {
// Managed-image HTTP reuses this captured route from its focused extension file.
// Carrying the snapshot forward prevents endpoint or TLS rediscovery after suspension.
let route: Route
fileprivate let socketGeneration: UInt64
let socketGeneration: UInt64
fileprivate let client: GatewayChannelActor
}
@@ -163,11 +163,14 @@ actor GatewayConnection {
private var shutdownGeneration: UInt64 = 0
// Callback work keeps the physical socket epoch that decoded it. Retiring
// that epoch prevents delayed pushes from entering a replacement socket.
private var activeSocketGeneration: UInt64?
var activeSocketGeneration: UInt64?
private var lastRetiredSocketGeneration: UInt64?
private var subscribers: [UUID: AsyncStream<GatewayPush>.Continuation] = [:]
private var lastSnapshot: HelloOk?
var realtimeTalkSubscribers: [
UInt64: [UUID: AsyncStream<GatewayPush>.Continuation]
] = [:]
var lastSnapshot: HelloOk?
var canvasPluginSurfaceURL: String?
struct CanvasPluginSurfaceRefresh {
@@ -792,7 +795,7 @@ extension GatewayConnection {
return lease.route.activationOwnershipFingerprint
}
private func serverLeaseMatchesCurrentState(_ lease: ServerLease) -> Bool {
func serverLeaseMatchesCurrentState(_ lease: ServerLease) -> Bool {
self.routeMatchesConfiguredConnection(lease.route) &&
self.configuredConnection?.client === lease.client &&
self.activeSocketGeneration == lease.socketGeneration &&
@@ -927,6 +930,7 @@ extension GatewayConnection {
/// reentrant work could continue on a client whose replacement is in flight.
private func retireConfiguredConnection() -> GatewayChannelActor? {
self.routeGeneration &+= 1
self.finishRealtimeTalkSubscribers()
self.resetSocketGeneration()
self.lastSnapshot = nil
self.resetCanvasPluginSurfaceState()
@@ -970,6 +974,7 @@ extension GatewayConnection {
guard routeGeneration == self.routeGeneration,
retireSocketGeneration(socketGeneration)
else { return }
self.finishRealtimeTalkSubscribers(socketGeneration: socketGeneration)
self.lastSnapshot = nil
self.resetCanvasPluginSurfaceState()
}
@@ -1222,6 +1227,24 @@ extension GatewayConnection {
for (_, continuation) in self.subscribers {
continuation.yield(push)
}
if let socketGeneration = self.activeSocketGeneration {
var terminatedSubscriberIDs: [UUID] = []
for (id, continuation) in self.realtimeTalkSubscribers[socketGeneration] ?? [:] {
switch continuation.yield(push) {
case .enqueued:
break
case .dropped, .terminated:
continuation.finish()
terminatedSubscriberIDs.append(id)
@unknown default:
continuation.finish()
terminatedSubscriberIDs.append(id)
}
}
for id in terminatedSubscriberIDs {
self.removeRealtimeTalkSubscriber(id, socketGeneration: socketGeneration)
}
}
}
private func canonicalizeSessionKey(_ raw: String) -> String {
@@ -0,0 +1,871 @@
import AudioToolbox
@preconcurrency import AVFoundation
import CoreAudio
import Foundation
import OpenClawKit
import OSLog
@MainActor
final class MacRealtimeTalkAudioCapture: RealtimeTalkAudioCapturing {
private static let frameBufferSize: AVAudioFrameCount = 2048
private let logger = Logger(subsystem: "ai.openclaw", category: "talk.realtime.capture")
private let selectedInputUID: @MainActor () -> String?
private let deliveryGate = TalkGenerationDeliveryGate()
private var audioEngine: AVAudioEngine?
private var inputNode: AVAudioInputNode?
private var audioInputObserver: AudioInputDeviceObserver?
private var audioOutputObserver: MacRealtimeTalkOutputRouteObserver?
private var activeInputResolution: AudioInputDeviceResolution?
private var targetSampleRate: Double?
private var onAudio: (@Sendable (RealtimeTalkAudioFrame) -> Void)?
private var onFailure: (@MainActor (String) -> Void)?
private var tapInstalled = false
private var suppressInputDuringOutput = true
private var outputRouteDecisionState = MacRealtimeTalkOutputRouteDecisionState()
private var outputRouteObservationGeneration: UInt64 = 0
#if DEBUG
private var testOutputRouteCallbackHandled: (@Sendable () -> Void)?
#endif
var suppressesInputDuringOutput: Bool {
self.suppressInputDuringOutput
}
init(selectedInputUID: @escaping @MainActor () -> String? = {
AppStateStore.shared.voiceWakeMicID
}) {
self.selectedInputUID = selectedInputUID
}
@MainActor deinit {
self.stop()
}
func start(
targetSampleRate: Double,
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void,
onFailure: @escaping @MainActor (String) -> Void) throws
{
guard targetSampleRate.isFinite, targetSampleRate > 0 else {
throw MacRealtimeTalkAudioCaptureError.invalidTargetSampleRate
}
self.stop()
self.targetSampleRate = targetSampleRate
self.onAudio = onAudio
self.onFailure = onFailure
self.startOutputRouteObserver()
do {
try self.startCaptureEngine(targetSampleRate: targetSampleRate, onAudio: onAudio)
self.startDeviceObserver()
} catch {
self.stop()
throw error
}
}
func stop() {
// Close delivery before removing the tap. A callback already running on Core Audio's
// queue must finish before stop returns, and later callbacks must drop their frames.
self.deliveryGate.deactivate()
self.audioInputObserver?.stop()
self.audioInputObserver = nil
self.retireOutputRouteObserver()
self.teardownEngine()
self.targetSampleRate = nil
self.onAudio = nil
self.onFailure = nil
}
private func startCaptureEngine(
targetSampleRate: Double,
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void) throws
{
let selection = AudioInputDeviceObserver.resolveSelection(self.selectedInputUID())
// AVAudioEngine materializes inputNode from the system default before CurrentDevice can bind.
// Without a usable default, accessing inputNode can SIGABRT even when another UID is alive.
guard selection.resolvedUID != nil, AudioInputDeviceObserver.hasUsableDefaultInputDevice() else {
throw MacRealtimeTalkAudioCaptureError.inputUnavailable
}
do {
try self.configureEngine(
selection: selection,
targetSampleRate: targetSampleRate,
onAudio: onAudio,
enableVoiceProcessing: true)
} catch {
self.logger.warning(
"realtime processed input setup failed; retrying without voice processing: " +
"\(error.localizedDescription, privacy: .public)")
self.deliveryGate.deactivate()
self.teardownEngine()
try self.configureEngine(
selection: selection,
targetSampleRate: targetSampleRate,
onAudio: onAudio,
enableVoiceProcessing: false)
}
}
private func configureEngine(
selection: AudioInputDeviceResolution,
targetSampleRate: Double,
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void,
enableVoiceProcessing: Bool) throws
{
let engine = AVAudioEngine()
self.audioEngine = engine
let input = engine.inputNode
self.inputNode = input
if enableVoiceProcessing {
try input.setVoiceProcessingEnabled(true)
}
let activeResolution = self.bindSelectedInputIfNeeded(selection, to: input)
guard activeResolution.resolvedUID != nil else {
throw MacRealtimeTalkAudioCaptureError.inputUnavailable
}
let format = input.outputFormat(forBus: 0)
guard format.commonFormat == .pcmFormatFloat32,
!format.isInterleaved,
format.channelCount > 0,
format.sampleRate > 0
else {
throw MacRealtimeTalkAudioCaptureError.invalidInputFormat
}
let deliveryToken = self.deliveryGate.activate()
input.installTap(
onBus: 0,
bufferSize: Self.frameBufferSize,
format: format,
block: MacRealtimeTalkTapHandlerFactory.make(
targetSampleRate: targetSampleRate,
deliveryGate: self.deliveryGate,
deliveryToken: deliveryToken,
onAudio: onAudio))
self.tapInstalled = true
engine.prepare()
try engine.start()
self.activeInputResolution = activeResolution
}
private func bindSelectedInputIfNeeded(
_ selection: AudioInputDeviceResolution,
to input: AVAudioInputNode) -> AudioInputDeviceResolution
{
guard selection.shouldBindSelectedDevice, let selectedUID = selection.resolvedUID else {
return selection
}
guard let audioUnit = input.audioUnit,
var deviceID = AudioInputDeviceObserver.inputDeviceID(forUID: selectedUID)
else {
self.logger.warning("realtime selected input could not be resolved; using system default")
return self.defaultFallback(for: selection)
}
let status = AudioUnitSetProperty(
audioUnit,
kAudioOutputUnitProperty_CurrentDevice,
kAudioUnitScope_Global,
0,
&deviceID,
UInt32(MemoryLayout<AudioObjectID>.size))
guard status == noErr else {
self.logger.warning(
"realtime selected input binding failed status=\(status); using system default")
return self.defaultFallback(for: selection)
}
self.logger.info(
"realtime selected input bound uid=\(selectedUID, privacy: .private(mask: .hash))")
return selection
}
private func defaultFallback(
for selection: AudioInputDeviceResolution) -> AudioInputDeviceResolution
{
AudioInputDeviceResolution(
selectedUID: selection.selectedUID,
resolvedUID: AudioInputDeviceObserver.resolveSelection(nil).resolvedUID,
fellBackToSystemDefault: selection.selectedUID != nil)
}
private func startDeviceObserver() {
let observer = AudioInputDeviceObserver()
observer.start { [weak self] in
Task { @MainActor [weak self] in
self?.audioInputDevicesDidChange()
}
}
self.audioInputObserver = observer
}
private func startOutputRouteObserver() {
let observer = MacRealtimeTalkOutputRouteObserver()
let onChange = self.replaceOutputRouteObserver(observer)
observer.start(onChange: onChange)
}
private func replaceOutputRouteObserver(
_ observer: MacRealtimeTalkOutputRouteObserver) -> @Sendable (MacRealtimeTalkOutputRoute?) -> Void
{
self.outputRouteObservationGeneration &+= 1
let generation = self.outputRouteObservationGeneration
self.audioOutputObserver = observer
#if DEBUG
let onHandled = self.testOutputRouteCallbackHandled
self.testOutputRouteCallbackHandled = nil
#endif
return { [weak self, observerID = ObjectIdentifier(observer)] route in
Task { @MainActor [weak self] in
#if DEBUG
defer { onHandled?() }
#endif
guard let self,
generation == self.outputRouteObservationGeneration,
self.audioOutputObserver.map(ObjectIdentifier.init) == observerID
else { return }
self.updateOutputRoute(route)
}
}
}
private func retireOutputRouteObserver() {
self.outputRouteObservationGeneration &+= 1
self.audioOutputObserver?.stop()
self.audioOutputObserver = nil
self.suppressInputDuringOutput = true
self.outputRouteDecisionState.reset()
}
private func updateOutputRoute(_ route: MacRealtimeTalkOutputRoute?) {
guard let decision = self.outputRouteDecisionState.update(route: route) else { return }
self.suppressInputDuringOutput = decision.suppressesInputDuringOutput
self.logger.info(
"realtime output route decision \(decision.redactedDescription, privacy: .public)")
}
private func audioInputDevicesDidChange() {
guard let targetSampleRate, let onAudio else { return }
let desiredResolution = AudioInputDeviceObserver.resolveSelection(self.selectedInputUID())
guard desiredResolution != self.activeInputResolution ||
self.activeInputResolution?.shouldRestart(
availableUIDs: AudioInputDeviceObserver.aliveInputDeviceUIDs(),
defaultUID: AudioInputDeviceObserver.defaultInputDeviceUID()) == true
else { return }
self.logger.warning("realtime active/default input changed; restarting capture")
self.restartCaptureAfterInputChange {
try self.startCaptureEngine(targetSampleRate: targetSampleRate, onAudio: onAudio)
}
}
private func restartCaptureAfterInputChange(_ restart: () throws -> Void) {
self.deliveryGate.deactivate()
self.teardownEngine()
do {
try restart()
} catch {
self.logger.error(
"realtime input restart failed: \(error.localizedDescription, privacy: .public)")
let onFailure = self.onFailure
self.stop()
onFailure?(String(
format: String(localized: "Realtime microphone became unavailable: %@"),
error.localizedDescription))
}
}
private func teardownEngine() {
if self.tapInstalled, let inputNode {
inputNode.removeTap(onBus: 0)
}
self.tapInstalled = false
self.audioEngine?.stop()
self.audioEngine = nil
inputNode = nil
self.activeInputResolution = nil
}
#if DEBUG
func _test_replaceOutputRouteObserver()
-> (
callback: @Sendable (MacRealtimeTalkOutputRoute?) -> Void,
handled: AsyncStream<Void>)
{
let handled = AsyncStream<Void>.makeStream(bufferingPolicy: .bufferingNewest(1))
self.testOutputRouteCallbackHandled = {
handled.continuation.yield()
handled.continuation.finish()
}
return (self.replaceOutputRouteObserver(MacRealtimeTalkOutputRouteObserver()), handled.stream)
}
#endif
}
struct MacRealtimeTalkOutputRoute: Equatable, Sendable {
let transportType: UInt32
let terminalTypes: [UInt32]
let selectedDataSource: MacRealtimeTalkOutputDataSource
}
enum MacRealtimeTalkOutputDataSource: Equatable, Sendable {
case unsupported
case failed
case selected(kinds: [UInt32])
}
enum MacRealtimeTalkOutputRouteDecisionReason: String, Sendable {
case routeUnavailable = "route-unavailable"
case dataSourceReadFailed = "data-source-read-failed"
case transportNotAllowlisted = "transport-not-allowlisted"
case outputKindUnavailable = "output-kind-unavailable"
case outputKindNotHeadphones = "output-kind-not-headphones"
case isolatedHeadphones = "isolated-headphones"
}
struct MacRealtimeTalkOutputRouteDecision: Equatable, Sendable {
let suppressesInputDuringOutput: Bool
let reason: MacRealtimeTalkOutputRouteDecisionReason
let transportType: UInt32?
let effectiveKinds: [UInt32]
let selectedDataSource: MacRealtimeTalkOutputDataSource?
var redactedDescription: String {
let transport = self.transportType.map(MacRealtimeTalkFourCC.describe) ?? "unavailable"
let kinds = MacRealtimeTalkFourCC.describe(self.effectiveKinds)
let source = switch self.selectedDataSource {
case .unsupported:
"unsupported"
case .failed:
"failed"
case let .selected(sourceKinds):
Self.selectedDataSourceTag(sourceKinds)
case nil:
"unavailable"
}
return "transport=\(transport) kinds=\(kinds) source=\(source) " +
"suppression=\(self.suppressesInputDuringOutput) reason=\(self.reason.rawValue)"
}
private static func selectedDataSourceTag(_ kinds: [UInt32]) -> String {
"selected:" + MacRealtimeTalkFourCC.describe(kinds)
}
}
enum MacRealtimeTalkOutputRoutePolicy {
static func decision(
for route: MacRealtimeTalkOutputRoute?) -> MacRealtimeTalkOutputRouteDecision
{
guard let route else {
return MacRealtimeTalkOutputRouteDecision(
suppressesInputDuringOutput: true,
reason: .routeUnavailable,
transportType: nil,
effectiveKinds: [],
selectedDataSource: nil)
}
if route.selectedDataSource == .failed {
return self.decision(
route: route,
effectiveKinds: [],
suppressesInput: true,
reason: .dataSourceReadFailed)
}
// The selected source is the active routing fact. Stream terminals are only a
// fallback for devices that expose no data-source property.
let effectiveKinds: [UInt32] = switch route.selectedDataSource {
case .unsupported:
route.terminalTypes.sorted()
case let .selected(kinds):
kinds.sorted()
case .failed:
[]
}
let allowlistedTransports: Set<UInt32> = [
kAudioDeviceTransportTypeBuiltIn,
kAudioDeviceTransportTypeUSB,
kAudioDeviceTransportTypeBluetooth,
kAudioDeviceTransportTypeBluetoothLE,
]
guard allowlistedTransports.contains(route.transportType) else {
return self.decision(
route: route,
effectiveKinds: effectiveKinds,
suppressesInput: true,
reason: .transportNotAllowlisted)
}
guard !effectiveKinds.isEmpty else {
return self.decision(
route: route,
effectiveKinds: [],
suppressesInput: true,
reason: .outputKindUnavailable)
}
guard effectiveKinds.allSatisfy({ $0 == kAudioStreamTerminalTypeHeadphones }) else {
return self.decision(
route: route,
effectiveKinds: effectiveKinds,
suppressesInput: true,
reason: .outputKindNotHeadphones)
}
return self.decision(
route: route,
effectiveKinds: effectiveKinds,
suppressesInput: false,
reason: .isolatedHeadphones)
}
private static func decision(
route: MacRealtimeTalkOutputRoute,
effectiveKinds: [UInt32],
suppressesInput: Bool,
reason: MacRealtimeTalkOutputRouteDecisionReason) -> MacRealtimeTalkOutputRouteDecision
{
let selectedDataSource = switch route.selectedDataSource {
case .unsupported:
MacRealtimeTalkOutputDataSource.unsupported
case .failed:
MacRealtimeTalkOutputDataSource.failed
case let .selected(kinds):
MacRealtimeTalkOutputDataSource.selected(kinds: kinds.sorted())
}
return MacRealtimeTalkOutputRouteDecision(
suppressesInputDuringOutput: suppressesInput,
reason: reason,
transportType: route.transportType,
effectiveKinds: effectiveKinds,
selectedDataSource: selectedDataSource)
}
}
struct MacRealtimeTalkOutputRouteDecisionState {
private(set) var current: MacRealtimeTalkOutputRouteDecision?
mutating func update(route: MacRealtimeTalkOutputRoute?) -> MacRealtimeTalkOutputRouteDecision? {
let next = MacRealtimeTalkOutputRoutePolicy.decision(for: route)
guard next != self.current else { return nil }
self.current = next
return next
}
mutating func reset() {
self.current = nil
}
}
private enum MacRealtimeTalkFourCC {
static func describe(_ values: [UInt32]) -> String {
"[" + values.map(self.describe).joined(separator: ",") + "]"
}
static func describe(_ value: UInt32) -> String {
let bytes = [
UInt8((value >> 24) & 0xFF),
UInt8((value >> 16) & 0xFF),
UInt8((value >> 8) & 0xFF),
UInt8(value & 0xFF),
]
guard bytes.allSatisfy({ (0x20...0x7E).contains($0) }) else {
return String(format: "0x%08X", value)
}
return "'\(String(bytes: bytes, encoding: .ascii) ?? "????")'"
}
}
private struct MacRealtimeTalkAudioPropertyObservation {
let objectID: AudioObjectID
let address: AudioObjectPropertyAddress
let listener: AudioObjectPropertyListenerBlock
init?(
objectID: AudioObjectID,
selector: AudioObjectPropertySelector,
scope: AudioObjectPropertyScope,
listener: @escaping AudioObjectPropertyListenerBlock)
{
var address = AudioObjectPropertyAddress(
mSelector: selector,
mScope: scope,
mElement: kAudioObjectPropertyElementMain)
guard AudioObjectAddPropertyListenerBlock(
objectID,
&address,
DispatchQueue.main,
listener) == noErr
else { return nil }
self.objectID = objectID
self.address = address
self.listener = listener
}
func stop() {
var address = self.address
_ = AudioObjectRemovePropertyListenerBlock(
self.objectID,
&address,
DispatchQueue.main,
self.listener)
}
}
final class MacRealtimeTalkOutputRouteObserver: @unchecked Sendable {
private let logger = Logger(subsystem: "ai.openclaw", category: "talk.realtime.output-route")
private var defaultOutputObservation: MacRealtimeTalkAudioPropertyObservation?
private var dataSourceObservation: MacRealtimeTalkAudioPropertyObservation?
private var warningReported = false
func start(onChange: @escaping @Sendable (MacRealtimeTalkOutputRoute?) -> Void) {
guard self.defaultOutputObservation == nil else { return }
let listener: AudioObjectPropertyListenerBlock = { [weak self] _, _ in
self?.bindCurrentOutput(onChange: onChange)
}
guard let observation = MacRealtimeTalkAudioPropertyObservation(
objectID: AudioObjectID(kAudioObjectSystemObject),
selector: kAudioHardwarePropertyDefaultOutputDevice,
scope: kAudioObjectPropertyScopeGlobal,
listener: listener)
else {
self.reportWarningOnce("default-output-listener-failed")
onChange(nil)
return
}
self.defaultOutputObservation = observation
self.bindCurrentOutput(onChange: onChange)
}
func stop() {
self.defaultOutputObservation?.stop()
self.defaultOutputObservation = nil
self.dataSourceObservation?.stop()
self.dataSourceObservation = nil
}
private func bindCurrentOutput(
onChange: @escaping @Sendable (MacRealtimeTalkOutputRoute?) -> Void)
{
self.dataSourceObservation?.stop()
self.dataSourceObservation = nil
guard let deviceID = Self.defaultOutputDeviceID() else {
self.reportWarningOnce("default-output-read-failed")
onChange(nil)
return
}
guard let route = Self.currentRoute(deviceID: deviceID) else {
self.reportWarningOnce("output-route-read-failed")
onChange(nil)
return
}
if route.selectedDataSource == .failed {
self.reportWarningOnce("data-source-read-failed")
}
guard Self.hasDataSourceProperty(deviceID: deviceID) else {
onChange(route)
return
}
let listener: AudioObjectPropertyListenerBlock = { [weak self] _, _ in
guard let self else { return }
guard let refreshedRoute = Self.currentRoute(deviceID: deviceID) else {
self.reportWarningOnce("output-route-read-failed")
onChange(nil)
return
}
if refreshedRoute.selectedDataSource == .failed {
self.reportWarningOnce("data-source-read-failed")
}
onChange(refreshedRoute)
}
guard let observation = MacRealtimeTalkAudioPropertyObservation(
objectID: deviceID,
selector: kAudioDevicePropertyDataSource,
scope: kAudioDevicePropertyScopeOutput,
listener: listener)
else {
// A supported source can change without the device ID changing. If it cannot
// be observed, poison the route so the suppression policy remains fail-closed.
self.reportWarningOnce("data-source-listener-failed")
onChange(MacRealtimeTalkOutputRoute(
transportType: route.transportType,
terminalTypes: route.terminalTypes,
selectedDataSource: .failed))
return
}
self.dataSourceObservation = observation
onChange(route)
}
private static func currentRoute(deviceID: AudioObjectID) -> MacRealtimeTalkOutputRoute? {
guard let transportType = uint32Property(
objectID: deviceID,
selector: kAudioDevicePropertyTransportType,
scope: kAudioObjectPropertyScopeGlobal)
else { return nil }
let terminalTypes = self.outputTerminalTypes(deviceID: deviceID)
return MacRealtimeTalkOutputRoute(
transportType: transportType,
terminalTypes: terminalTypes,
selectedDataSource: self.selectedDataSource(deviceID: deviceID))
}
private static func defaultOutputDeviceID() -> AudioObjectID? {
var address = AudioObjectPropertyAddress(
mSelector: kAudioHardwarePropertyDefaultOutputDevice,
mScope: kAudioObjectPropertyScopeGlobal,
mElement: kAudioObjectPropertyElementMain)
var deviceID = AudioObjectID(0)
var size = UInt32(MemoryLayout<AudioObjectID>.size)
let status = AudioObjectGetPropertyData(
AudioObjectID(kAudioObjectSystemObject),
&address,
0,
nil,
&size,
&deviceID)
return status == noErr && deviceID != 0 ? deviceID : nil
}
private static func outputTerminalTypes(deviceID: AudioObjectID) -> [UInt32] {
var address = AudioObjectPropertyAddress(
mSelector: kAudioDevicePropertyStreams,
mScope: kAudioDevicePropertyScopeOutput,
mElement: kAudioObjectPropertyElementMain)
var size: UInt32 = 0
guard AudioObjectGetPropertyDataSize(deviceID, &address, 0, nil, &size) == noErr,
size > 0,
Int(size) % MemoryLayout<AudioStreamID>.size == 0
else { return [] }
var streamIDs = [AudioStreamID](
repeating: 0,
count: Int(size) / MemoryLayout<AudioStreamID>.size)
guard AudioObjectGetPropertyData(
deviceID,
&address,
0,
nil,
&size,
&streamIDs) == noErr
else { return [] }
var terminalTypes: [UInt32] = []
terminalTypes.reserveCapacity(streamIDs.count)
for streamID in streamIDs {
guard let terminalType = self.uint32Property(
objectID: streamID,
selector: kAudioStreamPropertyTerminalType,
scope: kAudioObjectPropertyScopeGlobal)
else { return [] }
terminalTypes.append(terminalType)
}
return terminalTypes
}
private static func hasDataSourceProperty(deviceID: AudioObjectID) -> Bool {
var address = AudioObjectPropertyAddress(
mSelector: kAudioDevicePropertyDataSource,
mScope: kAudioDevicePropertyScopeOutput,
mElement: kAudioObjectPropertyElementMain)
return AudioObjectHasProperty(deviceID, &address)
}
private static func selectedDataSource(
deviceID: AudioObjectID) -> MacRealtimeTalkOutputDataSource
{
var address = AudioObjectPropertyAddress(
mSelector: kAudioDevicePropertyDataSource,
mScope: kAudioDevicePropertyScopeOutput,
mElement: kAudioObjectPropertyElementMain)
guard AudioObjectHasProperty(deviceID, &address) else { return .unsupported }
var size: UInt32 = 0
guard AudioObjectGetPropertyDataSize(deviceID, &address, 0, nil, &size) == noErr,
size > 0,
Int(size) % MemoryLayout<UInt32>.size == 0
else { return .failed }
var sourceIDs = [UInt32](
repeating: 0,
count: Int(size) / MemoryLayout<UInt32>.size)
guard AudioObjectGetPropertyData(
deviceID,
&address,
0,
nil,
&size,
&sourceIDs) == noErr,
!sourceIDs.isEmpty
else { return .failed }
var kinds: [UInt32] = []
kinds.reserveCapacity(sourceIDs.count)
for sourceID in sourceIDs {
guard let kind = self.dataSourceKind(deviceID: deviceID, sourceID: sourceID)
else { return .failed }
kinds.append(kind)
}
return .selected(kinds: kinds)
}
private static func dataSourceKind(
deviceID: AudioObjectID,
sourceID: UInt32) -> UInt32?
{
var address = AudioObjectPropertyAddress(
mSelector: kAudioDevicePropertyDataSourceKindForID,
mScope: kAudioDevicePropertyScopeOutput,
mElement: kAudioObjectPropertyElementMain)
var input = sourceID
var output: UInt32 = 0
var status = kAudioHardwareUnspecifiedError
withUnsafeMutablePointer(to: &input) { inputPointer in
withUnsafeMutablePointer(to: &output) { outputPointer in
var translation = AudioValueTranslation(
mInputData: UnsafeMutableRawPointer(inputPointer),
mInputDataSize: UInt32(MemoryLayout<UInt32>.size),
mOutputData: UnsafeMutableRawPointer(outputPointer),
mOutputDataSize: UInt32(MemoryLayout<UInt32>.size))
var size = UInt32(MemoryLayout<AudioValueTranslation>.size)
status = AudioObjectGetPropertyData(
deviceID,
&address,
0,
nil,
&size,
&translation)
}
}
return status == noErr ? output : nil
}
private static func uint32Property(
objectID: AudioObjectID,
selector: AudioObjectPropertySelector,
scope: AudioObjectPropertyScope) -> UInt32?
{
var address = AudioObjectPropertyAddress(
mSelector: selector,
mScope: scope,
mElement: kAudioObjectPropertyElementMain)
var value: UInt32 = 0
var size = UInt32(MemoryLayout<UInt32>.size)
let status = AudioObjectGetPropertyData(
objectID,
&address,
0,
nil,
&size,
&value)
return status == noErr ? value : nil
}
private func reportWarningOnce(_ reason: String) {
guard !self.warningReported else { return }
self.warningReported = true
self.logger.warning(
"realtime output route observation degraded reason=\(reason, privacy: .public)")
}
}
enum MacRealtimeTalkAudioCaptureError: LocalizedError {
case invalidTargetSampleRate
case inputUnavailable
case invalidInputFormat
var errorDescription: String? {
switch self {
case .invalidTargetSampleRate: String(localized: "Realtime Talk requested an invalid audio sample rate")
case .inputUnavailable: String(localized: "Selected input and system default are unavailable")
case .invalidInputFormat: String(localized: "Selected audio input has no usable Float32 format")
}
}
}
enum MacRealtimeTalkAudioFrameEncoder {
nonisolated static func encode(
buffer: AVAudioPCMBuffer,
targetSampleRate: Double,
timestampMs: Double) -> RealtimeTalkAudioFrame?
{
let inputSampleRate = buffer.format.sampleRate
guard targetSampleRate.isFinite, targetSampleRate > 0
else { return nil }
let data = RealtimeTalkPCM16Encoder.encode(
buffer: buffer,
inputSampleRate: inputSampleRate,
targetSampleRate: targetSampleRate)
guard !data.isEmpty else { return nil }
return RealtimeTalkAudioFrame(
data: data,
timestampMs: timestampMs,
rms: Float(TalkAudioLevel.pcm16RMS(data)))
}
}
enum MacRealtimeTalkTapHandlerFactory {
/// AVAudioEngine invokes tap blocks on a realtime audio queue. Build the block from a
/// nonisolated context so Swift does not inherit MacRealtimeTalkAudioCapture's MainActor
/// executor and trap when Core Audio calls it off the main thread.
nonisolated static func make(
targetSampleRate: Double,
deliveryGate: TalkGenerationDeliveryGate,
deliveryToken: UInt64,
onAudio: @escaping @Sendable (RealtimeTalkAudioFrame) -> Void) -> AVAudioNodeTapBlock
{
{ buffer, _ in
guard deliveryGate.isActive(deliveryToken) else { return }
let frame = MacRealtimeTalkAudioFrameEncoder.encode(
buffer: buffer,
targetSampleRate: targetSampleRate,
timestampMs: ProcessInfo.processInfo.systemUptime * 1000)
guard let frame else { return }
deliveryGate.deliver(ifActive: deliveryToken) {
onAudio(frame)
}
}
}
}
final class TalkGenerationDeliveryGate: @unchecked Sendable {
private let lock = NSLock()
private var generation: UInt64 = 0
private var active = true
func activate() -> UInt64 {
self.lock.lock()
defer { self.lock.unlock() }
self.generation &+= 1
self.active = true
return self.generation
}
func deactivate() {
self.lock.lock()
self.generation &+= 1
self.active = false
self.lock.unlock()
}
func isActive(_ generation: UInt64) -> Bool {
self.lock.lock()
defer { self.lock.unlock() }
return self.active && self.generation == generation
}
@discardableResult
func deliver(ifActive generation: UInt64, _ body: () -> Void) -> Bool {
self.lock.lock()
defer { self.lock.unlock() }
guard self.active, self.generation == generation else { return false }
body()
return true
}
}
@@ -16,6 +16,18 @@ struct TalkModeGatewayConfigState {
let referenceAudioPath: String?
let referenceText: String?
let seamColorHex: String?
let realtimeProvider: String?
let realtimeModelId: String?
let realtimeSpeakerVoice: String?
let realtimeMode: String?
let realtimeTransport: String?
let realtimeBrain: String?
var hasGatewayRealtimeRelayTuple: Bool {
self.realtimeMode == "realtime" &&
self.realtimeTransport == "gateway-relay" &&
self.realtimeBrain == "agent-consult"
}
}
enum TalkModeGatewayConfigParser {
@@ -62,6 +74,22 @@ enum TalkModeGatewayConfigParser {
.trimmingCharacters(in: .whitespacesAndNewlines)
let referenceText = activeConfig?["referenceText"]?.stringValue?
.trimmingCharacters(in: .whitespacesAndNewlines)
let realtime = talk?["realtime"]?.dictionaryValue
let realtimeProviders = realtime?["providers"]?.dictionaryValue
let realtimeProvider = Self.firstString(realtime, keys: ["provider"])
?? Self.singleRealtimeProviderId(realtimeProviders)
let realtimeProviderConfig = Self.realtimeProviderConfig(
providers: realtimeProviders,
provider: realtimeProvider)
let realtimeModelId = Self.firstString(realtime, keys: ["model"])
?? Self.firstString(realtimeProviderConfig, keys: ["model"])
let realtimeSpeakerVoice = Self.firstString(
realtime,
keys: ["speakerVoice", "voice"])
?? Self.firstString(realtimeProviderConfig, keys: ["speakerVoice", "voice"])
let realtimeMode = Self.firstString(realtime, keys: ["mode"])?.lowercased()
let realtimeTransport = Self.firstString(realtime, keys: ["transport"])?.lowercased()
let realtimeBrain = Self.firstString(realtime, keys: ["brain"])?.lowercased()
let resolvedVoice: String? = if activeProvider == defaultProvider {
(voice?.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty == false ? voice : nil) ??
(envVoice?.isEmpty == false ? envVoice : nil) ??
@@ -90,7 +118,13 @@ enum TalkModeGatewayConfigParser {
apiKey: resolvedApiKey,
referenceAudioPath: referenceAudioPath?.isEmpty == false ? referenceAudioPath : nil,
referenceText: referenceText?.isEmpty == false ? referenceText : nil,
seamColorHex: rawSeam.isEmpty ? nil : rawSeam)
seamColorHex: rawSeam.isEmpty ? nil : rawSeam,
realtimeProvider: realtimeProvider,
realtimeModelId: realtimeModelId,
realtimeSpeakerVoice: realtimeSpeakerVoice,
realtimeMode: realtimeMode,
realtimeTransport: realtimeTransport,
realtimeBrain: realtimeBrain)
}
static func fallback(
@@ -119,6 +153,50 @@ enum TalkModeGatewayConfigParser {
apiKey: resolvedApiKey,
referenceAudioPath: nil,
referenceText: nil,
seamColorHex: nil)
seamColorHex: nil,
realtimeProvider: nil,
realtimeModelId: nil,
realtimeSpeakerVoice: nil,
realtimeMode: nil,
realtimeTransport: nil,
realtimeBrain: nil)
}
private static func firstString(
_ config: [String: AnyCodable]?,
keys: [String]) -> String?
{
guard let config else { return nil }
for key in keys {
let value = config[key]?.stringValue?.trimmingCharacters(in: .whitespacesAndNewlines)
if value?.isEmpty == false { return value }
}
return nil
}
private static func singleRealtimeProviderId(_ providers: [String: AnyCodable]?) -> String? {
guard let providers, providers.count == 1 else { return nil }
let provider = providers.keys.first?.trimmingCharacters(in: .whitespacesAndNewlines)
return provider?.isEmpty == false ? provider : nil
}
private static func realtimeProviderConfig(
providers: [String: AnyCodable]?,
provider: String?) -> [String: AnyCodable]?
{
guard let providers else { return nil }
if let provider {
if let exact = providers[provider]?.dictionaryValue {
return exact
}
return providers.first { key, _ in
key.trimmingCharacters(in: .whitespacesAndNewlines)
.caseInsensitiveCompare(provider) == .orderedSame
}?.value.dictionaryValue
}
if providers.count == 1 {
return providers.values.first?.dictionaryValue
}
return nil
}
}
@@ -0,0 +1,816 @@
import Foundation
import OpenClawChatUI
import OpenClawKit
import OSLog
private enum RealtimeRelayConfigurationError: LocalizedError {
case incompatible
var errorDescription: String? {
"Gateway realtime Talk configuration changed before startup"
}
}
extension TalkModeRuntime {
#if DEBUG
typealias VoiceWakePermissionProvider = @Sendable () async -> Bool
typealias RealtimeAudioCaptureProvider =
@MainActor @Sendable () -> any RealtimeTalkAudioCapturing
#endif
private enum ScheduledRealtimeRecoveryState: Equatable {
case cancelled
case waitingForStartToFinish
case ready
}
private static let realtimeStableSessionSeconds: TimeInterval = 30
private static let realtimeRestartDelaysNanoseconds: [UInt64] = [500_000_000, 2_000_000_000]
static func realtimeRestartAttempt(
previousRapidRestarts: Int,
activeDuration: TimeInterval) -> Int
{
activeDuration >= self.realtimeStableSessionSeconds ? 1 : previousRapidRestarts + 1
}
static func realtimeRestartDelayNanoseconds(attempt: Int) -> UInt64? {
guard attempt > 0, attempt <= self.realtimeRestartDelaysNanoseconds.count else { return nil }
return self.realtimeRestartDelaysNanoseconds[attempt - 1]
}
func stop(
reconfigurationGeneration expectedReconfigurationGeneration: UInt64?,
lifecycleGeneration expectedLifecycleGeneration: Int?) async
{
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
self.pendingRealtimeRelayStartLifecycleGeneration = nil
self.resetRealtimeRecoveryState()
self.realtimeRelayGeneration &+= 1
self.realtimeRelayStartGeneration = nil
let realtimeSession = self.detachResourcesForRealtimeStop()
if let realtimeSession {
await MainActor.run { realtimeSession.stop() }
}
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
await stopSpeaking(
reason: .manual,
reconfigurationGeneration: expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
let projectionGeneration = self.realtimeRelayGeneration
_ = await MainActor.run {
self.realtimeRelayDeliveryGate.deliver(ifActive: projectionGeneration) {
TalkModeController.shared.updateLevel(0)
TalkModeController.shared.updatePartialTranscript("")
TalkModeController.shared.updatePhase(.idle)
}
}
}
func ownsReconfiguration(
_ expectedReconfigurationGeneration: UInt64?,
lifecycleGeneration expectedLifecycleGeneration: Int?) -> Bool
{
(expectedReconfigurationGeneration.map { $0 == self.realtimeReconfigurationGeneration } ?? true) &&
(expectedLifecycleGeneration.map { $0 == self.lifecycleGeneration } ?? true)
}
func beginRealtimeReconfiguration() -> (generation: UInt64, lifecycleGeneration: Int) {
self.lifecycleGeneration &+= 1
self.realtimeReconfigurationGeneration &+= 1
return (self.realtimeReconfigurationGeneration, self.lifecycleGeneration)
}
func realtimeRelayPreferenceDidChange() async {
guard isEnabled else { return }
let reconfiguration = self.beginRealtimeReconfiguration()
await self.stop(
reconfigurationGeneration: reconfiguration.generation,
lifecycleGeneration: reconfiguration.lifecycleGeneration)
guard self.isCurrent(reconfiguration.lifecycleGeneration) else { return }
await self.start()
}
func start() async {
let gen = lifecycleGeneration
#if DEBUG
guard self.voiceWakeSupportedProvider() else { return }
let hasVoiceWakePermission = await self.voiceWakePermissionProvider()
#else
guard voiceWakeSupported else { return }
let hasVoiceWakePermission = await PermissionManager.ensureVoiceWakePermissions(interactive: true)
#endif
guard hasVoiceWakePermission else {
logger.error("talk runtime not starting: permissions missing")
return
}
macOSRealtimeRelayOptIn = await MainActor.run {
AppStateStore.shared.talkRealtimeRelayEnabled
}
let bypassRealtime = bypassRealtimeOnNextStart
bypassRealtimeOnNextStart = false
if !macOSRealtimeRelayOptIn || bypassRealtime {
await reloadConfig()
}
guard isCurrent(gen) else { return }
if isPaused {
phase = .idle
await MainActor.run {
TalkModeController.shared.updateLevel(0)
TalkModeController.shared.updatePhase(.idle)
}
return
}
var nativeFallbackStatus: String?
if self.macOSRealtimeRelayOptIn, !bypassRealtime {
let fallbackRecognitionGeneration = recognitionGeneration
let fallbackRealtimeRelayGeneration = realtimeRelayGeneration &+ 1
do {
try await self.startRealtimeRelay(generation: gen)
return
} catch is CancellationError {
if self.consumePendingRealtimeRelayStart() {
await self.start()
}
return
} catch {
pendingRealtimeRelayStartLifecycleGeneration = nil
guard isCurrent(gen), !isPaused,
recognitionGeneration == fallbackRecognitionGeneration,
realtimeRelayGeneration == fallbackRealtimeRelayGeneration,
realtimeRelayStartGeneration == nil,
realtimeSession == nil
else { return }
let fallbackConfig = await fetchTalkConfig()
guard await self.applyNativeFallbackTalkConfig(
fallbackConfig,
lifecycleGeneration: gen,
recognitionGeneration: fallbackRecognitionGeneration,
relayGeneration: fallbackRealtimeRelayGeneration)
else { return }
logger.error(
"talk realtime unavailable; using native fallback: " +
"\(error.localizedDescription, privacy: .public)")
nativeFallbackStatus = String(localized: "Realtime unavailable — using native speech")
}
}
await self.startNativeFallback(generation: gen, status: nativeFallbackStatus)
}
func consumePendingRealtimeRelayStart() -> Bool {
guard let generation = pendingRealtimeRelayStartLifecycleGeneration else { return false }
pendingRealtimeRelayStartLifecycleGeneration = nil
return generation == lifecycleGeneration && isEnabled && !isPaused &&
realtimeSession == nil && realtimeRelayStartGeneration == nil &&
self.macOSRealtimeRelayOptIn
}
private func startNativeFallback(generation: Int, status: String? = nil) async {
let relayGeneration = realtimeRelayGeneration
let recognitionStarted = await startRecognition(lifecycleGeneration: generation)
guard await self.commitNativeFallback(
recognitionStarted: recognitionStarted,
lifecycleGeneration: generation,
recognitionGeneration: recognitionGeneration,
relayGeneration: relayGeneration,
status: status)
else { return }
guard recognitionStarted else { return }
startAudioInputObserver()
startSilenceMonitor()
}
func commitNativeFallback(
recognitionStarted: Bool,
lifecycleGeneration: Int,
recognitionGeneration: Int,
relayGeneration: UInt64,
status: String?) async -> Bool
{
let ownsFallback = {
self.canCommitRecognitionStart(
lifecycleGeneration: lifecycleGeneration,
recognitionAttempt: recognitionGeneration) &&
self.realtimeRelayGeneration == relayGeneration &&
self.realtimeRelayStartGeneration == nil && self.realtimeSession == nil
}
guard ownsFallback() else { return false }
phase = recognitionStarted ? .listening : .idle
return await self.projectRealtimeRelay(relayGeneration, nil) {
if recognitionStarted, let status {
TalkModeController.shared.updatePartialTranscript(status)
} else if !recognitionStarted {
TalkModeController.shared.updatePartialTranscript(
String(localized: "Realtime unavailable — native speech could not start"))
}
TalkModeController.shared.updatePhase(recognitionStarted ? .listening : .idle)
}
}
func inputDeviceSelectionDidChange() async {
if let realtimeSession {
guard isEnabled, !isPaused else { return }
let relayGeneration = realtimeRelayGeneration
do {
try await MainActor.run {
try realtimeSession.setInputPaused(true)
try realtimeSession.setInputPaused(false)
}
} catch {
logger.error(
"talk realtime input restart failed: \(error.localizedDescription, privacy: .public)")
await self.handleRealtimeInputRestartFailure(
error.localizedDescription,
relayGeneration: relayGeneration)
}
return
}
guard isEnabled, !isPaused, phase == .listening else { return }
logger.info("talk input selection changed; restarting capture")
let lifecycleGeneration = self.lifecycleGeneration
_ = await startRecognition(lifecycleGeneration: lifecycleGeneration)
}
func shouldAttemptRealtimeRelay() -> Bool {
guard macOSRealtimeRelayOptIn else {
logger.debug("talk macOS realtime relay disabled locally; using native speech")
return false
}
guard hasGatewayRealtimeRelayTuple else {
logger.warning(
"talk macOS realtime relay opted in but Gateway tuple is incompatible: " +
"mode=\(realtimeMode ?? "missing", privacy: .public) " +
"transport=\(realtimeTransport ?? "missing", privacy: .public) " +
"brain=\(realtimeBrain ?? "missing", privacy: .public); using native fallback")
return false
}
return Self.shouldUseRealtimeRelay(
localOptIn: macOSRealtimeRelayOptIn,
hasGatewayRealtimeRelayTuple: hasGatewayRealtimeRelayTuple)
}
static func shouldUseRealtimeRelay(
localOptIn: Bool,
hasGatewayRealtimeRelayTuple: Bool) -> Bool
{
localOptIn && hasGatewayRealtimeRelayTuple
}
func startRealtimeRelay(generation: Int) async throws {
let relayGeneration = try beginRealtimeRelayStart()
defer {
if self.realtimeRelayStartGeneration == relayGeneration {
self.realtimeRelayStartGeneration = nil
}
}
let bootstrap: GatewayConnection.RealtimeTalkBootstrap
do {
bootstrap = try await self.realtimeTalkBootstrapProvider()
} catch {
try await self.applyRealtimeTalkConfig(
self.fallbackTalkConfig(),
lifecycleGeneration: generation,
relayGeneration: relayGeneration)
throw error
}
guard isCurrent(generation), !isPaused,
realtimeRelayGeneration == relayGeneration
else { throw CancellationError() }
let config = self.parseTalkConfig(bootstrap.configSnapshot)
try await self.applyRealtimeTalkConfig(
config,
lifecycleGeneration: generation,
relayGeneration: relayGeneration)
guard self.shouldAttemptRealtimeRelay() else {
throw RealtimeRelayConfigurationError.incompatible
}
let session = try await makeRealtimeRelaySession(
bootstrap: bootstrap,
lifecycleGeneration: generation,
relayGeneration: relayGeneration)
try await ownAndStartRealtimeSession(
session,
lifecycleGeneration: generation,
relayGeneration: relayGeneration,
start: { session in try await session.start() })
realtimeSessionReadyAt = Date()
phase = .listening
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.updatePartialTranscript("")
TalkModeController.shared.updatePhase(.listening)
}
logger.info(
"talk realtime ready provider=\(realtimeProvider ?? "default", privacy: .public) " +
"model=\(realtimeModelId ?? "default", privacy: .public)")
}
func applyRealtimeTalkConfig(
_ config: TalkModeGatewayConfigState,
lifecycleGeneration: Int,
relayGeneration: UInt64) async throws
{
let locale = await MainActor.run { () -> String? in
guard self.realtimeRelayDeliveryGate.deliver(ifActive: relayGeneration, {
AppStateStore.shared.seamColorHex = config.seamColorHex
}) else { return nil }
return AppStateStore.shared.voiceWakeLocaleID
}
#if DEBUG
if let checkpoint = self.realtimeConfigApplicationCheckpoint {
await checkpoint()
}
#endif
guard let locale,
self.isCurrent(lifecycleGeneration),
!self.isPaused,
self.realtimeRelayGeneration == relayGeneration,
self.realtimeRelayStartGeneration == relayGeneration
else { throw CancellationError() }
self.commitTalkConfig(config, locale: locale)
}
func applyNativeFallbackTalkConfig(
_ config: TalkModeGatewayConfigState,
lifecycleGeneration: Int,
recognitionGeneration: Int,
relayGeneration: UInt64) async -> Bool
{
let locale = await MainActor.run { AppStateStore.shared.voiceWakeLocaleID }
#if DEBUG
if let checkpoint = self.realtimeConfigApplicationCheckpoint {
await checkpoint()
}
#endif
guard self.isCurrent(lifecycleGeneration),
!self.isPaused,
self.recognitionGeneration == recognitionGeneration,
self.realtimeRelayGeneration == relayGeneration,
self.realtimeRelayStartGeneration == nil,
self.realtimeSession == nil
else { return false }
self.commitTalkConfig(config, locale: locale)
return true
}
func fallbackTalkConfig() -> TalkModeGatewayConfigState {
let env = ProcessInfo.processInfo.environment
return TalkModeGatewayConfigParser.fallback(
defaultModelIdFallback: Self.defaultModelIdFallback,
defaultSilenceTimeoutMs: Self.defaultSilenceTimeoutMs,
envVoice: env["ELEVENLABS_VOICE_ID"],
sagVoice: env["SAG_VOICE_ID"],
envApiKey: env["ELEVENLABS_API_KEY"])
}
private func beginRealtimeRelayStart() throws -> UInt64 {
guard realtimeSession == nil, realtimeRelayStartGeneration == nil else {
throw CancellationError()
}
realtimeRelayGeneration &+= 1
let relayGeneration = realtimeRelayGeneration
realtimeRelayStartGeneration = relayGeneration
return relayGeneration
}
private func makeRealtimeRelaySession(
bootstrap: GatewayConnection.RealtimeTalkBootstrap,
lifecycleGeneration: Int,
relayGeneration: UInt64) async throws -> RealtimeTalkRelaySession
{
guard isCurrent(lifecycleGeneration), !isPaused,
realtimeRelayGeneration == relayGeneration
else { throw CancellationError() }
let activeSessionKey = await MainActor.run {
WebChatManager.shared.activeSessionKey
}
let sessionKey: String = if let activeSessionKey {
activeSessionKey
} else {
bootstrap.sessionKey
}
let options = RealtimeTalkRelaySession.Options(
sessionKey: sessionKey,
provider: realtimeProvider,
model: realtimeModelId,
voice: realtimeSpeakerVoice)
#if DEBUG
let audioCaptureProvider = self.realtimeAudioCaptureProvider
#endif
return await MainActor.run {
#if DEBUG
let audioCapture = audioCaptureProvider()
#else
let audioCapture = MacRealtimeTalkAudioCapture()
#endif
return RealtimeTalkRelaySession(
transport: bootstrap.transport,
options: options,
audioCapture: audioCapture,
pcmPlayer: RealtimePCMStreamingAudioPlayer(),
onStatus: { [weak self] status in
Task { await self?.handleRealtimeStatus(status, relayGeneration: relayGeneration) }
},
onIssue: { [weak self] issue in
Task { await self?.handleRealtimeIssue(issue, relayGeneration: relayGeneration) }
},
onTermination: { [weak self] termination in
Task {
await self?.handleRealtimeTermination(
termination,
relayGeneration: relayGeneration)
}
},
onSpeakingChanged: { [weak self] speaking in
Task {
await self?.handleRealtimeSpeakingChanged(
speaking,
relayGeneration: relayGeneration)
}
},
onInputLevel: { [weak self] level in
Task { await self?.handleRealtimeInputLevel(level, relayGeneration: relayGeneration) }
},
onOutputLevel: { [weak self] level in
Task { await self?.handleRealtimeOutputLevel(level, relayGeneration: relayGeneration) }
},
onTranscript: { [weak self] transcript in
Task {
await self?.handleRealtimeTranscript(
transcript,
relayGeneration: relayGeneration)
}
})
}
}
private func ownAndStartRealtimeSession(
_ session: RealtimeTalkRelaySession,
lifecycleGeneration: Int,
relayGeneration: UInt64,
start: @MainActor @Sendable (RealtimeTalkRelaySession) async throws -> Void) async throws
{
// Construction crosses executors. Claim ownership only after every lifecycle and
// attempt fact is revalidated, then publish before start can suspend.
guard isCurrent(lifecycleGeneration), !isPaused,
realtimeRelayGeneration == relayGeneration,
realtimeRelayStartGeneration == relayGeneration,
realtimeSession == nil
else {
await MainActor.run { session.stop() }
throw CancellationError()
}
realtimeSession = session
do {
try await start(session)
} catch {
await MainActor.run { session.stop() }
if realtimeSession === session {
realtimeSession = nil
}
guard isCurrent(lifecycleGeneration), !isPaused,
realtimeRelayGeneration == relayGeneration,
realtimeRelayStartGeneration == relayGeneration
else {
throw CancellationError()
}
throw error
}
guard isCurrent(lifecycleGeneration), !isPaused,
realtimeRelayGeneration == relayGeneration,
realtimeSession === session
else {
await MainActor.run { session.stop() }
if realtimeSession === session {
realtimeSession = nil
}
throw CancellationError()
}
}
private func handleRealtimeStatus(_ status: String, relayGeneration: UInt64) {
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session)
else { return }
logger.debug("talk realtime status=\(status, privacy: .public)")
}
private func handleRealtimeIssue(_ issue: RealtimeTalkRelayIssue, relayGeneration: UInt64) async {
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session)
else { return }
logger.error(
"talk realtime issue code=\(issue.code, privacy: .public) " +
"message=\(issue.message, privacy: .public)")
_ = await self.projectRealtimeRelay(relayGeneration, session) { TalkModeController.shared
.updatePartialTranscript(issue.message)
}
}
func handleRealtimeInputRestartFailure(
_ message: String,
relayGeneration: UInt64) async
{
let issue = RealtimeTalkRelayIssue(
code: "audio_input_unavailable",
message: message,
provider: realtimeProvider,
model: realtimeModelId,
transport: "gateway-relay",
phase: "audio-input")
await handleRealtimeIssue(issue, relayGeneration: relayGeneration)
await handleRealtimeTermination(
.audioInputFailed(message: issue.message),
relayGeneration: relayGeneration)
}
func setRealtimeInputPaused(
_ paused: Bool,
session: RealtimeTalkRelaySession,
relayGeneration: UInt64) async -> Bool
{
do {
try await MainActor.run {
try session.setInputPaused(paused)
}
return true
} catch {
logger.error(
"talk realtime pause transition failed: \(error.localizedDescription, privacy: .public)")
await self.handleRealtimeInputRestartFailure(
error.localizedDescription,
relayGeneration: relayGeneration)
return false
}
}
func handleRealtimeTermination(
_ termination: RealtimeTalkRelayTermination,
relayGeneration: UInt64) async
{
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session)
else { return }
logger.warning(
"talk realtime terminated=\(String(describing: termination), privacy: .public)")
let activeDuration = realtimeSessionReadyAt.map { Date().timeIntervalSince($0) } ?? 0
realtimeRelayGeneration &+= 1
let terminalGeneration = realtimeRelayGeneration
// Session-owned terminations close before signalling; runtime-initiated ones do not.
// stop() is idempotent, so closing here keeps a dead relay and its event subscription
// from outliving their owner while recovery starts a replacement session.
await MainActor.run { session.stop() }
guard self.ownsRealtimeRelay(terminalGeneration, session) else { return }
realtimeSession = nil
realtimeSessionReadyAt = nil
phase = .idle
let shouldRecover = isEnabled && !isPaused
let attempt = Self.realtimeRestartAttempt(
previousRapidRestarts: rapidRealtimeRestartCount,
activeDuration: activeDuration)
let delay = Self.realtimeRestartDelayNanoseconds(attempt: attempt)
guard await self.projectRealtimeRelay(terminalGeneration, nil, {
TalkModeController.shared.updateLevel(0)
TalkModeController.shared.updateSpeakingLevel(nil)
TalkModeController.shared.updatePhase(.idle)
if shouldRecover {
TalkModeController.shared.updatePartialTranscript(delay == nil
? String(localized: "Realtime disconnected repeatedly — using native speech")
: String(localized: "Realtime disconnected — reconnecting…"))
}
}) else { return }
let lifecycleGeneration = self.lifecycleGeneration
let restartGeneration = realtimeRestartGeneration &+ 1
guard shouldRecover, self.ownsRealtimeRelay(terminalGeneration, nil),
isEnabled, !isPaused
else { return }
rapidRealtimeRestartCount = attempt
realtimeRestartGeneration = restartGeneration
bypassRealtimeOnNextStart = delay == nil
self.scheduleRealtimeRecovery(
after: delay,
lifecycleGeneration: lifecycleGeneration,
restartGeneration: restartGeneration)
}
func handleRealtimeSpeakingChanged(_ speaking: Bool, relayGeneration: UInt64) async {
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session),
isEnabled,
!self.isPaused
else { return }
if speaking {
phase = .speaking
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.updatePhase(.speaking)
}
} else if !isPaused {
phase = .listening
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.updatePhase(.listening)
}
}
}
func handleRealtimeInputLevel(_ level: Double, relayGeneration: UInt64) async {
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session),
isEnabled,
!self.isPaused
else { return }
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.updateLevel(level)
}
}
func handleRealtimeOutputLevel(_ level: Double?, relayGeneration: UInt64) async {
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session),
isEnabled,
!self.isPaused
else { return }
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.updateSpeakingLevel(level)
}
}
func handleRealtimeTranscript(
_ transcript: RealtimeTalkTranscript,
relayGeneration: UInt64) async
{
guard let session = realtimeSession,
ownsRealtimeRelay(relayGeneration, session),
isEnabled,
!self.isPaused
else { return }
let text = transcript.text.trimmingCharacters(in: .whitespacesAndNewlines)
guard !text.isEmpty else { return }
guard transcript.role == "user" else { return }
if transcript.isFinal {
phase = .thinking
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.commitTranscript(text)
TalkModeController.shared.updatePhase(.thinking)
}
} else {
_ = await self.projectRealtimeRelay(relayGeneration, session) {
TalkModeController.shared.updatePartialTranscript(text)
}
}
}
func ownsRealtimeRelay(_ generation: UInt64, _ session: RealtimeTalkRelaySession?) -> Bool {
realtimeRelayGeneration == generation && realtimeSession === session
}
func projectRealtimeRelay(
_ generation: UInt64,
_ session: RealtimeTalkRelaySession?,
_ body: @escaping @MainActor @Sendable () -> Void) async -> Bool
{
guard self.ownsRealtimeRelay(generation, session) else { return false }
return await MainActor.run {
self.realtimeRelayDeliveryGate.deliver(ifActive: generation, body)
}
}
func resetRealtimeRecoveryState() {
self.cancelScheduledRealtimeRecovery()
realtimeSessionReadyAt = nil
rapidRealtimeRestartCount = 0
bypassRealtimeOnNextStart = false
}
func cancelScheduledRealtimeRecovery() {
realtimeRestartGeneration &+= 1
realtimeRestartTask?.cancel()
realtimeRestartTask = nil
}
private func scheduleRealtimeRecovery(
after delayNanoseconds: UInt64?,
lifecycleGeneration: Int,
restartGeneration: UInt64)
{
realtimeRestartTask?.cancel()
realtimeRestartTask = Task { [weak self] in
if let delayNanoseconds {
do {
try await Task.sleep(nanoseconds: delayNanoseconds)
} catch {
return
}
}
while let self {
switch await self.scheduledRealtimeRecoveryState(
lifecycleGeneration: lifecycleGeneration,
restartGeneration: restartGeneration)
{
case .cancelled:
return
case .waitingForStartToFinish:
do {
try await Task.sleep(nanoseconds: 50_000_000)
} catch {
return
}
case .ready:
await self.performScheduledRealtimeRecovery(
lifecycleGeneration: lifecycleGeneration,
restartGeneration: restartGeneration)
return
}
}
}
}
private func scheduledRealtimeRecoveryState(
lifecycleGeneration: Int,
restartGeneration: UInt64) -> ScheduledRealtimeRecoveryState
{
guard self.lifecycleGeneration == lifecycleGeneration,
realtimeRestartGeneration == restartGeneration,
isEnabled,
!isPaused,
realtimeSession == nil
else { return .cancelled }
return realtimeRelayStartGeneration == nil ? .ready : .waitingForStartToFinish
}
private func performScheduledRealtimeRecovery(
lifecycleGeneration: Int,
restartGeneration: UInt64) async
{
guard self.scheduledRealtimeRecoveryState(
lifecycleGeneration: lifecycleGeneration,
restartGeneration: restartGeneration) == .ready
else { return }
realtimeRestartTask = nil
await self.start()
}
}
#if DEBUG
extension TalkModeRuntime {
func _test_beginRecognitionAttempt(lifecycleGeneration: Int) -> Int? {
self.beginRecognitionAttempt(lifecycleGeneration: lifecycleGeneration)
}
func _test_setRecognitionCleanupProbe(_ probe: (@Sendable () -> Void)?) {
self.recognitionCleanupProbe = probe
}
func _test_setVoiceWakeReadiness(supported: Bool, permissionGranted: Bool) {
self.voiceWakeSupportedProvider = { supported }
self.voiceWakePermissionProvider = { permissionGranted }
}
func _test_setRealtimeAudioCaptureProvider(
_ provider: @escaping RealtimeAudioCaptureProvider)
{
self.realtimeAudioCaptureProvider = provider
}
func _test_setRealtimeConfigApplicationCheckpoint(
_ checkpoint: (@Sendable () async -> Void)?)
{
self.realtimeConfigApplicationCheckpoint = checkpoint
}
func _test_enableRealtimeRelaySelection() {
(macOSRealtimeRelayOptIn, hasGatewayRealtimeRelayTuple) = (true, true)
}
func _test_prepareEnabledLifecycle() -> Int {
isEnabled = true
isPaused = false
lifecycleGeneration &+= 1
return lifecycleGeneration
}
func _test_prepareEnabledRealtimeSessionForClose(
_ session: RealtimeTalkRelaySession) -> UInt64
{
self.cancelScheduledRealtimeRecovery()
isEnabled = true
isPaused = false
lifecycleGeneration &+= 1
realtimeRelayGeneration &+= 1
realtimeSession = session
realtimeSessionReadyAt = nil
rapidRealtimeRestartCount = 0
bypassRealtimeOnNextStart = false
return realtimeRelayGeneration
}
}
#endif
+387 -219
View File
@@ -9,6 +9,8 @@ import Speech
actor TalkModeRuntime {
static let shared = TalkModeRuntime()
typealias RealtimeTalkBootstrapProvider =
@Sendable () async throws -> GatewayConnection.RealtimeTalkBootstrap
enum PlaybackPlan: Equatable {
case elevenLabsThenSystemVoice(apiKey: String, voiceId: String)
@@ -22,13 +24,13 @@ actor TalkModeRuntime {
case fallback
}
private let logger = Logger(subsystem: "ai.openclaw", category: "talk.runtime")
private let ttsLogger = Logger(subsystem: "ai.openclaw", category: "talk.tts")
private static let defaultModelIdFallback = "eleven_v3"
private static let defaultTalkProvider = "elevenlabs"
private static let mlxTalkProvider = "mlx"
private static let systemTalkProvider = "system"
private static let defaultSilenceTimeoutMs = TalkDefaults.silenceTimeoutMs
let logger = Logger(subsystem: "ai.openclaw", category: "talk.runtime")
let ttsLogger = Logger(subsystem: "ai.openclaw", category: "talk.tts")
static let defaultModelIdFallback = "eleven_v3"
static let defaultTalkProvider = "elevenlabs"
static let mlxTalkProvider = "mlx"
static let systemTalkProvider = "system"
static let defaultSilenceTimeoutMs = TalkDefaults.silenceTimeoutMs
private final class RMSMeter: @unchecked Sendable {
private let lock = NSLock()
@@ -54,16 +56,16 @@ actor TalkModeRuntime {
private var activeInputResolution: AudioInputDeviceResolution?
private var recognitionRequest: SFSpeechAudioBufferRecognitionRequest?
private var recognitionTask: SFSpeechRecognitionTask?
private var recognitionGeneration: Int = 0
var recognitionGeneration: Int = 0
private var rmsTask: Task<Void, Never>?
private let rmsMeter = RMSMeter()
private var captureTask: Task<Void, Never>?
private var silenceTask: Task<Void, Never>?
private var phase: TalkModePhase = .idle
private var isEnabled = false
private var isPaused = false
private var lifecycleGeneration: Int = 0
var phase: TalkModePhase = .idle
var isEnabled = false
var isPaused = false
var lifecycleGeneration: Int = 0
private var lastHeard: Date?
private var noiseFloorRMS: Double = 1e-4
@@ -79,6 +81,42 @@ actor TalkModeRuntime {
private var defaultOutputFormat: String?
private var interruptOnSpeech: Bool = true
private var activeTalkProvider = TalkModeRuntime.defaultTalkProvider
var realtimeProvider: String?
var realtimeModelId: String?
var realtimeSpeakerVoice: String?
var realtimeMode: String?
var realtimeTransport: String?
var realtimeBrain: String?
var hasGatewayRealtimeRelayTuple = false
var macOSRealtimeRelayOptIn = false
var realtimeSession: RealtimeTalkRelaySession?
var realtimeSessionReadyAt: Date?
var rapidRealtimeRestartCount = 0
var bypassRealtimeOnNextStart = false
let realtimeRelayDeliveryGate = TalkGenerationDeliveryGate()
var realtimeRelayGeneration: UInt64 = 0 {
didSet { _ = self.realtimeRelayDeliveryGate.activate() }
}
var realtimeRelayStartGeneration: UInt64?
var pendingRealtimeRelayStartLifecycleGeneration: Int?
var realtimeRestartGeneration: UInt64 = 0
var realtimeRestartTask: Task<Void, Never>?
var realtimeReconfigurationGeneration: UInt64 = 0
let realtimeTalkBootstrapProvider: RealtimeTalkBootstrapProvider
#if DEBUG
var voiceWakeSupportedProvider: @Sendable () -> Bool = { voiceWakeSupported }
var voiceWakePermissionProvider: VoiceWakePermissionProvider = {
await PermissionManager.ensureVoiceWakePermissions(interactive: true)
}
var realtimeAudioCaptureProvider: RealtimeAudioCaptureProvider = {
MacRealtimeTalkAudioCapture()
}
var realtimeConfigApplicationCheckpoint: (@Sendable () async -> Void)?
var recognitionCleanupProbe: (@Sendable () -> Void)?
#endif
private var speechLocaleID: String?
private var lastInterruptedAtSeconds: Double?
private var voiceAliases: [String: String] = [:]
@@ -93,8 +131,10 @@ actor TalkModeRuntime {
private let minSpeechRMS: Double = 1e-3
private let speechBoostFactor: Double = 6.0
static func configureRecognitionRequest(_ request: SFSpeechAudioBufferRecognitionRequest) {
SpeechRecognitionRequestPolicy.configureInteractiveTranscription(request)
init(realtimeTalkBootstrapProvider: @escaping RealtimeTalkBootstrapProvider = {
try await GatewayConnection.shared.acquireRealtimeTalkBootstrap()
}) {
self.realtimeTalkBootstrapProvider = realtimeTalkBootstrapProvider
}
// MARK: - Lifecycle
@@ -103,8 +143,9 @@ actor TalkModeRuntime {
guard enabled != self.isEnabled else { return }
self.isEnabled = enabled
self.lifecycleGeneration &+= 1
resetRealtimeRecoveryState()
if enabled {
await self.start()
await start()
} else {
await self.stop()
}
@@ -117,83 +158,117 @@ 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.macOSRealtimeRelayOptIn {
self.pendingRealtimeRelayStartLifecycleGeneration = self.lifecycleGeneration
return
}
if paused, realtimeSession == nil {
cancelScheduledRealtimeRecovery()
}
if let realtimeSession {
let relayGeneration = self.realtimeRelayGeneration
if paused {
await MainActor.run { realtimeSession.setOutputPaused(true) }
guard self.isPaused,
self.realtimeRelayGeneration == relayGeneration,
self.realtimeSession === realtimeSession,
await setRealtimeInputPaused(
true,
session: realtimeSession,
relayGeneration: relayGeneration)
else { return }
self.lastTranscript = ""
self.lastHeard = nil
self.lastSpeechEnergyAt = nil
self.phase = .idle
_ = await projectRealtimeRelay(relayGeneration, realtimeSession) {
TalkModeController.shared.updateLevel(0)
TalkModeController.shared.updateSpeakingLevel(nil)
TalkModeController.shared.updatePartialTranscript("")
TalkModeController.shared.updatePhase(.idle)
}
} else {
guard await 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
}
self.phase = .listening
_ = await projectRealtimeRelay(relayGeneration, realtimeSession) {
TalkModeController.shared.updatePhase(.listening)
}
}
return
}
if !paused, self.macOSRealtimeRelayOptIn {
await start()
return
}
if paused {
self.lastTranscript = ""
self.lastHeard = nil
self.lastSpeechEnergyAt = nil
await MainActor.run { TalkModeController.shared.updatePartialTranscript("") }
await self.stopRecognition()
self.stopRecognition()
return
}
if self.phase == .idle || self.phase == .listening {
await self.startRecognition()
let lifecycleGeneration = self.lifecycleGeneration
guard await self.startRecognition(lifecycleGeneration: lifecycleGeneration),
self.isCurrent(lifecycleGeneration), !self.isPaused else { return }
self.phase = .listening
await MainActor.run { TalkModeController.shared.updatePhase(.listening) }
self.startSilenceMonitor()
}
}
private func isCurrent(_ generation: Int) -> Bool {
func isCurrent(_ generation: Int) -> Bool {
generation == self.lifecycleGeneration && self.isEnabled
}
private func start() async {
let gen = self.lifecycleGeneration
guard voiceWakeSupported else { return }
guard await PermissionManager.ensureVoiceWakePermissions(interactive: true) else {
self.logger.error("talk runtime not starting: permissions missing")
return
}
self.startAudioInputObserver()
await reloadConfig()
guard self.isCurrent(gen) else { return }
if self.isPaused {
self.phase = .idle
await MainActor.run {
TalkModeController.shared.updateLevel(0)
TalkModeController.shared.updatePhase(.idle)
}
return
}
await self.startRecognition()
guard self.isCurrent(gen) else { return }
self.phase = .listening
await MainActor.run { TalkModeController.shared.updatePhase(.listening) }
self.startSilenceMonitor()
func stop() async {
await self.stop(reconfigurationGeneration: nil, lifecycleGeneration: nil)
}
private func stop() async {
func detachResourcesForRealtimeStop() -> RealtimeTalkRelaySession? {
let realtimeSession = self.realtimeSession
self.realtimeSession = nil
self.audioInputObserver?.stop()
self.audioInputObserver = nil
self.captureTask?.cancel()
self.captureTask = nil
self.silenceTask?.cancel()
self.silenceTask = nil
// Stop audio before changing phase (stopSpeaking is gated on .speaking).
await stopSpeaking(reason: .manual)
self.lastTranscript = ""
self.lastHeard = nil
self.lastSpeechEnergyAt = nil
self.phase = .idle
await self.stopRecognition()
await MainActor.run {
TalkModeController.shared.updateLevel(0)
TalkModeController.shared.updatePartialTranscript("")
TalkModeController.shared.updatePhase(.idle)
}
self.stopRecognition()
return realtimeSession
}
func inputDeviceSelectionDidChange() async {
guard self.isEnabled, !self.isPaused, self.phase == .listening else { return }
self.logger.info("talk input selection changed; restarting capture")
await self.startRecognition()
}
private func startAudioInputObserver() {
func startAudioInputObserver() {
guard self.audioInputObserver == nil else { return }
let observer = AudioInputDeviceObserver()
observer.start {
@@ -204,9 +279,10 @@ actor TalkModeRuntime {
private func audioInputDevicesDidChange() async {
guard self.isEnabled, !self.isPaused, self.phase == .listening else { return }
let lifecycleGeneration = self.lifecycleGeneration
let availableUIDs = AudioInputDeviceObserver.aliveInputDeviceUIDs()
guard let activeInputResolution else {
await self.startRecognition()
_ = await self.startRecognition(lifecycleGeneration: lifecycleGeneration)
return
}
guard activeInputResolution.shouldRestart(
@@ -215,7 +291,7 @@ actor TalkModeRuntime {
else { return }
self.logger.warning("talk active/default input changed; restarting capture")
await self.startRecognition()
_ = await self.startRecognition(lifecycleGeneration: lifecycleGeneration)
}
// MARK: - Speech recognition
@@ -228,12 +304,13 @@ actor TalkModeRuntime {
let generation: Int
}
private func startRecognition() async {
await self.stopRecognition()
self.recognitionGeneration &+= 1
let generation = self.recognitionGeneration
func startRecognition(lifecycleGeneration: Int) async -> Bool {
guard let recognitionAttempt = beginRecognitionAttempt(
lifecycleGeneration: lifecycleGeneration)
else { return false }
let voiceWakeLocale = await MainActor.run { AppStateStore.shared.voiceWakeLocaleID }
let selectedInputUID = await MainActor.run { AppStateStore.shared.voiceWakeMicID }
let supportedLocaleIDs = Set(SFSpeechRecognizer.supportedLocales().map(\.identifier))
let localeID = TalkConfigParsing.resolvedSpeechRecognitionLocaleID(
preferredLocaleIDs: [
@@ -242,71 +319,97 @@ actor TalkModeRuntime {
Locale.autoupdatingCurrent.identifier,
],
supportedLocaleIDs: supportedLocaleIDs)
self.recognizer = localeID
let recognizer = localeID
.map { SFSpeechRecognizer(locale: Locale(identifier: $0)) }
?? SFSpeechRecognizer()
guard let recognizer, recognizer.isAvailable else {
self.logger.error("talk recognizer unavailable")
return
return false
}
self.logger.debug("talk recognizer locale=\(recognizer.locale.identifier, privacy: .public)")
let request = SFSpeechAudioBufferRecognitionRequest()
Self.configureRecognitionRequest(request)
self.recognitionRequest = request
let selectedInputUID = await MainActor.run { AppStateStore.shared.voiceWakeMicID }
let selection = AudioInputDeviceObserver.resolveSelection(selectedInputUID)
// Preparation above crosses MainActor. Hardware can become owned/running only
// after this lifecycle and recognition attempt are revalidated together.
guard self.canCommitRecognitionStart(
lifecycleGeneration: lifecycleGeneration,
recognitionAttempt: recognitionAttempt) else { return false }
// AVAudioEngine materializes inputNode from the system default before CurrentDevice can bind.
// Without a usable default, accessing inputNode can SIGABRT even when another UID is alive.
guard selection.resolvedUID != nil, AudioInputDeviceObserver.hasUsableDefaultInputDevice() else {
self.audioEngine = nil
self.activeInputResolution = nil
self.logger.error("talk mode: no usable audio input device")
return
return false
}
do {
try self.configureAudioEngine(
selection: selection,
request: request,
enableVoiceProcessing: true)
} catch {
self.logger.warning(
"talk processed input setup failed; retrying without voice processing: " +
"\(error.localizedDescription, privacy: .public)")
self.discardAudioEngine()
do {
try self.configureAudioEngine(
let started = TalkRecognitionCaptureLifecycle.start(
isCurrent: {
self.canCommitRecognitionStart(
lifecycleGeneration: lifecycleGeneration,
recognitionAttempt: recognitionAttempt)
},
prepare: { enableVoiceProcessing in
try self.prepareStartedRecognitionCapture(
selection: selection,
request: request,
enableVoiceProcessing: false)
} catch {
self.discardAudioEngine()
self.logger.error(
"talk audio engine start failed: \(error.localizedDescription, privacy: .public)")
return
}
}
enableVoiceProcessing: enableVoiceProcessing)
},
discard: { $0.discard() },
publish: { preparedCapture in
self.recognizer = recognizer
self.recognitionRequest = preparedCapture.request
self.audioEngine = preparedCapture.engine
self.activeInputResolution = preparedCapture.activeInputResolution
self.recognitionTask = recognizer.recognitionTask(
with: preparedCapture.request,
resultHandler: { [weak self, recognitionAttempt] result, error in
guard let self else { return }
let segments = result?.bestTranscription.segments ?? []
let transcript = result?.bestTranscription.formattedString
let update = RecognitionUpdate(
transcript: transcript,
hasConfidence: segments.contains { $0.confidence > 0.6 },
isFinal: result?.isFinal ?? false,
errorDescription: error?.localizedDescription,
generation: recognitionAttempt)
Task { await self.handleRecognition(update) }
})
},
onFailure: { enableVoiceProcessing, error in
if enableVoiceProcessing {
self.logger.warning(
"talk processed input start failed; retrying without voice processing: " +
"\(error.localizedDescription, privacy: .public)")
} else {
self.logger.error(
"talk audio engine start failed: \(error.localizedDescription, privacy: .public)")
}
})
guard started else { return false }
self.startRMSTicker(meter: self.rmsMeter)
self.recognitionTask = recognizer.recognitionTask(with: request) { [weak self, generation] result, error in
guard let self else { return }
let segments = result?.bestTranscription.segments ?? []
let transcript = result?.bestTranscription.formattedString
let update = RecognitionUpdate(
transcript: transcript,
hasConfidence: segments.contains { $0.confidence > 0.6 },
isFinal: result?.isFinal ?? false,
errorDescription: error?.localizedDescription,
generation: generation)
Task { await self.handleRecognition(update) }
}
return true
}
private func stopRecognition() async {
func beginRecognitionAttempt(lifecycleGeneration: Int) -> Int? {
guard self.isCurrent(lifecycleGeneration), !self.isPaused else { return nil }
self.recognitionGeneration &+= 1
let recognitionAttempt = self.recognitionGeneration
self.discardRecognitionResources()
return recognitionAttempt
}
func canCommitRecognitionStart(lifecycleGeneration: Int, recognitionAttempt: Int) -> Bool {
self.isCurrent(lifecycleGeneration) && !self.isPaused && self.recognitionGeneration == recognitionAttempt
}
private func stopRecognition() {
self.recognitionGeneration &+= 1
self.discardRecognitionResources()
}
func discardRecognitionResources() {
#if DEBUG
self.recognitionCleanupProbe?()
#endif
self.recognitionTask?.cancel()
self.recognitionTask = nil
self.recognitionRequest?.endAudio()
@@ -368,7 +471,7 @@ actor TalkModeRuntime {
// MARK: - Silence handling
private func startSilenceMonitor() {
func startSilenceMonitor() {
self.silenceTask?.cancel()
self.silenceTask = Task { [weak self] in
await self?.silenceLoop()
@@ -417,7 +520,7 @@ actor TalkModeRuntime {
if sendChime != .none {
await MainActor.run { VoiceWakeChimePlayer.play(sendChime, reason: "talk.send") }
}
await self.stopRecognition()
self.stopRecognition()
await sendAndSpeak(text)
}
@@ -451,49 +554,55 @@ actor TalkModeRuntime {
return selection
}
private func configureAudioEngine(
private func prepareStartedRecognitionCapture(
selection: AudioInputDeviceResolution,
request: SFSpeechAudioBufferRecognitionRequest,
enableVoiceProcessing: Bool) throws
enableVoiceProcessing: Bool)
throws -> PreparedRecognitionCapture
{
let request = SFSpeechAudioBufferRecognitionRequest()
TalkRecognitionCaptureLifecycle.configure(request)
let audioEngine = AVAudioEngine()
self.audioEngine = audioEngine
let input = audioEngine.inputNode
if enableVoiceProcessing {
do {
var tapInstalled = false
do {
if enableVoiceProcessing {
try input.setVoiceProcessingEnabled(true)
} catch {
// Aggregate devices can reject voice processing; capture still works without it.
self.logger.warning(
"talk voice processing unavailable: \(error.localizedDescription, privacy: .public)")
}
}
let activeResolution = self.bindSelectedInputIfNeeded(selection, to: input)
guard activeResolution.resolvedUID != nil else {
throw TalkAudioInputError.unavailable
}
let activeResolution = self.bindSelectedInputIfNeeded(selection, to: input)
guard activeResolution.resolvedUID != nil else {
throw TalkAudioInputError.unavailable
}
let format = input.outputFormat(forBus: 0)
guard format.channelCount > 0, format.sampleRate > 0 else {
throw TalkAudioInputError.invalidFormat
let format = input.outputFormat(forBus: 0)
guard format.channelCount > 0, format.sampleRate > 0 else {
throw TalkAudioInputError.invalidFormat
}
input.removeTap(onBus: 0)
let meter = self.rmsMeter
input.installTap(
onBus: 0,
bufferSize: 2048,
format: format)
{ [weak request, meter] buffer, _ in
request?.append(SpeechAudioBufferNormalizer.speechCompatibleBuffer(from: buffer))
meter.set(TalkAudioLevel.rms(buffer: buffer))
}
tapInstalled = true
audioEngine.prepare()
try audioEngine.start()
return PreparedRecognitionCapture(
request: request,
engine: audioEngine,
activeInputResolution: activeResolution)
} catch {
request.endAudio()
if tapInstalled {
input.removeTap(onBus: 0)
}
audioEngine.stop()
throw error
}
input.removeTap(onBus: 0)
let meter = self.rmsMeter
input.installTap(onBus: 0, bufferSize: 2048, format: format) { [weak request, meter] buffer, _ in
request?.append(SpeechAudioBufferNormalizer.speechCompatibleBuffer(from: buffer))
meter.set(TalkAudioLevel.rms(buffer: buffer))
}
audioEngine.prepare()
try audioEngine.start()
self.activeInputResolution = activeResolution
}
private func discardAudioEngine() {
self.audioEngine?.inputNode.removeTap(onBus: 0)
self.audioEngine?.stop()
self.audioEngine = nil
self.activeInputResolution = nil
}
private func defaultFallback(for selection: AudioInputDeviceResolution) -> AudioInputDeviceResolution {
@@ -504,18 +613,6 @@ actor TalkModeRuntime {
}
}
private enum TalkAudioInputError: LocalizedError {
case unavailable
case invalidFormat
var errorDescription: String? {
switch self {
case .unavailable: "Selected input and system default are unavailable"
case .invalidFormat: "Selected audio input has no usable format"
}
}
}
// MARK: - Gateway + TTS
extension TalkModeRuntime {
@@ -583,8 +680,9 @@ extension TalkModeRuntime {
guard let assistantText
else {
self.logger.warning("talk assistant text missing after timeout")
guard await self.startRecognition(lifecycleGeneration: gen),
self.isCurrent(gen), !self.isPaused else { return }
await self.startListening()
await self.startRecognition()
return
}
guard self.isCurrent(gen) else { return }
@@ -611,8 +709,10 @@ extension TalkModeRuntime {
}
return
}
let lifecycleGeneration = self.lifecycleGeneration
guard await self.startRecognition(lifecycleGeneration: lifecycleGeneration),
self.isCurrent(lifecycleGeneration), !self.isPaused else { return }
await self.startListening()
await self.startRecognition()
}
private func buildPrompt(transcript: String) -> String {
@@ -1113,8 +1213,8 @@ extension TalkModeRuntime {
}
private func prepareForPlayback(generation: Int) async -> Bool {
await self.startRecognition()
return self.isCurrent(generation)
guard await self.startRecognition(lifecycleGeneration: generation) else { return false }
return self.isCurrent(generation) && !self.isPaused
}
private func resolveVoiceId(preferred: String?, apiKey: String) async -> String? {
@@ -1170,13 +1270,60 @@ extension TalkModeRuntime {
return value.allSatisfy { $0.isLetter || $0.isNumber || $0 == "-" || $0 == "_" }
}
func stopSpeaking(reason: TalkStopReason) async {
func stopSpeaking(
reason: TalkStopReason,
reconfigurationGeneration expectedReconfigurationGeneration: UInt64? = nil,
lifecycleGeneration expectedLifecycleGeneration: Int? = nil) async
{
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
if let realtimeSession {
let relayReason = switch reason {
case .userTap: "user"
case .speech: "barge-in"
case .manual: "shutdown"
}
let cancelled = await MainActor.run {
realtimeSession.cancelOutput(reason: relayReason)
}
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
if cancelled, reason != .manual, !self.isPaused {
self.phase = .listening
await MainActor.run { TalkModeController.shared.updatePhase(.listening) }
}
return
}
let usePCM = self.lastPlaybackWasPCM
let remoteInterruptedAt = usePCM ? await stopPCM() : await stopMP3()
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
_ = usePCM ? await stopMP3() : await stopPCM()
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
let localInterruptedAt = await stopTalkAudio()
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
await TalkSystemSpeechSynthesizer.shared.stop()
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
await stopMLXVoice()
guard self.ownsReconfiguration(
expectedReconfigurationGeneration,
lifecycleGeneration: expectedLifecycleGeneration)
else { return }
guard self.phase == .speaking else { return }
let interruptedAt = remoteInterruptedAt ?? localInterruptedAt
if reason == .speech, let interruptedAt {
@@ -1308,10 +1455,72 @@ extension TalkModeRuntime {
await TalkMLXSpeechSynthesizer.shared.cancelCurrent()
}
func parseTalkConfig(_ snap: ConfigSnapshot) -> TalkModeGatewayConfigState {
let env = ProcessInfo.processInfo.environment
let envVoice = env["ELEVENLABS_VOICE_ID"]?.trimmingCharacters(in: .whitespacesAndNewlines)
let sagVoice = env["SAG_VOICE_ID"]?.trimmingCharacters(in: .whitespacesAndNewlines)
let envApiKey = env["ELEVENLABS_API_KEY"]?.trimmingCharacters(in: .whitespacesAndNewlines)
let parsed = TalkModeGatewayConfigParser.parse(
snapshot: snap,
defaultProvider: Self.defaultTalkProvider,
defaultModelIdFallback: Self.defaultModelIdFallback,
defaultSilenceTimeoutMs: Self.defaultSilenceTimeoutMs,
envVoice: envVoice,
sagVoice: sagVoice,
envApiKey: envApiKey)
if parsed.missingResolvedPayload {
self.ttsLogger.info("talk config ignored: normalized payload missing talk.resolved")
}
if parsed.activeProvider == Self.defaultTalkProvider {
self.ttsLogger.info("talk config provider from talk.resolved")
} else if parsed.activeProvider == Self.mlxTalkProvider ||
parsed.activeProvider == Self.systemTalkProvider
{
self.ttsLogger.info(
"talk provider \(parsed.activeProvider, privacy: .public) active")
} else {
self.ttsLogger
.info(
"""
talk provider \(parsed.activeProvider, privacy: .public) uses gateway talk.speak \
with system voice fallback
""")
}
return parsed
}
func fetchTalkConfig() async -> TalkModeGatewayConfigState {
do {
let snap: ConfigSnapshot = try await GatewayConnection.shared.requestDecoded(
method: .talkConfig,
params: ["includeSecrets": AnyCodable(true)],
timeoutMs: 8000)
return self.parseTalkConfig(snap)
} catch {
return self.fallbackTalkConfig()
}
}
// MARK: - Config
private func reloadConfig() async {
func reloadConfig() async {
let cfg = await fetchTalkConfig()
await self.applyTalkConfig(cfg)
self.macOSRealtimeRelayOptIn = await MainActor.run {
AppStateStore.shared.talkRealtimeRelayEnabled
}
}
func applyTalkConfig(_ cfg: TalkModeGatewayConfigState) async {
let locale = await MainActor.run {
AppStateStore.shared.seamColorHex = cfg.seamColorHex
return AppStateStore.shared.voiceWakeLocaleID
}
self.commitTalkConfig(cfg, locale: locale)
}
func commitTalkConfig(_ cfg: TalkModeGatewayConfigState, locale: String) {
self.defaultVoiceId = cfg.voiceId
self.voiceAliases = cfg.voiceAliases
if !self.voiceOverrideActive {
@@ -1324,8 +1533,14 @@ extension TalkModeRuntime {
self.defaultOutputFormat = cfg.outputFormat
self.interruptOnSpeech = cfg.interruptOnSpeech
self.activeTalkProvider = cfg.activeProvider
self.realtimeProvider = cfg.realtimeProvider
self.realtimeModelId = cfg.realtimeModelId
self.realtimeSpeakerVoice = cfg.realtimeSpeakerVoice
self.realtimeMode = cfg.realtimeMode
self.realtimeTransport = cfg.realtimeTransport
self.realtimeBrain = cfg.realtimeBrain
self.hasGatewayRealtimeRelayTuple = cfg.hasGatewayRealtimeRelayTuple
let configuredSilenceMs = cfg.silenceTimeoutMs
let locale = await MainActor.run { AppStateStore.shared.voiceWakeLocaleID }
let isCJKLocale = locale.hasPrefix("ko") || locale.hasPrefix("ja") || locale.hasPrefix("zh")
let effectiveSilenceMs = isCJKLocale ? max(configuredSilenceMs, 2000) : configuredSilenceMs
if isCJKLocale, configuredSilenceMs < 2000 {
@@ -1351,7 +1566,11 @@ extension TalkModeRuntime {
"apiKey=\(hasApiKey, privacy: .public) " +
"interrupt=\(cfg.interruptOnSpeech, privacy: .public) " +
"silenceTimeoutMs=\(cfg.silenceTimeoutMs, privacy: .public) " +
"speechLocale=\(cfg.speechLocaleID ?? "device", privacy: .public)")
"speechLocale=\(cfg.speechLocaleID ?? "device", privacy: .public) " +
"realtimeMode=\(cfg.realtimeMode ?? "off", privacy: .public) " +
"realtimeTransport=\(cfg.realtimeTransport ?? "default", privacy: .public) " +
"realtimeBrain=\(cfg.realtimeBrain ?? "default", privacy: .public) " +
"macOSRealtimeOptIn=\(self.macOSRealtimeRelayOptIn, privacy: .public)")
}
static func selectTalkProviderConfig(
@@ -1364,57 +1583,6 @@ extension TalkModeRuntime {
TalkConfigParsing.resolvedSilenceTimeoutMs(talk, fallback: self.defaultSilenceTimeoutMs)
}
private func fetchTalkConfig() async -> TalkModeGatewayConfigState {
let env = ProcessInfo.processInfo.environment
let envVoice = env["ELEVENLABS_VOICE_ID"]?.trimmingCharacters(in: .whitespacesAndNewlines)
let sagVoice = env["SAG_VOICE_ID"]?.trimmingCharacters(in: .whitespacesAndNewlines)
let envApiKey = env["ELEVENLABS_API_KEY"]?.trimmingCharacters(in: .whitespacesAndNewlines)
do {
let snap: ConfigSnapshot = try await GatewayConnection.shared.requestDecoded(
method: .talkConfig,
params: ["includeSecrets": AnyCodable(true)],
timeoutMs: 8000)
let parsed = TalkModeGatewayConfigParser.parse(
snapshot: snap,
defaultProvider: Self.defaultTalkProvider,
defaultModelIdFallback: Self.defaultModelIdFallback,
defaultSilenceTimeoutMs: Self.defaultSilenceTimeoutMs,
envVoice: envVoice,
sagVoice: sagVoice,
envApiKey: envApiKey)
if parsed.missingResolvedPayload {
self.ttsLogger.info("talk config ignored: normalized payload missing talk.resolved")
}
await MainActor.run {
AppStateStore.shared.seamColorHex = parsed.seamColorHex
}
if parsed.activeProvider == Self.defaultTalkProvider {
self.ttsLogger.info("talk config provider from talk.resolved")
} else if parsed.activeProvider == Self.mlxTalkProvider ||
parsed.activeProvider == Self.systemTalkProvider
{
self.ttsLogger.info(
"talk provider \(parsed.activeProvider, privacy: .public) active")
} else {
self.ttsLogger
.info(
"""
talk provider \(parsed.activeProvider, privacy: .public) uses gateway talk.speak \
with system voice fallback
""")
}
return parsed
} catch {
return TalkModeGatewayConfigParser.fallback(
defaultModelIdFallback: Self.defaultModelIdFallback,
defaultSilenceTimeoutMs: Self.defaultSilenceTimeoutMs,
envVoice: envVoice,
sagVoice: sagVoice,
envApiKey: envApiKey)
}
}
// MARK: - Audio level handling
private func noteAudioLevel(rms: Double) async {
@@ -0,0 +1,57 @@
@preconcurrency import AVFoundation
import Foundation
import Speech
struct PreparedRecognitionCapture {
let request: SFSpeechAudioBufferRecognitionRequest
let engine: AVAudioEngine
let activeInputResolution: AudioInputDeviceResolution
func discard() {
self.request.endAudio()
self.engine.inputNode.removeTap(onBus: 0)
self.engine.stop()
}
}
enum TalkAudioInputError: LocalizedError {
case unavailable
case invalidFormat
var errorDescription: String? {
switch self {
case .unavailable: "Selected input and system default are unavailable"
case .invalidFormat: "Selected audio input has no usable format"
}
}
}
enum TalkRecognitionCaptureLifecycle {
static func configure(_ request: SFSpeechAudioBufferRecognitionRequest) {
SpeechRecognitionRequestPolicy.configureInteractiveTranscription(request)
}
static func start<Capture>(
isCurrent: () -> Bool,
prepare: (_ enableVoiceProcessing: Bool) throws -> Capture,
discard: (Capture) -> Void,
publish: (Capture) -> Void,
onFailure: (_ enableVoiceProcessing: Bool, _ error: Error) -> Void) -> Bool
{
for enableVoiceProcessing in [true, false] {
guard isCurrent() else { return false }
do {
let capture = try prepare(enableVoiceProcessing)
guard isCurrent() else {
discard(capture)
return false
}
publish(capture)
return true
} catch {
onFailure(enableVoiceProcessing, error)
}
}
return false
}
}
@@ -182,6 +182,8 @@ struct VoiceWakeSettings: View {
subtitle: "Start listening while you hold the key and show the preview overlay.",
binding: self.$state.voicePushToTalkEnabled)
self.realtimeRelayToggle
if self.state.voicePushToTalkEnabled, self.state.talkEnabled {
SettingsCardRow(
title: "Push-to-talk paused",
@@ -930,6 +932,18 @@ private struct TriggerPhraseHelpRow: View {
}
}
extension VoiceWakeSettings {
private var realtimeRelayToggle: some View {
SettingsCardToggleRow(
title: "Use realtime Gateway relay",
subtitle: """
Use the Gateway's configured realtime voice session on this Mac. \
Requires realtime, gateway-relay, and agent-consult in Talk settings.
""",
binding: self.$state.talkRealtimeRelayEnabled)
}
}
#if DEBUG
struct VoiceWakeSettings_Previews: PreviewProvider {
static var previews: some View {
@@ -21,13 +21,11 @@ private func makeGatewayGenerationSnapshot(version: String) -> HelloOk {
statedir: nil,
sessiondefaults: nil,
authmode: nil,
updateavailable: nil
),
updateavailable: nil),
controluitabs: nil,
pluginsurfaceurls: nil,
auth: [:],
policy: [:]
)
policy: [:])
}
private func gatewayGenerationSnapshotVersion(_ push: GatewayPush?) -> String? {
@@ -40,15 +38,15 @@ private final class WebSocketMessageRecorder: @unchecked Sendable {
private var messages: [URLSessionWebSocketTask.Message] = []
func append(_ message: URLSessionWebSocketTask.Message) {
lock.lock()
self.lock.lock()
defer { self.lock.unlock() }
messages.append(message)
self.messages.append(message)
}
func snapshot() -> [URLSessionWebSocketTask.Message] {
lock.lock()
self.lock.lock()
defer { self.lock.unlock() }
return messages
return self.messages
}
}
@@ -67,24 +65,24 @@ final class GatewayConnectionEndpointSource: @unchecked Sendable {
}
func setEndpoint(_ endpoint: GatewayConnection.EndpointSnapshot) {
lock.lock()
self.lock.lock()
self.endpoint = endpoint
lock.unlock()
self.lock.unlock()
}
func snapshot() -> GatewayConnection.EndpointSnapshot {
lock.lock()
self.lock.lock()
defer { self.lock.unlock() }
return endpoint
return self.endpoint
}
func setURL(_ url: URL) {
lock.lock()
self.lock.lock()
let config = self.endpoint.config
self.endpoint = GatewayConnection.EndpointSnapshot(
config: (url: url, token: config.token, password: config.password),
routeAuthority: nil)
lock.unlock()
self.lock.unlock()
}
}
@@ -95,10 +93,10 @@ private actor GatewayConnectionSuspensionGate {
private var releaseWaiters: [CheckedContinuation<Void, Never>] = []
func suspend() async {
didStart = true
startWaiters.forEach { $0.resume() }
startWaiters.removeAll()
if !isOpen {
self.didStart = true
self.startWaiters.forEach { $0.resume() }
self.startWaiters.removeAll()
if !self.isOpen {
await withCheckedContinuation { continuation in
self.releaseWaiters.append(continuation)
}
@@ -106,16 +104,16 @@ private actor GatewayConnectionSuspensionGate {
}
func waitUntilStarted() async {
guard !didStart else { return }
guard !self.didStart else { return }
await withCheckedContinuation { continuation in
self.startWaiters.append(continuation)
}
}
func open() {
isOpen = true
releaseWaiters.forEach { $0.resume() }
releaseWaiters.removeAll()
self.isOpen = true
self.releaseWaiters.forEach { $0.resume() }
self.releaseWaiters.removeAll()
}
}
@@ -129,14 +127,13 @@ private func makeTestGatewayConnection() -> (GatewayConnection, GatewayTestWebSo
configProvider: {
(url: URL(string: "ws://127.0.0.1:1")!, token: nil, password: nil)
},
sessionBox: WebSocketSessionBox(session: session)
)
sessionBox: WebSocketSessionBox(session: session))
return (connection, session)
}
private func makeRecordingGatewayConnection(
responseData: @escaping @Sendable (String) -> Data
) -> (GatewayConnection, WebSocketMessageRecorder) {
responseData: @escaping @Sendable (String) -> Data) -> (GatewayConnection, WebSocketMessageRecorder)
{
let recorder = WebSocketMessageRecorder()
let session = GatewayTestWebSocketSession(taskFactory: {
GatewayTestWebSocketTask(sendHook: { task, message, sendIndex in
@@ -151,16 +148,15 @@ private func makeRecordingGatewayConnection(
configProvider: {
(url: URL(string: "ws://127.0.0.1:1")!, token: nil, password: nil)
},
sessionBox: WebSocketSessionBox(session: session)
)
sessionBox: WebSocketSessionBox(session: session))
return (connection, recorder)
}
private func makeRouteLifecycleConnection(
url: URL,
token: String? = nil,
password: String? = nil
) -> (GatewayConnectionEndpointSource, GatewayConnectionSuspensionGate, GatewayConnection) {
password: String? = nil) -> (GatewayConnectionEndpointSource, GatewayConnectionSuspensionGate, GatewayConnection)
{
let source = GatewayConnectionEndpointSource(url: url, token: token, password: password)
let gate = GatewayConnectionSuspensionGate()
let connection = GatewayConnection(
@@ -169,8 +165,7 @@ private func makeRouteLifecycleConnection(
clientShutdown: { client in
await gate.suspend()
await client.shutdown()
}
)
})
return (source, gate, connection)
}
@@ -182,8 +177,8 @@ private enum SuspendedConfigOperation {
private func assertConfigLookupCannotRecreateRoute(
url: URL,
operation: SuspendedConfigOperation
) async {
operation: SuspendedConfigOperation) async
{
let gate = GatewayConnectionSuspensionGate()
let config: GatewayConnection.Config = (url: url, token: nil, password: nil)
let connection = GatewayConnection(
@@ -191,8 +186,7 @@ private func assertConfigLookupCannotRecreateRoute(
await gate.suspend()
return config
},
sessionBox: WebSocketSessionBox(session: GatewayTestWebSocketSession())
)
sessionBox: WebSocketSessionBox(session: GatewayTestWebSocketSession()))
let operationTask = Task {
switch operation {
case .request:
@@ -330,6 +324,251 @@ private func assertConfigLookupCannotRecreateRoute(
}
}
@Test func `realtime talk transport pins requests to its server lease`() async throws {
let recorder = WebSocketMessageRecorder()
let session = GatewayTestWebSocketSession(taskFactory: {
GatewayTestWebSocketTask(
sendHook: { task, message, sendIndex in
recorder.append(message)
guard sendIndex > 0,
let data = Self.messageData(message),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
let id = frame["id"] as? String
else { return }
task.emitReceiveSuccess(.data(GatewayWebSocketTestSupport.okResponseData(id: id)))
},
receiveHook: { task, receiveIndex in
if receiveIndex == 0 {
return .data(GatewayWebSocketTestSupport.connectChallengeData())
}
let id = task.snapshotConnectRequestID() ?? "connect"
return .data(GatewayWebSocketTestSupport.connectOkData(id: id))
})
})
let connection = GatewayConnection(
configProvider: {
(
url: URL(string: "wss://gateway.example.invalid:9443")!,
token: "test-token-placeholder",
password: nil)
},
sessionBox: WebSocketSessionBox(session: session))
try await connection.refresh()
let transport = try await connection.acquireRealtimeTalkTransport()
#expect(await transport.isCurrent())
let events = await transport.subscribeServerEvents(10)
let nextEvent = Task {
var iterator = events.makeAsyncIterator()
return await iterator.next()
}
_ = try await transport.request(
"talk.session.close",
["sessionId": AnyCodable("talk-session-1")],
4321)
let talkRequest = try #require(recorder.snapshot().first { message in
guard let data = Self.messageData(message),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any]
else { return false }
return frame["method"] as? String == "talk.session.close"
})
let talkRequestData = try #require(Self.messageData(talkRequest))
let talkRequestFrame = try #require(
JSONSerialization.jsonObject(with: talkRequestData) as? [String: Any])
let params = try #require(talkRequestFrame["params"] as? [String: Any])
#expect(params["sessionId"] as? String == "talk-session-1")
let socketGeneration = try #require(await connection._test_activeSocketGeneration())
await connection._test_handleDisconnect(socketGeneration: socketGeneration)
let eventAfterDisconnect = try await AsyncTimeout.withTimeout(
seconds: 1,
onTimeout: { CancellationError() },
operation: { await nextEvent.value })
#expect(eventAfterDisconnect == nil)
await connection.shutdown()
try await connection.refresh()
let successor = try await connection.acquireRealtimeTalkTransport()
#expect(await !(transport.isCurrent()))
#expect(await successor.isCurrent())
await #expect(throws: (any Error).self) {
_ = try await transport.request("talk.session.close", nil, 4321)
}
_ = try await successor.request("talk.session.close", nil, 4321)
#expect(await successor.isCurrent())
await connection.shutdown()
}
@Test func `realtime bootstrap never moves config onto a replacement route`() async throws {
let recorder = WebSocketMessageRecorder()
let configRequestEntered = AsyncStream<Void>.makeStream(bufferingPolicy: .bufferingNewest(1))
let releaseConfig = AsyncTestGate()
let source = try GatewayConnectionEndpointSource(
url: #require(URL(string: "wss://route-a.example.invalid:9443")))
let session = GatewayTestWebSocketSession(taskFactory: {
GatewayTestWebSocketTask(
sendHook: { task, message, sendIndex in
recorder.append(message)
guard sendIndex > 0,
let data = Self.messageData(message),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
let id = frame["id"] as? String
else { return }
let method = frame["method"] as? String
let configRequestCount = recorder.snapshot().filter { recorded in
guard let data = Self.messageData(recorded),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any]
else { return false }
return frame["method"] as? String == "talk.config"
}.count
if method == "talk.config", configRequestCount == 1 {
configRequestEntered.continuation.yield()
configRequestEntered.continuation.finish()
await releaseConfig.wait()
}
let sessionKey = configRequestCount > 1 ? "route-b" : "route-a"
let response = Data(
"""
{"type":"res","id":"\(id)","ok":true,"payload":{"config":{
"session":{"mainKey":"\(sessionKey)"},
"talk":{"realtime":{"provider":"openai","mode":"realtime",
"transport":"gateway-relay","brain":"agent-consult"}}
}}}
""".utf8)
task.emitReceiveSuccess(.data(response))
},
receiveHook: { task, receiveIndex in
if receiveIndex == 0 {
return .data(GatewayWebSocketTestSupport.connectChallengeData())
}
let id = task.snapshotConnectRequestID() ?? "connect"
return .data(GatewayWebSocketTestSupport.connectOkData(id: id))
})
})
let connection = GatewayConnection(
testEndpointProvider: { source.snapshot() },
sessionBox: WebSocketSessionBox(session: session))
let entryTask = Task {
var iterator = configRequestEntered.stream.makeAsyncIterator()
return await iterator.next()
}
let staleBootstrap = Task {
try await connection.acquireRealtimeTalkBootstrap()
}
do {
_ = try await AsyncTimeout.withTimeout(
seconds: 1,
onTimeout: { CancellationError() },
operation: { await entryTask.value })
try source.setURL(#require(URL(string: "wss://route-b.example.invalid:9443")))
try await connection.refresh()
releaseConfig.open()
await #expect(throws: (any Error).self) {
_ = try await staleBootstrap.value
}
let preReplacementConfigCount = recorder.snapshot().filter { recorded in
guard let data = Self.messageData(recorded),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any]
else { return false }
return frame["method"] as? String == "talk.config"
}.count
#expect(preReplacementConfigCount == 1)
let replacement = try await connection.acquireRealtimeTalkBootstrap()
#expect(replacement.sessionKey == "route-b")
#expect(await replacement.transport.isCurrent())
let configRequests = recorder.snapshot().compactMap { message -> [String: Any]? in
guard let data = Self.messageData(message),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
frame["method"] as? String == "talk.config"
else { return nil }
return frame
}
#expect(configRequests.count == 2)
for request in configRequests {
let params = try #require(request["params"] as? [String: Any])
#expect(params["includeSecrets"] == nil)
}
} catch {
releaseConfig.open()
entryTask.cancel()
staleBootstrap.cancel()
_ = await entryTask.value
_ = try? await staleBootstrap.value
await connection.shutdown()
throw error
}
releaseConfig.open()
entryTask.cancel()
staleBootstrap.cancel()
_ = await entryTask.value
_ = try? await staleBootstrap.value
await connection.shutdown()
}
@Test func `realtime talk event overflow terminates its bounded subscription`() async throws {
let session = GatewayTestWebSocketSession(taskFactory: {
GatewayTestWebSocketTask(
sendHook: { task, message, sendIndex in
guard sendIndex > 0,
let data = Self.messageData(message),
let frame = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
let id = frame["id"] as? String
else { return }
task.emitReceiveSuccess(.data(GatewayWebSocketTestSupport.okResponseData(id: id)))
},
receiveHook: { task, receiveIndex in
if receiveIndex == 0 {
return .data(GatewayWebSocketTestSupport.connectChallengeData())
}
let id = task.snapshotConnectRequestID() ?? "connect"
return .data(GatewayWebSocketTestSupport.connectOkData(id: id))
})
})
let connection = GatewayConnection(
configProvider: {
(
url: URL(string: "wss://gateway.example.invalid:9443")!,
token: "test-token-placeholder",
password: nil)
},
sessionBox: WebSocketSessionBox(session: session))
try await connection.refresh()
let transport = try await connection.acquireRealtimeTalkTransport()
let events = await transport.subscribeServerEvents(1)
let socketGeneration = try #require(await connection._test_activeSocketGeneration())
for seq in 1...20 {
await connection._test_handlePush(
.event(EventFrame(
type: "event",
event: "talk.event",
payload: AnyCodable(["seq": seq]),
seq: seq,
stateversion: nil)),
socketGeneration: socketGeneration)
}
let terminalRead = Task {
var iterator = events.makeAsyncIterator()
var received: [EventFrame] = []
while let event = await iterator.next() {
received.append(event)
}
return received
}
let received = try await AsyncTimeout.withTimeout(
seconds: 1,
onTimeout: { CancellationError() },
operation: { await terminalRead.value })
#expect(!received.isEmpty)
#expect(received.count <= 2)
await connection.shutdown()
}
@Test func `operator widget capability refresh is shared and retained`() async throws {
let rawOldSurface = "http://127.0.0.1:18789/__openclaw__/cap/old-token"
let rawNewSurface = "http://127.0.0.1:18789/__openclaw__/cap/new-token"
@@ -407,14 +646,12 @@ private func assertConfigLookupCannotRecreateRoute(
method: "wizard.cancel",
code: "INVALID_REQUEST",
message: "wizard not found",
details: nil
)
details: nil)
let locked = GatewayResponseError(
method: "wizard.cancel",
code: "INVALID_REQUEST",
message: "wizard cancellation is locked",
details: nil
)
details: nil)
#expect(GatewayConnection.wizardCancellationOutcome(after: notFound) == .absent)
#expect(GatewayConnection.wizardCancellationOutcome(after: locked) == .unresolved)
@@ -465,19 +702,16 @@ private func assertConfigLookupCannotRecreateRoute(
connectionMode: .remote,
remoteTransport: .direct,
remoteURL: urlA.absoluteString,
remoteTarget: ""
))
remoteTarget: ""))
let ownerB = try #require(GatewayDiscoveryPreferences.deviceAuthGatewayID(
connectionMode: .remote,
remoteTransport: .direct,
remoteURL: urlB.absoluteString,
remoteTarget: ""
))
remoteTarget: ""))
try await assertDeviceTokenIsolation(
try await self.assertDeviceTokenIsolation(
routeA: (urlA, ownerA),
routeB: (urlB, ownerB)
)
routeB: (urlB, ownerB))
}
@Test func `SSH endpoint never receives another route device token`() async throws {
@@ -486,19 +720,16 @@ private func assertConfigLookupCannotRecreateRoute(
connectionMode: .remote,
remoteTransport: .ssh,
remoteURL: "",
remoteTarget: "operator@gateway-a.example"
))
remoteTarget: "operator@gateway-a.example"))
let ownerB = try #require(GatewayDiscoveryPreferences.deviceAuthGatewayID(
connectionMode: .remote,
remoteTransport: .ssh,
remoteURL: "",
remoteTarget: "operator@gateway-b.example"
))
remoteTarget: "operator@gateway-b.example"))
try await assertDeviceTokenIsolation(
try await self.assertDeviceTokenIsolation(
routeA: (tunnelURL, ownerA),
routeB: (tunnelURL, ownerB)
)
routeB: (tunnelURL, ownerB))
}
@Test func `retired socket callbacks cannot mutate cache or subscribers`() async {
@@ -510,29 +741,24 @@ private func assertConfigLookupCannotRecreateRoute(
await connection._test_handlePush(
.snapshot(makeGatewayGenerationSnapshot(version: "socket-1")),
routeGeneration: routeGeneration,
socketGeneration: 1
)
socketGeneration: 1)
await connection._test_handleDisconnect(
routeGeneration: routeGeneration,
socketGeneration: 1
)
socketGeneration: 1)
#expect(await connection.cachedGatewayVersion() == nil)
await connection._test_handlePush(
.snapshot(makeGatewayGenerationSnapshot(version: "stale-socket-1")),
routeGeneration: routeGeneration,
socketGeneration: 1
)
socketGeneration: 1)
await connection._test_handlePush(
.snapshot(makeGatewayGenerationSnapshot(version: "socket-2")),
routeGeneration: routeGeneration,
socketGeneration: 2
)
socketGeneration: 2)
await connection._test_handlePush(
.snapshot(makeGatewayGenerationSnapshot(version: "late-socket-1")),
routeGeneration: routeGeneration,
socketGeneration: 1
)
socketGeneration: 1)
let firstPush = await iterator.next()
let secondPush = await iterator.next()
@@ -553,13 +779,11 @@ private func assertConfigLookupCannotRecreateRoute(
await connection._test_handlePush(
.snapshot(makeGatewayGenerationSnapshot(version: "replaced-route")),
routeGeneration: replacedRouteGeneration,
socketGeneration: 1
)
socketGeneration: 1)
await connection._test_handlePush(
.snapshot(makeGatewayGenerationSnapshot(version: "current-route")),
routeGeneration: currentRouteGeneration,
socketGeneration: 1
)
socketGeneration: 1)
let push = await iterator.next()
#expect(gatewayGenerationSnapshotVersion(push) == "current-route")
@@ -710,8 +934,7 @@ private func assertConfigLookupCannotRecreateRoute(
channel: .last,
timeoutSeconds: nil,
idempotencyKey: "idem-1",
voiceWakeTrigger: " "
))
voiceWakeTrigger: " "))
await connection.shutdown()
#expect(result.ok == true)
@@ -747,8 +970,7 @@ private func assertConfigLookupCannotRecreateRoute(
message: "hello",
thinking: nil,
idempotencyKey: "chat-1",
attachments: []
)
attachments: [])
await connection.shutdown()
guard let chatMessage = recorder.snapshot().reversed().first(where: { message in
@@ -784,25 +1006,21 @@ private func assertConfigLookupCannotRecreateRoute(
@Test(arguments: [
(
#"{"defaultId":"main","mainKey":"main","scope":"per-sender","agents":[{"id":"main","model":{"primary":"openai/gpt-5.5"}}]}"#,
"openai/gpt-5.5"
),
"openai/gpt-5.5"),
(
#"{"defaultId":"work","mainKey":"main","scope":"per-sender","agents":[{"id":"main","model":{"primary":"openai/gpt-5.5"}},{"id":"work","model":{"primary":"anthropic/claude-opus-4-8"}}]}"#,
"anthropic/claude-opus-4-8"
),
"anthropic/claude-opus-4-8"),
(
#"{"defaultId":"main","mainKey":"main","scope":"per-sender","agents":[{"id":"main"},{"id":"work","model":{"primary":"openai/gpt-5.5"}}]}"#,
nil
),
nil),
(
#"{"defaultId":"main","mainKey":"main","scope":"per-sender","agents":[{"id":"main","model":{"primary":" "}}]}"#,
nil
),
nil),
])
func `configured inference model follows the default agent`(
json: String,
expected: String?
) throws {
expected: String?) throws
{
#expect(try GatewayConnection.decodeConfiguredInferenceModel(Data(json.utf8)) == expected)
}
@@ -924,8 +1142,8 @@ private func assertConfigLookupCannotRecreateRoute(
private func assertDeviceTokenIsolation(
routeA: (url: URL, owner: String),
routeB: (url: URL, owner: String)
) async throws {
routeB: (url: URL, owner: String)) async throws
{
#expect(routeA.owner != routeB.owner)
let tempDir = FileManager.default.temporaryDirectory
.appendingPathComponent(UUID().uuidString, isDirectory: true)
@@ -938,8 +1156,7 @@ private func assertConfigLookupCannotRecreateRoute(
let routeAAuth = try await self.connectAuth(
route: routeA,
storedDeviceToken: routeAToken,
unscopedToken: unscopedToken
)
unscopedToken: unscopedToken)
#expect(routeAAuth?["token"] as? String == routeAToken)
#expect(routeAAuth?["token"] as? String != unscopedToken)
@@ -952,8 +1169,8 @@ private func assertConfigLookupCannotRecreateRoute(
private func connectAuth(
route: (url: URL, owner: String),
storedDeviceToken: String? = nil,
unscopedToken: String? = nil
) async throws -> [String: Any]? {
unscopedToken: String? = nil) async throws -> [String: Any]?
{
let recorder = WebSocketMessageRecorder()
let session = GatewayTestWebSocketSession(taskFactory: {
GatewayTestWebSocketTask(sendHook: { task, message, sendIndex in
@@ -971,35 +1188,29 @@ private func assertConfigLookupCannotRecreateRoute(
guard DeviceAuthStore.storeTokenPersisted(
deviceId: identity.deviceId,
role: "operator",
token: unscopedToken
),
token: unscopedToken),
DeviceAuthStore.storeTokenPersisted(
deviceId: identity.deviceId,
role: "operator",
token: storedDeviceToken,
gatewayID: route.owner
)
gatewayID: route.owner)
else {
throw NSError(
domain: "GatewayConnectionControlTests",
code: 1,
userInfo: [NSLocalizedDescriptionKey: "failed to persist device auth fixture"]
)
userInfo: [NSLocalizedDescriptionKey: "failed to persist device auth fixture"])
}
}
return GatewayConnection.EndpointSnapshot(
config: (url: route.url, token: nil, password: nil),
routeAuthority: nil,
deviceAuthGatewayID: route.owner
)
deviceAuthGatewayID: route.owner)
},
sessionBox: WebSocketSessionBox(session: session)
)
sessionBox: WebSocketSessionBox(session: session))
_ = try await connection.request(
method: "health",
params: nil,
retryTransportFailures: false
)
retryTransportFailures: false)
await connection.shutdown()
for message in recorder.snapshot() {
@@ -0,0 +1,389 @@
@preconcurrency import AVFoundation
import Foundation
import OpenClawKit
import Testing
@testable import OpenClaw
private final class SendableTapHandler: @unchecked Sendable {
private let handler: AVAudioNodeTapBlock
init(_ handler: @escaping AVAudioNodeTapBlock) {
self.handler = handler
}
func callAsFunction(_ buffer: AVAudioPCMBuffer, _ time: AVAudioTime) {
self.handler(buffer, time)
}
}
struct MacRealtimeTalkAudioCaptureTests {
@Test func `encoder downmixes resamples and emits little endian pcm16`() throws {
let buffer = try makeFloatBuffer(
sampleRate: 48000,
channels: [
[0, 1, -1, 0.5],
[0, 1, -1, -0.5],
])
let frame = try #require(MacRealtimeTalkAudioFrameEncoder.encode(
buffer: buffer,
targetSampleRate: 24000,
timestampMs: 1234))
#expect(frame.timestampMs == 1234)
#expect(frame.data.count == 4)
#expect(self.samples(in: frame.data) == [0, -32767])
#expect(abs(frame.rms - Float(1.0 / 2.0.squareRoot())) < 0.0001)
}
@Test func `encoder interpolates and clamps samples`() throws {
let buffer = try makeFloatBuffer(
sampleRate: 24000,
channels: [[2, 0, -2]])
let frame = try #require(MacRealtimeTalkAudioFrameEncoder.encode(
buffer: buffer,
targetSampleRate: 48000,
timestampMs: 0))
#expect(self.samples(in: frame.data) == [32767, 32767, 0, -32767, -32767, -32767])
}
@Test func `encoder rejects empty and invalid target buffers`() throws {
let format = try #require(AVAudioFormat(
commonFormat: .pcmFormatFloat32,
sampleRate: 48000,
channels: 1,
interleaved: false))
let empty = try #require(AVAudioPCMBuffer(pcmFormat: format, frameCapacity: 1))
#expect(MacRealtimeTalkAudioFrameEncoder.encode(
buffer: empty,
targetSampleRate: 24000,
timestampMs: 0) == nil)
#expect(MacRealtimeTalkAudioFrameEncoder.encode(
buffer: empty,
targetSampleRate: 0,
timestampMs: 0) == nil)
}
@Test func `delivery gate invalidates prior capture generations`() {
let gate = TalkGenerationDeliveryGate()
let first = gate.activate()
var deliveries = 0
gate.deliver(ifActive: first) { deliveries += 1 }
gate.deactivate()
gate.deliver(ifActive: first) { deliveries += 1 }
let second = gate.activate()
gate.deliver(ifActive: first) { deliveries += 1 }
gate.deliver(ifActive: second) { deliveries += 1 }
#expect(deliveries == 2)
#expect(!gate.isActive(first))
#expect(gate.isActive(second))
}
@Test func `tap handler can run on a realtime audio queue`() throws {
let buffer = try makeFloatBuffer(
sampleRate: 48000,
channels: [[0, 0.5, -0.5, 0]])
let gate = TalkGenerationDeliveryGate()
let token = gate.activate()
let sink = RealtimeTalkFrameSink()
let handler = SendableTapHandler(MacRealtimeTalkTapHandlerFactory.make(
targetSampleRate: 24000,
deliveryGate: gate,
deliveryToken: token,
onAudio: { sink.append($0) }))
let finished = DispatchSemaphore(value: 0)
DispatchQueue(label: "talk.realtime.tap-test").async {
handler(buffer, AVAudioTime(sampleTime: 0, atRate: 48000))
finished.signal()
}
#expect(finished.wait(timeout: .now() + 2) == .success)
#expect(sink.count == 1)
}
@Test @MainActor func `capture rejects invalid target sample rate before touching hardware`() {
let capture = MacRealtimeTalkAudioCapture(selectedInputUID: { nil })
#expect(capture.suppressesInputDuringOutput)
#expect(throws: MacRealtimeTalkAudioCaptureError.self) {
try capture.start(
targetSampleRate: 0,
onAudio: { _ in },
onFailure: { _ in })
}
#expect(capture.suppressesInputDuringOutput)
}
@Test @MainActor func `queued route callback is ignored after observation retires`() async throws {
let capture = MacRealtimeTalkAudioCapture(selectedInputUID: { nil })
let probe = capture._test_replaceOutputRouteObserver()
probe.callback(self.headphonesRoute())
capture.stop()
try await self.waitForHandledCallback(probe.handled)
#expect(capture.suppressesInputDuringOutput)
}
@Test @MainActor func `retired observer callback is ignored after replacement`() async throws {
let capture = MacRealtimeTalkAudioCapture(selectedInputUID: { nil })
let retired = capture._test_replaceOutputRouteObserver()
_ = capture._test_replaceOutputRouteObserver()
retired.callback(self.headphonesRoute())
try await self.waitForHandledCallback(retired.handled)
#expect(capture.suppressesInputDuringOutput)
}
@Test @MainActor func `current observer enables headphone barge in`() async throws {
let capture = MacRealtimeTalkAudioCapture(selectedInputUID: { nil })
let probe = capture._test_replaceOutputRouteObserver()
probe.callback(self.headphonesRoute())
try await self.waitForHandledCallback(probe.handled)
#expect(!capture.suppressesInputDuringOutput)
}
@Test func `allowlisted transports preserve barge in only for headphones`() {
let transports = [
kAudioDeviceTransportTypeBuiltIn,
kAudioDeviceTransportTypeUSB,
kAudioDeviceTransportTypeBluetooth,
kAudioDeviceTransportTypeBluetoothLE,
]
for transport in transports {
let decision = MacRealtimeTalkOutputRoutePolicy.decision(for: self.route(
transport: transport,
terminals: [kAudioStreamTerminalTypeHeadphones]))
#expect(!decision.suppressesInputDuringOutput)
#expect(decision.reason == .isolatedHeadphones)
}
}
@Test func `non allowlisted transports fail closed even with headphone metadata`() {
let transports = [
kAudioDeviceTransportTypeAggregate,
kAudioDeviceTransportTypeVirtual,
kAudioDeviceTransportTypeUnknown,
kAudioDeviceTransportTypeHDMI,
kAudioDeviceTransportTypeDisplayPort,
]
for transport in transports {
let decision = MacRealtimeTalkOutputRoutePolicy.decision(for: self.route(
transport: transport,
terminals: [kAudioStreamTerminalTypeHeadphones]))
#expect(decision.suppressesInputDuringOutput)
#expect(decision.reason == .transportNotAllowlisted)
}
}
@Test func `allowlisted routes fail closed for empty speaker and mixed kinds`() {
let terminalTables: [([UInt32], MacRealtimeTalkOutputRouteDecisionReason)] = [
([], .outputKindUnavailable),
([kAudioStreamTerminalTypeSpeaker], .outputKindNotHeadphones),
(
[kAudioStreamTerminalTypeHeadphones, kAudioStreamTerminalTypeSpeaker],
.outputKindNotHeadphones),
]
for (terminals, reason) in terminalTables {
let decision = MacRealtimeTalkOutputRoutePolicy.decision(for: self.route(
transport: kAudioDeviceTransportTypeUSB,
terminals: terminals))
#expect(decision.suppressesInputDuringOutput)
#expect(decision.reason == reason)
}
}
@Test func `selected data source overrides stream terminal metadata both ways`() {
let selectedHeadphones = self.route(
transport: kAudioDeviceTransportTypeBuiltIn,
terminals: [kAudioStreamTerminalTypeSpeaker],
source: .selected(kinds: [kAudioStreamTerminalTypeHeadphones]))
let selectedSpeaker = self.route(
transport: kAudioDeviceTransportTypeBuiltIn,
terminals: [kAudioStreamTerminalTypeHeadphones],
source: .selected(kinds: [kAudioStreamTerminalTypeSpeaker]))
let headphonesDecision = MacRealtimeTalkOutputRoutePolicy.decision(for: selectedHeadphones)
let speakerDecision = MacRealtimeTalkOutputRoutePolicy.decision(for: selectedSpeaker)
#expect(!headphonesDecision.suppressesInputDuringOutput)
#expect(headphonesDecision.reason == .isolatedHeadphones)
#expect(speakerDecision.suppressesInputDuringOutput)
#expect(speakerDecision.reason == .outputKindNotHeadphones)
}
@Test func `supported data source read failure fails closed`() {
let route = self.route(
transport: kAudioDeviceTransportTypeBluetooth,
terminals: [kAudioStreamTerminalTypeHeadphones],
source: .failed)
let decision = MacRealtimeTalkOutputRoutePolicy.decision(for: route)
#expect(decision.suppressesInputDuringOutput)
#expect(decision.reason == .dataSourceReadFailed)
#expect(decision.effectiveKinds.isEmpty)
}
@Test func `missing route fails closed with a stable reason`() {
let decision = MacRealtimeTalkOutputRoutePolicy.decision(for: nil)
#expect(decision.suppressesInputDuringOutput)
#expect(decision.reason == .routeUnavailable)
#expect(decision.redactedDescription ==
"transport=unavailable kinds=[] source=unavailable " +
"suppression=true reason=route-unavailable")
}
@Test func `route decision log contains only redacted mechanical metadata`() {
let decision = MacRealtimeTalkOutputRoutePolicy.decision(for: self.route(
transport: kAudioDeviceTransportTypeBuiltIn,
terminals: [kAudioStreamTerminalTypeSpeaker],
source: .selected(kinds: [kAudioStreamTerminalTypeHeadphones])))
#expect(decision.redactedDescription ==
"transport='bltn' kinds=['hdph'] source=selected:['hdph'] " +
"suppression=false reason=isolated-headphones")
}
@Test func `route decision state deduplicates and recomputes changes`() {
var state = MacRealtimeTalkOutputRouteDecisionState()
let headphones = self.route(
transport: kAudioDeviceTransportTypeUSB,
terminals: [kAudioStreamTerminalTypeHeadphones])
let speakers = self.route(
transport: kAudioDeviceTransportTypeUSB,
terminals: [kAudioStreamTerminalTypeSpeaker])
let initial = state.update(route: headphones)
#expect(initial?.suppressesInputDuringOutput == false)
#expect(state.update(route: headphones) == nil)
let changed = state.update(route: speakers)
#expect(changed?.suppressesInputDuringOutput == true)
#expect(changed?.reason == .outputKindNotHeadphones)
state.reset()
#expect(state.current == nil)
let mixedKinds = [kAudioStreamTerminalTypeSpeaker, kAudioStreamTerminalTypeHeadphones]
let firstOrder = self.route(
transport: kAudioDeviceTransportTypeUSB,
terminals: [],
source: .selected(kinds: mixedKinds))
let reversedOrder = self.route(
transport: kAudioDeviceTransportTypeUSB,
terminals: [],
source: .selected(kinds: Array(mixedKinds.reversed())))
#expect(state.update(route: firstOrder) != nil)
#expect(state.update(route: reversedOrder) == nil)
}
@Test func `capture can be released away from the main actor`() async {
let holder = await MainActor.run {
OffMainActorCaptureHolder(MacRealtimeTalkAudioCapture(selectedInputUID: { nil }))
}
await Task.detached {
holder.releaseCapture()
}.value
await MainActor.run {}
}
private func makeFloatBuffer(
sampleRate: Double,
channels: [[Float]]) throws -> AVAudioPCMBuffer
{
let frameCount = try #require(channels.first?.count)
#expect(channels.allSatisfy { $0.count == frameCount })
let format = try #require(AVAudioFormat(
commonFormat: .pcmFormatFloat32,
sampleRate: sampleRate,
channels: AVAudioChannelCount(channels.count),
interleaved: false))
let buffer = try #require(AVAudioPCMBuffer(
pcmFormat: format,
frameCapacity: AVAudioFrameCount(frameCount)))
buffer.frameLength = AVAudioFrameCount(frameCount)
let output = try #require(buffer.floatChannelData)
for (channelIndex, samples) in channels.enumerated() {
for (sampleIndex, sample) in samples.enumerated() {
output[channelIndex][sampleIndex] = sample
}
}
return buffer
}
private func samples(in data: Data) -> [Int16] {
data.withUnsafeBytes { raw in
raw.bindMemory(to: Int16.self).map { Int16(littleEndian: $0) }
}
}
private func route(
transport: UInt32,
terminals: [UInt32],
source: MacRealtimeTalkOutputDataSource = .unsupported) -> MacRealtimeTalkOutputRoute
{
MacRealtimeTalkOutputRoute(
transportType: transport,
terminalTypes: terminals,
selectedDataSource: source)
}
private func headphonesRoute() -> MacRealtimeTalkOutputRoute {
self.route(
transport: kAudioDeviceTransportTypeBuiltIn,
terminals: [kAudioStreamTerminalTypeHeadphones])
}
private func waitForHandledCallback(_ stream: AsyncStream<Void>) async throws {
let handled = try await AsyncTimeout.withTimeout(
seconds: 1,
onTimeout: { CancellationError() },
operation: {
for await _ in stream.prefix(1) {
return true
}
return false
})
#expect(handled)
}
}
private final class RealtimeTalkFrameSink: @unchecked Sendable {
private let lock = NSLock()
private var frames: [RealtimeTalkAudioFrame] = []
var count: Int {
self.lock.lock()
defer { self.lock.unlock() }
return self.frames.count
}
func append(_ frame: RealtimeTalkAudioFrame) {
self.lock.lock()
self.frames.append(frame)
self.lock.unlock()
}
}
private final class OffMainActorCaptureHolder: @unchecked Sendable {
private var capture: MacRealtimeTalkAudioCapture?
init(_ capture: MacRealtimeTalkAudioCapture) {
self.capture = capture
}
func releaseCapture() {
self.capture = nil
}
}
@@ -53,4 +53,116 @@ struct TalkModeGatewayConfigTests {
#expect(parsed.referenceAudioPath == "/tmp/reference.wav")
#expect(parsed.referenceText == "reference transcript")
}
@Test func `realtime config uses top level overrides and normalizes control values`() {
let snapshot = Self.snapshot(talk: [
"realtime": [
"provider": " OpenAI ",
"providers": [
"openai": [
"model": "provider-model",
"speakerVoice": "alloy",
],
],
"model": " gpt-live-1-codex ",
"speakerVoice": " cedar ",
"mode": " Realtime ",
"transport": " Gateway-Relay ",
"brain": " Agent-Consult ",
],
])
let parsed = Self.parse(snapshot)
#expect(parsed.realtimeProvider == "OpenAI")
#expect(parsed.realtimeModelId == "gpt-live-1-codex")
#expect(parsed.realtimeSpeakerVoice == "cedar")
#expect(parsed.realtimeMode == "realtime")
#expect(parsed.realtimeTransport == "gateway-relay")
#expect(parsed.realtimeBrain == "agent-consult")
#expect(parsed.hasGatewayRealtimeRelayTuple)
}
@Test func `realtime config infers its sole provider and reads provider defaults`() {
let snapshot = Self.snapshot(talk: [
"realtime": [
"providers": [
"openai": [
"model": "gpt-realtime-2.1",
"voice": "marin",
],
],
"mode": "realtime",
],
])
let parsed = Self.parse(snapshot)
#expect(parsed.realtimeProvider == "openai")
#expect(parsed.realtimeModelId == "gpt-realtime-2.1")
#expect(parsed.realtimeSpeakerVoice == "marin")
#expect(parsed.realtimeMode == "realtime")
#expect(parsed.realtimeTransport == nil)
#expect(parsed.realtimeBrain == nil)
#expect(!parsed.hasGatewayRealtimeRelayTuple)
}
@Test func `realtime provider config lookup is case insensitive`() {
let snapshot = Self.snapshot(talk: [
"realtime": [
"provider": "OPENAI",
"providers": [
"openai": [
"model": "gpt-live-1-codex",
"speakerVoice": "cedar",
],
],
],
])
let parsed = Self.parse(snapshot)
#expect(parsed.realtimeProvider == "OPENAI")
#expect(parsed.realtimeModelId == "gpt-live-1-codex")
#expect(parsed.realtimeSpeakerVoice == "cedar")
}
@Test func `fallback has no realtime selection`() {
let parsed = TalkModeGatewayConfigParser.fallback(
defaultModelIdFallback: "eleven_v3",
defaultSilenceTimeoutMs: TalkDefaults.silenceTimeoutMs,
envVoice: nil,
sagVoice: nil,
envApiKey: nil)
#expect(parsed.realtimeProvider == nil)
#expect(parsed.realtimeModelId == nil)
#expect(parsed.realtimeSpeakerVoice == nil)
#expect(parsed.realtimeMode == nil)
#expect(parsed.realtimeTransport == nil)
#expect(parsed.realtimeBrain == nil)
}
private static func snapshot(talk: [String: Any]) -> ConfigSnapshot {
ConfigSnapshot(
path: nil,
exists: true,
raw: nil,
hash: nil,
parsed: nil,
valid: true,
config: ["talk": AnyCodable(talk)],
issues: nil)
}
private static func parse(_ snapshot: ConfigSnapshot) -> TalkModeGatewayConfigState {
TalkModeGatewayConfigParser.parse(
snapshot: snapshot,
defaultProvider: "elevenlabs",
defaultModelIdFallback: "eleven_v3",
defaultSilenceTimeoutMs: TalkDefaults.silenceTimeoutMs,
envVoice: nil,
sagVoice: nil,
envApiKey: nil)
}
}
File diff suppressed because it is too large Load Diff
+69
View File
@@ -58,6 +58,75 @@ stopping Talk releases the camera and microphone tracks.
- Replies are written to WebChat (same as typing).
- **Interrupt on speech** (default on): if the user talks while the assistant is speaking, playback stops and the interruption timestamp is noted for the next prompt.
## Realtime Talk over the Gateway relay (macOS)
macOS defaults to the native path above: Apple Speech recognition, Gateway chat, and `talk.speak`
playback. It switches to a streamed realtime session only when `talk.realtime` selects all three
of these together:
| Key | Required value |
| ----------- | --------------- |
| `mode` | `realtime` |
| `transport` | `gateway-relay` |
| `brain` | `agent-consult` |
Any other combination — including a partially set one — keeps the native path.
```json5
{
talk: {
realtime: {
provider: "openai",
providers: {
openai: {
model: "gpt-realtime-2.1",
speakerVoice: "cedar",
},
},
mode: "realtime",
transport: "gateway-relay",
brain: "agent-consult",
},
},
}
```
The Mac must also opt in locally with **Settings > Voice & Talk > Use realtime Gateway relay**.
This preference defaults off and stays on that Mac; Gateway config alone never activates the
streamed path. Keep `transport: "webrtc"` for browser or iOS client-owned sessions; macOS uses
the relay only when the config explicitly selects `gateway-relay`.
The Gateway must also advertise `gateway-relay` and `agent-consult` for the selected provider in
`talk.catalog`. Realtime requires macOS 26 or newer, matching Voice Wake; on older versions the
Talk and Voice Wake controls are unavailable.
### When realtime cannot start
Talk never silently sits idle. If the relay fails to start — no Gateway route, rejected
credentials, or an unsupported model — the failure is logged, the overlay shows the reason, and
Talk falls back to the native speech path for that session.
Once a session is running, a dropped relay reconnects on a bounded retry schedule (roughly 0.5 s
then 2 s). If those attempts are exhausted, the overlay reports
`Realtime disconnected repeatedly — using native speech` and the next start bypasses realtime.
Losing the microphone mid-session closes the relay and takes the same route.
Relay output cancellation is turn-scoped. Clients copy the current `turnId` from the
`talk.event` audio envelope. Matching ids return `applied`, stale ids return `stale`, and
sessions without an active turn return `idle`. Older clients that omit `turnId` still cancel
the current turn:
```json
{
"method": "talk.session.cancelOutput",
"params": {
"sessionId": "relay-session-id",
"turnId": "turn-7",
"reason": "barge-in"
}
}
```
## Voice directives in replies
The assistant can prefix a reply with a single JSON line to control voice: