diff --git a/app/src/main/AndroidManifest.xml b/app/src/main/AndroidManifest.xml
index e9a318d..16d35ca 100644
--- a/app/src/main/AndroidManifest.xml
+++ b/app/src/main/AndroidManifest.xml
@@ -11,6 +11,10 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/app/src/main/java/dev/kaizen/app/live/AudioPipeline.kt b/app/src/main/java/dev/kaizen/app/live/AudioPipeline.kt
new file mode 100644
index 0000000..9d626a0
--- /dev/null
+++ b/app/src/main/java/dev/kaizen/app/live/AudioPipeline.kt
@@ -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()
+ }
+}
diff --git a/app/src/main/java/dev/kaizen/app/live/LiveCallService.kt b/app/src/main/java/dev/kaizen/app/live/LiveCallService.kt
new file mode 100644
index 0000000..834b0ca
--- /dev/null
+++ b/app/src/main/java/dev/kaizen/app/live/LiveCallService.kt
@@ -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 = _status
+
+ private val _subtitle = MutableStateFlow("")
+ val subtitle: StateFlow = _subtitle
+
+ private val _amplitude = MutableStateFlow(0f)
+ val amplitude: StateFlow = _amplitude
+
+ private val _userSpeaking = MutableStateFlow(false)
+ val userSpeaking: StateFlow = _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,
+}
diff --git a/app/src/main/java/dev/kaizen/app/live/LiveClient.kt b/app/src/main/java/dev/kaizen/app/live/LiveClient.kt
new file mode 100644
index 0000000..34fd548
--- /dev/null
+++ b/app/src/main/java/dev/kaizen/app/live/LiveClient.kt
@@ -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(extraBufferCapacity = 64)
+ val events: SharedFlow = _events
+
+ // ─── State ────────────────────────────────────────────────────────────
+
+ private val ws = AtomicReference(null)
+ private val sessionId = AtomicReference(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()
+ }
+ }
+}
diff --git a/app/src/main/java/dev/kaizen/app/live/LiveProtocol.kt b/app/src/main/java/dev/kaizen/app/live/LiveProtocol.kt
new file mode 100644
index 0000000..4242f4e
--- /dev/null
+++ b/app/src/main/java/dev/kaizen/app/live/LiveProtocol.kt
@@ -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
+ }
+}