Add Live Call transport layer (P10 foundation)
LiveProtocol — type-safe Kaizen WS protocol serialization/parsing, mirrors lib/live/protocol.ts 1:1 (setup/resume/audio/text/interrupt/end + all 12 server event types). LiveClient — OkHttp WebSocket client with auto-reconnect (exponential backoff, max 5 attempts), resume protocol (sessionId + lastSeqId), backpressure handling, terminal error detection. 15s ping interval keeps mobile NAT alive. AudioPipeline — bidirectional audio: AudioRecord (16kHz PCM16 mono, VOICE_COMMUNICATION for hardware AEC/noise suppression) → 100ms chunks → Base64. AudioTrack (24kHz PCM16 mono, LOW_LATENCY, MODE_STREAM) ← server chunks. Mic-ducking during KI speech (RMS threshold gating). LiveCallService — Foreground Service (MICROPHONE type), survives rotation/backgrounding. Notification with duration timer + end button. Status state machine (IDLE→CONNECTING→LISTENING→SPEAKING). Orchestrates LiveClient + AudioPipeline, emits onTurnEnd for tree integration.
This commit is contained in:
Bruno Deanoz
parent
25fbb17d18
commit
65ea3a5656
5 changed files with 890 additions and 0 deletions
|
|
@ -11,6 +11,10 @@
|
|||
<!-- Required for voice recording (mic button in chat) -->
|
||||
<uses-permission android:name="android.permission.RECORD_AUDIO" />
|
||||
|
||||
<!-- Required for Live Call foreground service (mic access while backgrounded) -->
|
||||
<uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
|
||||
<uses-permission android:name="android.permission.FOREGROUND_SERVICE_MICROPHONE" />
|
||||
|
||||
<application
|
||||
android:allowBackup="true"
|
||||
android:dataExtractionRules="@xml/data_extraction_rules"
|
||||
|
|
@ -32,6 +36,11 @@
|
|||
<category android:name="android.intent.category.LAUNCHER" />
|
||||
</intent-filter>
|
||||
</activity>
|
||||
|
||||
<service
|
||||
android:name=".live.LiveCallService"
|
||||
android:foregroundServiceType="microphone"
|
||||
android:exported="false" />
|
||||
</application>
|
||||
|
||||
</manifest>
|
||||
210
app/src/main/java/dev/kaizen/app/live/AudioPipeline.kt
Normal file
210
app/src/main/java/dev/kaizen/app/live/AudioPipeline.kt
Normal file
|
|
@ -0,0 +1,210 @@
|
|||
package dev.kaizen.app.live
|
||||
|
||||
import android.Manifest
|
||||
import android.content.Context
|
||||
import android.content.pm.PackageManager
|
||||
import android.media.AudioAttributes
|
||||
import android.media.AudioFormat
|
||||
import android.media.AudioRecord
|
||||
import android.media.AudioTrack
|
||||
import android.media.MediaRecorder
|
||||
import android.util.Base64
|
||||
import android.util.Log
|
||||
import androidx.core.content.ContextCompat
|
||||
import kotlinx.coroutines.*
|
||||
import java.nio.ByteBuffer
|
||||
import java.nio.ByteOrder
|
||||
import kotlin.math.abs
|
||||
import kotlin.math.sqrt
|
||||
|
||||
/**
|
||||
* Bidirectional audio pipeline for Live Call.
|
||||
*
|
||||
* Capture: AudioRecord (16kHz PCM16 mono, VOICE_COMMUNICATION for hardware AEC)
|
||||
* → 100ms chunks → Base64 → [onChunk] callback.
|
||||
*
|
||||
* Playback: AudioTrack (24kHz PCM16 mono, LOW_LATENCY, MODE_STREAM)
|
||||
* ← PCM16 chunks from server.
|
||||
*
|
||||
* Mic-ducking: gain reduced to 10% while the KI speaks (prevents echo feedback
|
||||
* loop on devices with weak AEC). Not a digital gain — we skip sending chunks
|
||||
* whose RMS is below the ducking threshold.
|
||||
*/
|
||||
class AudioPipeline(private val context: Context) {
|
||||
|
||||
companion object {
|
||||
private const val TAG = "AudioPipeline"
|
||||
|
||||
private const val CAPTURE_SAMPLE_RATE = 16_000
|
||||
private const val PLAYBACK_SAMPLE_RATE = 24_000
|
||||
private const val CHANNEL_IN = AudioFormat.CHANNEL_IN_MONO
|
||||
private const val CHANNEL_OUT = AudioFormat.CHANNEL_OUT_MONO
|
||||
private const val ENCODING = AudioFormat.ENCODING_PCM_16BIT
|
||||
|
||||
private const val CHUNK_MS = 100
|
||||
private const val CAPTURE_CHUNK_SAMPLES = CAPTURE_SAMPLE_RATE * CHUNK_MS / 1000 // 1600
|
||||
private const val CAPTURE_CHUNK_BYTES = CAPTURE_CHUNK_SAMPLES * 2 // 3200
|
||||
|
||||
private const val DUCKING_GAIN = 0.10f
|
||||
}
|
||||
|
||||
var onChunk: ((base64Pcm: String) -> Unit)? = null
|
||||
var onAmplitude: ((Float) -> Unit)? = null
|
||||
|
||||
private var recorder: AudioRecord? = null
|
||||
private var track: AudioTrack? = null
|
||||
private var captureJob: Job? = null
|
||||
|
||||
@Volatile private var ducking = false
|
||||
@Volatile private var capturing = false
|
||||
|
||||
// ─── Capture ──────────────────────────────────────────────────────────
|
||||
|
||||
fun startCapture(scope: CoroutineScope): Boolean {
|
||||
if (ContextCompat.checkSelfPermission(context, Manifest.permission.RECORD_AUDIO) != PackageManager.PERMISSION_GRANTED) {
|
||||
Log.w(TAG, "RECORD_AUDIO permission not granted")
|
||||
return false
|
||||
}
|
||||
|
||||
val minBuf = AudioRecord.getMinBufferSize(CAPTURE_SAMPLE_RATE, CHANNEL_IN, ENCODING)
|
||||
if (minBuf == AudioRecord.ERROR || minBuf == AudioRecord.ERROR_BAD_VALUE) {
|
||||
Log.e(TAG, "Invalid min buffer size: $minBuf")
|
||||
return false
|
||||
}
|
||||
|
||||
val rec = AudioRecord(
|
||||
MediaRecorder.AudioSource.VOICE_COMMUNICATION,
|
||||
CAPTURE_SAMPLE_RATE,
|
||||
CHANNEL_IN,
|
||||
ENCODING,
|
||||
maxOf(minBuf * 4, CAPTURE_CHUNK_BYTES * 4),
|
||||
)
|
||||
|
||||
if (rec.state != AudioRecord.STATE_INITIALIZED) {
|
||||
Log.e(TAG, "AudioRecord init failed")
|
||||
rec.release()
|
||||
return false
|
||||
}
|
||||
|
||||
recorder = rec
|
||||
rec.startRecording()
|
||||
capturing = true
|
||||
|
||||
captureJob = scope.launch(Dispatchers.IO) {
|
||||
val buffer = ShortArray(CAPTURE_CHUNK_SAMPLES)
|
||||
val byteBuffer = ByteBuffer.allocate(CAPTURE_CHUNK_BYTES).order(ByteOrder.LITTLE_ENDIAN)
|
||||
|
||||
while (isActive && capturing) {
|
||||
val read = rec.read(buffer, 0, CAPTURE_CHUNK_SAMPLES)
|
||||
if (read <= 0) continue
|
||||
|
||||
val rms = computeRms(buffer, read)
|
||||
onAmplitude?.invoke(rms)
|
||||
|
||||
if (ducking && rms < DUCKING_GAIN) continue
|
||||
|
||||
byteBuffer.clear()
|
||||
for (i in 0 until read) byteBuffer.putShort(buffer[i])
|
||||
|
||||
val base64 = Base64.encodeToString(byteBuffer.array(), 0, read * 2, Base64.NO_WRAP)
|
||||
onChunk?.invoke(base64)
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
fun stopCapture() {
|
||||
capturing = false
|
||||
captureJob?.cancel()
|
||||
captureJob = null
|
||||
try {
|
||||
recorder?.stop()
|
||||
} catch (_: IllegalStateException) {}
|
||||
recorder?.release()
|
||||
recorder = null
|
||||
}
|
||||
|
||||
// ─── Playback ─────────────────────────────────────────────────────────
|
||||
|
||||
fun initPlayback(): Boolean {
|
||||
val minBuf = AudioTrack.getMinBufferSize(PLAYBACK_SAMPLE_RATE, CHANNEL_OUT, ENCODING)
|
||||
if (minBuf == AudioTrack.ERROR || minBuf == AudioTrack.ERROR_BAD_VALUE) {
|
||||
Log.e(TAG, "Invalid playback buffer size: $minBuf")
|
||||
return false
|
||||
}
|
||||
|
||||
val t = AudioTrack.Builder()
|
||||
.setAudioAttributes(
|
||||
AudioAttributes.Builder()
|
||||
.setUsage(AudioAttributes.USAGE_ASSISTANT)
|
||||
.setContentType(AudioAttributes.CONTENT_TYPE_SPEECH)
|
||||
.build()
|
||||
)
|
||||
.setAudioFormat(
|
||||
AudioFormat.Builder()
|
||||
.setSampleRate(PLAYBACK_SAMPLE_RATE)
|
||||
.setEncoding(ENCODING)
|
||||
.setChannelMask(CHANNEL_OUT)
|
||||
.build()
|
||||
)
|
||||
.setBufferSizeInBytes(maxOf(minBuf * 4, 24_000 * 2))
|
||||
.setPerformanceMode(AudioTrack.PERFORMANCE_MODE_LOW_LATENCY)
|
||||
.setTransferMode(AudioTrack.MODE_STREAM)
|
||||
.build()
|
||||
|
||||
if (t.state != AudioTrack.STATE_INITIALIZED) {
|
||||
Log.e(TAG, "AudioTrack init failed")
|
||||
t.release()
|
||||
return false
|
||||
}
|
||||
|
||||
track = t
|
||||
t.play()
|
||||
return true
|
||||
}
|
||||
|
||||
fun playChunk(base64Pcm: String) {
|
||||
val bytes = Base64.decode(base64Pcm, Base64.NO_WRAP)
|
||||
track?.write(bytes, 0, bytes.size)
|
||||
}
|
||||
|
||||
fun flushPlayback() {
|
||||
track?.pause()
|
||||
track?.flush()
|
||||
track?.play()
|
||||
}
|
||||
|
||||
fun stopPlayback() {
|
||||
try {
|
||||
track?.pause()
|
||||
track?.flush()
|
||||
} catch (_: IllegalStateException) {}
|
||||
track?.release()
|
||||
track = null
|
||||
}
|
||||
|
||||
// ─── Ducking ──────────────────────────────────────────────────────────
|
||||
|
||||
fun setDucking(enabled: Boolean) {
|
||||
ducking = enabled
|
||||
}
|
||||
|
||||
// ─── Cleanup ──────────────────────────────────────────────────────────
|
||||
|
||||
fun release() {
|
||||
stopCapture()
|
||||
stopPlayback()
|
||||
}
|
||||
|
||||
// ─── Helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
private fun computeRms(buffer: ShortArray, count: Int): Float {
|
||||
var sum = 0.0
|
||||
for (i in 0 until count) {
|
||||
val sample = buffer[i].toDouble() / Short.MAX_VALUE
|
||||
sum += sample * sample
|
||||
}
|
||||
return sqrt(sum / count).toFloat()
|
||||
}
|
||||
}
|
||||
304
app/src/main/java/dev/kaizen/app/live/LiveCallService.kt
Normal file
304
app/src/main/java/dev/kaizen/app/live/LiveCallService.kt
Normal file
|
|
@ -0,0 +1,304 @@
|
|||
package dev.kaizen.app.live
|
||||
|
||||
import android.app.Notification
|
||||
import android.app.NotificationChannel
|
||||
import android.app.NotificationManager
|
||||
import android.app.PendingIntent
|
||||
import android.app.Service
|
||||
import android.content.Context
|
||||
import android.content.Intent
|
||||
import android.content.pm.ServiceInfo
|
||||
import android.os.Binder
|
||||
import android.os.IBinder
|
||||
import android.util.Log
|
||||
import dev.kaizen.app.MainActivity
|
||||
import dev.kaizen.app.R
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
|
||||
/**
|
||||
* Foreground Service for Live Call — keeps mic + WS alive when the app is
|
||||
* backgrounded or the screen rotates. Shows a persistent notification with
|
||||
* call duration and an end button.
|
||||
*
|
||||
* Lifecycle: started by ChatScreen → binds → [startCall] → audio flows
|
||||
* → [endCall] or notification "Beenden" → cleanup → stopSelf.
|
||||
*/
|
||||
class LiveCallService : Service() {
|
||||
|
||||
companion object {
|
||||
private const val TAG = "LiveCallService"
|
||||
private const val CHANNEL_ID = "kaizen_live_call"
|
||||
private const val NOTIFICATION_ID = 9001
|
||||
private const val ACTION_END = "dev.kaizen.app.LIVE_CALL_END"
|
||||
|
||||
fun intent(context: Context): Intent = Intent(context, LiveCallService::class.java)
|
||||
}
|
||||
|
||||
// ─── Public state ─────────────────────────────────────────────────────
|
||||
|
||||
private val _status = MutableStateFlow(CallStatus.IDLE)
|
||||
val status: StateFlow<CallStatus> = _status
|
||||
|
||||
private val _subtitle = MutableStateFlow("")
|
||||
val subtitle: StateFlow<String> = _subtitle
|
||||
|
||||
private val _amplitude = MutableStateFlow(0f)
|
||||
val amplitude: StateFlow<Float> = _amplitude
|
||||
|
||||
private val _userSpeaking = MutableStateFlow(false)
|
||||
val userSpeaking: StateFlow<Boolean> = _userSpeaking
|
||||
|
||||
var onTurnEnd: ((ServerEvent.TurnEnd) -> Unit)? = null
|
||||
var onError: ((String, String?) -> Unit)? = null
|
||||
|
||||
// ─── Internal ─────────────────────────────────────────────────────────
|
||||
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate)
|
||||
private var client: LiveClient? = null
|
||||
private var audio: AudioPipeline? = null
|
||||
private var durationJob: Job? = null
|
||||
private var startTimeMs = 0L
|
||||
|
||||
// ─── Binder ───────────────────────────────────────────────────────────
|
||||
|
||||
inner class LocalBinder : Binder() {
|
||||
val service: LiveCallService get() = this@LiveCallService
|
||||
}
|
||||
|
||||
private val binder = LocalBinder()
|
||||
override fun onBind(intent: Intent?): IBinder = binder
|
||||
|
||||
// ─── Lifecycle ────────────────────────────────────────────────────────
|
||||
|
||||
override fun onCreate() {
|
||||
super.onCreate()
|
||||
createNotificationChannel()
|
||||
}
|
||||
|
||||
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
|
||||
if (intent?.action == ACTION_END) {
|
||||
endCall()
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
override fun onDestroy() {
|
||||
endCall()
|
||||
scope.cancel()
|
||||
super.onDestroy()
|
||||
}
|
||||
|
||||
// ─── Start / End ──────────────────────────────────────────────────────
|
||||
|
||||
fun startCall(baseUrl: String, token: String, model: String, voice: String, conversationId: String?) {
|
||||
if (_status.value != CallStatus.IDLE) return
|
||||
|
||||
_status.value = CallStatus.CONNECTING
|
||||
|
||||
startForeground(NOTIFICATION_ID, buildNotification("Verbinde…"), ServiceInfo.FOREGROUND_SERVICE_TYPE_MICROPHONE)
|
||||
|
||||
val pipeline = AudioPipeline(this)
|
||||
audio = pipeline
|
||||
|
||||
val liveClient = LiveClient(baseUrl, token, scope)
|
||||
client = liveClient
|
||||
|
||||
pipeline.onAmplitude = { amp -> _amplitude.value = amp }
|
||||
|
||||
scope.launch {
|
||||
liveClient.events.collect { event -> handleEvent(event, pipeline) }
|
||||
}
|
||||
|
||||
pipeline.onChunk = { base64 -> liveClient.sendAudio(base64) }
|
||||
|
||||
liveClient.connect(model, voice, conversationId)
|
||||
}
|
||||
|
||||
fun endCall() {
|
||||
if (_status.value == CallStatus.IDLE) return
|
||||
|
||||
client?.end()
|
||||
audio?.release()
|
||||
durationJob?.cancel()
|
||||
|
||||
client?.dispose()
|
||||
client = null
|
||||
audio = null
|
||||
durationJob = null
|
||||
|
||||
_status.value = CallStatus.IDLE
|
||||
_subtitle.value = ""
|
||||
_amplitude.value = 0f
|
||||
_userSpeaking.value = false
|
||||
|
||||
stopForeground(STOP_FOREGROUND_REMOVE)
|
||||
stopSelf()
|
||||
}
|
||||
|
||||
fun sendText(text: String) {
|
||||
client?.sendText(text)
|
||||
}
|
||||
|
||||
fun interrupt() {
|
||||
client?.interrupt()
|
||||
}
|
||||
|
||||
// ─── Event handling ───────────────────────────────────────────────────
|
||||
|
||||
private fun handleEvent(event: ServerEvent, pipeline: AudioPipeline) {
|
||||
when (event) {
|
||||
is ServerEvent.Ready -> {
|
||||
_status.value = CallStatus.LISTENING
|
||||
|
||||
if (!pipeline.initPlayback()) {
|
||||
Log.e(TAG, "Playback init failed")
|
||||
}
|
||||
if (!pipeline.startCapture(scope)) {
|
||||
Log.e(TAG, "Capture init failed")
|
||||
endCall()
|
||||
return
|
||||
}
|
||||
|
||||
startTimeMs = System.currentTimeMillis()
|
||||
startDurationTimer()
|
||||
updateNotification("Anruf aktiv")
|
||||
}
|
||||
|
||||
is ServerEvent.Resumed -> {
|
||||
_status.value = CallStatus.LISTENING
|
||||
updateNotification("Anruf fortgesetzt")
|
||||
}
|
||||
|
||||
is ServerEvent.Audio -> {
|
||||
if (_status.value != CallStatus.SPEAKING) {
|
||||
_status.value = CallStatus.SPEAKING
|
||||
pipeline.setDucking(true)
|
||||
}
|
||||
pipeline.playChunk(event.data)
|
||||
}
|
||||
|
||||
is ServerEvent.Text -> {
|
||||
_subtitle.value += event.delta
|
||||
}
|
||||
|
||||
is ServerEvent.Transcript -> {
|
||||
if (event.final) {
|
||||
_subtitle.value = event.text
|
||||
}
|
||||
}
|
||||
|
||||
is ServerEvent.Vad -> {
|
||||
_userSpeaking.value = event.speaking
|
||||
if (event.speaking) {
|
||||
_status.value = CallStatus.LISTENING
|
||||
_subtitle.value = ""
|
||||
pipeline.setDucking(false)
|
||||
}
|
||||
}
|
||||
|
||||
is ServerEvent.Interrupted -> {
|
||||
pipeline.flushPlayback()
|
||||
pipeline.setDucking(false)
|
||||
_status.value = CallStatus.LISTENING
|
||||
}
|
||||
|
||||
is ServerEvent.TurnEnd -> {
|
||||
pipeline.setDucking(false)
|
||||
_status.value = CallStatus.LISTENING
|
||||
onTurnEnd?.invoke(event)
|
||||
}
|
||||
|
||||
is ServerEvent.ToolCall -> {
|
||||
if (event.status == "running") {
|
||||
_subtitle.value = "${event.name}…"
|
||||
}
|
||||
}
|
||||
|
||||
is ServerEvent.Backpressure -> {
|
||||
// LiveClient handles throttling internally
|
||||
}
|
||||
|
||||
is ServerEvent.GoAway -> {
|
||||
if (event.reason == "timeout") {
|
||||
endCall()
|
||||
}
|
||||
}
|
||||
|
||||
is ServerEvent.Error -> {
|
||||
onError?.invoke(event.code, event.message)
|
||||
if (event.code in setOf("auth_failed", "restricted_model", "restricted_feature", "budget_exceeded", "concurrent_limit")) {
|
||||
endCall()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Duration timer ───────────────────────────────────────────────────
|
||||
|
||||
private fun startDurationTimer() {
|
||||
durationJob?.cancel()
|
||||
durationJob = scope.launch {
|
||||
while (isActive) {
|
||||
delay(1000)
|
||||
val elapsed = (System.currentTimeMillis() - startTimeMs) / 1000
|
||||
val min = elapsed / 60
|
||||
val sec = elapsed % 60
|
||||
updateNotification(String.format("Anruf · %d:%02d", min, sec))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Notification ─────────────────────────────────────────────────────
|
||||
|
||||
private fun createNotificationChannel() {
|
||||
val channel = NotificationChannel(
|
||||
CHANNEL_ID,
|
||||
getString(R.string.app_name),
|
||||
NotificationManager.IMPORTANCE_LOW,
|
||||
).apply {
|
||||
description = "Live Call"
|
||||
setShowBadge(false)
|
||||
}
|
||||
getSystemService(NotificationManager::class.java)?.createNotificationChannel(channel)
|
||||
}
|
||||
|
||||
private fun buildNotification(text: String): Notification {
|
||||
val openIntent = PendingIntent.getActivity(
|
||||
this, 0,
|
||||
Intent(this, MainActivity::class.java).apply {
|
||||
flags = Intent.FLAG_ACTIVITY_SINGLE_TOP
|
||||
},
|
||||
PendingIntent.FLAG_IMMUTABLE,
|
||||
)
|
||||
|
||||
val endIntent = PendingIntent.getService(
|
||||
this, 1,
|
||||
Intent(this, LiveCallService::class.java).apply { action = ACTION_END },
|
||||
PendingIntent.FLAG_IMMUTABLE,
|
||||
)
|
||||
|
||||
return Notification.Builder(this, CHANNEL_ID)
|
||||
.setSmallIcon(R.drawable.ic_launcher_foreground)
|
||||
.setContentTitle(getString(R.string.app_name))
|
||||
.setContentText(text)
|
||||
.setContentIntent(openIntent)
|
||||
.addAction(Notification.Action.Builder(null, "Beenden", endIntent).build())
|
||||
.setOngoing(true)
|
||||
.build()
|
||||
}
|
||||
|
||||
private fun updateNotification(text: String) {
|
||||
getSystemService(NotificationManager::class.java)
|
||||
?.notify(NOTIFICATION_ID, buildNotification(text))
|
||||
}
|
||||
}
|
||||
|
||||
enum class CallStatus {
|
||||
IDLE,
|
||||
CONNECTING,
|
||||
LISTENING,
|
||||
SPEAKING,
|
||||
}
|
||||
211
app/src/main/java/dev/kaizen/app/live/LiveClient.kt
Normal file
211
app/src/main/java/dev/kaizen/app/live/LiveClient.kt
Normal file
|
|
@ -0,0 +1,211 @@
|
|||
package dev.kaizen.app.live
|
||||
|
||||
import android.util.Log
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.SharedFlow
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import java.util.concurrent.atomic.AtomicReference
|
||||
|
||||
/**
|
||||
* WebSocket client for the Kaizen Live Call protocol.
|
||||
*
|
||||
* Connects to `wss://{baseUrl}/ws/live` with Bearer auth, speaks the
|
||||
* provider-agnostic JSON protocol defined in `lib/live/protocol.ts`.
|
||||
* Handles auto-reconnect with exponential backoff + resume.
|
||||
*
|
||||
* Thread-safe: OkHttp dispatches callbacks on its own threads,
|
||||
* all state mutations go through atomic refs or the coroutine scope.
|
||||
*/
|
||||
class LiveClient(
|
||||
private val baseUrl: String,
|
||||
private val token: String,
|
||||
private val scope: CoroutineScope,
|
||||
) {
|
||||
companion object {
|
||||
private const val TAG = "LiveClient"
|
||||
private const val INITIAL_BACKOFF_MS = 500L
|
||||
private const val MAX_BACKOFF_MS = 8_000L
|
||||
private const val MAX_RECONNECT_ATTEMPTS = 5
|
||||
}
|
||||
|
||||
// ─── Public event stream ──────────────────────────────────────────────
|
||||
|
||||
private val _events = MutableSharedFlow<ServerEvent>(extraBufferCapacity = 64)
|
||||
val events: SharedFlow<ServerEvent> = _events
|
||||
|
||||
// ─── State ────────────────────────────────────────────────────────────
|
||||
|
||||
private val ws = AtomicReference<WebSocket?>(null)
|
||||
private val sessionId = AtomicReference<String?>(null)
|
||||
private val lastSeqId = AtomicInteger(0)
|
||||
private val reconnectAttempts = AtomicInteger(0)
|
||||
private var chunkIntervalMs = 100L
|
||||
|
||||
@Volatile private var active = false
|
||||
@Volatile private var pendingSetup: SetupParams? = null
|
||||
|
||||
private data class SetupParams(
|
||||
val model: String,
|
||||
val voice: String,
|
||||
val conversationId: String?,
|
||||
)
|
||||
|
||||
private val client = OkHttpClient.Builder()
|
||||
.readTimeout(0, TimeUnit.MILLISECONDS)
|
||||
.pingInterval(15, TimeUnit.SECONDS)
|
||||
.connectTimeout(10, TimeUnit.SECONDS)
|
||||
.build()
|
||||
|
||||
// ─── Connect ──────────────────────────────────────────────────────────
|
||||
|
||||
fun connect(model: String, voice: String, conversationId: String?) {
|
||||
active = true
|
||||
pendingSetup = SetupParams(model, voice, conversationId)
|
||||
reconnectAttempts.set(0)
|
||||
openSocket()
|
||||
}
|
||||
|
||||
private fun openSocket() {
|
||||
if (!active) return
|
||||
|
||||
val wsUrl = baseUrl
|
||||
.replace("https://", "wss://")
|
||||
.replace("http://", "ws://") + "/ws/live"
|
||||
|
||||
val request = Request.Builder()
|
||||
.url(wsUrl)
|
||||
.header("Authorization", "Bearer $token")
|
||||
.build()
|
||||
|
||||
val socket = client.newWebSocket(request, Listener())
|
||||
ws.set(socket)
|
||||
}
|
||||
|
||||
// ─── Send ─────────────────────────────────────────────────────────────
|
||||
|
||||
fun sendAudio(base64Pcm: String) {
|
||||
ws.get()?.send(encodeAudio(base64Pcm))
|
||||
}
|
||||
|
||||
fun sendText(text: String) {
|
||||
ws.get()?.send(encodeText(text))
|
||||
}
|
||||
|
||||
fun interrupt() {
|
||||
ws.get()?.send(encodeInterrupt())
|
||||
}
|
||||
|
||||
fun end() {
|
||||
active = false
|
||||
ws.get()?.send(encodeEnd())
|
||||
ws.getAndSet(null)?.close(1000, "end")
|
||||
}
|
||||
|
||||
fun dispose() {
|
||||
active = false
|
||||
ws.getAndSet(null)?.cancel()
|
||||
client.dispatcher.executorService.shutdown()
|
||||
}
|
||||
|
||||
// ─── Reconnect ────────────────────────────────────────────────────────
|
||||
|
||||
private fun scheduleReconnect() {
|
||||
if (!active) return
|
||||
val attempt = reconnectAttempts.incrementAndGet()
|
||||
if (attempt > MAX_RECONNECT_ATTEMPTS) {
|
||||
Log.w(TAG, "Max reconnect attempts reached")
|
||||
scope.launch { _events.emit(ServerEvent.Error("reconnect_failed", "Max reconnect attempts")) }
|
||||
active = false
|
||||
return
|
||||
}
|
||||
|
||||
val backoff = (INITIAL_BACKOFF_MS * (1 shl (attempt - 1))).coerceAtMost(MAX_BACKOFF_MS)
|
||||
Log.d(TAG, "Reconnecting in ${backoff}ms (attempt $attempt)")
|
||||
|
||||
scope.launch {
|
||||
delay(backoff)
|
||||
if (active) openSocket()
|
||||
}
|
||||
}
|
||||
|
||||
// ─── WebSocket Listener ───────────────────────────────────────────────
|
||||
|
||||
private inner class Listener : WebSocketListener() {
|
||||
|
||||
override fun onOpen(webSocket: WebSocket, response: Response) {
|
||||
Log.d(TAG, "WebSocket opened")
|
||||
reconnectAttempts.set(0)
|
||||
|
||||
val sid = sessionId.get()
|
||||
if (sid != null) {
|
||||
webSocket.send(encodeResume(sid, lastSeqId.get()))
|
||||
} else {
|
||||
val setup = pendingSetup ?: return
|
||||
webSocket.send(encodeSetup(setup.model, setup.voice, setup.conversationId))
|
||||
}
|
||||
}
|
||||
|
||||
override fun onMessage(webSocket: WebSocket, text: String) {
|
||||
val event = parseServerEvent(text) ?: return
|
||||
|
||||
when (event) {
|
||||
is ServerEvent.Ready -> {
|
||||
sessionId.set(event.sessionId)
|
||||
}
|
||||
is ServerEvent.Resumed -> {
|
||||
sessionId.set(event.sessionId)
|
||||
}
|
||||
is ServerEvent.Audio -> {
|
||||
lastSeqId.set(maxOf(lastSeqId.get(), event.seq))
|
||||
}
|
||||
is ServerEvent.Text -> {
|
||||
lastSeqId.set(maxOf(lastSeqId.get(), event.seq))
|
||||
}
|
||||
is ServerEvent.Backpressure -> {
|
||||
chunkIntervalMs = if (event.throttleMs > 0) event.throttleMs.toLong() else 100L
|
||||
}
|
||||
is ServerEvent.GoAway -> {
|
||||
if (event.reason == "shutdown") {
|
||||
scope.launch {
|
||||
delay(500)
|
||||
if (active) openSocket()
|
||||
}
|
||||
}
|
||||
}
|
||||
is ServerEvent.Error -> {
|
||||
val terminal = event.code in setOf(
|
||||
"auth_failed", "restricted_model", "restricted_feature",
|
||||
"budget_exceeded", "concurrent_limit",
|
||||
)
|
||||
if (terminal) active = false
|
||||
}
|
||||
else -> {}
|
||||
}
|
||||
|
||||
scope.launch { _events.emit(event) }
|
||||
}
|
||||
|
||||
override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
|
||||
webSocket.close(code, reason)
|
||||
}
|
||||
|
||||
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
|
||||
Log.d(TAG, "WebSocket closed: $code $reason")
|
||||
ws.compareAndSet(webSocket, null)
|
||||
if (active && code != 1000) scheduleReconnect()
|
||||
}
|
||||
|
||||
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
|
||||
Log.w(TAG, "WebSocket failure: ${t.message}")
|
||||
ws.compareAndSet(webSocket, null)
|
||||
if (active) scheduleReconnect()
|
||||
}
|
||||
}
|
||||
}
|
||||
156
app/src/main/java/dev/kaizen/app/live/LiveProtocol.kt
Normal file
156
app/src/main/java/dev/kaizen/app/live/LiveProtocol.kt
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
package dev.kaizen.app.live
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.boolean
|
||||
import kotlinx.serialization.json.booleanOrNull
|
||||
import kotlinx.serialization.json.double
|
||||
import kotlinx.serialization.json.doubleOrNull
|
||||
import kotlinx.serialization.json.int
|
||||
import kotlinx.serialization.json.intOrNull
|
||||
import kotlinx.serialization.json.jsonObject
|
||||
import kotlinx.serialization.json.jsonPrimitive
|
||||
import kotlinx.serialization.json.long
|
||||
import kotlinx.serialization.json.longOrNull
|
||||
|
||||
private val json = Json { ignoreUnknownKeys = true; encodeDefaults = false }
|
||||
|
||||
// ─── Client → Server ──────────────────────────────────────────────────────
|
||||
|
||||
@Serializable
|
||||
data class SetupMessage(
|
||||
val type: String = "setup",
|
||||
val model: String,
|
||||
val voice: String,
|
||||
val conversationId: String? = null,
|
||||
val tools: Boolean? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class ResumeMessage(
|
||||
val type: String = "resume",
|
||||
val sessionId: String,
|
||||
val lastSeqId: Int? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class AudioMessage(val type: String = "audio", val data: String)
|
||||
|
||||
@Serializable
|
||||
data class TextMessage(val type: String = "text", val text: String)
|
||||
|
||||
@Serializable
|
||||
data class SimpleMessage(val type: String)
|
||||
|
||||
fun encodeSetup(model: String, voice: String, conversationId: String?, tools: Boolean? = true): String =
|
||||
json.encodeToString(SetupMessage.serializer(), SetupMessage(model = model, voice = voice, conversationId = conversationId, tools = tools))
|
||||
|
||||
fun encodeResume(sessionId: String, lastSeqId: Int?): String =
|
||||
json.encodeToString(ResumeMessage.serializer(), ResumeMessage(sessionId = sessionId, lastSeqId = lastSeqId))
|
||||
|
||||
fun encodeAudio(base64Pcm: String): String =
|
||||
json.encodeToString(AudioMessage.serializer(), AudioMessage(data = base64Pcm))
|
||||
|
||||
fun encodeText(text: String): String =
|
||||
json.encodeToString(TextMessage.serializer(), TextMessage(text = text))
|
||||
|
||||
fun encodeInterrupt(): String = """{"type":"interrupt"}"""
|
||||
fun encodeEnd(): String = """{"type":"end"}"""
|
||||
|
||||
// ─── Server → Client ──────────────────────────────────────────────────────
|
||||
|
||||
data class LiveUsage(
|
||||
val inputTokens: Int = 0,
|
||||
val outputTokens: Int = 0,
|
||||
val cost: Double? = null,
|
||||
val durationMs: Long = 0,
|
||||
)
|
||||
|
||||
sealed interface ServerEvent {
|
||||
data class Ready(val sessionId: String) : ServerEvent
|
||||
data class Resumed(val sessionId: String) : ServerEvent
|
||||
data class Audio(val data: String, val seq: Int) : ServerEvent
|
||||
data class Text(val delta: String, val seq: Int) : ServerEvent
|
||||
data class Transcript(val text: String, val final: Boolean) : ServerEvent
|
||||
data class Vad(val speaking: Boolean) : ServerEvent
|
||||
data object Interrupted : ServerEvent
|
||||
data class TurnEnd(
|
||||
val usage: LiveUsage,
|
||||
val userText: String,
|
||||
val assistantText: String,
|
||||
val userMessageId: String? = null,
|
||||
val assistantMessageId: String? = null,
|
||||
) : ServerEvent
|
||||
data class ToolCall(val name: String, val status: String) : ServerEvent
|
||||
data class Backpressure(val throttleMs: Int) : ServerEvent
|
||||
data class GoAway(val remainingMs: Int, val reason: String) : ServerEvent
|
||||
data class Error(val code: String, val message: String?) : ServerEvent
|
||||
}
|
||||
|
||||
fun parseServerEvent(raw: String): ServerEvent? {
|
||||
val obj: JsonObject
|
||||
try {
|
||||
obj = json.parseToJsonElement(raw).jsonObject
|
||||
} catch (_: Exception) {
|
||||
return null
|
||||
}
|
||||
|
||||
val type = obj["type"]?.jsonPrimitive?.content ?: return null
|
||||
|
||||
return when (type) {
|
||||
"ready" -> ServerEvent.Ready(
|
||||
sessionId = obj["sessionId"]?.jsonPrimitive?.content ?: return null,
|
||||
)
|
||||
"resumed" -> ServerEvent.Resumed(
|
||||
sessionId = obj["sessionId"]?.jsonPrimitive?.content ?: return null,
|
||||
)
|
||||
"audio" -> ServerEvent.Audio(
|
||||
data = obj["data"]?.jsonPrimitive?.content ?: return null,
|
||||
seq = obj["seq"]?.jsonPrimitive?.intOrNull ?: 0,
|
||||
)
|
||||
"text" -> ServerEvent.Text(
|
||||
delta = obj["delta"]?.jsonPrimitive?.content ?: return null,
|
||||
seq = obj["seq"]?.jsonPrimitive?.intOrNull ?: 0,
|
||||
)
|
||||
"transcript" -> ServerEvent.Transcript(
|
||||
text = obj["text"]?.jsonPrimitive?.content ?: "",
|
||||
final = obj["final"]?.jsonPrimitive?.booleanOrNull ?: false,
|
||||
)
|
||||
"vad" -> ServerEvent.Vad(
|
||||
speaking = obj["speaking"]?.jsonPrimitive?.booleanOrNull ?: false,
|
||||
)
|
||||
"interrupted" -> ServerEvent.Interrupted
|
||||
"turn_end" -> {
|
||||
val u = obj["usage"]?.jsonObject
|
||||
ServerEvent.TurnEnd(
|
||||
usage = LiveUsage(
|
||||
inputTokens = u?.get("inputTokens")?.jsonPrimitive?.intOrNull ?: 0,
|
||||
outputTokens = u?.get("outputTokens")?.jsonPrimitive?.intOrNull ?: 0,
|
||||
cost = u?.get("cost")?.jsonPrimitive?.doubleOrNull,
|
||||
durationMs = u?.get("durationMs")?.jsonPrimitive?.longOrNull ?: 0,
|
||||
),
|
||||
userText = obj["userText"]?.jsonPrimitive?.content ?: "",
|
||||
assistantText = obj["assistantText"]?.jsonPrimitive?.content ?: "",
|
||||
userMessageId = obj["userMessageId"]?.jsonPrimitive?.content,
|
||||
assistantMessageId = obj["assistantMessageId"]?.jsonPrimitive?.content,
|
||||
)
|
||||
}
|
||||
"tool_call" -> ServerEvent.ToolCall(
|
||||
name = obj["name"]?.jsonPrimitive?.content ?: "",
|
||||
status = obj["status"]?.jsonPrimitive?.content ?: "",
|
||||
)
|
||||
"backpressure" -> ServerEvent.Backpressure(
|
||||
throttleMs = obj["throttleMs"]?.jsonPrimitive?.intOrNull ?: 0,
|
||||
)
|
||||
"go_away" -> ServerEvent.GoAway(
|
||||
remainingMs = obj["remainingMs"]?.jsonPrimitive?.intOrNull ?: 0,
|
||||
reason = obj["reason"]?.jsonPrimitive?.content ?: "",
|
||||
)
|
||||
"error" -> ServerEvent.Error(
|
||||
code = obj["code"]?.jsonPrimitive?.content ?: "unknown",
|
||||
message = obj["message"]?.jsonPrimitive?.content,
|
||||
)
|
||||
else -> null
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue