Send pre-connect audio (#622)

- Adds infrastructure and public API for the agent pre-connect feature
based on data channels

livekit-examples/multimodal-agent-python#18

### Building blocks:
- `LocalAudioTrackRecorder` - a simple stream-based wrapper for
`startLocalRecording`, can be used in isolation
- `PreConnectAudioBuffer` - handles the `Room` state automatically (if a
`Room?` is passed) by setting participant attributes and flushing the
recording into the data stream 🚰, can be used manually via `start/stop`
methods
- `Room.startCapturingBeforeConnecting` - a convenience one-liner to get
the default behavior; start automatically, stop when the agent joins

### Usage

```swift
.onAppear {
    try? AudioManager.shared.setRecordingAlwaysPreparedMode(true)
}

+

private func connect() {
    Task(priority: .userInitiated) {
        isConnecting = true

        if enablePreConnect {
            try? await room.startCapturingBeforeConnecting()
            isConnecting = false
        }
```

Author: pblazej

.nanpa/pre-connect-audio.kdl

@@ -0,0 +1,2 @@
+minor type="added" "Added LocalAudioTrackRecorder"
+minor type="added" "Added the possibility to capture pre-connect audio and send it to agents via data streams"

Sources/LiveKit/Core/PreConnectAudioBuffer.swift

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2025 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import AVFAudio
+import Foundation
+
+/// A buffer that captures audio before connecting to the server,
+/// and sends it on certain ``RoomDelegate`` events.
+@objc
+public final class PreConnectAudioBuffer: NSObject, Loggable {
+    /// The default participant attribute key used to indicate that the audio buffer is active.
+    @objc
+    public static let attributeKey = "lk.agent.pre-connect-audio"
+
+    /// The default data topic used to send the audio buffer.
+    @objc
+    public static let dataTopic = "lk.agent.pre-connect-audio-buffer"
+
+    /// The room instance to listen for events.
+    @objc
+    public let room: Room?
+
+    /// The audio recorder instance.
+    @objc
+    public let recorder: LocalAudioTrackRecorder
+
+    private let state = StateSync<State>(State())
+    private struct State {
+        var audioStream: LocalAudioTrackRecorder.Stream?
+    }
+
+    /// Initialize the audio buffer with a room instance.
+    /// - Parameters:
+    ///   - room: The room instance to listen for events.
+    ///   - recorder: The audio recorder to use for capturing.
+    @objc
+    public init(room: Room?,
+                recorder: LocalAudioTrackRecorder = LocalAudioTrackRecorder(
+                    track: LocalAudioTrack.createTrack(),
+                    format: .pcmFormatInt16, // supported by agent plugins
+                    sampleRate: 24000, // supported by agent plugins
+                    maxSize: 10 * 1024 * 1024 // arbitrary max recording size of 10MB
+                ))
+    {
+        self.room = room
+        self.recorder = recorder
+        super.init()
+    }
+
+    deinit {
+        stopRecording()
+        room?.remove(delegate: self)
+    }
+
+    /// Start capturing audio and listening to ``RoomDelegate`` events.
+    @objc
+    public func startRecording() async throws {
+        room?.add(delegate: self)
+
+        let stream = try await recorder.start()
+        log("Started capturing audio", .info)
+        state.mutate { state in
+            state.audioStream = stream
+        }
+    }
+
+    /// Stop capturing audio.
+    /// - Parameters:
+    ///   - flush: If `true`, the audio stream will be flushed immediately without sending.
+    @objc
+    public func stopRecording(flush: Bool = false) {
+        recorder.stop()
+        log("Stopped capturing audio", .info)
+        if flush, let stream = state.audioStream {
+            Task {
+                for await _ in stream {}
+            }
+        }
+    }
+}
+
+// MARK: - RoomDelegate
+
+extension PreConnectAudioBuffer: RoomDelegate {
+    public func roomDidConnect(_ room: Room) {
+        Task {
+            try? await setParticipantAttribute(room: room)
+        }
+    }
+
+    public func room(_ room: Room, participant _: LocalParticipant, remoteDidSubscribeTrack _: LocalTrackPublication) {
+        stopRecording()
+        Task {
+            try? await sendAudioData(to: room)
+        }
+    }
+
+    /// Set the participant attribute to indicate that the audio buffer is active.
+    /// - Parameters:
+    ///   - key: The key to set the attribute.
+    ///   - room: The room instance to set the attribute.
+    @objc
+    public func setParticipantAttribute(key _: String = attributeKey, room: Room) async throws {
+        var attributes = room.localParticipant.attributes
+        attributes[Self.attributeKey] = "true"
+        try await room.localParticipant.set(attributes: attributes)
+        log("Set participant attribute", .info)
+    }
+
+    /// Send the audio data to the room.
+    /// - Parameters:
+    ///   - room: The room instance to send the audio data.
+    ///   - topic: The topic to send the audio data.
+    @objc
+    public func sendAudioData(to room: Room, on topic: String = dataTopic) async throws {
+        guard let audioStream = state.audioStream else {
+            throw LiveKitError(.invalidState, message: "Audio stream is nil")
+        }
+
+        let streamOptions = StreamByteOptions(
+            topic: topic,
+            attributes: [
+                "sampleRate": "\(recorder.sampleRate)",
+                "channels": "\(recorder.channels)",
+            ]
+        )
+        let writer = try await room.localParticipant.streamBytes(options: streamOptions)
+        try await writer.write(audioStream.collect())
+        try await writer.close()
+        log("Sent audio data", .info)
+
+        room.remove(delegate: self)
+    }
+}

Sources/LiveKit/Core/Room+PreConnect.swift

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2025 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+
+public extension Room {
+    /// Start capturing audio before connecting to the server,
+    /// so that it's not lost when the connection is established.
+    /// It will be automatically sent via data stream to the other participant
+    /// using the `PreConnectAudioBuffer.dataTopic` when the local track is subscribed.
+    /// - See: ``PreConnectAudioBuffer``
+    /// - Note: Use ``AudioManager/setRecordingAlwaysPreparedMode(_:)`` to request microphone permissions early.
+    func startCapturingBeforeConnecting() async throws {
+        try await preConnectBuffer.startRecording()
+    }
+}

Sources/LiveKit/Core/Room.swift

@@ -116,6 +116,12 @@ public class Room: NSObject, ObservableObject, Loggable {
         try await self?.send(dataPacket: packet)
     }
 
+    // MARK: - PreConnect
+
+    lazy var preConnectBuffer = PreConnectAudioBuffer(room: self)
+
+    // MARK: - Queue
+
     var _blockProcessQueue = DispatchQueue(label: "LiveKitSDK.engine.pendingBlocks",
                                            qos: .default)
 
@@ -395,6 +401,10 @@ extension Room {
             e2eeManager.cleanUp()
         }
 
+        if disconnectError != nil {
+            preConnectBuffer.stopRecording(flush: true)
+        }
+
         // Reset state
         _state.mutate {
             // if isFullReconnect, keep connection related states

Sources/LiveKit/DataStream/Outgoing/OutgoingStreamManager.swift

@@ -246,7 +246,7 @@ actor OutgoingStreamManager: Loggable {
     }
 
     /// Maximum number of bytes to send in a single chunk.
-    private static let chunkSize = 15000
+    private static let chunkSize = 15 * 1024
 
     /// Default MIME type to use for text streams.
     fileprivate static let textMimeType = "text/plain"

Sources/LiveKit/Extensions/AVAudioPCMBuffer.swift

@@ -133,4 +133,37 @@ public extension AVAudioPCMBuffer {
 
         return outputBuffer
     }
+
+    func toData() -> Data? {
+        switch format.commonFormat {
+        case .pcmFormatInt16:
+            guard let channelData = int16ChannelData else { return nil }
+            return interleave(channelData: channelData)
+        case .pcmFormatInt32:
+            guard let channelData = int32ChannelData else { return nil }
+            return interleave(channelData: channelData)
+        case .pcmFormatFloat32:
+            guard let channelData = floatChannelData else { return nil }
+            return interleave(channelData: channelData)
+        default:
+            return nil
+        }
+    }
+
+    private func interleave<T>(channelData: UnsafePointer<UnsafeMutablePointer<T>>) -> Data {
+        let channels = Int(format.channelCount)
+        let frameLength = Int(frameLength)
+        var interleavedBuffer = [T](repeating: channelData[0][0], count: frameLength * channels)
+
+        for frame in 0 ..< frameLength {
+            for channel in 0 ..< channels {
+                let channelPtr = channelData[channel]
+                interleavedBuffer[frame * channels + channel] = channelPtr[frame]
+            }
+        }
+
+        return interleavedBuffer.withUnsafeBytes { bufferPointer in
+            Data(bufferPointer)
+        }
+    }
 }

Sources/LiveKit/Track/AudioManager.swift

@@ -244,6 +244,9 @@ public class AudioManager: Loggable {
     /// This will per persisted accross Rooms and connections.
     public var isRecordingAlwaysPreparedMode: Bool { RTC.audioDeviceModule.isRecordingAlwaysPreparedMode }
 
+    /// Keep recording initialized (mic input) and pre-warm voice processing etc.
+    /// Mic permission is required and dialog will appear if not already granted.
+    /// This will per persisted accross Rooms and connections.
     public func setRecordingAlwaysPreparedMode(_ enabled: Bool) throws {
         let result = RTC.audioDeviceModule.setRecordingAlwaysPreparedMode(enabled)
         try checkAdmResult(code: result)