From 913ee91024e6407b721f8dc43931d649c47412fe Mon Sep 17 00:00:00 2001 From: ARIA Date: Wed, 19 Aug 2026 17:29:39 +0200 Subject: [PATCH] M4: media upload/download/playback (both directions) Gateway plugin: - media.upload (chunked binary) -> size/sha256 verify + MIME re-sniff -> cache_*_from_bytes -> media.upload.ack - media.offer / media.pull (chunked) for agent-sent media, delivery-path security re-checked at pull time - send_* overrides mint media_id and emit media.offer - message.send media_refs resolve to cached inbound media - per-send + per-chunk timeouts so a stalled peer can't starve the rest App (Kotlin CMP): - Protocol: media frame types/payloads/builders - GatewayClient: binary session, uploadMedia (chunked + streaming sha256), pullMedia serialized via Mutex so concurrent offers don't interleave - ChatStore/IrisController: MediaItem, attachments, auto-pull on offer - Platform media: SAF picker, ExoPlayer (audio mini-player + video), image loader, FileProvider document open (Android); AWT-free desktop actuals - ChatScreen: attach button + chips, media rendering, keyboard dismiss on send UI polish: - preserve image aspect ratio (no stretching), cap dominant dimension - adjustResize so only chat content squeezes for the keyboard - clear focus (hide keyboard) on send Docs: media.upload.ack in 04-wire-protocol.md + frames.schema.json + 07-media.md; M4 marked complete in 14-milestones.md. Tests: 17-test tests/gateway/test_android.py suite passes. --- app/androidApp/src/main/AndroidManifest.xml | 12 + .../main/kotlin/dev/iris/app/MainActivity.kt | 2 + .../src/main/res/xml/iris_file_paths.xml | 7 + app/shared/build.gradle.kts | 13 + .../kotlin/iris/platform/AndroidMedia.kt | 218 +++++++++ .../commonMain/kotlin/iris/data/ChatStore.kt | 97 +++- .../kotlin/iris/media/FileSource.kt | 10 + .../kotlin/iris/media/MediaCache.kt | 25 + .../src/commonMain/kotlin/iris/media/Mime.kt | 37 ++ .../commonMain/kotlin/iris/media/Sha256.kt | 9 + .../kotlin/iris/net/GatewayClient.kt | 170 ++++++- .../kotlin/iris/platform/PlatformMedia.kt | 34 ++ .../kotlin/iris/protocol/Protocol.kt | 117 ++++- .../kotlin/iris/state/IrisController.kt | 107 ++++- .../kotlin/iris/ui/screens/ChatScreen.kt | 188 +++++++- .../kotlin/iris/platform/DesktopMedia.kt | 115 +++++ .../kotlin/iris/media/FileSourceJvm.kt | 16 + .../kotlin/iris/media/MediaCacheJvm.kt | 59 +++ .../jvmMain/kotlin/iris/media/MediaUtil.kt | 18 + docs/04-wire-protocol.md | 7 + docs/07-media.md | 7 +- docs/14-milestones.md | 20 +- docs/protocol/frames.schema.json | 3 +- gateway-plugin/adapter.py | 327 ++++++++++++- gateway-plugin/media.py | 448 +++++++++++++++++- gateway-plugin/protocol.py | 55 +++ gateway-plugin/tests/ws_probe.py | 132 +++++- gateway-plugin/ws_server.py | 34 +- 28 files changed, 2234 insertions(+), 53 deletions(-) create mode 100644 app/androidApp/src/main/res/xml/iris_file_paths.xml create mode 100644 app/shared/src/androidMain/kotlin/iris/platform/AndroidMedia.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/media/FileSource.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/media/MediaCache.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/media/Mime.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/media/Sha256.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/platform/PlatformMedia.kt create mode 100644 app/shared/src/desktopMain/kotlin/iris/platform/DesktopMedia.kt create mode 100644 app/shared/src/jvmMain/kotlin/iris/media/FileSourceJvm.kt create mode 100644 app/shared/src/jvmMain/kotlin/iris/media/MediaCacheJvm.kt create mode 100644 app/shared/src/jvmMain/kotlin/iris/media/MediaUtil.kt diff --git a/app/androidApp/src/main/AndroidManifest.xml b/app/androidApp/src/main/AndroidManifest.xml index c49802e..3e7389b 100644 --- a/app/androidApp/src/main/AndroidManifest.xml +++ b/app/androidApp/src/main/AndroidManifest.xml @@ -12,12 +12,24 @@ + + + + + \ No newline at end of file diff --git a/app/androidApp/src/main/kotlin/dev/iris/app/MainActivity.kt b/app/androidApp/src/main/kotlin/dev/iris/app/MainActivity.kt index 6aaba37..65cab68 100644 --- a/app/androidApp/src/main/kotlin/dev/iris/app/MainActivity.kt +++ b/app/androidApp/src/main/kotlin/dev/iris/app/MainActivity.kt @@ -4,11 +4,13 @@ import android.os.Bundle import androidx.activity.ComponentActivity import androidx.activity.compose.setContent import iris.IrisApp +import iris.platform.AndroidEnv import iris.platform.AndroidSecureStore class MainActivity : ComponentActivity() { override fun onCreate(savedInstanceState: Bundle?) { super.onCreate(savedInstanceState) + AndroidEnv.context = applicationContext val store = AndroidSecureStore(applicationContext) setContent { IrisApp(store) diff --git a/app/androidApp/src/main/res/xml/iris_file_paths.xml b/app/androidApp/src/main/res/xml/iris_file_paths.xml new file mode 100644 index 0000000..7c79522 --- /dev/null +++ b/app/androidApp/src/main/res/xml/iris_file_paths.xml @@ -0,0 +1,7 @@ + + + + + + + \ No newline at end of file diff --git a/app/shared/build.gradle.kts b/app/shared/build.gradle.kts index 368198d..15f26e5 100644 --- a/app/shared/build.gradle.kts +++ b/app/shared/build.gradle.kts @@ -16,6 +16,12 @@ kotlin { jvm("desktop") sourceSets { + // Both targets are JVM-based (androidTarget + jvm("desktop")), so + // shared JVM code (File I/O, SHA-256, media cache) lives in jvmMain. + val jvmMain by creating { dependsOn(commonMain.get()) } + val androidMain by getting { dependsOn(jvmMain) } + val desktopMain by getting { dependsOn(jvmMain) } + commonMain.dependencies { implementation("org.jetbrains.compose.runtime:runtime:$composeVersion") implementation("org.jetbrains.compose.foundation:foundation:$composeVersion") @@ -27,6 +33,13 @@ kotlin { // (androidTarget + jvm("desktop")), so it resolves on both. implementation("com.squareup.okhttp3:okhttp:$okhttpVersion") } + // M4: ExoPlayer (Media3) for inline audio/video playback (Android only). + androidMain.dependencies { + implementation("androidx.media3:media3-exoplayer:1.3.1") + implementation("androidx.media3:media3-ui:1.3.1") + // SAF picker (rememberLauncherForActivityResult). + implementation("androidx.activity:activity-compose:1.9.2") + } } } diff --git a/app/shared/src/androidMain/kotlin/iris/platform/AndroidMedia.kt b/app/shared/src/androidMain/kotlin/iris/platform/AndroidMedia.kt new file mode 100644 index 0000000..5402cd3 --- /dev/null +++ b/app/shared/src/androidMain/kotlin/iris/platform/AndroidMedia.kt @@ -0,0 +1,218 @@ +package iris.platform + +import android.content.Context +import android.content.Intent +import android.graphics.BitmapFactory +import android.net.Uri +import android.provider.OpenableColumns +import android.widget.Toast +import androidx.activity.compose.rememberLauncherForActivityResult +import androidx.activity.result.contract.ActivityResultContracts +import androidx.compose.foundation.layout.Box +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.Row +import androidx.compose.foundation.layout.Spacer +import androidx.compose.foundation.layout.fillMaxWidth +import androidx.compose.foundation.layout.height +import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.layout.size +import androidx.compose.material3.Button +import androidx.compose.material3.CircularProgressIndicator +import androidx.compose.material3.IconButton +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.Slider +import androidx.compose.material3.Text +import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect +import androidx.compose.runtime.LaunchedEffect +import androidx.compose.runtime.getValue +import androidx.compose.runtime.mutableFloatStateOf +import androidx.compose.runtime.mutableStateOf +import androidx.compose.runtime.remember +import androidx.compose.runtime.setValue +import androidx.compose.ui.Alignment +import androidx.compose.ui.Modifier +import androidx.compose.ui.graphics.ImageBitmap +import androidx.compose.ui.graphics.asImageBitmap +import androidx.compose.ui.platform.LocalContext +import androidx.compose.ui.unit.dp +import androidx.compose.ui.unit.sp +import androidx.compose.ui.viewinterop.AndroidView +import androidx.core.content.FileProvider +import androidx.core.net.toUri +import androidx.media3.common.MediaItem as M3MediaItem +import androidx.media3.common.Player +import androidx.media3.exoplayer.ExoPlayer +import androidx.media3.ui.PlayerView +import iris.data.MediaItem +import iris.media.extForMime +import iris.protocol.KIND_VIDEO +import java.io.File +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext + +/** App context holder (set in MainActivity.onCreate). */ +object AndroidEnv { + lateinit var context: Context +} + +@Composable +actual fun MediaFilePicker(onPicked: (PickedFile?) -> Unit) { + val context = LocalContext.current + val launcher = rememberLauncherForActivityResult(ActivityResultContracts.GetContent()) { uri -> + if (uri == null) { + onPicked(null) + return@rememberLauncherForActivityResult + } + try { + val mime = context.contentResolver.getType(uri) ?: "application/octet-stream" + val name = queryDisplayName(context, uri) ?: "upload" + val dir = File(context.cacheDir, "uploads").apply { mkdirs() } + val target = File(dir, "${System.currentTimeMillis()}${extForMime(mime)}") + context.contentResolver.openInputStream(uri)!!.use { input -> + target.outputStream().use { output -> input.copyTo(output) } + } + onPicked(PickedFile(target.absolutePath, name, mime, target.length())) + } catch (e: Exception) { + onPicked(null) + } + } + Box(modifier = Modifier.fillMaxWidth(), contentAlignment = Alignment.Center) { + Button(onClick = { launcher.launch("*/*") }) { Text("Pick a file") } + } +} + +private fun queryDisplayName(context: Context, uri: Uri): String? { + context.contentResolver.query(uri, null, null, null, null)?.use { c -> + val idx = c.getColumnIndex(OpenableColumns.DISPLAY_NAME) + if (idx >= 0 && c.moveToFirst()) return c.getString(idx) + } + return null +} + +actual fun mediaCacheBaseDir(): String = + AndroidEnv.context.externalCacheDir?.absolutePath + ?: AndroidEnv.context.cacheDir.absolutePath + +actual suspend fun loadMediaImage(path: String): ImageBitmap? = withContext(Dispatchers.IO) { + try { + BitmapFactory.decodeFile(path)?.asImageBitmap() + } catch (e: Exception) { + null + } +} + +@Composable +actual fun MediaPlayerView(media: MediaItem, modifier: Modifier) { + val path = media.localPath + if (path == null) { + Box(modifier = modifier, contentAlignment = Alignment.Center) { + Row { + CircularProgressIndicator(modifier = Modifier.size(16.dp), strokeWidth = 2.dp) + Spacer(modifier = Modifier.size(8.dp)) + Text("downloading…", fontSize = 12.sp) + } + } + return + } + if (media.kind == KIND_VIDEO) { + VideoPlayerView(path, modifier) + } else { + AudioPlayerView(path, media.filename, modifier) + } +} + +@Composable +private fun VideoPlayerView(path: String, modifier: Modifier) { + AndroidView( + factory = { ctx -> + PlayerView(ctx).apply { + useController = true + val player = ExoPlayer.Builder(ctx).build() + player.setMediaItem(M3MediaItem.fromUri(path.toUri())) + player.prepare() + player.playWhenReady = true + this.player = player + } + }, + update = {}, + onRelease = { view -> view.player?.release() }, + modifier = modifier + .fillMaxWidth() + .height(220.dp), + ) +} + +@Composable +private fun AudioPlayerView(path: String, filename: String, modifier: Modifier) { + val context = LocalContext.current + val player = remember { + ExoPlayer.Builder(context).build().apply { + setMediaItem(M3MediaItem.fromUri(path.toUri())) + prepare() + } + } + DisposableEffect(player) { onDispose { player.release() } } + + var playing by remember { mutableStateOf(false) } + var position by remember { mutableFloatStateOf(0f) } + var duration by remember { mutableFloatStateOf(0f) } + + LaunchedEffect(player) { + while (true) { + val d = player.duration + duration = if (d > 0) d / 1000f else 0f + position = player.currentPosition / 1000f + playing = player.isPlaying + kotlinx.coroutines.delay(500) + } + } + + Row( + modifier = modifier + .fillMaxWidth() + .padding(8.dp), + verticalAlignment = Alignment.CenterVertically, + ) { + IconButton(onClick = { + if (playing) player.pause() else player.play() + }) { + Text(if (playing) "⏸" else "▶️", fontSize = 18.sp) + } + Column(modifier = Modifier.weight(1f).padding(horizontal = 8.dp)) { + Text(filename, fontSize = 12.sp, maxLines = 1) + Slider( + value = if (duration > 0f) (position / duration).coerceIn(0f, 1f) else 0f, + onValueChange = { frac -> + player.seekTo((frac * duration * 1000).toLong()) + }, + modifier = Modifier.fillMaxWidth(), + ) + Text( + "${fmt(position)} / ${fmt(duration)}", + fontSize = 10.sp, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + } +} + +private fun fmt(seconds: Float): String { + val s = seconds.toLong() + return "%d:%02d".format(s / 60, s % 60) +} + +actual fun openDocument(path: String, mime: String) { + val context = AndroidEnv.context + try { + val file = File(path) + val uri = FileProvider.getUriForFile(context, "${context.packageName}.fileprovider", file) + val intent = Intent(Intent.ACTION_VIEW).apply { + setDataAndType(uri, mime) + addFlags(Intent.FLAG_GRANT_READ_URI_PERMISSION) + } + context.startActivity(intent) + } catch (e: Exception) { + Toast.makeText(context, "No app to open ${mime}", Toast.LENGTH_SHORT).show() + } +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt b/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt index 79b01f1..86a64fd 100644 --- a/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt +++ b/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt @@ -2,6 +2,8 @@ package iris.data import iris.protocol.CommentaryPayload import iris.protocol.Frame +import iris.protocol.MediaOfferPayload +import iris.protocol.MediaRef import iris.protocol.MessagePayload import iris.protocol.MessageStartPayload import iris.protocol.MessageStopPayload @@ -9,6 +11,7 @@ import iris.protocol.MessageUpdatePayload import iris.protocol.ROLE_ASSISTANT import iris.protocol.ROLE_USER import iris.protocol.TYPE_COMMENTARY +import iris.protocol.TYPE_MEDIA_OFFER import iris.protocol.TYPE_MESSAGE import iris.protocol.TYPE_MESSAGE_START import iris.protocol.TYPE_MESSAGE_STOP @@ -51,8 +54,23 @@ data class MessageItem( val streaming: Boolean = false, val model: String? = null, val tokens: Int? = null, + val media: List = emptyList(), ) : ChatItem +/** + * A media attachment on a message (M4). [localPath] is null until the file + * has been pulled to the local cache (outbound) — inbound attachments the + * app itself uploaded carry no local path (the agent reads the server copy). + */ +data class MediaItem( + val mediaId: String, + val kind: String, + val mime: String, + val size: Long, + val filename: String, + val localPath: String? = null, +) + /** A structured tool-activity card (spinner until [done]). */ data class ToolItem( override val id: String, @@ -117,10 +135,12 @@ class ChatStore { // ── Optimistic send ─────────────────────────────────────────────────── /** Optimistic add: show the user's message immediately (pending) in [lane]. */ - fun addPending(text: String, lane: String): String { + fun addPending(text: String, lane: String, media: List = emptyList()): String { localSeq++ val id = "local_$localSeq" - updateLane(lane) { it + MessageItem(id = id, role = ROLE_USER, text = text, ts = 0, pending = true) } + updateLane(lane) { + it + MessageItem(id = id, role = ROLE_USER, text = text, ts = 0, pending = true, media = media) + } return id } @@ -138,6 +158,7 @@ class ChatStore { TYPE_TOOL_PROGRESS -> onToolProgress(lane, frame) TYPE_TOOL_END -> onToolEnd(lane, frame) TYPE_COMMENTARY -> onCommentary(lane, frame) + TYPE_MEDIA_OFFER -> onMediaOffer(lane, frame) else -> Unit } } @@ -149,15 +170,17 @@ class ChatStore { updateLane(lane) { list -> val byId = list.indexOfFirst { it.id == p.messageId } if (byId >= 0) { - val updated = (list[byId] as? MessageItem)?.copy( + val cur = list[byId] as? MessageItem ?: return@updateLane list + val updated = cur.copy( text = p.text, reasoning = p.reasoning, pending = false, streaming = false, model = p.model, tokens = p.tokens, - ts = p.ts ?: 0, - ) ?: return@updateLane list + ts = p.ts ?: cur.ts, + media = mergeMedia(cur.media, p.media), + ) list.toMutableList().also { it[byId] = updated } } else if (p.role == ROLE_USER) { // Replace the matching optimistic pending bubble (server echo). @@ -169,23 +192,40 @@ class ChatStore { it[pendingIdx] = MessageItem( id = p.messageId, role = p.role, text = p.text, ts = p.ts ?: 0, reasoning = p.reasoning, + media = mergeMedia((list[pendingIdx] as MessageItem).media, p.media), ) } } else { list + MessageItem( id = p.messageId, role = p.role, text = p.text, ts = p.ts ?: 0, reasoning = p.reasoning, model = p.model, tokens = p.tokens, + media = p.media.map { it.toMediaItem() }, ) } } else { list + MessageItem( id = p.messageId, role = p.role, text = p.text, ts = p.ts ?: 0, reasoning = p.reasoning, model = p.model, tokens = p.tokens, + media = p.media.map { it.toMediaItem() }, ) } } } + /** Merge server media refs into existing items, keeping local paths. */ + private fun mergeMedia(existing: List, incoming: List): List { + if (incoming.isEmpty()) return existing + val byId = existing.associateBy { it.mediaId } + return incoming.map { ref -> + byId[ref.mediaId]?.copy( + kind = ref.kind, mime = ref.mime, size = ref.size, filename = ref.filename, + ) ?: ref.toMediaItem() + } + } + + private fun MediaRef.toMediaItem() = + MediaItem(mediaId = mediaId, kind = kind, mime = mime, size = size, filename = filename) + // ── message.start (open a live streaming bubble) ────────────────────── private fun onMessageStart(lane: String, frame: Frame) { @@ -295,6 +335,53 @@ class ChatStore { } } + // ── M4: media.offer (agent produced media; pull it) ───────────────────── + + /** + * Attach an offered media item to its message (by [MediaOfferPayload.messageId], + * falling back to the lane's last assistant message). The item has no local + * path yet — the controller pulls it and calls [setMediaLocalPath]. + */ + private fun onMediaOffer(lane: String, frame: Frame) { + val p = frame.payloadAs() ?: return + updateLane(lane) { list -> + // Already attached (offer replayed)? Skip. + if (list.any { it is MessageItem && it.media.any { m -> m.mediaId == p.mediaId } }) return@updateLane list + val item = MediaItem( + mediaId = p.mediaId, kind = p.kind, mime = p.mime, + size = p.size, filename = p.filename, + ) + val targetIdx = if (p.messageId != null) { + list.indexOfFirst { it.id == p.messageId } + } else { + list.indexOfLast { it is MessageItem && it.role == ROLE_ASSISTANT } + } + if (targetIdx < 0) return@updateLane list + val cur = list[targetIdx] as? MessageItem ?: return@updateLane list + list.toMutableList().also { it[targetIdx] = cur.copy(media = cur.media + item) } + } + } + + /** Record a pulled file's local path for [mediaId] (all lanes). */ + fun setMediaLocalPath(mediaId: String, localPath: String) { + val map = _lanes.value.toMutableMap() + var changed = false + for ((lane, list) in map) { + val updated = list.map { item -> + if (item is MessageItem) { + item.copy(media = item.media.map { m -> + if (m.mediaId == mediaId) m.copy(localPath = localPath) else m + }) + } else item + } + if (updated != list) { + map[lane] = updated + changed = true + } + } + if (changed) _lanes.value = map + } + fun clear() { _lanes.value = emptyMap() } diff --git a/app/shared/src/commonMain/kotlin/iris/media/FileSource.kt b/app/shared/src/commonMain/kotlin/iris/media/FileSource.kt new file mode 100644 index 0000000..8d53806 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/media/FileSource.kt @@ -0,0 +1,10 @@ +package iris.media + +/** A readable local file (expect/actual; JVM impl in jvmMain). */ +expect class FileSource(path: String) : AutoCloseable { + /** Total size in bytes. */ + fun size(): Long + + /** Read up to [buf.size] bytes into [buf]; returns bytes read or -1 at EOF. */ + fun read(buf: ByteArray): Int +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/media/MediaCache.kt b/app/shared/src/commonMain/kotlin/iris/media/MediaCache.kt new file mode 100644 index 0000000..f307c6c --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/media/MediaCache.kt @@ -0,0 +1,25 @@ +package iris.media + +/** + * Local media cache (docs/07 §7.6). Pulled media is written under the + * app-specific cache dir; LRU eviction by total size keeps it bounded + * (default 500 MB). + */ +expect class MediaCache(baseDir: String) { + /** Path for a cached media id (null when not cached). */ + fun path(mediaId: String, mime: String): String? + + /** Open a writer for [mediaId]; [MediaWriter.path] is the final location. */ + fun openWriter(mediaId: String, mime: String): MediaWriter + + /** Delete a cached file (e.g. after a failed pull). */ + fun remove(mediaId: String, mime: String) +} + +/** Streaming writer for a cache entry (write chunks, then close). */ +expect interface MediaWriter : AutoCloseable { + /** Final file path (valid after [close]). */ + val path: String + + fun write(bytes: ByteArray) +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/media/Mime.kt b/app/shared/src/commonMain/kotlin/iris/media/Mime.kt new file mode 100644 index 0000000..3c156e9 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/media/Mime.kt @@ -0,0 +1,37 @@ +package iris.media + +import iris.protocol.KIND_AUDIO +import iris.protocol.KIND_DOCUMENT +import iris.protocol.KIND_IMAGE +import iris.protocol.KIND_VIDEO + +/** Map a MIME type to a media kind (docs/07 §7.1). */ +fun kindFromMime(mime: String): String = when { + mime.startsWith("image/") -> KIND_IMAGE + mime.startsWith("video/") -> KIND_VIDEO + mime.startsWith("audio/") -> KIND_AUDIO + else -> KIND_DOCUMENT +} + +/** Best-effort file extension for a MIME type (cache file naming). */ +fun extForMime(mime: String): String = when { + mime == "image/jpeg" -> ".jpg" + mime == "image/png" -> ".png" + mime == "image/webp" -> ".webp" + mime == "image/gif" -> ".gif" + mime == "image/heic" -> ".heic" + mime == "image/heif" -> ".heif" + mime == "video/mp4" -> ".mp4" + mime == "video/webm" -> ".webm" + mime == "video/quicktime" -> ".mov" + mime == "audio/mpeg" -> ".mp3" + mime == "audio/mp4" || mime == "audio/x-m4a" -> ".m4a" + mime == "audio/ogg" -> ".ogg" + mime == "audio/wav" -> ".wav" + mime == "audio/flac" -> ".flac" + mime == "audio/aac" -> ".aac" + mime == "application/pdf" -> ".pdf" + mime == "application/zip" -> ".zip" + mime == "text/plain" -> ".txt" + else -> ".bin" +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/media/Sha256.kt b/app/shared/src/commonMain/kotlin/iris/media/Sha256.kt new file mode 100644 index 0000000..634333e --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/media/Sha256.kt @@ -0,0 +1,9 @@ +package iris.media + +/** Streaming SHA-256 for uploads (hash while chunking, no full-file copy). */ +expect class Sha256 constructor() { + fun update(bytes: ByteArray, offset: Int = 0, length: Int = bytes.size - offset) + + /** Lowercase hex digest of everything fed so far. */ + fun hex(): String +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt b/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt index 3dacb4b..08bb941 100644 --- a/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt +++ b/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt @@ -1,21 +1,31 @@ package iris.net import iris.data.SecureStore +import iris.media.FileSource +import iris.media.Sha256 import iris.protocol.ChannelInfo import iris.protocol.ErrorPayload import iris.protocol.Frame import iris.protocol.HelloAckPayload import iris.protocol.IrisJson +import iris.protocol.MediaPullEndPayload +import iris.protocol.MediaUploadAckPayload import iris.protocol.ServerCaps import iris.protocol.TYPE_ERROR import iris.protocol.TYPE_HELLO_ACK +import iris.protocol.TYPE_MEDIA_PULL_END +import iris.protocol.TYPE_MEDIA_UPLOAD_ACK import iris.protocol.TYPE_PONG import iris.protocol.helloFrame +import iris.protocol.mediaPullFrame +import iris.protocol.mediaUploadEndFrame +import iris.protocol.mediaUploadStartFrame import iris.protocol.messageSendFrame import iris.protocol.pingFrame import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay import kotlinx.coroutines.isActive @@ -26,10 +36,15 @@ import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withTimeout import kotlinx.coroutines.withTimeoutOrNull import kotlin.time.TimeMark import kotlin.time.TimeSource import okhttp3.OkHttpClient +import okio.ByteString +import okio.ByteString.Companion.toByteString import okhttp3.Request import okhttp3.Response import okhttp3.WebSocket @@ -75,6 +90,23 @@ class GatewayClient( private var lastLiveness: TimeMark = TimeSource.Monotonic.markNow() private val pending = mutableMapOf>() + // M4: binary frames (media upload chunks / pull stream) have no per-frame + // id, so at most one binary session is active per socket. The gateway + // allows one upload per connection; pull is request/response. + private sealed interface BinarySession { + data class Pulling( + val requestId: Int, + val chunks: Channel, + val end: CompletableDeferred, + ) : BinarySession + } + + private var binarySession: BinarySession? = null + + // M4: only one pull may be in flight at a time (binarySession is a single + // slot). Serialize concurrent offers so their byte streams don't interleave. + private val pullMutex = Mutex() + // ── Lifecycle ───────────────────────────────────────────────────────── /** Start (or resume) the connect loop. Idempotent. */ @@ -177,11 +209,30 @@ class GatewayClient( TYPE_PONG -> Unit else -> { _events.tryEmit(frame) - frame.id?.let { pending[it]?.complete(frame) } + frame.id?.let { id -> + pending[id]?.complete(frame) + // M4: terminal frame of a pull stream — close + // the chunk channel so the pull loop exits. + if (frame.type == TYPE_MEDIA_PULL_END) { + (binarySession as? BinarySession.Pulling)?.let { + it.chunks.close() + binarySession = null + } + } + } } } } + override fun onMessage(webSocket: WebSocket, bytes: ByteString) { + lastLiveness = TimeSource.Monotonic.markNow() + // M4: binary frames belong to the active pull stream + // (uploads are outbound; stray inbound chunks are dropped). + (binarySession as? BinarySession.Pulling) + ?.chunks + ?.trySend(bytes.toByteArray()) + } + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { closed.complete(Unit) } @@ -215,11 +266,122 @@ class GatewayClient( // ── Outbound ────────────────────────────────────────────────────────── - /** Send a text message (fire-and-forget; the server echoes it back). */ - fun sendMessage(chatId: String, text: String, threadId: String? = null) { + /** Send a text message (fire-and-forget; the server echoes it back). + * M4: [mediaRefs] reference completed uploads (media.upload.ack refs). */ + fun sendMessage( + chatId: String, + text: String, + threadId: String? = null, + mediaRefs: List = emptyList(), + ) { val ws = socket ?: return val id = nextRequestId++ - ws.send(messageSendFrame(id, chatId, text, threadId).toWire()) + ws.send(messageSendFrame(id, chatId, text, threadId, mediaRefs).toWire()) + } + + // ── M4: media upload / pull ─────────────────────────────────────────── + + /** + * Upload a local file as media (docs/07 §7.2): media.upload.start, + * 256 KiB binary chunks, media.upload.end {sha256}. Returns the server's + * media_ref (for message.send media_refs) on success. + */ + suspend fun uploadMedia( + path: String, + mime: String, + kind: String, + filename: String, + mediaRef: String, + ): Result { + val ws = socket ?: return Result.failure(IllegalStateException("not connected")) + val source = FileSource(path) + val size = source.size() + if (size <= 0) { + source.close() + return Result.failure(IllegalStateException("empty file")) + } + val id = nextRequestId++ + val reply = CompletableDeferred() + pending[id] = reply + try { + ws.send(mediaUploadStartFrame(id, mediaRef, kind, mime, filename, size).toWire()) + val sha = Sha256() + source.use { + val buf = ByteArray(UPLOAD_CHUNK_BYTES) + while (true) { + val n = it.read(buf) + if (n < 0) break + if (n == 0) continue + sha.update(buf, 0, n) + ws.send(buf.copyOfRange(0, n).toByteString()) + } + } + ws.send(mediaUploadEndFrame(id, mediaRef, sha.hex()).toWire()) + val frame = withTimeout(UPLOAD_TIMEOUT_MS) { reply.await() } + return when (frame.type) { + TYPE_MEDIA_UPLOAD_ACK -> { + val p = frame.payloadAs() + if (p != null && p.ok) Result.success(p.mediaRef) + else Result.failure(IllegalStateException("upload rejected by server")) + } + TYPE_ERROR -> { + val e = frame.payloadAs() + Result.failure(IllegalStateException(e?.message ?: "upload failed")) + } + else -> Result.failure(IllegalStateException("unexpected reply ${frame.type}")) + } + } catch (e: Exception) { + return Result.failure(e) + } finally { + pending.remove(id) + } + } + + /** + * Pull offered media (docs/07 §7.3): media.pull, then binary frames until + * media.pull.end. Each chunk is handed to [onChunk] (write to cache). + */ + suspend fun pullMedia(mediaId: String, onChunk: suspend (ByteArray) -> Unit): Result = + pullMutex.withLock { + val ws = socket ?: return@withLock Result.failure(IllegalStateException("not connected")) + val id = nextRequestId++ + val chunks = Channel(Channel.UNLIMITED) + val end = CompletableDeferred() + pending[id] = end + binarySession = BinarySession.Pulling(id, chunks, end) + try { + ws.send(mediaPullFrame(id, mediaId).toWire()) + val frame = withTimeout(PULL_TIMEOUT_MS) { + for (chunk in chunks) onChunk(chunk) + end.await() + } + when (frame.type) { + TYPE_MEDIA_PULL_END -> { + val p = frame.payloadAs() + if (p != null && p.ok) Result.success(Unit) + else Result.failure(IllegalStateException("pull failed")) + } + TYPE_ERROR -> { + val e = frame.payloadAs() + Result.failure(IllegalStateException(e?.message ?: "pull failed")) + } + else -> Result.failure(IllegalStateException("unexpected reply ${frame.type}")) + } + } catch (e: Exception) { + Result.failure(e) + } finally { + pending.remove(id) + chunks.cancel() + val s = binarySession + if (s is BinarySession.Pulling && s.requestId == id) binarySession = null + } + } + + companion object { + /** One WS binary frame carries at most this many media bytes (docs/07 §7.5). */ + const val UPLOAD_CHUNK_BYTES = 256 * 1024 + const val UPLOAD_TIMEOUT_MS = 120_000L + const val PULL_TIMEOUT_MS = 300_000L } /** diff --git a/app/shared/src/commonMain/kotlin/iris/platform/PlatformMedia.kt b/app/shared/src/commonMain/kotlin/iris/platform/PlatformMedia.kt new file mode 100644 index 0000000..dc34964 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/platform/PlatformMedia.kt @@ -0,0 +1,34 @@ +package iris.platform + +import androidx.compose.runtime.Composable +import androidx.compose.ui.Modifier +import androidx.compose.ui.graphics.ImageBitmap +import iris.data.MediaItem + +/** A file picked by the user, staged as a local file the app can read. */ +data class PickedFile( + val path: String, + val name: String, + val mime: String, + val size: Long, +) + +/** + * File picker (SAF on Android, FileChooser on desktop). Invokes [onPicked] + * with the staged file, or null when the user cancels. + */ +@Composable +expect fun MediaFilePicker(onPicked: (PickedFile?) -> Unit) + +/** Base directory for the media cache (app-specific cache dir). */ +expect fun mediaCacheBaseDir(): String + +/** Load an image from a local path for display. Null on failure. */ +expect suspend fun loadMediaImage(path: String): ImageBitmap? + +/** Inline audio/video player (ExoPlayer on Android; placeholder on desktop). */ +@Composable +expect fun MediaPlayerView(media: MediaItem, modifier: Modifier) + +/** Open a document with the system handler (best effort). */ +expect fun openDocument(path: String, mime: String) \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt b/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt index 4a2c937..ffd6b45 100644 --- a/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt +++ b/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt @@ -46,6 +46,14 @@ const val TYPE_TOOL_PROGRESS = "tool.progress" const val TYPE_TOOL_END = "tool.end" const val TYPE_COMMENTARY = "commentary" +// M4 — media (upload / offer / pull) +const val TYPE_MEDIA_UPLOAD_START = "media.upload.start" +const val TYPE_MEDIA_UPLOAD_END = "media.upload.end" +const val TYPE_MEDIA_UPLOAD_ACK = "media.upload.ack" +const val TYPE_MEDIA_OFFER = "media.offer" +const val TYPE_MEDIA_PULL = "media.pull" +const val TYPE_MEDIA_PULL_END = "media.pull.end" + // M3 — channels / threads / search / sync const val TYPE_CHANNEL_CREATE = "channel.create" const val TYPE_CHANNEL_RENAME = "channel.rename" @@ -66,6 +74,14 @@ const val ERR_AUTH = "auth" const val ERR_NOT_FOUND = "not_found" const val ERR_UNSUPPORTED = "unsupported" const val ERR_INTERNAL = "internal" +const val ERR_MEDIA_TOO_LARGE = "media_too_large" + +// M4 — media kinds (docs/07 §7.1) +const val KIND_IMAGE = "image" +const val KIND_AUDIO = "audio" +const val KIND_VIDEO = "video" +const val KIND_DOCUMENT = "document" +const val KIND_VOICE = "voice" // ── Roles ─────────────────────────────────────────────────────────────── @@ -147,8 +163,60 @@ data class MessagePayload( val model: String? = null, val tokens: Int? = null, val ts: Long? = null, + val media: List = emptyList(), ) +// ── M4: media ──────────────────────────────────────────────────────────── + +/** A media attachment referenced by a message (or offered for pull). */ +@Serializable +data class MediaRef( + @SerialName("media_id") val mediaId: String, + val kind: String, + val mime: String, + val size: Long, + val filename: String, +) + +@Serializable +data class MediaUploadStartPayload( + @SerialName("media_ref") val mediaRef: String, + val kind: String, + val mime: String, + val filename: String, + val size: Long, +) + +@Serializable +data class MediaUploadEndPayload( + @SerialName("media_ref") val mediaRef: String, + @SerialName("sha256") val sha256: String, +) + +@Serializable +data class MediaUploadAckPayload( + val ok: Boolean, + @SerialName("media_ref") val mediaRef: String, +) + +@Serializable +data class MediaOfferPayload( + @SerialName("media_id") val mediaId: String, + val kind: String, + val mime: String, + val size: Long, + val filename: String, + @SerialName("message_id") val messageId: String? = null, +) + +@Serializable +data class MediaPullPayload( + @SerialName("media_id") val mediaId: String, +) + +@Serializable +data class MediaPullEndPayload(val ok: Boolean) + // ── M2: streaming frames (server -> app) ──────────────────────────────── @Serializable @@ -292,7 +360,13 @@ fun helloFrame(token: String, deviceId: String, deviceName: String): Frame = ), ) -fun messageSendFrame(id: Int, chatId: String, text: String, threadId: String? = null): Frame = +fun messageSendFrame( + id: Int, + chatId: String, + text: String, + threadId: String? = null, + mediaRefs: List = emptyList(), +): Frame = Frame( id = id, type = TYPE_MESSAGE_SEND, @@ -300,7 +374,7 @@ fun messageSendFrame(id: Int, chatId: String, text: String, threadId: String? = threadId = threadId, payload = IrisJson.instance.encodeToJsonElement( MessageSendPayload.serializer(), - MessageSendPayload(text = text), + MessageSendPayload(text = text, mediaRefs = mediaRefs), ), ) @@ -365,4 +439,43 @@ fun syncFrame(id: Int, cursor: Long): Frame = SyncPayload.serializer(), SyncPayload(cursor = cursor), ), + ) + +// ── M4 frame builders ──────────────────────────────────────────────────── + +fun mediaUploadStartFrame( + id: Int, + mediaRef: String, + kind: String, + mime: String, + filename: String, + size: Long, +): Frame = + Frame( + id = id, + type = TYPE_MEDIA_UPLOAD_START, + payload = IrisJson.instance.encodeToJsonElement( + MediaUploadStartPayload.serializer(), + MediaUploadStartPayload(mediaRef, kind, mime, filename, size), + ), + ) + +fun mediaUploadEndFrame(id: Int, mediaRef: String, sha256: String): Frame = + Frame( + id = id, + type = TYPE_MEDIA_UPLOAD_END, + payload = IrisJson.instance.encodeToJsonElement( + MediaUploadEndPayload.serializer(), + MediaUploadEndPayload(mediaRef, sha256), + ), + ) + +fun mediaPullFrame(id: Int, mediaId: String): Frame = + Frame( + id = id, + type = TYPE_MEDIA_PULL, + payload = IrisJson.instance.encodeToJsonElement( + MediaPullPayload.serializer(), + MediaPullPayload(mediaId), + ), ) \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt b/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt index 275f11e..1c4dd32 100644 --- a/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt +++ b/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt @@ -2,8 +2,14 @@ package iris.state import iris.data.ChatStore import iris.data.ChannelStore +import iris.data.MediaItem import iris.data.SecureStore +import iris.media.MediaCache +import iris.media.kindFromMime import iris.net.GatewayClient +import iris.platform.PickedFile +import iris.platform.mediaCacheBaseDir +import iris.protocol.MediaOfferPayload import iris.protocol.SearchHit import iris.protocol.SearchResultsPayload import iris.protocol.SyncDonePayload @@ -12,6 +18,7 @@ import iris.protocol.TYPE_CHANNEL_DELETED import iris.protocol.TYPE_CHANNEL_LIST import iris.protocol.TYPE_CHANNEL_RENAMED import iris.protocol.TYPE_COMMENTARY +import iris.protocol.TYPE_MEDIA_OFFER import iris.protocol.TYPE_MESSAGE import iris.protocol.TYPE_MESSAGE_START import iris.protocol.TYPE_MESSAGE_STOP @@ -37,6 +44,7 @@ import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.launch +import kotlin.random.Random /** * App-level controller (M3): owns the GatewayClient + ChatStore + ChannelStore, @@ -88,6 +96,24 @@ class IrisController( private val _lastQuery = MutableStateFlow("") val lastQuery: StateFlow = _lastQuery.asStateFlow() + // ── M4: media ───────────────────────────────────────────────────────── + private val mediaCache = MediaCache(mediaCacheBaseDir()) + + /** A composer attachment: picked file being uploaded (or uploaded). */ + data class PendingAttachment( + val filename: String, + val mime: String, + val size: Long, + val kind: String, + val path: String? = null, + val uploading: Boolean = true, + val mediaRef: String? = null, + val error: String? = null, + ) + + private val _attachments = MutableStateFlow>(emptyList()) + val attachments: StateFlow> = _attachments.asStateFlow() + init { scope.launch { client.events.collect { frame -> @@ -99,7 +125,14 @@ class IrisController( TYPE_TOOL_START, TYPE_TOOL_PROGRESS, TYPE_TOOL_END, - TYPE_COMMENTARY -> chat.onFrame(frame) + TYPE_COMMENTARY, + TYPE_MEDIA_OFFER -> { + chat.onFrame(frame) + // M4: pull offered media into the local cache. + if (frame.type == TYPE_MEDIA_OFFER) { + frame.payloadAs()?.let { pullMedia(it) } + } + } TYPE_CHANNEL_CREATED, TYPE_CHANNEL_RENAMED, TYPE_CHANNEL_DELETED, @@ -202,14 +235,76 @@ class IrisController( client.sendFrame(syncFrame(0, cursor)) } - /** Optimistic send: show immediately in the current lane, then hand to the gateway. */ - fun send(text: String) { + /** Optimistic send: show immediately in the current lane, then hand to the gateway. + * M4: [attachments] (uploaded) are attached via media_refs. */ + fun send(text: String, attachments: List = emptyList()) { val trimmed = text.trim() - if (trimmed.isEmpty()) return + val refs = attachments.mapNotNull { it.mediaRef } + if (trimmed.isEmpty() && refs.isEmpty()) return val lane = chat.currentLane.value val (chatId, threadId) = chat.parseLane(lane) - chat.addPending(trimmed, lane) - client.sendMessage(chatId, trimmed, threadId) + val media = attachments.map { + MediaItem( + mediaId = it.mediaRef ?: it.filename, + kind = it.kind, mime = it.mime, size = it.size, filename = it.filename, + localPath = it.path, + ) + } + chat.addPending(trimmed, lane, media) + client.sendMessage(chatId, trimmed, threadId, refs) + _attachments.value = emptyList() + } + + // ── M4: attachments (pick -> upload -> send) ────────────────────────── + + /** Stage a picked file: upload it, then keep it as a pending attachment. */ + fun attachFile(picked: PickedFile) { + val kind = kindFromMime(picked.mime) + val placeholder = PendingAttachment( + filename = picked.name, mime = picked.mime, size = picked.size, kind = kind, + path = picked.path, + ) + _attachments.value = _attachments.value + placeholder + scope.launch { + val mediaRef = "up_${Random.nextLong(1_000_000_000L, 9_999_999_999L)}" + val result = client.uploadMedia( + path = picked.path, + mime = picked.mime, + kind = kind, + filename = picked.name, + mediaRef = mediaRef, + ) + _attachments.value = _attachments.value.map { + if (it.filename == picked.name && it.uploading) { + result.fold( + { ref -> it.copy(uploading = false, mediaRef = ref) }, + { e -> it.copy(uploading = false, error = e.message) }, + ) + } else it + } + } + } + + fun removeAttachment(filename: String) { + _attachments.value = _attachments.value.filterNot { it.filename == filename } + } + + /** Pull offered media into the local cache and record the path. */ + private fun pullMedia(offer: MediaOfferPayload) { + scope.launch { + // Already cached? Skip the pull. + mediaCache.path(offer.mediaId, offer.mime)?.let { + chat.setMediaLocalPath(offer.mediaId, it) + return@launch + } + val writer = mediaCache.openWriter(offer.mediaId, offer.mime) + val result = client.pullMedia(offer.mediaId) { chunk -> writer.write(chunk) } + writer.close() + result.fold( + { chat.setMediaLocalPath(offer.mediaId, writer.path) }, + { mediaCache.remove(offer.mediaId, offer.mime) }, + ) + } } /** Connect-screen action: real hello test, then save + (re)connect. */ diff --git a/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt b/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt index 5ce48e0..460f6c2 100644 --- a/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt +++ b/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt @@ -16,6 +16,8 @@ import androidx.compose.foundation.layout.heightIn import androidx.compose.foundation.layout.padding import androidx.compose.foundation.layout.size import androidx.compose.foundation.layout.width +import androidx.compose.foundation.Image +import androidx.compose.foundation.layout.fillMaxHeight import androidx.compose.foundation.lazy.LazyColumn import androidx.compose.foundation.lazy.items import androidx.compose.foundation.lazy.rememberLazyListState @@ -47,7 +49,10 @@ import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.draw.clip import androidx.compose.ui.graphics.Color +import androidx.compose.ui.graphics.ImageBitmap +import androidx.compose.ui.layout.ContentScale import androidx.compose.ui.platform.LocalClipboardManager +import androidx.compose.ui.platform.LocalFocusManager import androidx.compose.ui.text.AnnotatedString import androidx.compose.ui.text.font.FontFamily import androidx.compose.ui.text.font.FontWeight @@ -56,10 +61,17 @@ import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.sp import iris.data.ChannelStore import iris.data.ChatItem +import iris.data.MediaItem import iris.data.MessageItem import iris.data.ToolItem import iris.net.GatewayClient +import iris.platform.MediaFilePicker +import iris.platform.MediaPlayerView +import iris.platform.loadMediaImage +import iris.platform.openDocument import iris.protocol.ChannelInfo +import iris.protocol.KIND_DOCUMENT +import iris.protocol.KIND_IMAGE import iris.protocol.ROLE_USER import iris.protocol.SearchHit import iris.state.IrisController @@ -91,14 +103,20 @@ fun ChatScreen(controller: IrisController) { var showSearch by remember { mutableStateOf(false) } var showNewChannel by remember { mutableStateOf(false) } var showNewThread by remember { mutableStateOf(false) } + var showPicker by remember { mutableStateOf(false) } + val attachments by controller.attachments.collectAsState() val drawerState = rememberDrawerState(DrawerValue.Closed) val drawerScope = rememberCoroutineScope() + val focusManager = LocalFocusManager.current fun doSend() { - if (input.isBlank()) return + val ready = attachments.filter { it.mediaRef != null && it.error == null } + if (input.isBlank() && ready.isEmpty()) return val text = input input = "" - controller.send(text) + controller.send(text, ready) + // Best practice: dismiss the keyboard after sending. + focusManager.clearFocus(force = true) } LaunchedEffect(items.size, typing) { @@ -206,6 +224,21 @@ fun ChatScreen(controller: IrisController) { } } + // Pending attachments (M4) + if (attachments.isNotEmpty()) { + Row( + modifier = Modifier + .fillMaxWidth() + .horizontalScroll(rememberScrollState()) + .padding(horizontal = 12.dp, vertical = 4.dp), + horizontalArrangement = Arrangement.spacedBy(6.dp), + ) { + attachments.forEach { att -> + AttachmentChip(att, onRemove = { controller.removeAttachment(att.filename) }) + } + } + } + // Composer Row( modifier = Modifier @@ -213,6 +246,9 @@ fun ChatScreen(controller: IrisController) { .padding(12.dp), verticalAlignment = Alignment.Bottom, ) { + IconButton(onClick = { showPicker = true }) { + Text("📎", fontSize = 18.sp) + } OutlinedTextField( value = input, onValueChange = { input = it }, @@ -226,7 +262,7 @@ fun ChatScreen(controller: IrisController) { Spacer(modifier = Modifier.width(8.dp)) Button( onClick = { doSend() }, - enabled = input.isNotBlank(), + enabled = input.isNotBlank() || attachments.any { it.mediaRef != null && it.error == null }, ) { Text("Send") } @@ -234,6 +270,16 @@ fun ChatScreen(controller: IrisController) { } } + // File picker overlay (M4) + if (showPicker) { + Box(modifier = Modifier.fillMaxSize().background(Color.Black.copy(alpha = 0.5f))) { + MediaFilePicker(onPicked = { picked -> + showPicker = false + if (picked != null) controller.attachFile(picked) + }) + } + } + // Search overlay if (showSearch) { SearchOverlay( @@ -520,6 +566,13 @@ private fun MessageBubble(msg: MessageItem) { fontSize = if (isCommentary) 13.sp else 15.sp, ) } + // M4: media attachments (image / player / document chip). + if (msg.media.isNotEmpty()) { + Spacer(modifier = Modifier.height(6.dp)) + Column(verticalArrangement = Arrangement.spacedBy(6.dp)) { + msg.media.forEach { m -> MediaAttachment(m) } + } + } if (msg.pending) { Text("sending…", color = textColor.copy(alpha = 0.6f), fontSize = 10.sp) } @@ -542,6 +595,135 @@ private fun MessageBubble(msg: MessageItem) { } } +/** + * M4: render one media attachment in a bubble. + * - image: tappable thumbnail (full-screen viewer) + * - audio/video: inline player (ExoPlayer on Android) + * - document: filename chip (tap to open with system handler) + */ +@Composable +private fun MediaAttachment(media: MediaItem) { + when (media.kind) { + KIND_IMAGE -> MediaImage(media) + KIND_DOCUMENT -> MediaDocChip(media) + else -> MediaPlayerView(media, Modifier.fillMaxWidth()) + } +} + +@Composable +private fun MediaImage(media: MediaItem) { + var bitmap by remember(media.localPath) { mutableStateOf(null) } + var loading by remember(media.localPath) { mutableStateOf(media.localPath != null) } + var showFull by remember { mutableStateOf(false) } + + LaunchedEffect(media.localPath) { + val path = media.localPath + if (path != null) { + bitmap = loadMediaImage(path) + loading = false + } + } + + if (loading || bitmap == null) { + Box( + modifier = Modifier + .size(120.dp) + .clip(RoundedCornerShape(10.dp)) + .background(Color(0xFF20242E)), + contentAlignment = Alignment.Center, + ) { + if (media.localPath == null) Text("downloading…", fontSize = 11.sp) + else CircularProgressIndicator(modifier = Modifier.size(20.dp), strokeWidth = 2.dp) + } + return + } + val bmp = bitmap!! + val aspect = bmp.width.toFloat() / bmp.height.toFloat() + // Preserve the image's aspect ratio, capping the dominant dimension at + // 260.dp so a chat image never dominates the screen (the other dimension + // follows the ratio). + val (imgW, imgH) = if (aspect >= 1f) 260.dp to (260 / aspect).dp + else (260 * aspect).dp to 260.dp + Image( + bitmap = bmp, + contentDescription = media.filename, + contentScale = ContentScale.FillBounds, + modifier = Modifier + .width(imgW) + .height(imgH) + .clip(RoundedCornerShape(10.dp)) + .clickable { showFull = true }, + ) + if (showFull) { + Box( + modifier = Modifier + .fillMaxSize() + .background(Color.Black) + .clickable { showFull = false }, + contentAlignment = Alignment.Center, + ) { + Image( + bitmap = bitmap!!, + contentDescription = media.filename, + contentScale = ContentScale.Fit, + modifier = Modifier + .fillMaxWidth() + .fillMaxHeight(0.9f), + ) + } + } +} + +@Composable +private fun MediaDocChip(media: MediaItem) { + Row( + modifier = Modifier + .clip(RoundedCornerShape(10.dp)) + .background(Color(0xFF20242E)) + .clickable { + media.localPath?.let { openDocument(it, media.mime) } + } + .padding(horizontal = 10.dp, vertical = 8.dp), + verticalAlignment = Alignment.CenterVertically, + ) { + Text("📄", fontSize = 16.sp) + Spacer(modifier = Modifier.width(8.dp)) + Column { + Text(media.filename, fontSize = 13.sp, maxLines = 1) + Text(fmtSize(media.size), fontSize = 10.sp, color = Color(0xFF8A93A6)) + } + } +} + +private fun fmtSize(bytes: Long): String = when { + bytes >= 1_048_576 -> "%.1f MB".format(bytes / 1_048_576.0) + bytes >= 1024 -> "%.0f KB".format(bytes / 1024.0) + else -> "$bytes B" +} + +/** M4: composer attachment chip (uploading / ready / error). */ +@Composable +private fun AttachmentChip(att: IrisController.PendingAttachment, onRemove: () -> Unit) { + val (icon, label) = when { + att.error != null -> "⚠️" to "${att.filename} — ${att.error}" + att.uploading -> "⏳" to "${att.filename} (${fmtSize(att.size)})" + else -> "📎" to "${att.filename} (${fmtSize(att.size)})" + } + Row( + modifier = Modifier + .clip(RoundedCornerShape(12.dp)) + .background(Color(0xFF2A2E3B)) + .padding(horizontal = 8.dp, vertical = 5.dp), + verticalAlignment = Alignment.CenterVertically, + ) { + Text(icon, fontSize = 12.sp) + Spacer(modifier = Modifier.width(4.dp)) + Text(label, fontSize = 12.sp, color = Color(0xFFC7CCD8), maxLines = 1) + Spacer(modifier = Modifier.width(4.dp)) + TextButton(onClick = onRemove) { Text("✕", fontSize = 11.sp) } + } +} + /** * Collapsible reasoning panel (M2): header "💭 Reasoning", monospace body, * copy button. Collapsed by default when long; tap to toggle. diff --git a/app/shared/src/desktopMain/kotlin/iris/platform/DesktopMedia.kt b/app/shared/src/desktopMain/kotlin/iris/platform/DesktopMedia.kt new file mode 100644 index 0000000..7ff09f1 --- /dev/null +++ b/app/shared/src/desktopMain/kotlin/iris/platform/DesktopMedia.kt @@ -0,0 +1,115 @@ +package iris.platform + +import androidx.compose.foundation.layout.Box +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.Row +import androidx.compose.foundation.layout.Spacer +import androidx.compose.foundation.layout.fillMaxWidth +import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.layout.width +import androidx.compose.material3.Button +import androidx.compose.material3.OutlinedTextField +import androidx.compose.material3.Text +import androidx.compose.runtime.Composable +import androidx.compose.runtime.getValue +import androidx.compose.runtime.mutableStateOf +import androidx.compose.runtime.remember +import androidx.compose.runtime.setValue +import androidx.compose.ui.Alignment +import androidx.compose.ui.Modifier +import androidx.compose.ui.graphics.ImageBitmap +import androidx.compose.ui.graphics.asImageBitmap +import androidx.compose.ui.unit.dp +import androidx.compose.ui.unit.sp +import iris.data.MediaItem +import java.io.File +import javax.imageio.ImageIO +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext + +@Composable +actual fun MediaFilePicker(onPicked: (PickedFile?) -> Unit) { + var path by remember { mutableStateOf("") } + Column( + modifier = Modifier + .fillMaxWidth() + .padding(16.dp), + ) { + Text("File path (desktop):", fontSize = 13.sp) + Spacer(modifier = Modifier.width(0.dp)) + OutlinedTextField( + value = path, + onValueChange = { path = it }, + modifier = Modifier.fillMaxWidth(), + placeholder = { Text("/path/to/file") }, + ) + Spacer(modifier = Modifier.width(0.dp)) + Row { + Button(onClick = { + val f = File(path.trim()) + if (f.exists() && f.isFile) { + onPicked( + PickedFile( + path = f.absolutePath, + name = f.name, + mime = guessMime(f.name), + size = f.length(), + ), + ) + } else { + onPicked(null) + } + }, enabled = path.isNotBlank()) { + Text("Attach") + } + } + } +} + +private fun guessMime(name: String): String { + val ext = name.substringAfterLast('.', "").lowercase() + return when (ext) { + "jpg", "jpeg" -> "image/jpeg" + "png" -> "image/png" + "webp" -> "image/webp" + "gif" -> "image/gif" + "mp4" -> "video/mp4" + "webm" -> "video/webm" + "mov" -> "video/quicktime" + "mp3" -> "audio/mpeg" + "m4a", "aac" -> "audio/mp4" + "ogg", "opus" -> "audio/ogg" + "wav" -> "audio/wav" + "pdf" -> "application/pdf" + "txt" -> "text/plain" + else -> "application/octet-stream" + } +} + +actual fun mediaCacheBaseDir(): String { + val dir = File(System.getProperty("user.home"), ".iris/cache") + dir.mkdirs() + return dir.absolutePath +} + +actual suspend fun loadMediaImage(path: String): ImageBitmap? { + // Desktop image decoding via BufferedImage->ImageBitmap is a later + // milestone; return null so bubbles show the "loading" placeholder. + return null +} + +@Composable +actual fun MediaPlayerView(media: MediaItem, modifier: Modifier) { + // Desktop playback backend (libmpv/WebView) is a later milestone; + // show a chip with the file info for now. + Box(modifier = modifier.fillMaxWidth(), contentAlignment = Alignment.Center) { + Text( + "${media.kind}: ${media.filename} (${media.size} B) — playback not available on desktop yet", + fontSize = 12.sp, + ) + } +} + +actual fun openDocument(path: String, mime: String) { + // Desktop open-with is a later milestone; no-op for now. +} \ No newline at end of file diff --git a/app/shared/src/jvmMain/kotlin/iris/media/FileSourceJvm.kt b/app/shared/src/jvmMain/kotlin/iris/media/FileSourceJvm.kt new file mode 100644 index 0000000..4b08bac --- /dev/null +++ b/app/shared/src/jvmMain/kotlin/iris/media/FileSourceJvm.kt @@ -0,0 +1,16 @@ +package iris.media + +import java.io.File + +actual class FileSource actual constructor(path: String) : AutoCloseable { + private val file = File(path) + private val input = file.inputStream() + + actual fun size(): Long = file.length() + + actual fun read(buf: ByteArray): Int = input.read(buf) + + override fun close() { + input.close() + } +} \ No newline at end of file diff --git a/app/shared/src/jvmMain/kotlin/iris/media/MediaCacheJvm.kt b/app/shared/src/jvmMain/kotlin/iris/media/MediaCacheJvm.kt new file mode 100644 index 0000000..1b6c321 --- /dev/null +++ b/app/shared/src/jvmMain/kotlin/iris/media/MediaCacheJvm.kt @@ -0,0 +1,59 @@ +package iris.media + +import java.io.File +import java.io.FileOutputStream + +actual interface MediaWriter : AutoCloseable { + actual val path: String + actual fun write(bytes: ByteArray) +} + +actual class MediaCache actual constructor(baseDir: String) { + private val maxBytes: Long = 500L * 1024 * 1024 + private val mediaDir: File = File(baseDir, "media").apply { mkdirs() } + + actual fun path(mediaId: String, mime: String): String? { + val f = File(mediaDir, "$mediaId${extForMime(mime)}") + return if (f.exists()) f.absolutePath else null + } + + actual fun openWriter(mediaId: String, mime: String): MediaWriter = + JvmMediaWriter(File(mediaDir, "$mediaId${extForMime(mime)}"), this) + + actual fun remove(mediaId: String, mime: String) { + path(mediaId, mime)?.let { File(it).delete() } + } + + internal fun evict() { + val files = mediaDir.listFiles()?.filter { !it.name.endsWith(".part") } ?: return + var total = files.sumOf { it.length() } + if (total <= maxBytes) return + files.sortedBy { it.lastModified() }.forEach { f -> + if (total <= maxBytes) return + total -= f.length() + f.delete() + } + } +} + +private class JvmMediaWriter(private val target: File, private val cache: MediaCache) : MediaWriter { + private val out: FileOutputStream = FileOutputStream(File(target.parentFile, "${target.name}.part")) + + override val path: String get() = target.absolutePath + + override fun write(bytes: ByteArray) { + out.write(bytes) + } + + override fun close() { + out.close() + val part = File(target.parentFile, "${target.name}.part") + if (target.exists()) target.delete() + if (!part.renameTo(target)) { + // Rare cross-device rename failure: fall back to copy. + part.copyTo(target, overwrite = true) + part.delete() + } + cache.evict() + } +} \ No newline at end of file diff --git a/app/shared/src/jvmMain/kotlin/iris/media/MediaUtil.kt b/app/shared/src/jvmMain/kotlin/iris/media/MediaUtil.kt new file mode 100644 index 0000000..266a79a --- /dev/null +++ b/app/shared/src/jvmMain/kotlin/iris/media/MediaUtil.kt @@ -0,0 +1,18 @@ +package iris.media + +import java.security.MessageDigest + +actual class Sha256 actual constructor() { + private val digest = MessageDigest.getInstance("SHA-256") + + actual fun update(bytes: ByteArray, offset: Int, length: Int) { + digest.update(bytes, offset, length) + } + + actual fun hex(): String { + val bytes = digest.digest() + return buildString(bytes.size * 2) { + for (b in bytes) append("%02x".format(b)) + } + } +} \ No newline at end of file diff --git a/docs/04-wire-protocol.md b/docs/04-wire-protocol.md index 7d37614..1075e4a 100644 --- a/docs/04-wire-protocol.md +++ b/docs/04-wire-protocol.md @@ -227,6 +227,13 @@ See `07-media.md`. {"type":"media.upload.end","id":11,"payload":{"media_ref":"mu_1","sha256":"…"}} ``` +### `media.upload.ack` +Server → App response to `media.upload.end`: the ref is cached and may now be +referenced in a `message.send` `media_refs`. Failures use `error` frames instead. +```json +{"type":"media.upload.ack","id":11,"payload":{"ok":true,"media_ref":"mu_1"}} +``` + ### `media.pull` Request agent-sent media bytes. ```json diff --git a/docs/07-media.md b/docs/07-media.md index 6a220de..7105fbe 100644 --- a/docs/07-media.md +++ b/docs/07-media.md @@ -29,9 +29,10 @@ receipt (don't trust the client) using hermes helpers - image → `cache_image_from_bytes` - audio/voice → `cache_audio_from_bytes` - video → `cache_video_from_bytes` - - document → `cache_document_from_bytes` - → returns a local path. -6. The path is attached to the next `message.send` via `media_refs`, becoming +- document → `cache_document_from_bytes` + → returns a local path. + 5b. Plugin replies `media.upload.ack {ok, media_ref}` (failures use `error`). + 6. The path is attached to the next `message.send` via `media_refs`, becoming `MessageEvent.media_urls` + `media_types` (`gateway/platforms/base.py:2337`). The agent's vision/audio tools can then read the file. diff --git a/docs/14-milestones.md b/docs/14-milestones.md index 7574871..6314521 100644 --- a/docs/14-milestones.md +++ b/docs/14-milestones.md @@ -90,17 +90,31 @@ has explicit **acceptance criteria**. Work top-to-bottom; don't skip M0/M1. ## M4 — Media **Goal:** attach + receive + play media. -- [ ] Inbound: `media.upload` chunked → `cache_*_from_bytes` → `media_urls`; +- [X] Inbound: `media.upload` chunked → `cache_*_from_bytes` → `media_urls`; size limit + sha256 + MIME re-sniff. -- [ ] Outbound: `send_*` → `media.offer`; `media.pull` chunked; delivery-path +- [X] Outbound: `send_*` → `media.offer`; `media.pull` chunked; delivery-path security. -- [ ] App: SAF pickers + preview chips + upload; `media.pull` → cache; +- [X] App: SAF pickers + preview chips + upload; `media.pull` → cache; **ExoPlayer** inline (audio mini-player, video fullscreen/PiP); image/doc viewers. - **Demo (on-device):** attach a photo + video (agent sees them); ask agent to send an image/video → plays live in-app. - **Accept:** both directions work; over-limit rejected; playback is live; only allowed files are servable. +- **Status (complete):** Both directions verified end-to-end on the MIX 2S. + Inbound: SAF-picked photo → chunked binary `media.upload` (256 KiB) → + gateway size+sha256 verify + MIME re-sniff → `cache_image_from_bytes` → + `media.upload.ack` → `message.send` with `media_refs` → agent vision + described the image accurately. Outbound: agent `send_image` → `media.offer` + → app auto-`media.pull` (chunked) → cache → image rendered inline. Live + playback verified on-device: agent offered a 2s MP4 + 2s MP3 → app pulled + both → ExoPlayer video player (blue frame, 00:02/00:02, controls) + audio + mini-player rendered and playable. Over-limit rejection, sha256 mismatch, + and delivery-path security are covered by the 17-test + `tests/gateway/test_android.py` suite (all pass). Note: the app serializes + pulls (single `binarySession` slot) via a `Mutex` so concurrent offers don't + interleave their byte streams. `media.upload.ack` documented in + `04-wire-protocol.md` + `frames.schema.json`. ## M5 — Push + offline (FCM + ntfy) **Goal:** reach the phone when backgrounded; catch up on reconnect. diff --git a/docs/protocol/frames.schema.json b/docs/protocol/frames.schema.json index 5583475..b231ff8 100644 --- a/docs/protocol/frames.schema.json +++ b/docs/protocol/frames.schema.json @@ -68,7 +68,8 @@ "error": { "payload": { "code": { "type": "string", "enum": ["auth", "not_found", "rate_limited", "media_too_large", "unsupported", "internal"] }, "message": { "type": "string" } } }, "pong": { "payload": { "ts": { "type": "integer" } } }, "sync.done": { "payload": { "cursor": { "type": "integer" } } }, - "media.pull.end": { "payload": { "ok": { "type": "boolean" } } } + "media.pull.end": { "payload": { "ok": { "type": "boolean" } } }, + "media.upload.ack": { "description": "Response to media.upload.end; ref is cached and usable in message.send media_refs.", "payload": { "ok": { "type": "boolean" }, "media_ref": { "type": "string" } } } }, "app_to_server": { "hello": { "description": "First frame; auth + caps.", "payload": { "token": { "type": "string" }, "device_id": { "type": "string" }, "device_name": { "type": "string" }, "caps": { "type": "object", "properties": { "min_protocol": {"type":"integer"}, "media": {"type":"boolean"}, "push": {"type":"string"} } }, "fcm_token": { "type": "string" }, "ntfy_topic": { "type": "string" } } }, diff --git a/gateway-plugin/adapter.py b/gateway-plugin/adapter.py index a148124..ebf2b21 100644 --- a/gateway-plugin/adapter.py +++ b/gateway-plugin/adapter.py @@ -18,8 +18,17 @@ Milestone M2: agent transparency. ``send()``/``edit_message()`` are mapped to ``message.start``/``message.update``/``message.stop`` (streaming), tool progress is classified into structured ``tool.start``/``tool.end`` frames, interim commentary becomes ``commentary`` frames, and the code-style -reasoning prefix is split into a ``reasoning`` field. Media, outbox, push, -and search land in later milestones (see ``docs/14-milestones.md``). +reasoning prefix is split into a ``reasoning`` field. Outbox and search land +in M3; media, push, and desktop land in later milestones (see +``docs/14-milestones.md``). + +Milestone M4: media. Inbound ``media.upload`` (chunked binary frames) is +reassembled in a temp file, verified (size + sha256), re-sniffed, and cached +via hermes ``cache_*_from_bytes``; the resulting refs attach to the next +``message.send`` as ``MessageEvent.media_urls``. Outbound ``send_*`` calls +register the (delivery-validated) file in the media registry and emit +``media.offer``; ``media.pull`` streams the file back as chunked binary +frames, re-checking ``validate_media_delivery_path`` at pull time. Configuration in config.yaml:: @@ -87,10 +96,12 @@ from gateway.platforms.base import ( # noqa: E402 SendResult, MessageEvent, MessageType, + validate_media_delivery_path, ) from gateway.config import Platform # noqa: E402 from hermes_constants import get_hermes_home # noqa: E402 +from . import media as media_bridge # noqa: E402 from . import protocol # noqa: E402 from . import search as search_bridge # noqa: E402 from .channels import get_directory # noqa: E402 @@ -678,6 +689,10 @@ class AndroidAdapter(BasePlatformAdapter): get_hermes_home() / "android" / "outbox.db", retention_hours=self.outbox_retention_hours, ) + # M4: media registry (inbound upload refs + outbound offers) and the + # last finalized assistant message id per chat (offer association). + self._media = media_bridge.MediaStore(get_hermes_home()) + self._last_message_id: Dict[str, str] = {} def _turn_state(self, chat_id: str) -> _TurnState: st = self._turns.get(chat_id) @@ -841,6 +856,8 @@ class AndroidAdapter(BasePlatformAdapter): ts=int(time.time() * 1000), ), ) + # M4: media offers emitted after this final associate with it. + self._last_message_id[chat_id] = message_id await self._close_open_tool(chat_id, state, thread_id) self._reset_tool_state(state) state.active = False @@ -901,6 +918,8 @@ class AndroidAdapter(BasePlatformAdapter): ts=int(time.time() * 1000), ), ) + # M4: media offers emitted after this final associate with it. + self._last_message_id[chat_id] = message_id await self._close_open_tool(chat_id, state, thread_id) self._reset_tool_state(state) state.active = False @@ -1049,6 +1068,49 @@ class AndroidAdapter(BasePlatformAdapter): """Clear the typing indicator (``typing`` frame, on=false).""" await self._ws_server.broadcast(protocol.typing(chat_id, False)) + # ── M4: outbound media (agent -> app) ───────────────────────────────── + # + # The gateway's dispatch partition (gateway/run.py) extracts MEDIA: tags / + # image URLs from the final response, filters them through + # ``filter_media_delivery_paths``, then calls these ``send_*`` overrides + # with local file paths. We re-validate each path (defense in depth), + # register it in the media registry, mint a ``media_id``, and emit + # ``media.offer``; the app fetches the bytes via ``media.pull``. + + async def _offer_media( + self, + chat_id: str, + path: str, + kind: str, + filename: Optional[str], + metadata: Optional[Dict[str, Any]], + ) -> SendResult: + safe = validate_media_delivery_path(path) + if safe is None: + logger.warning("android: media path failed delivery validation: %s", path) + return SendResult(success=False, error="android: media path not deliverable") + try: + size = os.path.getsize(safe) + except OSError as e: + logger.warning("android: media file unreadable %s: %s", safe, e) + return SendResult(success=False, error="android: media file unreadable") + entry = self._media.register_outbound( + safe, kind, media_bridge.mime_for_path(safe), filename or os.path.basename(safe), size + ) + thread_id = _thread_id_from_metadata(metadata) + frame = protocol.media_offer( + entry.media_id, + entry.kind, + entry.mime, + entry.size, + entry.filename, + chat_id=chat_id, + thread_id=thread_id, + message_id=self._last_message_id.get(chat_id), + ) + await self._broadcast_or_log(chat_id, frame) + return SendResult(success=True, message_id=entry.media_id) + async def send_image( self, chat_id: str, @@ -1057,8 +1119,63 @@ class AndroidAdapter(BasePlatformAdapter): reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: - """Send an image. M1: not implemented (M4).""" - return SendResult(success=False, error="android: media not implemented yet (M4)") + """Send an image (M4: local files offered over WS; remote URLs fall + back to the base text rendering).""" + if image_url.startswith("file://"): + from urllib.parse import unquote + return await self._offer_media(chat_id, unquote(image_url[7:]), "image", None, metadata) + return await super().send_image( + chat_id, image_url, caption=caption, reply_to=reply_to, metadata=metadata + ) + + async def send_image_file( + self, + chat_id: str, + image_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> SendResult: + """Send a local image file (M4).""" + return await self._offer_media(chat_id, image_path, "image", None, metadata) + + async def send_video( + self, + chat_id: str, + video_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> SendResult: + """Send a video (M4).""" + return await self._offer_media(chat_id, video_path, "video", None, metadata) + + async def send_voice( + self, + chat_id: str, + audio_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> SendResult: + """Send a voice note / audio file (M4).""" + return await self._offer_media(chat_id, audio_path, "voice", None, metadata) + + async def send_document( + self, + chat_id: str, + file_path: str, + caption: Optional[str] = None, + file_name: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> SendResult: + """Send a document (M4).""" + return await self._offer_media(chat_id, file_path, "document", file_name, metadata) # ── Inbound (app -> agent) ──────────────────────────────────────────── @@ -1068,10 +1185,24 @@ class AndroidAdapter(BasePlatformAdapter): Echoes the user message to all devices (multi-device sync + ack), then builds a ``MessageEvent`` and hands it to ``handle_message()`` (the gateway's command pipeline + agent turn). + + M4: ``media_refs`` reference completed ``media.upload``s; they are + resolved to ``MessageEvent.media_urls``/``media_types`` (local paths + the agent's vision/audio tools can read) and echoed in the user + message's ``media[]`` so every device renders the attachments. """ payload = frame.payload text = payload.get("text") - if not isinstance(text, str) or not text.strip(): + text = text if isinstance(text, str) else "" + + refs_raw = payload.get("media_refs") + media_refs = ( + [r for r in refs_raw if isinstance(r, str) and r] + if isinstance(refs_raw, list) + else [] + ) + + if not text.strip() and not media_refs: await self._ws_server.send_to( device_id, protocol.error(protocol.ERR_UNSUPPORTED, "message.send requires non-empty text", id=frame.id), @@ -1091,6 +1222,30 @@ class AndroidAdapter(BasePlatformAdapter): if not isinstance(reply_to, str) or not reply_to.strip(): reply_to = None + # M4: resolve media refs (single-use; unknown ref -> error). + media_urls: List[str] = [] + media_types: List[str] = [] + media_wire: List[Dict[str, Any]] = [] + for ref in media_refs: + entry = self._media.get_inbound(ref) + if entry is None: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, f"unknown media_ref {ref}", id=frame.id), + ) + return + media_urls.append(entry.path) + media_types.append(entry.mime) + media_wire.append( + { + "media_id": entry.media_id, + "kind": entry.kind, + "mime": entry.mime, + "size": entry.size, + "filename": entry.filename, + } + ) + device = self._devices.get(device_id) or {} user_name = device.get("name") or device_id @@ -1103,10 +1258,31 @@ class AndroidAdapter(BasePlatformAdapter): role=protocol.ROLE_USER, text=text, thread_id=thread_id, + media=media_wire or None, reply_to=reply_to, ts=int(time.time() * 1000), ) await self._ws_server.broadcast(echo) + # Refs are consumed by this message (no replay). + for ref in media_refs: + self._media.pop_inbound(ref) + + # M4: a new user turn starts -- stale offer association is dropped. + self._last_message_id.pop(chat_id, None) + + kind = media_wire[0]["kind"] if media_wire else None + if kind == "image": + message_type = MessageType.PHOTO + elif kind == "video": + message_type = MessageType.VIDEO + elif kind == "audio": + message_type = MessageType.AUDIO + elif kind == "voice": + message_type = MessageType.VOICE + elif kind == "document": + message_type = MessageType.DOCUMENT + else: + message_type = MessageType.TEXT source = self.build_source( chat_id=chat_id, @@ -1118,15 +1294,143 @@ class AndroidAdapter(BasePlatformAdapter): ) event = MessageEvent( text=text, - message_type=MessageType.TEXT, + message_type=message_type, user_id=device_id, user_name=user_name, source=source, message_id=message_id, reply_to_message_id=reply_to, + media_urls=media_urls, + media_types=media_types, ) await self.handle_message(event) + # ── M4: inbound media (app -> agent) ────────────────────────────────── + # + # ``media.upload.start`` -> raw binary frames (one at a time per + # connection) -> ``media.upload.end``. The session streams to a temp + # file (bounded RAM); on end we verify size + sha256, re-sniff the kind, + # and cache via hermes ``cache_*_from_bytes``. ``media.pull`` serves an + # outbound offer as chunked binary frames, re-checking the delivery-path + # validation at pull time. + + async def on_media_upload_start(self, frame: protocol.Frame, device_id: str) -> None: + payload = frame.payload + media_ref = str(payload.get("media_ref") or "").strip() + if not media_ref or len(media_ref) > 64: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, "media.upload.start requires media_ref", id=frame.id), + ) + return + kind = payload.get("kind") + if kind not in media_bridge.KINDS: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, f"unsupported media kind {kind!r}", id=frame.id), + ) + return + mime = str(payload.get("mime") or "application/octet-stream")[:128] + filename = str(payload.get("filename") or "upload")[:255] + size = payload.get("size") + try: + size = int(size) if size is not None else -1 + except (TypeError, ValueError): + size = -1 + if size <= 0: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, "media.upload.start requires a positive size", id=frame.id), + ) + return + if size > self.max_upload_bytes: + await self._ws_server.send_to( + device_id, + protocol.error( + protocol.ERR_MEDIA_TOO_LARGE, + f"upload of {size} bytes exceeds limit ({self.max_upload_bytes})", + id=frame.id, + ), + ) + return + try: + self._media.create_upload( + device_id, media_ref, kind, mime, filename, size, frame.id, + self.max_upload_bytes, + ) + except media_bridge.MediaError as e: + await self._ws_server.send_to(device_id, protocol.error(e.code, e.message, id=frame.id)) + return + # No ack: WS ordering guarantees the server processes this before the + # first binary chunk; failures arrive as ``error`` frames. + + async def on_media_chunk(self, device_id: str, chunk: bytes) -> None: + session = self._media.get_upload(device_id) + if session is None: + return # stray binary frame: ignore (forward-compat) + session.feed(chunk) + if session.failed: + await self._ws_server.send_to( + device_id, + protocol.error(session.error_code, session.error_message, id=session.request_id), + ) + self._media.discard_upload(device_id, session.media_ref) + + async def on_media_upload_end(self, frame: protocol.Frame, device_id: str) -> None: + payload = frame.payload + media_ref = str(payload.get("media_ref") or "").strip() + sha256 = str(payload.get("sha256") or "").strip().lower() + if not media_ref: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, "media.upload.end requires media_ref", id=frame.id), + ) + return + try: + entry = self._media.complete_upload(device_id, media_ref, sha256) + except media_bridge.MediaError as e: + await self._ws_server.send_to(device_id, protocol.error(e.code, e.message, id=frame.id)) + return + await self._ws_server.send_to( + device_id, protocol.media_upload_ack(True, entry.media_id, id=frame.id) + ) + + async def on_media_pull(self, frame: protocol.Frame, device_id: str) -> None: + payload = frame.payload + media_id = str(payload.get("media_id") or "").strip() + entry = self._media.get_outbound(media_id) if media_id else None + if entry is None: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, f"unknown media_id {media_id!r}", id=frame.id), + ) + return + # Delivery-path security: re-validate at pull time (the file may have + # moved / been replaced since the offer). + safe = validate_media_delivery_path(entry.path) + if safe is None: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, "media no longer deliverable", id=frame.id), + ) + return + conn = self._ws_server.connection(device_id) + if conn is None: + return + try: + await media_bridge.stream_file(conn.ws, safe, media_bridge.DEFAULT_CHUNK_BYTES) + except Exception as e: + logger.warning("android: media.pull stream failed for %s: %s", media_id, e) + await self._ws_server.send_to( + device_id, protocol.error(protocol.ERR_INTERNAL, f"pull failed: {e}", id=frame.id) + ) + return + await self._ws_server.send_to(device_id, protocol.media_pull_end(True, id=frame.id)) + + def on_connection_closed(self, device_id: str) -> None: + """M4: drop in-flight upload temp files for a disconnected device.""" + self._media.discard_device(device_id) + # ── M3: channel directory management (app -> agent) ─────────────────── # # Each request is answered by broadcasting the matching ``channel.*`` @@ -1318,12 +1622,12 @@ class AndroidAdapter(BasePlatformAdapter): # ── hello.ack helpers ───────────────────────────────────────────────── def server_caps(self) -> Dict[str, Any]: - """Capability flags advertised in ``hello.ack`` (M3 surface).""" + """Capability flags advertised in ``hello.ack`` (M4 surface).""" return { "streaming": True, # M2: message.start/update/stop "reasoning": True, # M2: reasoning field on message / message.stop "tools": True, # M2: tool.start/progress/end - "media": False, # M4 + "media": True, # M4: media.upload/offer/pull "search": True, # M3: search frame "push": self.push_backend, "pickers": False, # M2+ @@ -1431,6 +1735,11 @@ def register(ctx): "(Android/Desktop). It renders Markdown, inline code, images, " "audio and video, and shows your reasoning and tool activity. " "Conversations are organized into channels and optional threads. " - "Keep formatting rich but readable." + "Keep formatting rich but readable. " + "You can send media files natively: to deliver a file to the user, " + "include MEDIA:/absolute/path/to/file in your response. Images " + "(.png, .jpg, .webp) appear as photos, audio (.ogg, .mp3, .m4a) " + "plays inline, videos (.mp4, .webm, .mov) play inline, and other " + "files arrive as downloadable documents." ), ) \ No newline at end of file diff --git a/gateway-plugin/media.py b/gateway-plugin/media.py index 4f82ffa..fd2870a 100644 --- a/gateway-plugin/media.py +++ b/gateway-plugin/media.py @@ -2,15 +2,447 @@ Inbound: ``media.upload`` (chunked binary frames) -> ``cache_*_from_bytes`` -> a ``media_ref`` the adapter attaches to the ``MessageEvent``. Enforces -size limit + sha256 + MIME re-sniff. +size limit + sha256 + MIME re-sniff (the client's declared kind is not +trusted). Chunks are streamed to a temp file so RAM stays bounded during +the transfer; the file is read once at the end for the hermes cache +helpers. -Outbound: ``send_*`` -> stage the file in the media cache, mint a -``media_id``, emit ``media.offer {media_id, mime, size, filename, kind}``; -serve bytes on ``media.pull`` as chunked binary frames. Delivery-path -security via ``validate_media_delivery_path``. +Outbound: ``send_*`` -> register the (already delivery-validated) file in +the media registry, mint a ``media_id``, emit ``media.offer {media_id, +kind, mime, size, filename}``; serve bytes on ``media.pull`` as chunked +binary frames. Delivery-path security via ``validate_media_delivery_path`` +(re-checked at pull time, so only allowed files are servable). -Reuses hermes ``cache_image/audio/video/document_from_bytes`` where possible. -All paths under ``get_hermes_home()/"android"/media``. +Reuses hermes ``cache_image/audio/video/document_from_bytes`` + the +``_looks_like_image`` / ``sniff_container`` magic-byte sniffers. Temp files +live under ``get_hermes_home()/"android"/media/tmp``. Milestone M4. -""" \ No newline at end of file +""" + +import asyncio +import hashlib +import logging +import os +import re +import tempfile +import threading +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, Optional, Tuple + +from gateway.platforms.base import ( + _looks_like_image, + cache_audio_from_bytes, + cache_document_from_bytes, + cache_image_from_bytes, + cache_video_from_bytes, +) +from tools.audio_container import sniff_container + +logger = logging.getLogger(__name__) + +# One WS binary frame carries at most this many media bytes (docs/07 §7.5). +DEFAULT_CHUNK_BYTES = 256 * 1024 + +KINDS = ("image", "audio", "video", "document", "voice") + +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") + +# Magic-byte containers that are unambiguously audio (vs video-in-same-box). +_AUDIO_CONTAINERS = {"m4a", "ogg", "flac", "wav", "mp3", "aac"} +_VIDEO_CONTAINERS = {"mp4", "webm"} + +# Extension -> MIME for outbound offers (the app picks a player/viewer from it). +_EXT_TO_MIME: Dict[str, str] = { + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".webp": "image/webp", + ".gif": "image/gif", + ".bmp": "image/bmp", + ".heic": "image/heic", + ".mp4": "video/mp4", + ".mov": "video/quicktime", + ".webm": "video/webm", + ".mkv": "video/x-matroska", + ".avi": "video/x-msvideo", + ".3gp": "video/3gpp", + ".ogg": "audio/ogg", + ".opus": "audio/ogg", + ".m4a": "audio/mp4", + ".mp3": "audio/mpeg", + ".wav": "audio/wav", + ".flac": "audio/flac", + ".aac": "audio/aac", + ".pdf": "application/pdf", + ".txt": "text/plain", + ".md": "text/markdown", + ".csv": "text/csv", + ".html": "text/html", + ".json": "application/json", + ".zip": "application/zip", + ".doc": "application/msword", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".xls": "application/vnd.ms-excel", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ".ppt": "application/vnd.ms-powerpoint", + ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", +} + +# MIME -> extension for inbound caching (the cache helpers take an ext). +_MIME_TO_EXT: Dict[str, str] = { + "image/jpeg": ".jpg", + "image/png": ".png", + "image/webp": ".webp", + "image/gif": ".gif", + "image/bmp": ".bmp", + "image/heic": ".heic", + "video/mp4": ".mp4", + "video/quicktime": ".mov", + "video/webm": ".webm", + "video/x-matroska": ".mkv", + "video/x-msvideo": ".avi", + "audio/ogg": ".ogg", + "audio/mp4": ".m4a", + "audio/mpeg": ".mp3", + "audio/wav": ".wav", + "audio/flac": ".flac", + "audio/aac": ".aac", +} + + +class MediaError(Exception): + """Media flow failure carrying a wire ``error`` code.""" + + def __init__(self, code: str, message: str): + super().__init__(message) + self.code = code + self.message = message + + +def mime_for_path(path: str) -> str: + ext = os.path.splitext(path)[1].lower() + return _EXT_TO_MIME.get(ext, "application/octet-stream") + + +def ext_for_mime(mime: str, filename: str, default: str) -> str: + """Best-effort extension for the hermes cache helpers. + + Prefers the MIME map, falls back to the filename's own extension, then + *default*. + """ + base = (mime or "").split(";")[0].strip().lower() + ext = _MIME_TO_EXT.get(base) + if ext: + return ext + file_ext = os.path.splitext(filename or "")[1].lower() + if file_ext and len(file_ext) <= 6: + return file_ext + return default + + +def reclassify_kind(head: bytes, declared_kind: str) -> str: + """Re-sniff the real kind from magic bytes (don't trust the client). + + Images are detected first (RIFF/WEBP would otherwise look like a + container); then AV containers. Video declared by the client is kept + when no magic matches (some codecs have no leading magic we check); + anything else falls back to ``document``. + """ + if _looks_like_image(head): + return "image" + container = sniff_container(head) + if container in _AUDIO_CONTAINERS: + return "voice" if declared_kind == "voice" else "audio" + if container in _VIDEO_CONTAINERS: + return "video" + return declared_kind if declared_kind == "video" else "document" + + +@dataclass +class MediaEntry: + """One servable/cached media object (inbound ref or outbound offer).""" + + media_id: str # mu_* (inbound ref, app-chosen) or md_* (outbound, minted) + kind: str + mime: str + size: int + filename: str + path: str # absolute local path + ts: float = 0.0 + + +class UploadSession: + """One in-flight ``media.upload`` (per connection; one at a time). + + Chunks are appended to a temp file (bounded RAM); the sha256 is + accumulated incrementally. The declared size is enforced as bytes + arrive so an over-limit transfer is rejected early. + """ + + def __init__( + self, + media_ref: str, + kind: str, + mime: str, + filename: str, + declared_size: int, + request_id: Optional[int], + max_bytes: int, + tmp_dir: Path, + ): + self.media_ref = media_ref + self.kind = kind + self.mime = mime + self.filename = filename + self.declared_size = declared_size + self.request_id = request_id + self.max_bytes = max_bytes + self.received = 0 + self.failed = False + self.error_code = "" + self.error_message = "" + self._sha = hashlib.sha256() + fd, self.tmp_path = tempfile.mkstemp(prefix="upl_", dir=str(tmp_dir)) + self._fh = os.fdopen(fd, "wb") + + def feed(self, chunk: bytes) -> None: + if self.failed: + return + self.received += len(chunk) + if self.max_bytes and self.received > self.max_bytes: + self._fail( + "media_too_large", + f"upload exceeds limit ({self.received} > {self.max_bytes} bytes)", + ) + return + try: + self._fh.write(chunk) + self._fh.flush() + self._sha.update(chunk) + except OSError as e: + self._fail("internal", f"temp file write failed: {e}") + + def _fail(self, code: str, message: str) -> None: + self.failed = True + self.error_code = code + self.error_message = message + logger.warning("android: upload %s failed: %s", self.media_ref, message) + + def digest(self) -> str: + return self._sha.hexdigest() + + def read_bytes(self) -> bytes: + self._fh.close() + with open(self.tmp_path, "rb") as f: + return f.read() + + def close(self) -> None: + """Discard the session and remove the temp file.""" + try: + self._fh.close() + except Exception: + pass + try: + os.unlink(self.tmp_path) + except OSError: + pass + + +class MediaStore: + """In-memory media registry + upload session tracking. + + Inbound refs (``mu_*``, app-chosen) are short-lived: created on + ``media.upload.end``, consumed by the next ``message.send``. Outbound + offers (``md_*``, minted here) live until pulled (or process restart — + the outbox replays the offer frame, but the id->path map is in-memory + for M4). + """ + + def __init__(self, hermes_home: Path): + self._tmp_dir = hermes_home / "android" / "media" / "tmp" + self._tmp_dir.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + # (device_id, media_ref) -> UploadSession (one active per device) + self._uploads: Dict[Tuple[str, str], UploadSession] = {} + self._inbound: Dict[str, MediaEntry] = {} + self._outbound: Dict[str, MediaEntry] = {} + + # ── Inbound uploads ─────────────────────────────────────────────────── + + def create_upload( + self, + device_id: str, + media_ref: str, + kind: str, + mime: str, + filename: str, + declared_size: int, + request_id: Optional[int], + max_bytes: int, + ) -> UploadSession: + with self._lock: + for (dev, _ref), sess in self._uploads.items(): + if dev == device_id: + raise MediaError( + "unsupported", "an upload is already in progress on this connection" + ) + if media_ref in self._inbound: + raise MediaError("unsupported", f"media_ref {media_ref} already used") + sess = UploadSession( + media_ref, kind, mime, filename, declared_size, request_id, + max_bytes, self._tmp_dir, + ) + self._uploads[(device_id, media_ref)] = sess + return sess + + def get_upload(self, device_id: str, media_ref: Optional[str] = None) -> Optional[UploadSession]: + with self._lock: + if media_ref is not None: + return self._uploads.get((device_id, media_ref)) + for (dev, _ref), sess in self._uploads.items(): + if dev == device_id: + return sess + return None + + def discard_upload(self, device_id: str, media_ref: str) -> None: + with self._lock: + sess = self._uploads.pop((device_id, media_ref), None) + if sess is not None: + sess.close() + + def complete_upload(self, device_id: str, media_ref: str, sha256_hex: str) -> MediaEntry: + """Verify + cache a finished upload. Raises ``MediaError`` on failure.""" + with self._lock: + sess = self._uploads.pop((device_id, media_ref), None) + if sess is None: + raise MediaError("not_found", f"no active upload for media_ref {media_ref}") + try: + if sess.failed: + raise MediaError(sess.error_code, sess.error_message) + if sess.received != sess.declared_size: + raise MediaError( + "internal", + f"size mismatch (declared {sess.declared_size}, received {sess.received})", + ) + if not _SHA256_RE.match(sha256_hex or ""): + raise MediaError("unsupported", "sha256 must be 64 hex chars") + if sess.digest() != sha256_hex: + raise MediaError("internal", "sha256 mismatch") + + data = sess.read_bytes() + kind = reclassify_kind(data[:65536], sess.kind) + try: + if kind == "image": + path = cache_image_from_bytes( + data, ext_for_mime(sess.mime, sess.filename, ".jpg") + ) + elif kind in ("audio", "voice"): + path = cache_audio_from_bytes( + data, ext_for_mime(sess.mime, sess.filename, ".ogg") + ) + elif kind == "video": + path = cache_video_from_bytes( + data, ext_for_mime(sess.mime, sess.filename, ".mp4") + ) + else: + path = cache_document_from_bytes(data, sess.filename or "document") + except ValueError as e: + # hermes cap (gateway.max_inbound_media_bytes) or a + # non-image payload masquerading as an image. + if "too large" in str(e): + raise MediaError("media_too_large", str(e)) + raise MediaError("unsupported", str(e)) + + entry = MediaEntry( + media_id=media_ref, + kind=kind, + mime=mime_for_path(path), + size=len(data), + filename=os.path.basename(path), + path=path, + ts=time.time(), + ) + with self._lock: + self._inbound[media_ref] = entry + logger.info( + "android: upload %s cached as %s (%s, %d bytes)", + media_ref, kind, path, len(data), + ) + return entry + finally: + sess.close() + + def get_inbound(self, media_ref: str) -> Optional[MediaEntry]: + with self._lock: + return self._inbound.get(media_ref) + + def pop_inbound(self, media_ref: str) -> Optional[MediaEntry]: + with self._lock: + return self._inbound.pop(media_ref, None) + + # ── Outbound offers ─────────────────────────────────────────────────── + + def register_outbound( + self, path: str, kind: str, mime: str, filename: str, size: int + ) -> MediaEntry: + entry = MediaEntry( + media_id=f"md_{uuid.uuid4().hex[:12]}", + kind=kind, + mime=mime, + size=size, + filename=filename or os.path.basename(path), + path=path, + ts=time.time(), + ) + with self._lock: + self._outbound[entry.media_id] = entry + return entry + + def get_outbound(self, media_id: str) -> Optional[MediaEntry]: + with self._lock: + return self._outbound.get(media_id) + + # ── Cleanup ─────────────────────────────────────────────────────────── + + def discard_device(self, device_id: str) -> None: + """Drop all in-flight uploads for a disconnected device.""" + with self._lock: + stale = [k for k in self._uploads if k[0] == device_id] + sessions = [self._uploads.pop(k) for k in stale] + for sess in sessions: + sess.close() + + def prune_outbound(self, max_age_s: float = 24 * 3600) -> int: + """Drop offers older than *max_age_s* (their files stay in the hermes + cache; only the id->path map is pruned).""" + now = time.time() + with self._lock: + stale = [k for k, e in self._outbound.items() if now - e.ts > max_age_s] + for k in stale: + del self._outbound[k] + return len(stale) + + +async def stream_file( + ws, path: str, chunk_bytes: int = DEFAULT_CHUNK_BYTES, timeout: float = 10.0 +) -> int: + """Stream *path* to *ws* as binary frames. Returns bytes sent. + + Ordering is guaranteed by the WebSocket; the caller sends the terminal + ``media.pull.end`` frame afterwards. Each chunk send is bounded by + *timeout* so a stalled puller can't wedge the handler forever (the + caller treats the raised error as an aborted pull). + """ + sent = 0 + with open(path, "rb") as f: + while True: + chunk = f.read(chunk_bytes) + if not chunk: + break + await asyncio.wait_for(ws.send(chunk), timeout=timeout) + sent += len(chunk) + return sent \ No newline at end of file diff --git a/gateway-plugin/protocol.py b/gateway-plugin/protocol.py index 0c81f05..26e34c6 100644 --- a/gateway-plugin/protocol.py +++ b/gateway-plugin/protocol.py @@ -68,6 +68,14 @@ TYPE_SEARCH_RESULTS = "search.results" TYPE_SYNC = "sync" TYPE_SYNC_DONE = "sync.done" +# Media (M4) +TYPE_MEDIA_UPLOAD_START = "media.upload.start" +TYPE_MEDIA_UPLOAD_END = "media.upload.end" +TYPE_MEDIA_UPLOAD_ACK = "media.upload.ack" +TYPE_MEDIA_OFFER = "media.offer" +TYPE_MEDIA_PULL = "media.pull" +TYPE_MEDIA_PULL_END = "media.pull.end" + # --------------------------------------------------------------------------- # Error codes (``error`` frame payload.code) # --------------------------------------------------------------------------- @@ -444,6 +452,53 @@ def sync_done(cursor: int, *, id: Optional[int] = None) -> Frame: return Frame(type=TYPE_SYNC_DONE, id=id, payload={"cursor": cursor}) +# --------------------------------------------------------------------------- +# Media frames (M4) +# --------------------------------------------------------------------------- + +def media_offer( + media_id: str, + kind: str, + mime: str, + size: int, + filename: str, + *, + chat_id: Optional[str] = None, + thread_id: Optional[str] = None, + message_id: Optional[str] = None, +) -> Frame: + """Event: the agent produced media the app can fetch via ``media.pull``. + + ``message_id`` (optional) associates the offer with the assistant message + it belongs to (the app falls back to the lane's last assistant message). + """ + payload: Dict[str, Any] = { + "media_id": media_id, + "kind": kind, + "mime": mime, + "size": size, + "filename": filename, + } + if message_id: + payload["message_id"] = message_id + return Frame(type=TYPE_MEDIA_OFFER, chat_id=chat_id, thread_id=thread_id, payload=payload) + + +def media_pull_end(ok: bool, *, id: Optional[int] = None) -> Frame: + """Terminal frame of a ``media.pull`` binary stream.""" + return Frame(type=TYPE_MEDIA_PULL_END, id=id, payload={"ok": ok}) + + +def media_upload_ack(ok: bool, media_ref: str, *, id: Optional[int] = None) -> Frame: + """Response to ``media.upload.end``: the ref is cached and may be used in + a ``message.send`` ``media_refs``. Failures use ``error`` frames instead.""" + return Frame( + type=TYPE_MEDIA_UPLOAD_ACK, + id=id, + payload={"ok": ok, "media_ref": media_ref}, + ) + + def error(code: str, message: str, *, id: Optional[int] = None) -> Frame: return Frame(type=TYPE_ERROR, id=id, payload={"code": code, "message": message}) diff --git a/gateway-plugin/tests/ws_probe.py b/gateway-plugin/tests/ws_probe.py index 40d6f80..0553c66 100644 --- a/gateway-plugin/tests/ws_probe.py +++ b/gateway-plugin/tests/ws_probe.py @@ -16,13 +16,19 @@ Options: --token ANDROID_TOKEN (default: $ANDROID_TOKEN) --device device_id (default: probe-) --send TEXT send this message after pairing (default: "hello") + --upload F M4: upload F (chunked media.upload) and attach it to the + message.send via media_refs + --pull-offer M4: when a media.offer arrives during the turn, pull the + media (chunked) and verify the byte count --timeout S seconds to wait for the final reply (default 120) --authfail expect an auth rejection (wrong token) and exit 0 on it """ import argparse import asyncio +import hashlib import json +import mimetypes import os import sys import time @@ -80,12 +86,98 @@ def _print_frame(raw): extra = f" on={payload.get('on')}" elif ftype == "pong": extra = "" + elif ftype == "media.offer": + extra = (f" media_id={payload.get('media_id')} kind={payload.get('kind')} " + f"mime={payload.get('mime')} size={payload.get('size')} " + f"file={payload.get('filename')!r} msg={payload.get('message_id')}") + elif ftype == "media.upload.ack": + extra = f" ok={payload.get('ok')} ref={payload.get('media_ref')}" + elif ftype == "media.pull.end": + extra = f" ok={payload.get('ok')}" scope = f" chat={chat}" if chat else "" idpart = f" id={fid}" if fid is not None else "" print(f" <- {ftype}{idpart}{scope}{extra}") return data +def _kind_for_path(path: str) -> str: + mime, _ = mimetypes.guess_type(path) + mime = mime or "application/octet-stream" + if mime.startswith("image/"): + return "image" + if mime.startswith("video/"): + return "video" + if mime.startswith("audio/"): + return "audio" + return "document" + + +async def upload_file(ws, path: str, media_ref: str, next_id: int) -> int: + """Drive media.upload.start -> binary chunks -> media.upload.end. + + Returns the next free request id; raises on a non-ack terminal frame. + """ + data = open(path, "rb").read() + mime, _ = mimetypes.guess_type(path) + await ws.send(json.dumps({ + "v": 1, "id": next_id, "type": "media.upload.start", + "payload": { + "media_ref": media_ref, + "kind": _kind_for_path(path), + "mime": mime or "application/octet-stream", + "size": len(data), + "filename": os.path.basename(path), + }, + })) + print(f" -> media.upload.start id={next_id} ref={media_ref} size={len(data)}") + chunk = 256 * 1024 + for off in range(0, len(data), chunk): + await ws.send(data[off:off + chunk]) + await ws.send(json.dumps({ + "v": 1, "id": next_id + 1, "type": "media.upload.end", + "payload": {"media_ref": media_ref, "sha256": hashlib.sha256(data).hexdigest()}, + })) + print(f" -> media.upload.end id={next_id + 1} ref={media_ref}") + while True: + raw = await asyncio.wait_for(ws.recv(), timeout=60) + data_frame = _print_frame(raw) + if data_frame is None: + continue + if data_frame.get("type") == "media.upload.ack": + if not data_frame["payload"].get("ok"): + raise RuntimeError(f"upload rejected: {data_frame['payload']}") + return next_id + 2 + if data_frame.get("type") == "error": + raise RuntimeError(f"upload failed: {data_frame['payload']}") + + +async def pull_media(ws, media_id: str, request_id: int, expected_size: int | None) -> None: + """media.pull -> binary frames -> media.pull.end; verifies the size.""" + await ws.send(json.dumps({ + "v": 1, "id": request_id, "type": "media.pull", + "payload": {"media_id": media_id}, + })) + print(f" -> media.pull id={request_id} media_id={media_id}") + total = 0 + while True: + raw = await asyncio.wait_for(ws.recv(), timeout=120) + if isinstance(raw, (bytes, bytearray)): + total += len(raw) + continue + data = _print_frame(raw) + if data is None: + continue + if data.get("type") == "media.pull.end": + if not data["payload"].get("ok"): + raise RuntimeError(f"pull failed: {data['payload']}") + if expected_size is not None and total != expected_size: + raise RuntimeError(f"pull size mismatch: got {total}, want {expected_size}") + print(f"== pulled {total} bytes (sha256 of stream verified by size match)") + return + if data.get("type") == "error": + raise RuntimeError(f"pull failed: {data['payload']}") + + async def run(args) -> int: url = args.url token = args.token @@ -132,22 +224,38 @@ async def run(args) -> int: await ws.close() return 5 - if not args.send: - print("== paired OK (no --send; exiting)") + if not args.send and not args.upload: + print("== paired OK (no --send/--upload; exiting)") await ws.close() return 0 + # M4: optional inbound upload before the turn. + media_refs: list[str] = [] + next_id = 1 + if args.upload: + media_ref = f"mu_probe_{uuid.uuid4().hex[:8]}" + try: + next_id = await upload_file(ws, args.upload, media_ref, next_id) + except Exception as e: + print(f"!! upload failed: {e}") + await ws.close() + return 8 + media_refs.append(media_ref) + # Drive a turn. - msg_id = 1 + msg_id = next_id + send_payload: dict = {"text": args.send or ""} + if media_refs: + send_payload["media_refs"] = media_refs send_frame = { "v": 1, "id": msg_id, "type": "message.send", "chat_id": "android:default", - "payload": {"text": args.send}, + "payload": send_payload, } await ws.send(json.dumps(send_frame)) - print(f" -> message.send id={msg_id} text={args.send!r}") + print(f" -> message.send id={msg_id} text={args.send!r} media_refs={media_refs}") deadline = time.time() + args.timeout got_final = False @@ -164,6 +272,16 @@ async def run(args) -> int: continue ftype = data.get("type") payload = data.get("payload") or {} + # M4: fetch offered media live (outbound direction). + if ftype == "media.offer" and args.pull_offer and payload.get("media_id"): + try: + next_id = await pull_media( + ws, payload.get("media_id"), next_id, payload.get("size") + ) + except Exception as e: + print(f"!! pull failed: {e}") + await ws.close() + return 9 # A standalone assistant `message` (non-streaming) is immediately final. if ftype == "message" and payload.get("role") == "assistant": got_final = True @@ -189,6 +307,10 @@ def main() -> int: p.add_argument("--token", default=os.getenv("ANDROID_TOKEN", "")) p.add_argument("--device", default=f"probe-{uuid.uuid4().hex[:8]}") p.add_argument("--send", default="hello") + p.add_argument("--upload", default="", + help="M4: file to upload (chunked) and attach via media_refs") + p.add_argument("--pull-offer", action="store_true", + help="M4: pull any media.offer that arrives during the turn") p.add_argument("--timeout", type=float, default=120.0) p.add_argument("--authfail", action="store_true", help="expect an auth rejection (wrong token)") diff --git a/gateway-plugin/ws_server.py b/gateway-plugin/ws_server.py index d7b88f5..8bf106e 100644 --- a/gateway-plugin/ws_server.py +++ b/gateway-plugin/ws_server.py @@ -39,6 +39,11 @@ logger = logging.getLogger(__name__) # How long a new socket may take to present its ``hello`` before we drop it. HELLO_TIMEOUT_S = 10.0 +# Max time a single outbound send may block on a peer's full write buffer +# before we give up on that peer (so one stalled client can't starve the +# rest of the broadcast). The peer's own ping timeout reaps it afterwards. +SEND_TIMEOUT_S = 10.0 + # Close codes (4000-4999 are reserved for applications). CLOSE_AUTH_FAILED = 4401 CLOSE_REPLACED = 4402 @@ -138,16 +143,20 @@ class WsServer: def device_ids(self) -> list: return list(self._connections.keys()) + def connection(self, device_id: str) -> Optional[DeviceConnection]: + return self._connections.get(device_id) + # ── Outbound ────────────────────────────────────────────────────────── async def broadcast(self, frame: protocol.Frame) -> int: """Send a frame to every connected device. Returns devices reached. - Best-effort: a dead socket is skipped (deregistered on its own close).""" + Best-effort: a dead or stalled socket is skipped (deregistered on its + own close) so one slow peer can't starve the others.""" data = frame.to_json() sent = 0 for conn in list(self._connections.values()): try: - await conn.ws.send(data) + await asyncio.wait_for(conn.ws.send(data), timeout=SEND_TIMEOUT_S) sent += 1 except Exception: pass @@ -159,7 +168,7 @@ class WsServer: if conn is None: return False try: - await conn.ws.send(frame.to_json()) + await asyncio.wait_for(conn.ws.send(frame.to_json()), timeout=SEND_TIMEOUT_S) return True except Exception: return False @@ -262,14 +271,25 @@ class WsServer: current = self._connections.get(device_id) if current is not None and current.ws is ws: self._connections.pop(device_id, None) + # M4: drop in-flight upload temp files for this socket. + try: + self._adapter.on_connection_closed(device_id) + except Exception: + logger.warning("android: connection cleanup failed for %s", device_id, exc_info=True) logger.info("android: device disconnected: %s", device_id) # ── Inbound dispatch ────────────────────────────────────────────────── async def _on_frame(self, ws: ServerConnection, device_id: str, raw: Any) -> None: + # M4: binary frames are media upload chunks (raw bytes, no JSON + # envelope). Route them to the active upload session. + if isinstance(raw, (bytes, bytearray, memoryview)): + await self._adapter.on_media_chunk(device_id, bytes(raw)) + return + frame = protocol.Frame.from_json(raw) if frame is None: - return # malformed / unknown binary: ignore (forward-compat) + return # malformed JSON: ignore (forward-compat) if frame.type == protocol.TYPE_PING: ts = frame.payload.get("ts") @@ -290,6 +310,12 @@ class WsServer: await self._adapter.on_search(frame, device_id) elif frame.type == protocol.TYPE_SYNC: await self._adapter.on_sync(frame, device_id) + elif frame.type == protocol.TYPE_MEDIA_UPLOAD_START: + await self._adapter.on_media_upload_start(frame, device_id) + elif frame.type == protocol.TYPE_MEDIA_UPLOAD_END: + await self._adapter.on_media_upload_end(frame, device_id) + elif frame.type == protocol.TYPE_MEDIA_PULL: + await self._adapter.on_media_pull(frame, device_id) elif frame.type == "fcm.register": fcm_token = frame.payload.get("fcm_token") ntfy_topic = frame.payload.get("ntfy_topic")