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")