From 65ea3a5656002f19b6b8faef845a2b8ccef95c91 Mon Sep 17 00:00:00 2001 From: Bruno Deanoz Date: Wed, 24 Jun 2026 03:06:48 +0200 Subject: [PATCH] Add Live Call transport layer (P10 foundation) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- app/src/main/AndroidManifest.xml | 9 + .../java/dev/kaizen/app/live/AudioPipeline.kt | 210 ++++++++++++ .../dev/kaizen/app/live/LiveCallService.kt | 304 ++++++++++++++++++ .../java/dev/kaizen/app/live/LiveClient.kt | 211 ++++++++++++ .../java/dev/kaizen/app/live/LiveProtocol.kt | 156 +++++++++ 5 files changed, 890 insertions(+) create mode 100644 app/src/main/java/dev/kaizen/app/live/AudioPipeline.kt create mode 100644 app/src/main/java/dev/kaizen/app/live/LiveCallService.kt create mode 100644 app/src/main/java/dev/kaizen/app/live/LiveClient.kt create mode 100644 app/src/main/java/dev/kaizen/app/live/LiveProtocol.kt 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 + } +}