feat: LiveKit voice session, control channel, transcripts (Phase 2)
- VoiceSession: token->connect with fresh-room retry on timeout, agent discovery, agent_ready wait, mic enable/mute, text send over lk.chat - ControlChannel: outbound set_voice/set_model/get_models/set_mode/ media_state; inbound agent_ready/models/model_changed/task_event - TranscriptHandler: transcription segments + transcript topic finals, partial/final dedupe per role - ConversationView/OrbView: status orb, mic/text-mode toggles, composer - ServerConfig: plain ws:// signaling for private IPs (self-signed TLS not injectable into SDK WebSocket); Info.plist local networking
This commit is contained in:
snowspeeder committed
1 parent
378b8cb36f
commit
4986b6d7dc
14 files changed
+900
-50
No files matched your search
@@ -0,0 +1,231 @@
|
||||
import Foundation
|
||||
import LiveKit
|
||||
import os
|
||||
|
||||
private let logger = Logger(subsystem: "dev.shane.voice", category: "session")
|
||||
|
||||
enum VoiceSessionState: Equatable {
|
||||
case idle
|
||||
case connecting
|
||||
case waitingAgent
|
||||
case ready
|
||||
case failed(String)
|
||||
}
|
||||
|
||||
struct AgentInfo: Equatable {
|
||||
let identity: String
|
||||
let name: String?
|
||||
}
|
||||
|
||||
/// Room lifecycle: connect, agent discovery, agent_ready wait, reconnect-once,
|
||||
/// disconnect teardown. Ports the web UI's connect flow.
|
||||
@MainActor
|
||||
final class VoiceSession: ObservableObject {
|
||||
@Published private(set) var state: VoiceSessionState = .idle
|
||||
@Published private(set) var agent: AgentInfo?
|
||||
@Published private(set) var isMuted = false
|
||||
|
||||
private(set) var room = Room()
|
||||
let controlChannel = ControlChannel()
|
||||
let transcriptHandler = TranscriptHandler()
|
||||
private let config: ServerConfig
|
||||
private let tokenClient: TokenClient
|
||||
private var reconnectAttempted = false
|
||||
private var connectAttempts = 0
|
||||
private var agentWaitTask: Task<Void, Never>?
|
||||
|
||||
init(config: ServerConfig, tokenClient: TokenClient) {
|
||||
self.config = config
|
||||
self.tokenClient = tokenClient
|
||||
room.add(delegate: self)
|
||||
room.add(delegate: controlChannel)
|
||||
room.add(delegate: transcriptHandler)
|
||||
controlChannel.onEvent = { [weak self] _ in
|
||||
guard self != nil else { return }
|
||||
}
|
||||
}
|
||||
|
||||
var isReady: Bool { state == .ready }
|
||||
|
||||
func connect() async {
|
||||
await start()
|
||||
}
|
||||
|
||||
private func makeRoom() -> Room {
|
||||
let room = Room()
|
||||
room.add(delegate: self)
|
||||
room.add(delegate: controlChannel)
|
||||
room.add(delegate: transcriptHandler)
|
||||
return room
|
||||
}
|
||||
|
||||
private func start() async {
|
||||
state = .connecting
|
||||
connectAttempts = 0
|
||||
await attemptConnect()
|
||||
}
|
||||
|
||||
private func attemptConnect() async {
|
||||
let result: Result<Void, Error>
|
||||
do {
|
||||
let token = try await tokenClient.fetchToken()
|
||||
logger.log("token ok room=\(token.room, privacy: .public)")
|
||||
try await room.connect(url: config.livekitURL.absoluteString, token: token.token, connectOptions: makeConnectOptions())
|
||||
result = .success(())
|
||||
} catch {
|
||||
result = .failure(error)
|
||||
}
|
||||
switch result {
|
||||
case .success:
|
||||
logger.log("room connected")
|
||||
scanForAgent()
|
||||
startAgentWait()
|
||||
case .failure(let error):
|
||||
let isTimeout = (error as? LiveKitError)?.type == .timedOut
|
||||
if isTimeout, connectAttempts < 2 {
|
||||
connectAttempts += 1
|
||||
let attempt = connectAttempts
|
||||
logger.log("connect attempt \(attempt) timed out, retrying with fresh room")
|
||||
await teardownRoom()
|
||||
await attemptConnect()
|
||||
return
|
||||
}
|
||||
logger.error("connect failed: \(String(describing: error), privacy: .public)")
|
||||
state = .failed(error.localizedDescription)
|
||||
}
|
||||
}
|
||||
|
||||
private func makeConnectOptions() -> ConnectOptions {
|
||||
return ConnectOptions(
|
||||
socketConnectTimeoutInterval: 15,
|
||||
primaryTransportConnectTimeout: 20,
|
||||
publisherTransportConnectTimeout: 20
|
||||
)
|
||||
}
|
||||
|
||||
private func teardownRoom() async {
|
||||
await room.disconnect()
|
||||
room = makeRoom()
|
||||
}
|
||||
|
||||
private func handleAgentTimeout() async {
|
||||
guard state == .waitingAgent, !reconnectAttempted else {
|
||||
if state == .waitingAgent {
|
||||
state = .failed("Agent never joined. Is the agent worker running?")
|
||||
}
|
||||
return
|
||||
}
|
||||
reconnectAttempted = true
|
||||
await teardownRoom()
|
||||
await start()
|
||||
}
|
||||
|
||||
private func startAgentWait() {
|
||||
state = .waitingAgent
|
||||
agentWaitTask?.cancel()
|
||||
agentWaitTask = Task { [weak self] in
|
||||
try? await Task.sleep(for: .seconds(10))
|
||||
guard !Task.isCancelled else { return }
|
||||
await self?.handleAgentTimeout()
|
||||
}
|
||||
}
|
||||
|
||||
func agentReadyReceived() {
|
||||
guard state == .waitingAgent else {
|
||||
logger.log("agent_ready ignored in state \(String(describing: self.state), privacy: .public)")
|
||||
return
|
||||
}
|
||||
agentWaitTask?.cancel()
|
||||
state = .ready
|
||||
logger.log("state=ready, enabling mic")
|
||||
Task { await enableMicrophone() }
|
||||
}
|
||||
|
||||
private func scanForAgent() {
|
||||
for participant in room.remoteParticipants.values {
|
||||
if isAgentParticipant(participant) {
|
||||
foundAgent(participant)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func isAgentParticipant(_ participant: RemoteParticipant) -> Bool {
|
||||
participant.kind == .agent || participant.identity?.stringValue == "voice-assistant"
|
||||
}
|
||||
|
||||
private func foundAgent(_ participant: RemoteParticipant) {
|
||||
agent = AgentInfo(identity: participant.identity?.stringValue ?? "", name: participant.name)
|
||||
}
|
||||
|
||||
func enableMicrophone() async {
|
||||
for attempt in 0..<3 {
|
||||
do {
|
||||
try await room.localParticipant.setMicrophone(enabled: true)
|
||||
isMuted = false
|
||||
logger.log("mic enabled")
|
||||
return
|
||||
} catch {
|
||||
let message = String(describing: error)
|
||||
if attempt < 2 {
|
||||
logger.log("mic enable failed (\(message, privacy: .public)), retrying in 1s")
|
||||
try? await Task.sleep(for: .seconds(1))
|
||||
continue
|
||||
}
|
||||
logger.error("mic enable failed after retries: \(message, privacy: .public)")
|
||||
state = .failed("Microphone error: \(error.localizedDescription)")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func setMuted(_ muted: Bool) async {
|
||||
guard isReady else { return }
|
||||
do {
|
||||
try await room.localParticipant.setMicrophone(enabled: !muted)
|
||||
isMuted = muted
|
||||
} catch {
|
||||
state = .failed("Microphone error: \(error.localizedDescription)")
|
||||
}
|
||||
}
|
||||
|
||||
func sendText(_ text: String) async throws {
|
||||
try await room.localParticipant.sendText(text, for: "lk.chat")
|
||||
}
|
||||
|
||||
func disconnect() async {
|
||||
agentWaitTask?.cancel()
|
||||
state = .idle
|
||||
agent = nil
|
||||
isMuted = false
|
||||
reconnectAttempted = false
|
||||
await teardownRoom()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
extension VoiceSession: RoomDelegate {
|
||||
nonisolated func room(_ room: Room, participantDidConnect participant: RemoteParticipant) {
|
||||
Task { @MainActor [weak self] in
|
||||
guard let self, self.isAgentParticipant(participant) else { return }
|
||||
self.foundAgent(participant)
|
||||
}
|
||||
}
|
||||
|
||||
nonisolated func room(_ room: Room, participant: RemoteParticipant?, didReceiveData data: Data, forTopic topic: String, encryptionType: EncryptionType) {
|
||||
logger.log("data topic=\(topic, privacy: .public) bytes=\(data.count) payload=\(String(data: data.prefix(300), encoding: .utf8) ?? "<binary>", privacy: .public)")
|
||||
Task { @MainActor [weak self] in
|
||||
guard let self else { return }
|
||||
// Secondary transcript source (finals only).
|
||||
if topic == "transcript",
|
||||
let json = try? JSONDecoder().decode([String: JSONValue].self, from: data),
|
||||
json["type"]?.stringValue == "transcript",
|
||||
let text = json["text"]?.stringValue {
|
||||
let role: TranscriptEvent.Role = json["role"]?.stringValue == "agent" ? .agent : .user
|
||||
self.transcriptHandler.receiveDataTranscript(role: role, text: text)
|
||||
return
|
||||
}
|
||||
if self.controlChannel.handleTaskEvent(data) { return }
|
||||
_ = self.controlChannel.handleVoiceControl(data, session: self)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user