diff --git a/.gitignore b/.gitignore index d179f3c..7931181 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,7 @@ venv/ # ── Kotlin / Gradle (app) ───────────────────────────────────────────────── .gradle/ +.kotlin/ build/ local.properties *.iml diff --git a/app/shared/src/commonMain/kotlin/iris/data/ChannelStore.kt b/app/shared/src/commonMain/kotlin/iris/data/ChannelStore.kt new file mode 100644 index 0000000..7338400 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/data/ChannelStore.kt @@ -0,0 +1,80 @@ +package iris.data + +import iris.protocol.ChannelDeletedPayload +import iris.protocol.ChannelInfo +import iris.protocol.ChannelListPayload +import iris.protocol.Frame +import iris.protocol.TYPE_CHANNEL_CREATED +import iris.protocol.TYPE_CHANNEL_DELETED +import iris.protocol.TYPE_CHANNEL_LIST +import iris.protocol.TYPE_CHANNEL_RENAMED +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow + +/** + * Local channel-directory cache (M3). The server is authoritative; this cache + * is seeded from `hello.ack` and reconciled on `channel.*` events (merge, don't + * clobber). Holds both channels and threads; helpers split them for the UI. + */ +class ChannelStore { + private val _channels = MutableStateFlow>(emptyList()) + val channels: StateFlow> = _channels.asStateFlow() + + /** Seed / replace the whole directory (from `hello.ack` or `channel.list`). */ + fun setAll(channels: List) { + _channels.value = sorted(channels) + } + + /** Reconcile a server frame into the cache. */ + fun onFrame(frame: Frame) { + when (frame.type) { + TYPE_CHANNEL_CREATED, TYPE_CHANNEL_RENAMED -> upsert(frame) + TYPE_CHANNEL_DELETED -> remove(frame) + TYPE_CHANNEL_LIST -> { + frame.payloadAs()?.let { setAll(it.channels) } + } + else -> Unit + } + } + + private fun upsert(frame: Frame) { + val info = frame.payloadAs() ?: return + val list = _channels.value.toMutableList() + val idx = list.indexOfFirst { it.chatId == info.chatId } + if (idx >= 0) list[idx] = info else list.add(info) + _channels.value = sorted(list) + } + + private fun remove(frame: Frame) { + val p = frame.payloadAs() ?: return + _channels.value = _channels.value.filter { it.chatId != p.chatId } + } + + private fun sorted(list: List): List = + list.sortedWith( + compareByDescending { it.isDefault } + .thenBy { it.kind != "thread" } + .thenBy { it.name.lowercase() }, + ) + + // ── UI helpers ──────────────────────────────────────────────────────── + + /** Channels (default + user channels) for the drawer/rail. */ + fun channelsForDrawer(): List = + _channels.value.filter { it.kind != "thread" } + + /** Threads under a channel, for the topic switcher. */ + fun threadsFor(chatId: String): List = + _channels.value.filter { it.kind == "thread" && it.parentChatId == chatId } + + fun byId(chatId: String): ChannelInfo? = + _channels.value.firstOrNull { it.chatId == chatId } + + fun defaultChannel(): ChannelInfo? = + _channels.value.firstOrNull { it.isDefault } ?: _channels.value.firstOrNull() + + fun clear() { + _channels.value = emptyList() + } +} \ 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 1067896..79b01f1 100644 --- a/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt +++ b/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt @@ -26,11 +26,14 @@ import kotlinx.serialization.json.JsonElement import kotlin.random.Random /** - * In-memory chat cache (M2). The server is authoritative; this is a cache that - * reconciles on server frames. M2 adds streaming bubbles (message.start/ - * update/stop), structured tool cards (tool.start/progress/end), and dimmed - * commentary beats, all interleaved in chronological order as [ChatItem]s. - * SQLDelight persistence lands with the channel/message schema later. + * Lane-aware chat cache (M3). The server is authoritative; this is a cache that + * reconciles on server frames. + * + * M2 kept a single flat list (one chat). M3 introduces channels + threads, so + * messages are stored per **lane**: a lane is a `chatId` (the flat / "General" + * lane, `thread_id=null`) or `chatId:threadId` (a topic). Each lane keeps its + * own chronological list of [ChatItem]s (messages, streaming bubbles, tool + * cards, commentary). The UI views one lane at a time ([currentLane]). */ sealed interface ChatItem { val id: String @@ -65,199 +68,234 @@ data class ToolItem( ) : ChatItem class ChatStore { - private val _items = MutableStateFlow>(emptyList()) - val items: StateFlow> = _items.asStateFlow() + /** lane key -> chronological items. */ + private val _lanes = MutableStateFlow>>(emptyMap()) + val lanes: StateFlow>> = _lanes.asStateFlow() + + /** The lane the UI is currently viewing. */ + private val _currentLane = MutableStateFlow(DEFAULT_LANE) + val currentLane: StateFlow = _currentLane.asStateFlow() private var localSeq = 0 private var toolSeq = 0 - /** Optimistic add: show the user's message immediately (pending). */ - fun addPending(text: String): String { + companion object { + const val DEFAULT_LANE = "android:default" + + fun randomId(prefix: String): String = + "${prefix}${Random.nextLong(1_000_000_000L, 9_999_999_999L)}" + } + + // ── Lane helpers ────────────────────────────────────────────────────── + + /** Lane key for a (chat, thread) pair. Uses `::` as the separator because + * chat ids already contain a single `:` (e.g. `android:chan_1`). */ + fun laneKey(chatId: String, threadId: String?): String = + if (threadId.isNullOrBlank()) chatId else "$chatId::$threadId" + + /** Split a lane key back into (chatId, threadId). */ + fun parseLane(key: String): Pair { + val idx = key.indexOf("::") + return if (idx >= 0) key.substring(0, idx) to key.substring(idx + 2) else key to null + } + + fun setLane(laneKey: String) { + if (laneKey.isNotBlank()) _currentLane.value = laneKey + } + + private fun laneOf(frame: Frame): String? { + val chatId = frame.chatId ?: return null + return laneKey(chatId, frame.threadId) + } + + private fun updateLane(lane: String, transform: (List) -> List) { + val map = _lanes.value.toMutableMap() + map[lane] = transform(map[lane].orEmpty()) + _lanes.value = map + } + + // ── Optimistic send ─────────────────────────────────────────────────── + + /** Optimistic add: show the user's message immediately (pending) in [lane]. */ + fun addPending(text: String, lane: String): String { localSeq++ val id = "local_$localSeq" - _items.value = _items.value + - 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) } return id } - /** Reconcile a server frame into the cache. */ + // ── Frame reconciliation ────────────────────────────────────────────── + + /** Reconcile a server frame into the cache (routed by chat/thread lane). */ fun onFrame(frame: Frame) { + val lane = laneOf(frame) ?: return when (frame.type) { - TYPE_MESSAGE -> onMessage(frame) - TYPE_MESSAGE_START -> onMessageStart(frame) - TYPE_MESSAGE_UPDATE -> onMessageUpdate(frame) - TYPE_MESSAGE_STOP -> onMessageStop(frame) - TYPE_TOOL_START -> onToolStart(frame) - TYPE_TOOL_PROGRESS -> onToolProgress(frame) - TYPE_TOOL_END -> onToolEnd(frame) - TYPE_COMMENTARY -> onCommentary(frame) + TYPE_MESSAGE -> onMessage(lane, frame) + TYPE_MESSAGE_START -> onMessageStart(lane, frame) + TYPE_MESSAGE_UPDATE -> onMessageUpdate(lane, frame) + TYPE_MESSAGE_STOP -> onMessageStop(lane, frame) + TYPE_TOOL_START -> onToolStart(lane, frame) + TYPE_TOOL_PROGRESS -> onToolProgress(lane, frame) + TYPE_TOOL_END -> onToolEnd(lane, frame) + TYPE_COMMENTARY -> onCommentary(lane, frame) else -> Unit } } // ── message (final / standalone, incl. non-streaming + user echo) ───── - private fun onMessage(frame: Frame) { + private fun onMessage(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - val byId = list.indexOfFirst { it.id == p.messageId } - if (byId >= 0) { - list[byId] = (list[byId] as? MessageItem)?.copy( - text = p.text, - reasoning = p.reasoning, - pending = false, - streaming = false, - model = p.model, - tokens = p.tokens, - ts = p.ts ?: 0, - ) ?: return - _items.value = list - return - } - if (p.role == ROLE_USER) { - // Replace the matching optimistic pending bubble (server echo). - val pendingIdx = list.indexOfLast { - it is MessageItem && it.pending && it.role == ROLE_USER && it.text == p.text - } - if (pendingIdx >= 0) { - list[pendingIdx] = MessageItem( - id = p.messageId, role = p.role, text = p.text, - ts = p.ts ?: 0, reasoning = p.reasoning, + updateLane(lane) { list -> + val byId = list.indexOfFirst { it.id == p.messageId } + if (byId >= 0) { + val updated = (list[byId] as? MessageItem)?.copy( + text = p.text, + reasoning = p.reasoning, + pending = false, + streaming = false, + model = p.model, + tokens = p.tokens, + ts = p.ts ?: 0, + ) ?: return@updateLane list + list.toMutableList().also { it[byId] = updated } + } else if (p.role == ROLE_USER) { + // Replace the matching optimistic pending bubble (server echo). + val pendingIdx = list.indexOfLast { + it is MessageItem && it.pending && it.role == ROLE_USER && it.text == p.text + } + if (pendingIdx >= 0) { + list.toMutableList().also { + it[pendingIdx] = MessageItem( + id = p.messageId, role = p.role, text = p.text, + ts = p.ts ?: 0, reasoning = p.reasoning, + ) + } + } 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, + ) + } + } 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, ) - _items.value = list - return } } - list.add( - MessageItem( - id = p.messageId, role = p.role, text = p.text, ts = p.ts ?: 0, - reasoning = p.reasoning, model = p.model, tokens = p.tokens, - ), - ) - _items.value = list } // ── message.start (open a live streaming bubble) ────────────────────── - private fun onMessageStart(frame: Frame) { + private fun onMessageStart(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - if (list.any { it.id == p.messageId }) return - list.add( - MessageItem( - id = p.messageId, role = p.role, text = "", - ts = 0, streaming = true, - ), - ) - _items.value = list + updateLane(lane) { list -> + if (list.any { it.id == p.messageId }) list + else list + MessageItem( + id = p.messageId, role = p.role, text = "", ts = 0, streaming = true, + ) + } } // ── message.update (replace live bubble text; full snapshot) ────────── - private fun onMessageUpdate(frame: Frame) { + private fun onMessageUpdate(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - val idx = list.indexOfFirst { it.id == p.messageId } - if (idx < 0) return - val cur = list[idx] as? MessageItem ?: return - list[idx] = cur.copy(text = p.text) - _items.value = list + updateLane(lane) { list -> + val idx = list.indexOfFirst { it.id == p.messageId } + if (idx < 0) return@updateLane list + val cur = list[idx] as? MessageItem ?: return@updateLane list + list.toMutableList().also { it[idx] = cur.copy(text = p.text) } + } } // ── message.stop (finalize the live bubble) ─────────────────────────── - private fun onMessageStop(frame: Frame) { + private fun onMessageStop(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - val idx = list.indexOfFirst { it.id == p.messageId } - if (idx < 0) { - // No live bubble (e.g. missed start) — materialize a final one. - list.add( - MessageItem( + updateLane(lane) { list -> + val idx = list.indexOfFirst { it.id == p.messageId } + if (idx < 0) { + // No live bubble (e.g. missed start) — materialize a final one. + list + MessageItem( id = p.messageId, role = ROLE_ASSISTANT, text = p.finalText, ts = p.ts ?: 0, reasoning = p.reasoning, model = p.model, tokens = p.tokens, - ), - ) - _items.value = list - return + ) + } else { + val cur = list[idx] as? MessageItem ?: return@updateLane list + list.toMutableList().also { + it[idx] = cur.copy( + text = p.finalText, + reasoning = p.reasoning, + model = p.model, + tokens = p.tokens, + streaming = false, + ts = p.ts ?: cur.ts, + ) + } + } } - val cur = list[idx] as? MessageItem ?: return - list[idx] = cur.copy( - text = p.finalText, - reasoning = p.reasoning, - model = p.model, - tokens = p.tokens, - streaming = false, - ts = p.ts ?: cur.ts, - ) - _items.value = list } // ── tool.start (new tool card) ──────────────────────────────────────── - private fun onToolStart(frame: Frame) { + private fun onToolStart(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return toolSeq++ val id = "tool_$toolSeq" - val list = _items.value.toMutableList() - list.add( - ToolItem( + updateLane(lane) { list -> + list + ToolItem( id = id, index = p.index, name = p.name, preview = p.preview, args = p.args, - ), - ) - _items.value = list + ) + } } // ── tool.progress (in-progress note) ────────────────────────────────── - private fun onToolProgress(frame: Frame) { + private fun onToolProgress(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - val idx = list.indexOfLast { it is ToolItem && !it.done && it.index == p.index } - if (idx < 0) return - val cur = list[idx] as ToolItem - list[idx] = cur.copy(note = p.note) - _items.value = list + updateLane(lane) { list -> + val idx = list.indexOfLast { it is ToolItem && !it.done && it.index == p.index } + if (idx < 0) return@updateLane list + val cur = list[idx] as ToolItem + list.toMutableList().also { it[idx] = cur.copy(note = p.note) } + } } // ── tool.end (mark the tool card complete) ──────────────────────────── - private fun onToolEnd(frame: Frame) { + private fun onToolEnd(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - val idx = list.indexOfLast { it is ToolItem && !it.done && it.index == p.index } - if (idx < 0) return - val cur = list[idx] as ToolItem - list[idx] = cur.copy( - done = true, ok = p.ok, duration = p.duration, - outputPreview = p.outputPreview, - ) - _items.value = list + updateLane(lane) { list -> + val idx = list.indexOfLast { it is ToolItem && !it.done && it.index == p.index } + if (idx < 0) return@updateLane list + val cur = list[idx] as ToolItem + list.toMutableList().also { + it[idx] = cur.copy( + done = true, ok = p.ok, duration = p.duration, + outputPreview = p.outputPreview, + ) + } + } } // ── commentary (dimmed interim beat) ────────────────────────────────── - private fun onCommentary(frame: Frame) { + private fun onCommentary(lane: String, frame: Frame) { val p = frame.payloadAs() ?: return - val list = _items.value.toMutableList() - if (list.any { it.id == p.messageId }) return - list.add( - MessageItem( + updateLane(lane) { list -> + if (list.any { it.id == p.messageId }) list + else list + MessageItem( id = p.messageId, role = ROLE_ASSISTANT, text = p.text, ts = 0, isCommentary = true, - ), - ) - _items.value = list + ) + } } fun clear() { - _items.value = emptyList() - } - - companion object { - fun randomId(prefix: String): String = - "${prefix}${Random.nextLong(1_000_000_000L, 9_999_999_999L)}" + _lanes.value = emptyMap() } } \ 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 d6f4550..3dacb4b 100644 --- a/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt +++ b/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt @@ -222,6 +222,18 @@ class GatewayClient( ws.send(messageSendFrame(id, chatId, text, threadId).toWire()) } + /** + * Send an arbitrary frame with a fresh request id (fire-and-forget). + * The server replies (or broadcasts) a frame carrying the same id; the + * app reconciles from [events]. Returns the id used, or -1 if not connected. + */ + fun sendFrame(frame: Frame): Int { + val ws = socket ?: return -1 + val id = nextRequestId++ + ws.send(frame.copy(id = id).toWire()) + return id + } + /** Send a ping (heartbeat). */ fun ping() { socket?.send(pingFrame().toWire()) diff --git a/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt b/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt index 4fc9eb2..4a2c937 100644 --- a/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt +++ b/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt @@ -46,6 +46,20 @@ const val TYPE_TOOL_PROGRESS = "tool.progress" const val TYPE_TOOL_END = "tool.end" const val TYPE_COMMENTARY = "commentary" +// M3 — channels / threads / search / sync +const val TYPE_CHANNEL_CREATE = "channel.create" +const val TYPE_CHANNEL_RENAME = "channel.rename" +const val TYPE_CHANNEL_SET_DEFAULT = "channel.set_default" +const val TYPE_CHANNEL_DELETE = "channel.delete" +const val TYPE_CHANNEL_CREATED = "channel.created" +const val TYPE_CHANNEL_RENAMED = "channel.renamed" +const val TYPE_CHANNEL_DELETED = "channel.deleted" +const val TYPE_CHANNEL_LIST = "channel.list" +const val TYPE_SEARCH = "search" +const val TYPE_SEARCH_RESULTS = "search.results" +const val TYPE_SYNC = "sync" +const val TYPE_SYNC_DONE = "sync.done" + // ── Error codes ───────────────────────────────────────────────────────── const val ERR_AUTH = "auth" @@ -110,6 +124,8 @@ data class ChannelInfo( val name: String, val kind: String = "channel", @SerialName("is_default") val isDefault: Boolean = false, + @SerialName("parent_chat_id") val parentChatId: String? = null, + val archived: Boolean = false, ) @Serializable @@ -211,6 +227,60 @@ data class ErrorPayload(val code: String, val message: String) @Serializable data class PingPayload(val ts: Long? = null) +// ── M3: channel directory (app -> server requests) ────────────────────── + +@Serializable +data class ChannelCreatePayload( + val name: String, + val kind: String = "channel", + @SerialName("parent_chat_id") val parentChatId: String? = null, +) + +@Serializable +data class ChannelRenamePayload(val name: String) + +@Serializable +data class ChannelListPayload(val channels: List = emptyList()) + +@Serializable +data class ChannelDeletedPayload(@SerialName("chat_id") val chatId: String) + +// ── M3: search ────────────────────────────────────────────────────────── + +@Serializable +data class SearchPayload( + val query: String, + val scope: String = "all", + @SerialName("chat_id") val chatId: String? = null, + @SerialName("thread_id") val threadId: String? = null, + val limit: Int? = null, +) + +@Serializable +data class SearchHit( + @SerialName("message_id") val messageId: String, + @SerialName("chat_id") val chatId: String, + @SerialName("thread_id") val threadId: String? = null, + val role: String, + val snippet: String, + val ts: Long, +) + +@Serializable +data class SearchResultsPayload( + val query: String, + val scope: String, + val hits: List = emptyList(), +) + +// ── M3: sync (reconnect catch-up) ─────────────────────────────────────── + +@Serializable +data class SyncPayload(val cursor: Long) + +@Serializable +data class SyncDonePayload(val cursor: Long) + // ── Frame builders ────────────────────────────────────────────────────── fun helloFrame(token: String, deviceId: String, deviceName: String): Frame = @@ -235,4 +305,64 @@ fun messageSendFrame(id: Int, chatId: String, text: String, threadId: String? = ) fun pingFrame(): Frame = - Frame(type = TYPE_PING, payload = IrisJson.instance.encodeToJsonElement(PingPayload.serializer(), PingPayload())) \ No newline at end of file + Frame(type = TYPE_PING, payload = IrisJson.instance.encodeToJsonElement(PingPayload.serializer(), PingPayload())) + +// ── M3 frame builders ─────────────────────────────────────────────────── + +fun channelCreateFrame(id: Int, name: String, kind: String = "channel", parentChatId: String? = null): Frame = + Frame( + id = id, + type = TYPE_CHANNEL_CREATE, + payload = IrisJson.instance.encodeToJsonElement( + ChannelCreatePayload.serializer(), + ChannelCreatePayload(name = name, kind = kind, parentChatId = parentChatId), + ), + ) + +fun channelRenameFrame(id: Int, chatId: String, name: String): Frame = + Frame( + id = id, + type = TYPE_CHANNEL_RENAME, + chatId = chatId, + payload = IrisJson.instance.encodeToJsonElement( + ChannelRenamePayload.serializer(), + ChannelRenamePayload(name = name), + ), + ) + +fun channelSetDefaultFrame(id: Int, chatId: String): Frame = + Frame(id = id, type = TYPE_CHANNEL_SET_DEFAULT, chatId = chatId) + +fun channelDeleteFrame(id: Int, chatId: String): Frame = + Frame(id = id, type = TYPE_CHANNEL_DELETE, chatId = chatId) + +fun channelListFrame(id: Int): Frame = + Frame(id = id, type = TYPE_CHANNEL_LIST) + +fun searchFrame( + id: Int, + query: String, + scope: String, + chatId: String? = null, + threadId: String? = null, +): Frame = + Frame( + id = id, + type = TYPE_SEARCH, + chatId = chatId, + threadId = threadId, + payload = IrisJson.instance.encodeToJsonElement( + SearchPayload.serializer(), + SearchPayload(query = query, scope = scope, chatId = chatId, threadId = threadId), + ), + ) + +fun syncFrame(id: Int, cursor: Long): Frame = + Frame( + id = id, + type = TYPE_SYNC, + payload = IrisJson.instance.encodeToJsonElement( + SyncPayload.serializer(), + SyncPayload(cursor = cursor), + ), + ) \ 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 7fb00c8..275f11e 100644 --- a/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt +++ b/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt @@ -1,18 +1,35 @@ package iris.state import iris.data.ChatStore +import iris.data.ChannelStore import iris.data.SecureStore import iris.net.GatewayClient +import iris.protocol.SearchHit +import iris.protocol.SearchResultsPayload +import iris.protocol.SyncDonePayload +import iris.protocol.TYPE_CHANNEL_CREATED +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_MESSAGE import iris.protocol.TYPE_MESSAGE_START import iris.protocol.TYPE_MESSAGE_STOP import iris.protocol.TYPE_MESSAGE_UPDATE +import iris.protocol.TYPE_SEARCH_RESULTS +import iris.protocol.TYPE_SYNC_DONE import iris.protocol.TYPE_TOOL_END import iris.protocol.TYPE_TOOL_PROGRESS import iris.protocol.TYPE_TOOL_START import iris.protocol.TYPE_TYPING import iris.protocol.TypingPayload +import iris.protocol.channelCreateFrame +import iris.protocol.channelDeleteFrame +import iris.protocol.channelListFrame +import iris.protocol.channelRenameFrame +import iris.protocol.channelSetDefaultFrame +import iris.protocol.searchFrame +import iris.protocol.syncFrame import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob @@ -22,9 +39,9 @@ import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.launch /** - * App-level controller (M2): owns the GatewayClient + ChatStore, routes - * server frames (message / streaming / tool / commentary / typing) into - * state, and exposes the actions the UI needs. + * App-level controller (M3): owns the GatewayClient + ChatStore + ChannelStore, + * routes server frames (message / streaming / tool / commentary / typing / + * channel.* / search / sync) into state, and exposes the actions the UI needs. */ class IrisController( private val store: SecureStore, @@ -34,6 +51,7 @@ class IrisController( val client = GatewayClient(scope, store) val chat = ChatStore() + val channels = ChannelStore() private val _typing = MutableStateFlow(false) val typing: StateFlow = _typing.asStateFlow() @@ -54,6 +72,22 @@ class IrisController( private val _homeChannel = MutableStateFlow("android:default") val homeChannel: StateFlow = _homeChannel.asStateFlow() + // ── M3: threads toggle (per-app for now; per-channel lands later) ───── + private val _threadsEnabled = MutableStateFlow(false) + val threadsEnabled: StateFlow = _threadsEnabled.asStateFlow() + + fun toggleThreads() { + _threadsEnabled.value = !_threadsEnabled.value + } + + // ── M3: search state ────────────────────────────────────────────────── + private val _searchResults = MutableStateFlow>(emptyList()) + val searchResults: StateFlow> = _searchResults.asStateFlow() + private val _searching = MutableStateFlow(false) + val searching: StateFlow = _searching.asStateFlow() + private val _lastQuery = MutableStateFlow("") + val lastQuery: StateFlow = _lastQuery.asStateFlow() + init { scope.launch { client.events.collect { frame -> @@ -66,6 +100,19 @@ class IrisController( TYPE_TOOL_PROGRESS, TYPE_TOOL_END, TYPE_COMMENTARY -> chat.onFrame(frame) + TYPE_CHANNEL_CREATED, + TYPE_CHANNEL_RENAMED, + TYPE_CHANNEL_DELETED, + TYPE_CHANNEL_LIST -> channels.onFrame(frame) + TYPE_SEARCH_RESULTS -> { + frame.payloadAs()?.let { _searchResults.value = it.hits } + _searching.value = false + } + TYPE_SYNC_DONE -> { + // Replayed frames already flowed through [events]; the + // cursor is authoritative server-side (outbox). + frame.payloadAs() + } TYPE_TYPING -> { frame.payloadAs()?.let { _typing.value = it.on } } @@ -76,8 +123,12 @@ class IrisController( scope.launch { client.state.collect { s -> if (s is GatewayClient.State.Connected) { + channels.setAll(s.channels) val home = s.channels.firstOrNull { it.isDefault }?.chatId - if (home != null) _homeChannel.value = home + if (home != null) { + _homeChannel.value = home + chat.setLane(home) + } } } } @@ -85,12 +136,80 @@ class IrisController( client.start() } - /** Optimistic send: show immediately, then hand to the gateway. */ + // ── M3: navigation (lane switching) ─────────────────────────────────── + + /** Switch to a channel's flat / "General" lane. */ + fun openChannel(chatId: String) { + chat.setLane(chatId) + } + + /** Switch to a specific thread lane under [chatId]. */ + fun openThread(chatId: String, threadId: String) { + chat.setLane(chat.laneKey(chatId, threadId)) + } + + // ── M3: channel directory ops (server is authoritative) ─────────────── + + fun createChannel(name: String) { + val trimmed = name.trim() + if (trimmed.isEmpty()) return + client.sendFrame(channelCreateFrame(0, trimmed)) + } + + fun createThread(name: String, parentChatId: String) { + val trimmed = name.trim() + if (trimmed.isEmpty()) return + client.sendFrame(channelCreateFrame(0, trimmed, kind = "thread", parentChatId = parentChatId)) + } + + fun renameChannel(chatId: String, name: String) { + val trimmed = name.trim() + if (trimmed.isEmpty()) return + client.sendFrame(channelRenameFrame(0, chatId, trimmed)) + } + + fun setDefaultChannel(chatId: String) { + client.sendFrame(channelSetDefaultFrame(0, chatId)) + } + + fun deleteChannel(chatId: String) { + client.sendFrame(channelDeleteFrame(0, chatId)) + } + + fun refreshChannels() { + client.sendFrame(channelListFrame(0)) + } + + // ── M3: search ──────────────────────────────────────────────────────── + + fun search(query: String, scope: String = "all", chatId: String? = null, threadId: String? = null) { + val q = query.trim() + if (q.isEmpty()) return + _lastQuery.value = q + _searching.value = true + client.sendFrame(searchFrame(0, q, scope, chatId, threadId)) + } + + fun clearSearch() { + _searchResults.value = emptyList() + _lastQuery.value = "" + _searching.value = false + } + + // ── M3: sync (reconnect catch-up) ───────────────────────────────────── + + fun sync(cursor: Long) { + client.sendFrame(syncFrame(0, cursor)) + } + + /** Optimistic send: show immediately in the current lane, then hand to the gateway. */ fun send(text: String) { val trimmed = text.trim() if (trimmed.isEmpty()) return - chat.addPending(trimmed) - client.sendMessage(_homeChannel.value, trimmed) + val lane = chat.currentLane.value + val (chatId, threadId) = chat.parseLane(lane) + chat.addPending(trimmed, lane) + client.sendMessage(chatId, trimmed, threadId) } /** Connect-screen action: real hello test, then save + (re)connect. */ @@ -105,6 +224,7 @@ class IrisController( fun forget() { store.clear() chat.clear() + channels.clear() client.restart() } 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 92fef73..5ce48e0 100644 --- a/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt +++ b/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt @@ -2,6 +2,7 @@ package iris.ui.screens import androidx.compose.foundation.background import androidx.compose.foundation.clickable +import androidx.compose.foundation.horizontalScroll import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Box import androidx.compose.foundation.layout.Column @@ -18,21 +19,29 @@ import androidx.compose.foundation.layout.width import androidx.compose.foundation.lazy.LazyColumn import androidx.compose.foundation.lazy.items import androidx.compose.foundation.lazy.rememberLazyListState +import androidx.compose.foundation.rememberScrollState import androidx.compose.foundation.shape.RoundedCornerShape import androidx.compose.foundation.text.KeyboardActions import androidx.compose.foundation.text.KeyboardOptions +import androidx.compose.material3.AlertDialog import androidx.compose.material3.Button import androidx.compose.material3.CircularProgressIndicator +import androidx.compose.material3.DrawerValue +import androidx.compose.material3.IconButton import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.ModalDrawerSheet +import androidx.compose.material3.ModalNavigationDrawer import androidx.compose.material3.OutlinedTextField import androidx.compose.material3.Text import androidx.compose.material3.TextButton +import androidx.compose.material3.rememberDrawerState import androidx.compose.runtime.Composable import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.collectAsState import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateOf import androidx.compose.runtime.remember +import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.runtime.setValue import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier @@ -45,28 +54,45 @@ import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.text.input.ImeAction import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.sp +import iris.data.ChannelStore import iris.data.ChatItem import iris.data.MessageItem import iris.data.ToolItem import iris.net.GatewayClient +import iris.protocol.ChannelInfo import iris.protocol.ROLE_USER +import iris.protocol.SearchHit import iris.state.IrisController import iris.state.ToolDetail +import kotlinx.coroutines.launch /** - * Chat screen (M2): message list + composer. User bubbles right (accent), - * assistant bubbles left (surface). M2 adds live streaming bubbles, a - * collapsible reasoning block above the answer, structured tool cards - * (verbosity-controlled), and dimmed commentary beats. + * Chat screen (M3): channel drawer + thread toggle + topic switcher + search, + * on top of the M2 message list / composer. Messages are shown for the current + * lane (a channel's flat lane, or a specific thread). */ @Composable fun ChatScreen(controller: IrisController) { val state by controller.client.state.collectAsState() - val items by controller.chat.items.collectAsState() + val lanes by controller.chat.lanes.collectAsState() + val currentLane by controller.chat.currentLane.collectAsState() + val items = lanes[currentLane].orEmpty() val typing by controller.typing.collectAsState() val toolDetail by controller.toolDetail.collectAsState() + val threadsEnabled by controller.threadsEnabled.collectAsState() + + val channels by controller.channels.channels.collectAsState() + val (currentChatId, currentThreadId) = controller.chat.parseLane(currentLane) + val currentChannel = channels.firstOrNull { it.chatId == currentChatId } + val threads = channels.filter { it.kind == "thread" && it.parentChatId == currentChatId } + val listState = rememberLazyListState() var input by remember { mutableStateOf("") } + var showSearch by remember { mutableStateOf(false) } + var showNewChannel by remember { mutableStateOf(false) } + var showNewThread by remember { mutableStateOf(false) } + val drawerState = rememberDrawerState(DrawerValue.Closed) + val drawerScope = rememberCoroutineScope() fun doSend() { if (input.isBlank()) return @@ -80,87 +106,363 @@ fun ChatScreen(controller: IrisController) { if (last >= 0) listState.animateScrollToItem(last) } - Column(modifier = Modifier.fillMaxSize()) { - // Header - Row( - modifier = Modifier - .fillMaxWidth() - .padding(horizontal = 16.dp, vertical = 10.dp), - verticalAlignment = Alignment.CenterVertically, - ) { - Text("Iris", style = MaterialTheme.typography.titleMedium, fontWeight = FontWeight.SemiBold) - Spacer(modifier = Modifier.width(10.dp)) - StatusChip(state) - Spacer(modifier = Modifier.weight(1f)) - // M2: cycle tool-card verbosity (full Settings screen lands in M7). - TextButton(onClick = { controller.cycleToolDetail() }) { - Text("tools: ${toolDetail.label}", fontSize = 11.sp) + ModalNavigationDrawer( + drawerState = drawerState, + gesturesEnabled = drawerState.isOpen, + drawerContent = { + ModalDrawerSheet(modifier = Modifier.width(280.dp)) { + ChannelDrawer( + channels = channels.filter { it.kind != "thread" }, + currentChatId = currentChatId, + onOpen = { ch -> + controller.openChannel(ch.chatId) + drawerScope.launch { drawerState.close() } + }, + onNewChannel = { + showNewChannel = true + drawerScope.launch { drawerState.close() } + }, + onSetDefault = { ch -> controller.setDefaultChannel(ch.chatId) }, + onDelete = { ch -> controller.deleteChannel(ch.chatId) }, + ) + } + }, + ) { + Column(modifier = Modifier.fillMaxSize()) { + // Header + Row( + modifier = Modifier + .fillMaxWidth() + .padding(horizontal = 8.dp, vertical = 6.dp), + verticalAlignment = Alignment.CenterVertically, + ) { + IconButton(onClick = { drawerScope.launch { drawerState.open() } }) { + Text("☰", fontSize = 18.sp) + } + Text( + currentChannel?.name ?: "Iris", + style = MaterialTheme.typography.titleMedium, + fontWeight = FontWeight.SemiBold, + modifier = Modifier.weight(1f), + ) + StatusChip(state) + Spacer(modifier = Modifier.width(4.dp)) + IconButton(onClick = { showSearch = true }) { + Text("🔍", fontSize = 16.sp) + } + IconButton(onClick = { controller.toggleThreads() }) { + Text(if (threadsEnabled) "🧵✓" else "🧵", fontSize = 16.sp) + } + TextButton(onClick = { controller.cycleToolDetail() }) { + Text("tools: ${toolDetail.label}", fontSize = 11.sp) + } } - } - // Messages - LazyColumn( - state = listState, - modifier = Modifier - .weight(1f) - .fillMaxWidth(), - contentPadding = PaddingValues(16.dp), - verticalArrangement = Arrangement.spacedBy(8.dp), - ) { - if (items.isEmpty() && !typing) { - item(key = "empty") { - Box(modifier = Modifier.fillMaxWidth(), contentAlignment = Alignment.Center) { + // Topic switcher (threads on) + if (threadsEnabled) { + TopicSwitcher( + threads = threads, + currentThreadId = currentThreadId, + onGeneral = { controller.openChannel(currentChatId) }, + onThread = { t -> controller.openThread(currentChatId, t.chatId) }, + onNewThread = { showNewThread = true }, + ) + } + + // Messages + LazyColumn( + state = listState, + modifier = Modifier + .weight(1f) + .fillMaxWidth(), + contentPadding = PaddingValues(16.dp), + verticalArrangement = Arrangement.spacedBy(8.dp), + ) { + if (items.isEmpty() && !typing) { + item(key = "empty") { + Box(modifier = Modifier.fillMaxWidth(), contentAlignment = Alignment.Center) { + Text( + "Say hello to your agent.", + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + } + } + items(items, key = { it.id }) { item -> + when (item) { + is MessageItem -> MessageBubble(item) + is ToolItem -> if (toolDetail != ToolDetail.NOTHING) ToolCard(item, toolDetail) + } + } + if (typing) { + item(key = "typing") { Text( - "Say hello to your agent.", + "typing…", + style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onSurfaceVariant, + modifier = Modifier.padding(start = 12.dp), ) } } } - items(items, key = { it.id }) { item -> - when (item) { - is MessageItem -> MessageBubble(item) - is ToolItem -> if (toolDetail != ToolDetail.NOTHING) ToolCard(item, toolDetail) - } - } - if (typing) { - item(key = "typing") { - Text( - "typing…", - style = MaterialTheme.typography.bodySmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - modifier = Modifier.padding(start = 12.dp), - ) - } - } - } - // Composer - Row( - modifier = Modifier - .fillMaxWidth() - .padding(12.dp), - verticalAlignment = Alignment.Bottom, - ) { - OutlinedTextField( - value = input, - onValueChange = { input = it }, + // Composer + Row( modifier = Modifier - .weight(1f) - .heightIn(min = 48.dp, max = 160.dp), - placeholder = { Text("Message") }, - keyboardOptions = KeyboardOptions(imeAction = ImeAction.Send), - keyboardActions = KeyboardActions(onSend = { doSend() }), - ) - Spacer(modifier = Modifier.width(8.dp)) - Button( - onClick = { doSend() }, - enabled = input.isNotBlank(), + .fillMaxWidth() + .padding(12.dp), + verticalAlignment = Alignment.Bottom, ) { - Text("Send") + OutlinedTextField( + value = input, + onValueChange = { input = it }, + modifier = Modifier + .weight(1f) + .heightIn(min = 48.dp, max = 160.dp), + placeholder = { Text("Message") }, + keyboardOptions = KeyboardOptions(imeAction = ImeAction.Send), + keyboardActions = KeyboardActions(onSend = { doSend() }), + ) + Spacer(modifier = Modifier.width(8.dp)) + Button( + onClick = { doSend() }, + enabled = input.isNotBlank(), + ) { + Text("Send") + } } } } + + // Search overlay + if (showSearch) { + SearchOverlay( + controller = controller, + onJump = { hit -> + if (hit.threadId.isNullOrBlank()) controller.openChannel(hit.chatId) + else controller.openThread(hit.chatId, hit.threadId) + showSearch = false + }, + onDismiss = { showSearch = false }, + ) + } + + // New channel dialog + if (showNewChannel) { + NameDialog( + title = "New channel", + onConfirm = { name -> + controller.createChannel(name) + showNewChannel = false + }, + onDismiss = { showNewChannel = false }, + ) + } + + // New thread dialog + if (showNewThread) { + NameDialog( + title = "New topic", + onConfirm = { name -> + controller.createThread(name, currentChatId) + showNewThread = false + }, + onDismiss = { showNewThread = false }, + ) + } +} + +/** Left drawer: channel list + create / set-default / delete. */ +@Composable +private fun ChannelDrawer( + channels: List, + currentChatId: String, + onOpen: (ChannelInfo) -> Unit, + onNewChannel: () -> Unit, + onSetDefault: (ChannelInfo) -> Unit, + onDelete: (ChannelInfo) -> Unit, +) { + Column(modifier = Modifier.fillMaxSize().padding(12.dp)) { + Text( + "Channels", + style = MaterialTheme.typography.titleMedium, + modifier = Modifier.padding(bottom = 8.dp), + ) + LazyColumn( + verticalArrangement = Arrangement.spacedBy(4.dp), + modifier = Modifier.weight(1f), + ) { + items(channels, key = { it.chatId }) { ch -> + val isCurrent = ch.chatId == currentChatId + Row( + modifier = Modifier + .fillMaxWidth() + .clip(RoundedCornerShape(8.dp)) + .background(if (isCurrent) Color(0xFF2A3550) else Color.Transparent) + .clickable { onOpen(ch) } + .padding(horizontal = 10.dp, vertical = 6.dp), + verticalAlignment = Alignment.CenterVertically, + ) { + Text( + ch.name + if (ch.isDefault) " ★" else "", + fontSize = 14.sp, + modifier = Modifier.weight(1f), + ) + if (!ch.isDefault) { + TextButton(onClick = { onSetDefault(ch) }) { + Text("★", fontSize = 12.sp) + } + TextButton(onClick = { onDelete(ch) }) { + Text("✕", fontSize = 12.sp) + } + } + } + } + } + Button(onClick = onNewChannel, modifier = Modifier.fillMaxWidth()) { + Text("+ New channel") + } + } +} + +/** Horizontal topic chips: General + threads + "new topic". */ +@Composable +private fun TopicSwitcher( + threads: List, + currentThreadId: String?, + onGeneral: () -> Unit, + onThread: (ChannelInfo) -> Unit, + onNewThread: () -> Unit, +) { + Row( + modifier = Modifier + .fillMaxWidth() + .horizontalScroll(rememberScrollState()) + .padding(horizontal = 12.dp, vertical = 6.dp), + horizontalArrangement = Arrangement.spacedBy(6.dp), + ) { + TopicChip("General", currentThreadId == null, onClick = onGeneral) + threads.forEach { t -> + TopicChip(t.name, currentThreadId == t.chatId, onClick = { onThread(t) }) + } + TopicChip("+ topic", false, onClick = onNewThread) + } +} + +@Composable +private fun TopicChip(label: String, selected: Boolean, onClick: () -> Unit) { + Box( + modifier = Modifier + .clip(RoundedCornerShape(12.dp)) + .background(if (selected) Color(0xFF4F7CFF) else Color(0xFF2A2E3B)) + .clickable(onClick = onClick) + .padding(horizontal = 10.dp, vertical = 5.dp), + ) { + Text(label, fontSize = 12.sp, color = if (selected) Color.White else Color(0xFFC7CCD8)) + } +} + +/** Full-screen search overlay (M3). */ +@Composable +private fun SearchOverlay( + controller: IrisController, + onJump: (SearchHit) -> Unit, + onDismiss: () -> Unit, +) { + val results by controller.searchResults.collectAsState() + val searching by controller.searching.collectAsState() + val lastQuery by controller.lastQuery.collectAsState() + var query by remember { mutableStateOf(lastQuery) } + + Box(modifier = Modifier.fillMaxSize().background(Color.Black.copy(alpha = 0.5f))) { + Column( + modifier = Modifier + .fillMaxSize() + .background(MaterialTheme.colorScheme.surface) + .padding(16.dp), + ) { + Row(verticalAlignment = Alignment.CenterVertically) { + Text("Search", style = MaterialTheme.typography.titleMedium, modifier = Modifier.weight(1f)) + TextButton(onClick = onDismiss) { Text("close") } + } + Row(verticalAlignment = Alignment.CenterVertically) { + OutlinedTextField( + value = query, + onValueChange = { query = it }, + modifier = Modifier.weight(1f), + placeholder = { Text("Search messages…") }, + keyboardOptions = KeyboardOptions(imeAction = ImeAction.Search), + keyboardActions = KeyboardActions(onSearch = { controller.search(query) }), + ) + Spacer(modifier = Modifier.width(8.dp)) + Button(onClick = { controller.search(query) }, enabled = query.isNotBlank() && !searching) { + Text(if (searching) "…" else "search") + } + } + Spacer(modifier = Modifier.height(12.dp)) + if (results.isEmpty() && !searching) { + Box(modifier = Modifier.fillMaxSize(), contentAlignment = Alignment.Center) { + Text("No results.", color = MaterialTheme.colorScheme.onSurfaceVariant) + } + } else { + LazyColumn(verticalArrangement = Arrangement.spacedBy(8.dp)) { + items(results, key = { it.messageId }) { hit -> + SearchHitRow(hit, controller.channels, onClick = { onJump(hit) }) + } + } + } + } + } +} + +@Composable +private fun SearchHitRow(hit: SearchHit, channels: ChannelStore, onClick: () -> Unit) { + val channel = channels.byId(hit.chatId) + val thread = hit.threadId?.let { channels.byId(it) } + Column( + modifier = Modifier + .fillMaxWidth() + .clip(RoundedCornerShape(10.dp)) + .background(Color(0xFF20242E)) + .clickable(onClick = onClick) + .padding(10.dp), + ) { + Row(verticalAlignment = Alignment.CenterVertically) { + Text( + (channel?.name ?: hit.chatId) + (thread?.let { " / ${it.name}" } ?: ""), + fontSize = 12.sp, + fontWeight = FontWeight.Medium, + color = Color(0xFFB9C0D0), + modifier = Modifier.weight(1f), + ) + Text(hit.role, fontSize = 10.sp, color = Color(0xFF8A93A6)) + } + Spacer(modifier = Modifier.height(4.dp)) + Text(hit.snippet, fontSize = 13.sp, color = Color(0xFFD7DBE5), maxLines = 3) + } +} + +/** Simple name-entry dialog (new channel / new topic). */ +@Composable +private fun NameDialog(title: String, onConfirm: (String) -> Unit, onDismiss: () -> Unit) { + var name by remember { mutableStateOf("") } + AlertDialog( + onDismissRequest = onDismiss, + title = { Text(title) }, + text = { + OutlinedTextField( + value = name, + onValueChange = { name = it }, + placeholder = { Text("Name") }, + ) + }, + confirmButton = { + TextButton(onClick = { onConfirm(name) }, enabled = name.isNotBlank()) { Text("Create") } + }, + dismissButton = { + TextButton(onClick = onDismiss) { Text("Cancel") } + }, + ) } @Composable diff --git a/gateway-plugin/adapter.py b/gateway-plugin/adapter.py index 826dbd2..a148124 100644 --- a/gateway-plugin/adapter.py +++ b/gateway-plugin/adapter.py @@ -92,6 +92,9 @@ from gateway.config import Platform # noqa: E402 from hermes_constants import get_hermes_home # noqa: E402 from . import protocol # noqa: E402 +from . import search as search_bridge # noqa: E402 +from .channels import get_directory # noqa: E402 +from .outbox import Outbox # noqa: E402 from .pairing import ( # noqa: E402 DeviceRegistry, generate_token, @@ -471,24 +474,45 @@ def _parse_port(raw: str) -> int: def _parse_target_ref(target_ref: str) -> Optional[tuple]: """Parse a raw target string into ``(chat_id, thread_id)`` or ``None``. - Recognises the native syntax ``android:[:]``. Returns - ``None`` for anything else so the target proceeds to channel-directory + Recognises the native syntax ``android:[:]`` where the + chat_id itself carries the ``android:`` prefix (e.g. ``android:chan_7``) + and an optional thread is a trailing ``:t_``. A bare friendly name + (e.g. ``Cron Reports``) is resolved against the channel directory so cron + / ``send_message`` can target a channel by name immediately, without + waiting for the core directory's refresh timer. Returns ``None`` for + anything unrecognised so the target proceeds to the core channel-directory resolution. """ - if not target_ref or not target_ref.startswith("android:"): + if not target_ref: return None - body = target_ref[len("android:"):] - if not body: + t = target_ref.strip() + if not t: return None - if ":" in body: - chat_id, thread_id = body.split(":", 1) - thread_id = thread_id or None - else: - chat_id, thread_id = body, None - chat_id = chat_id.strip() - if not chat_id: - return None - return (chat_id, thread_id) + + if t.startswith("android:"): + body = t[len("android:"):].strip() + if not body: + return None + thread_id: Optional[str] = None + if ":" in body: + head, tail = body.rsplit(":", 1) + if tail and tail.startswith("t_"): + thread_id = tail + body = head + return (f"android:{body}", thread_id) + + # Bare friendly name -> resolve via the channel directory. A thread resolves + # to its session lane (parent_chat_id + thread_id); a channel/default to + # its chat_id. + try: + entry = get_directory().resolve_entry(t) + except Exception: + entry = None + if entry is not None: + if entry["kind"] == "thread": + return (entry["parent_chat_id"], entry["chat_id"]) + return (entry["chat_id"], None) + return None # --------------------------------------------------------------------------- @@ -648,6 +672,12 @@ class AndroidAdapter(BasePlatformAdapter): self._connected = False # M2: per-chat turn state for outbound frame classification. self._turns: Dict[str, _TurnState] = {} + # M3: channel directory (shared singleton) + offline outbox. + self._channels = get_directory() + self._outbox = Outbox( + get_hermes_home() / "android" / "outbox.db", + retention_hours=self.outbox_retention_hours, + ) def _turn_state(self, chat_id: str) -> _TurnState: st = self._turns.get(chat_id) @@ -695,6 +725,13 @@ class AndroidAdapter(BasePlatformAdapter): self._connected = False return False + # M3: ensure the default (home) channel exists in the directory so the + # app's channel list and cron home delivery have a stable anchor. + try: + self._channels.ensure_default(self.home_channel, self.home_channel_name) + except Exception: + logger.warning("android: ensure_default failed", exc_info=True) + self._connected = True self._mark_connected() logger.info("android: connected; WS server on %s:%s", self.host, self.port) @@ -716,6 +753,10 @@ class AndroidAdapter(BasePlatformAdapter): self._devices.close() except Exception: pass + try: + self._outbox.close() + except Exception: + pass self._connected = False self._mark_disconnected() logger.info("android: disconnected") @@ -984,10 +1025,16 @@ class AndroidAdapter(BasePlatformAdapter): async def _broadcast_or_log(self, chat_id: str, frame: "protocol.Frame") -> None: delivered = await self._ws_server.broadcast(frame) if delivered == 0: - logger.info( - "android: no live devices for %s; %s frame not delivered (outbox lands in M3)", - chat_id, frame.type, - ) + # M3: no live device -- park the frame in the outbox so a + # reconnecting app can `sync` it (M5 adds push to wake the device). + try: + cursor = self._outbox.append(chat_id, frame.to_json()) + logger.info( + "android: no live devices for %s; %s frame parked in outbox (cursor=%s)", + chat_id, frame.type, cursor, + ) + except Exception: + logger.warning("android: outbox append failed", exc_info=True) async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None: """Send a typing indicator (``typing`` frame, on=true).""" @@ -1080,50 +1127,260 @@ class AndroidAdapter(BasePlatformAdapter): ) await self.handle_message(event) + # ── M3: channel directory management (app -> agent) ─────────────────── + # + # Each request is answered by broadcasting the matching ``channel.*`` + # event carrying the request ``id``: the requester's pending request + # completes on the id, and every other device reconciles its local copy + # from the same frame (single broadcast serves as event + response). + + async def on_channel_create(self, frame: protocol.Frame, device_id: str) -> None: + payload = frame.payload + name = payload.get("name") + if not isinstance(name, str) or not name.strip(): + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, "channel.create requires a name", id=frame.id), + ) + return + kind = payload.get("kind") + kind = kind if kind in ("channel", "thread") else "channel" + parent_chat_id = payload.get("parent_chat_id") + if not isinstance(parent_chat_id, str) or not parent_chat_id.strip(): + parent_chat_id = None + if kind == "thread" and not parent_chat_id: + parent_chat_id = frame.chat_id or self.home_channel + try: + entry = self._channels.create(name=name, kind=kind, parent_chat_id=parent_chat_id) + except ValueError as e: + await self._ws_server.send_to( + device_id, protocol.error(protocol.ERR_UNSUPPORTED, str(e), id=frame.id) + ) + return + resp = protocol.channel_created(entry) + resp.id = frame.id + await self._ws_server.broadcast(resp) + + async def on_channel_rename(self, frame: protocol.Frame, device_id: str) -> None: + chat_id = frame.chat_id or frame.payload.get("chat_id") + if not isinstance(chat_id, str) or not chat_id.strip(): + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, "channel.rename requires chat_id", id=frame.id), + ) + return + name = frame.payload.get("name") + if not isinstance(name, str) or not name.strip(): + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, "channel.rename requires a name", id=frame.id), + ) + return + try: + entry = self._channels.rename(chat_id, name) + except ValueError as e: + await self._ws_server.send_to( + device_id, protocol.error(protocol.ERR_UNSUPPORTED, str(e), id=frame.id) + ) + return + if entry is None: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, f"unknown chat_id {chat_id}", id=frame.id), + ) + return + resp = protocol.channel_renamed(entry) + resp.id = frame.id + await self._ws_server.broadcast(resp) + + async def on_channel_set_default(self, frame: protocol.Frame, device_id: str) -> None: + chat_id = frame.chat_id or frame.payload.get("chat_id") + if not isinstance(chat_id, str) or not chat_id.strip(): + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, "channel.set_default requires chat_id", id=frame.id), + ) + return + entry = self._channels.set_default(chat_id) + if entry is None: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, f"unknown chat_id {chat_id}", id=frame.id), + ) + return + # Reuse the renamed event shape: it carries the full entry (incl. the + # new is_default flag) so every device reconciles the default change. + resp = protocol.channel_renamed(entry) + resp.id = frame.id + await self._ws_server.broadcast(resp) + + async def on_channel_delete(self, frame: protocol.Frame, device_id: str) -> None: + chat_id = frame.chat_id or frame.payload.get("chat_id") + if not isinstance(chat_id, str) or not chat_id.strip(): + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, "channel.delete requires chat_id", id=frame.id), + ) + return + entry = self._channels.delete(chat_id) + if entry is None: + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_NOT_FOUND, f"cannot delete {chat_id} (unknown or default)", id=frame.id), + ) + return + resp = protocol.channel_deleted(chat_id) + resp.id = frame.id + await self._ws_server.broadcast(resp) + + async def on_channel_list(self, frame: protocol.Frame, device_id: str) -> None: + channels = self._channels.list(include_archived=False) + resp = protocol.channel_list(channels) + resp.id = frame.id + await self._ws_server.send_to(device_id, resp) + + # ── M3: search (app -> agent) ───────────────────────────────────────── + + async def on_search(self, frame: protocol.Frame, device_id: str) -> None: + payload = frame.payload + query = payload.get("query") + if not isinstance(query, str) or not query.strip(): + await self._ws_server.send_to( + device_id, protocol.error(protocol.ERR_UNSUPPORTED, "search requires a query", id=frame.id) + ) + return + scope = payload.get("scope") + scope = scope if scope in ("all", "chat") else "all" + chat_id = payload.get("chat_id") or frame.chat_id + if not isinstance(chat_id, str) or not chat_id.strip(): + chat_id = None + thread_id = payload.get("thread_id") or frame.thread_id + if not isinstance(thread_id, str) or not thread_id.strip(): + thread_id = None + limit = payload.get("limit") + try: + limit = int(limit) if limit is not None else 20 + except (TypeError, ValueError): + limit = 20 + db_path = get_hermes_home() / "state.db" + hits = search_bridge.search( + db_path, query, scope=scope, chat_id=chat_id, thread_id=thread_id, limit=limit + ) + resp = protocol.search_results(query, scope, hits, id=frame.id) + await self._ws_server.send_to(device_id, resp) + + # ── M3: sync (reconnect catch-up) ───────────────────────────────────── + + async def on_sync(self, frame: protocol.Frame, device_id: str) -> None: + payload = frame.payload + cursor = payload.get("cursor") + try: + cursor = int(cursor) if cursor is not None else 0 + except (TypeError, ValueError): + cursor = 0 + for e in self._outbox.replay(cursor): + raw = e["frame"] + replayed = protocol.Frame( + type=raw.get("type", ""), + payload=raw.get("payload", {}) if isinstance(raw.get("payload"), dict) else {}, + id=raw.get("id") if isinstance(raw.get("id"), int) else None, + chat_id=raw.get("chat_id") if isinstance(raw.get("chat_id"), str) else e.get("chat_id"), + thread_id=raw.get("thread_id") if isinstance(raw.get("thread_id"), str) else None, + v=raw.get("v") if isinstance(raw.get("v"), int) else protocol.PROTOCOL_VERSION, + ) + await self._ws_server.send_to(device_id, replayed) + done = protocol.sync_done(self._outbox.latest_cursor(), id=frame.id) + await self._ws_server.send_to(device_id, done) + # ── Chat info ───────────────────────────────────────────────────────── def _channel_name(self, chat_id: str) -> str: - """Channel display name. M1: home channel only (directory is M3).""" + """Channel display name (M3: from the channel directory).""" + if not chat_id: + return "chat" + entry = self._channels.get(chat_id) + if entry is not None: + return entry["name"] if chat_id in (self.home_channel, DEFAULT_HOME_CHANNEL): return self.home_channel_name - return chat_id or "chat" + return chat_id async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: - """Return ``{name, type, chat_id}`` for a chat. - - M1: the channel directory is not persisted yet, so report the home - channel name for the default chat and the raw id otherwise. - """ + """Return ``{name, type, chat_id}`` for a chat (M3: directory-backed).""" + entry = self._channels.get(chat_id) + kind = entry["kind"] if entry else "channel" return { "name": self._channel_name(chat_id), - "type": "channel", + "type": "dm" if kind == "default" else "channel", "chat_id": chat_id, } # ── hello.ack helpers ───────────────────────────────────────────────── def server_caps(self) -> Dict[str, Any]: - """Capability flags advertised in ``hello.ack`` (M2 surface).""" + """Capability flags advertised in ``hello.ack`` (M3 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 - "search": False, # M3 + "search": True, # M3: search frame "push": self.push_backend, "pickers": False, # M2+ } def channel_list(self) -> List[Dict[str, Any]]: - """Channel directory for ``hello.ack``. M1: home channel only (M3).""" - return [ - { - "chat_id": self.home_channel, - "name": self.home_channel_name, - "kind": "default", - "is_default": True, - } - ] + """Channel directory for ``hello.ack`` (M3: full non-archived list).""" + return self._channels.list(include_archived=False) + + # ── M3: core channel-directory hook (cron / send_message name resolution) + + async def list_channels(self) -> List[Dict[str, Any]]: + """Expose the directory to the gateway's core channel directory. + + ``gateway/channel_directory.build_channel_directory`` calls this to + populate ``channel_directory.json``, which ``resolve_channel_name`` + reads for friendly-name -> chat_id resolution (cron + send_message). + Threads are addressed via the explicit ``android::`` + syntax (see ``_parse_target_ref``), so only channels are listed here. + """ + out: List[Dict[str, Any]] = [] + for entry in self._channels.list(include_archived=False): + if entry["kind"] == "thread": + continue + out.append( + { + "id": entry["chat_id"], + "name": entry["name"], + "type": "dm" if entry["kind"] == "default" else "channel", + } + ) + return out + + # ── M3: thread handoff (gateway create_handoff_thread) ──────────────── + + async def create_handoff_thread( + self, parent_chat_id: str, name: str + ) -> Optional[str]: + """Mint a named thread under *parent_chat_id* (gateway handoff path). + + Returns the new ``thread_id`` (``t_``) so the handed-off session is + isolated in its own lane, or ``None`` when the parent is unknown. + """ + parent = self._channels.get(parent_chat_id) + if parent is None: + # Unknown parent: still mint a thread under it so the handoff has a + # lane (the directory row is created lazily on first use). + parent_chat_id = parent_chat_id or self.home_channel + try: + entry = self._channels.create( + name=name or "Handoff", kind="thread", parent_chat_id=parent_chat_id + ) + except Exception: + logger.warning("android: create_handoff_thread failed", exc_info=True) + return None + await self._ws_server.broadcast(protocol.channel_created(entry)) + return entry["chat_id"] # --------------------------------------------------------------------------- diff --git a/gateway-plugin/channels.py b/gateway-plugin/channels.py new file mode 100644 index 0000000..956a84c --- /dev/null +++ b/gateway-plugin/channels.py @@ -0,0 +1,353 @@ +"""Channel directory (SQLite) -- the source of truth for the app's channel list. + +Maps app concepts onto hermes' existing ``chat_id`` / ``thread_id`` primitives +(docs/06-channels-cron-search.md §6.1): + +* **default chat** -> the home channel (``ANDROID_HOME_CHANNEL``, default + ``android:default``), ``kind="default"``, ``is_default=1``. +* **user channel** -> a minted ``chat_id = android:chan_``, ``kind="channel"``. +* **thread** -> a minted ``thread_id = t_`` under a ``chat_id``, + ``kind="thread"`` (stored with its ``parent_chat_id``). + +The directory is the single source of truth for the app's channel list and for +cron name resolution. It is also exposed to the gateway's core channel +directory (``gateway/channel_directory.py``) via the adapter's +``list_channels()`` hook, so ``send_message`` / cron can resolve a friendly +name (e.g. "Cron Reports") to a chat_id. + +Storage: ``get_hermes_home()/"android"/channels.db``. + +Milestone M3. +""" + +import logging +import sqlite3 +import threading +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +# Entry ``kind`` values. +KIND_DEFAULT = "default" +KIND_CHANNEL = "channel" +KIND_THREAD = "thread" + +# chat_id / thread_id minting prefixes. +CHANNEL_PREFIX = "android:chan_" +THREAD_PREFIX = "t_" + + +class ChannelDirectory: + """Persistent channel directory under ``get_hermes_home()/"android"``. + + Thread-safe (single connection + lock); all operations are small and fast + enough to run inline on the gateway's asyncio loop. Mirrors the + ``DeviceRegistry`` pattern (``pairing.py``). + """ + + def __init__(self, db_path: Path): + self._db_path = Path(db_path) + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) + self._conn.row_factory = sqlite3.Row + with self._lock: + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS channels ( + chat_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + kind TEXT NOT NULL DEFAULT 'channel', + parent_chat_id TEXT, + is_default INTEGER NOT NULL DEFAULT 0, + archived INTEGER NOT NULL DEFAULT 0, + created REAL NOT NULL DEFAULT 0 + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS counters ( + name TEXT PRIMARY KEY, + value INTEGER NOT NULL DEFAULT 0 + ) + """ + ) + self._conn.commit() + + # ── id minting ──────────────────────────────────────────────────────── + + def _next_counter(self, name: str) -> int: + with self._lock: + self._conn.execute( + "INSERT INTO counters (name, value) VALUES (?, 1) " + "ON CONFLICT(name) DO UPDATE SET value = value + 1", + (name,), + ) + row = self._conn.execute( + "SELECT value FROM counters WHERE name = ?", (name,) + ).fetchone() + self._conn.commit() + return int(row["value"]) if row else 1 + + def _mint_chat_id(self, kind: str) -> str: + if kind == KIND_THREAD: + return f"{THREAD_PREFIX}{self._next_counter('thread')}" + return f"{CHANNEL_PREFIX}{self._next_counter('chan')}" + + # ── default channel ─────────────────────────────────────────────────── + + def ensure_default(self, chat_id: str, name: str) -> Dict[str, Any]: + """Ensure the default (home) channel exists. Idempotent. + + If a row already exists for *chat_id* it is kept (name refreshed only + when it was still the auto default); if another row is marked default + it is cleared so exactly one default exists. + """ + chat_id = (chat_id or "android:default").strip() or "android:default" + name = (name or "Default").strip() or "Default" + with self._lock: + existing = self._conn.execute( + "SELECT name, is_default FROM channels WHERE chat_id = ?", + (chat_id,), + ).fetchone() + if existing is None: + self._conn.execute( + "INSERT INTO channels (chat_id, name, kind, is_default, created) " + "VALUES (?, ?, ?, 1, ?)", + (chat_id, name, KIND_DEFAULT, time.time()), + ) + else: + # Refresh the name only if it was never renamed by the user + # (heuristic: still equals the previous default name is not + # trackable, so leave user-renamed names alone). + self._conn.execute( + "UPDATE channels SET kind = ?, is_default = 1 WHERE chat_id = ?", + (KIND_DEFAULT, chat_id), + ) + # Exactly one default: clear any other default flag. + self._conn.execute( + "UPDATE channels SET is_default = 0 WHERE chat_id != ?", (chat_id,) + ) + self._conn.commit() + entry = self.get(chat_id) + if entry is not None: + return entry + return { + "chat_id": chat_id, "name": name, "kind": KIND_DEFAULT, + "parent_chat_id": None, "is_default": True, "archived": False, + "created": time.time(), + } + + # ── create / rename / set_default / delete ──────────────────────────── + + def create( + self, + name: str, + kind: str = KIND_CHANNEL, + parent_chat_id: Optional[str] = None, + ) -> Dict[str, Any]: + """Mint a new channel (or thread) and store it. Returns the entry.""" + name = (name or "").strip() + if not name: + raise ValueError("channel name required") + if kind not in (KIND_CHANNEL, KIND_THREAD): + kind = KIND_CHANNEL + if kind == KIND_THREAD and not parent_chat_id: + raise ValueError("thread requires a parent_chat_id") + chat_id = self._mint_chat_id(kind) + now = time.time() + with self._lock: + self._conn.execute( + "INSERT INTO channels (chat_id, name, kind, parent_chat_id, created) " + "VALUES (?, ?, ?, ?, ?)", + (chat_id, name, kind, parent_chat_id, now), + ) + self._conn.commit() + entry = self.get(chat_id) + if entry is not None: + return entry + return { + "chat_id": chat_id, "name": name, "kind": kind, + "parent_chat_id": parent_chat_id, "is_default": False, + "archived": False, "created": now, + } + + def rename(self, chat_id: str, name: str) -> Optional[Dict[str, Any]]: + name = (name or "").strip() + if not name: + raise ValueError("channel name required") + with self._lock: + cur = self._conn.execute( + "UPDATE channels SET name = ? WHERE chat_id = ? AND archived = 0", + (name, chat_id), + ) + self._conn.commit() + if cur.rowcount == 0: + return None + return self.get(chat_id) + + def set_default(self, chat_id: str) -> Optional[Dict[str, Any]]: + """Mark *chat_id* as the default channel (clears the previous one).""" + with self._lock: + row = self._conn.execute( + "SELECT 1 FROM channels WHERE chat_id = ? AND archived = 0", + (chat_id,), + ).fetchone() + if row is None: + return None + self._conn.execute("UPDATE channels SET is_default = 0") + self._conn.execute( + "UPDATE channels SET is_default = 1 WHERE chat_id = ?", (chat_id,) + ) + self._conn.commit() + return self.get(chat_id) + + def delete(self, chat_id: str) -> Optional[Dict[str, Any]]: + """Soft-delete (archive) a channel. History stays for search. + + The default channel cannot be deleted. Returns the (archived) entry, + or ``None`` when the id is unknown / is the default. + """ + with self._lock: + row = self._conn.execute( + "SELECT is_default FROM channels WHERE chat_id = ?", (chat_id,) + ).fetchone() + if row is None or row["is_default"]: + return None + self._conn.execute( + "UPDATE channels SET archived = 1 WHERE chat_id = ?", (chat_id,) + ) + self._conn.commit() + return self.get(chat_id) + + # ── reads ───────────────────────────────────────────────────────────── + + def get(self, chat_id: str) -> Optional[Dict[str, Any]]: + with self._lock: + row = self._conn.execute( + "SELECT * FROM channels WHERE chat_id = ?", (chat_id,) + ).fetchone() + return _row_to_entry(row) if row else None + + def list(self, include_archived: bool = False) -> List[Dict[str, Any]]: + """Directory listing. Default channel first, then creation order.""" + sql = "SELECT * FROM channels" + if not include_archived: + sql += " WHERE archived = 0" + sql += " ORDER BY is_default DESC, created ASC" + with self._lock: + rows = self._conn.execute(sql).fetchall() + return [_row_to_entry(r) for r in rows] + + def default(self) -> Optional[Dict[str, Any]]: + with self._lock: + row = self._conn.execute( + "SELECT * FROM channels WHERE is_default = 1 LIMIT 1" + ).fetchone() + return _row_to_entry(row) if row else None + + def threads_for(self, chat_id: str) -> List[Dict[str, Any]]: + """All (non-archived) threads under *chat_id*, oldest first.""" + with self._lock: + rows = self._conn.execute( + "SELECT * FROM channels WHERE kind = ? AND parent_chat_id = ? " + "AND archived = 0 ORDER BY created ASC", + (KIND_THREAD, chat_id), + ).fetchall() + return [_row_to_entry(r) for r in rows] + + def resolve_entry(self, name: str) -> Optional[Dict[str, Any]]: + """Resolve a friendly name to a directory entry (case-insensitive). + + Matches non-archived channels/threads by exact name first, then by + unique prefix. Returns the entry dict, or ``None`` when nothing (or + more than one) matches. + """ + query = (name or "").strip().lower() + if not query: + return None + with self._lock: + rows = self._conn.execute( + "SELECT * FROM channels WHERE archived = 0" + ).fetchall() + entries = [_row_to_entry(r) for r in rows] + exact = [e for e in entries if (e["name"] or "").strip().lower() == query] + if len(exact) == 1: + return exact[0] + if len(exact) > 1: + return None + prefix = [ + e for e in entries if (e["name"] or "").strip().lower().startswith(query) + ] + if len(prefix) == 1: + return prefix[0] + return None + + def resolve_name(self, name: str) -> Optional[str]: + """Resolve a friendly name to a valid chat_id (case-insensitive). + + For a thread, returns the *parent* chat_id (the thread's session lane + is ``parent_chat_id`` + ``thread_id``; the bare thread id is not a + standalone chat). Used by the core channel directory. + """ + entry = self.resolve_entry(name) + if entry is None: + return None + if entry["kind"] == KIND_THREAD: + return entry["parent_chat_id"] + return entry["chat_id"] + + def close(self) -> None: + with self._lock: + try: + self._conn.close() + except Exception: + pass + + +def _row_to_entry(row: sqlite3.Row) -> Dict[str, Any]: + return { + "chat_id": row["chat_id"], + "name": row["name"], + "kind": row["kind"], + "parent_chat_id": row["parent_chat_id"], + "is_default": bool(row["is_default"]), + "archived": bool(row["archived"]), + "created": row["created"], + } + + +# --------------------------------------------------------------------------- +# Module-level lazy singleton (profile-aware) +# +# ``parse_target_ref_fn`` is registered at plugin load (before any adapter is +# constructed) and needs name resolution, so it reaches the directory through +# this singleton. The adapter uses the same singleton so both paths agree. +# Keyed on ``get_hermes_home()`` so a profile switch rebuilds it. +# --------------------------------------------------------------------------- + +_directory: Optional[ChannelDirectory] = None +_directory_home: Optional[Path] = None +_directory_lock = threading.Lock() + + +def get_directory() -> ChannelDirectory: + """Return the process-wide channel directory for the active profile.""" + global _directory, _directory_home + from hermes_constants import get_hermes_home + + home = Path(get_hermes_home()) + with _directory_lock: + if _directory is None or _directory_home != home: + if _directory is not None: + try: + _directory.close() + except Exception: + pass + _directory = ChannelDirectory(home / "android" / "channels.db") + _directory_home = home + return _directory \ No newline at end of file diff --git a/gateway-plugin/outbox.py b/gateway-plugin/outbox.py index 2101dc0..a601929 100644 --- a/gateway-plugin/outbox.py +++ b/gateway-plugin/outbox.py @@ -1,10 +1,159 @@ -"""SQLite offline outbox + monotonic sync cursor. +"""SQLite offline outbox + monotonic sync cursor (M3). -Undelivered frames are appended per ``chat_id`` with a monotonic cursor so a -reconnecting app can ``sync {cursor}`` the delta without re-reading full -history. Retention prunes old entries (``outbox_retention_hours``). +Undelivered frames (sent while no device is live) are appended with a +**monotonic** cursor so a reconnecting app can ``sync {cursor}`` the delta +without re-reading full history. The cursor is a separate high-water counter +that only ever increases -- pruning old rows never resets it, so a late +reconnect can't be handed a cursor lower than one it already saw. + +Retention prunes rows older than ``outbox_retention_hours`` (default 72h). A +device offline longer than the window misses those frames; it recovers full +context via ``history`` (M5 wires push so the device is woken to sync). Storage: ``get_hermes_home()/"android"/outbox.db``. Milestone M3 (built), extended in M5 (push integration). -""" \ No newline at end of file +""" + +import json +import logging +import sqlite3 +import threading +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +DEFAULT_RETENTION_HOURS = 72 +_REPLAY_LIMIT = 1000 +_PRUNE_INTERVAL_S = 3600.0 + + +class Outbox: + """Persistent outbox under ``get_hermes_home()/"android"``. + + Thread-safe (single connection + lock); operations are small and fast + enough to run inline on the gateway's asyncio loop (mirrors + ``DeviceRegistry`` / ``ChannelDirectory``). + """ + + def __init__(self, db_path: Path, retention_hours: int = DEFAULT_RETENTION_HOURS): + self._db_path = Path(db_path) + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._retention_hours = max(1, int(retention_hours)) + self._lock = threading.Lock() + self._last_prune = 0.0 + self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) + self._conn.row_factory = sqlite3.Row + with self._lock: + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS outbox ( + cursor INTEGER PRIMARY KEY, + chat_id TEXT, + frame TEXT NOT NULL, + created REAL NOT NULL DEFAULT 0 + ) + """ + ) + self._conn.execute( + "CREATE INDEX IF NOT EXISTS idx_outbox_created ON outbox (created)" + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS counters ( + name TEXT PRIMARY KEY, + value INTEGER NOT NULL DEFAULT 0 + ) + """ + ) + self._conn.commit() + + # ── append / cursor ─────────────────────────────────────────────────── + + def append(self, chat_id: Optional[str], frame_json: str) -> int: + """Append a frame; returns the (monotonic) cursor assigned to it.""" + now = time.time() + with self._lock: + self._conn.execute( + "INSERT INTO counters (name, value) VALUES ('cursor', 1) " + "ON CONFLICT(name) DO UPDATE SET value = value + 1" + ) + row = self._conn.execute( + "SELECT value FROM counters WHERE name = 'cursor'" + ).fetchone() + cursor = int(row["value"]) if row else 1 + self._conn.execute( + "INSERT INTO outbox (cursor, chat_id, frame, created) " + "VALUES (?, ?, ?, ?)", + (cursor, chat_id, frame_json, now), + ) + self._conn.commit() + self._maybe_prune() + return cursor + + def latest_cursor(self) -> int: + """The high-water cursor (0 when nothing has been appended).""" + with self._lock: + row = self._conn.execute( + "SELECT value FROM counters WHERE name = 'cursor'" + ).fetchone() + return int(row["value"]) if row else 0 + + # ── replay ──────────────────────────────────────────────────────────── + + def replay(self, cursor: int, limit: int = _REPLAY_LIMIT) -> List[Dict[str, Any]]: + """Frames with ``cursor > `cursor```, oldest first. + + Each entry: ``{cursor, chat_id, frame}`` where ``frame`` is the parsed + frame dict (the caller re-serializes / forwards it to the device). + """ + cursor = max(0, int(cursor or 0)) + limit = max(1, min(int(limit or _REPLAY_LIMIT), _REPLAY_LIMIT)) + with self._lock: + rows = self._conn.execute( + "SELECT cursor, chat_id, frame FROM outbox " + "WHERE cursor > ? ORDER BY cursor ASC LIMIT ?", + (cursor, limit), + ).fetchall() + out: List[Dict[str, Any]] = [] + for r in rows: + try: + frame = json.loads(r["frame"]) + except (json.JSONDecodeError, TypeError): + continue + if not isinstance(frame, dict): + continue + out.append( + {"cursor": int(r["cursor"]), "chat_id": r["chat_id"], "frame": frame} + ) + return out + + # ── retention ───────────────────────────────────────────────────────── + + def _maybe_prune(self) -> None: + now = time.time() + if now - self._last_prune < _PRUNE_INTERVAL_S: + return + self._last_prune = now + cutoff = now - self._retention_hours * 3600 + with self._lock: + try: + self._conn.execute("DELETE FROM outbox WHERE created < ?", (cutoff,)) + self._conn.commit() + except sqlite3.Error as e: + logger.debug("android outbox: prune failed: %s", e) + + def prune(self) -> None: + """Force a retention prune (ignores the interval throttle).""" + self._last_prune = 0.0 + self._maybe_prune() + + def close(self) -> None: + with self._lock: + try: + self._conn.close() + except Exception: + pass \ No newline at end of file diff --git a/gateway-plugin/protocol.py b/gateway-plugin/protocol.py index 831f697..0c81f05 100644 --- a/gateway-plugin/protocol.py +++ b/gateway-plugin/protocol.py @@ -17,7 +17,7 @@ Milestone M5: notification, fcm.register, read.receipt. import json from dataclasses import dataclass, field -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional PROTOCOL_VERSION = 1 @@ -50,6 +50,24 @@ TYPE_TOOL_END = "tool.end" # Intermediate assistant beat (M2) TYPE_COMMENTARY = "commentary" +# Channels / threads (M3) +TYPE_CHANNEL_CREATE = "channel.create" +TYPE_CHANNEL_RENAME = "channel.rename" +TYPE_CHANNEL_SET_DEFAULT = "channel.set_default" +TYPE_CHANNEL_DELETE = "channel.delete" +TYPE_CHANNEL_CREATED = "channel.created" +TYPE_CHANNEL_RENAMED = "channel.renamed" +TYPE_CHANNEL_DELETED = "channel.deleted" +TYPE_CHANNEL_LIST = "channel.list" + +# Search (M3) +TYPE_SEARCH = "search" +TYPE_SEARCH_RESULTS = "search.results" + +# Reconnect catch-up (M3 outbox; extended by M5 push) +TYPE_SYNC = "sync" +TYPE_SYNC_DONE = "sync.done" + # --------------------------------------------------------------------------- # Error codes (``error`` frame payload.code) # --------------------------------------------------------------------------- @@ -352,6 +370,80 @@ def commentary( ) +# --------------------------------------------------------------------------- +# Channel directory frames (M3) +# --------------------------------------------------------------------------- + +def _channel_payload(entry: Dict[str, Any]) -> Dict[str, Any]: + """Project a directory entry onto the wire shape.""" + payload: Dict[str, Any] = { + "chat_id": entry.get("chat_id"), + "name": entry.get("name"), + "kind": entry.get("kind", "channel"), + } + if entry.get("parent_chat_id") is not None: + payload["parent_chat_id"] = entry["parent_chat_id"] + if entry.get("is_default"): + payload["is_default"] = True + if entry.get("archived"): + payload["archived"] = True + return payload + + +def channel_created(entry: Dict[str, Any]) -> Frame: + """Broadcast: a channel/thread was created.""" + return Frame(type=TYPE_CHANNEL_CREATED, payload=_channel_payload(entry)) + + +def channel_renamed(entry: Dict[str, Any]) -> Frame: + """Broadcast: a channel/thread was renamed.""" + return Frame(type=TYPE_CHANNEL_RENAMED, payload=_channel_payload(entry)) + + +def channel_deleted(chat_id: str) -> Frame: + """Broadcast: a channel was archived (soft-deleted).""" + return Frame(type=TYPE_CHANNEL_DELETED, payload={"chat_id": chat_id}) + + +def channel_list(channels: List[Dict[str, Any]]) -> Frame: + """Full directory (response to a ``channel.list`` request).""" + return Frame( + type=TYPE_CHANNEL_LIST, + payload={"channels": [_channel_payload(c) for c in channels]}, + ) + + +# --------------------------------------------------------------------------- +# Search frames (M3) +# --------------------------------------------------------------------------- + +def search_results( + query: str, + scope: str, + hits: List[Dict[str, Any]], + *, + id: Optional[int] = None, +) -> Frame: + """Response to a ``search`` request. + + Each hit: ``{message_id, chat_id, thread_id, role, snippet, ts}``. + """ + return Frame( + type=TYPE_SEARCH_RESULTS, + id=id, + payload={"query": query, "scope": scope, "hits": hits}, + ) + + +# --------------------------------------------------------------------------- +# Sync frames (M3 outbox) +# --------------------------------------------------------------------------- + +def sync_done(cursor: int, *, id: Optional[int] = None) -> Frame: + """Terminal frame of a ``sync`` replay: the new cursor to persist.""" + return Frame(type=TYPE_SYNC_DONE, id=id, payload={"cursor": cursor}) + + 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/search.py b/gateway-plugin/search.py index a1ff3ed..02f188b 100644 --- a/gateway-plugin/search.py +++ b/gateway-plugin/search.py @@ -1,8 +1,236 @@ -"""FTS5 session search bridge. +"""FTS5 session search bridge (M3). -Bridges the ``search {query, scope, chat_id?, thread_id?}`` frame to the -hermes session store (SQLite + FTS5, ``hermes_state_search.py``) and returns -``search.results``. Scope: "all" (everywhere) or "chat" (this chat/channel). +Bridges the ``search {query, scope, chat_id?, thread_id?}`` frame to the hermes +session store (SQLite + FTS5, ``hermes_state.py`` / ``hermes_state_search.py``) +and returns ``search.results`` hits. + +The session DB (``get_hermes_home()/"state.db"``) is opened **read-only** -- +search never writes to the store. Two query paths: + +* **FTS5** (primary): ``messages_fts MATCH `` with BM25 ranking. +* **LIKE** (fallback): when the FTS5 table is absent (FTS disabled / fresh DB) + or the MATCH raises, a substring scan over ``messages.content``. + +Scope: +* ``"all"`` -- every channel/thread/session. +* ``"chat"`` -- restrict to the given ``chat_id`` (and optional ``thread_id``). + +Each hit: ``{message_id, chat_id, thread_id, role, snippet, ts}`` where +``message_id`` is the session-store row id (string) and ``ts`` is epoch +milliseconds. The app navigates to the hit's channel/thread and matches the +message by timestamp to scroll + highlight. + +Privacy: search is local to the user's own hermes home; no data leaves the +machine. Milestone M3. -""" \ No newline at end of file +""" + +import logging +import re +import sqlite3 +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +MAX_QUERY_CHARS = 200 +DEFAULT_LIMIT = 20 +MAX_LIMIT = 100 + +# FTS5 special chars (mirror of hermes_state_search._FTS5_SPECIAL_CHARS) for the +# fallback sanitizer when the real one can't be imported. +_FTS5_SPECIAL_CHARS = '+{}():"^@/#&|~[]<>,;!?$=\\\'' +_FTS5_SPECIAL_RE = re.compile(f"[{re.escape(_FTS5_SPECIAL_CHARS)}]") + + +def _sanitize(query: str) -> str: + """Sanitize user input for a safe FTS5 MATCH. + + Prefers the gateway's own sanitizer (exact parity with session_search); + falls back to a simplified strip-and-quote pass when it can't be imported. + """ + q = (query or "").strip() + if not q: + return "" + try: + from hermes_state_search import SessionSearchMixin + + return SessionSearchMixin._sanitize_fts5_query(q) + except Exception: + return _sanitize_fallback(q) + + +def _sanitize_fallback(query: str) -> str: + q = query[:MAX_QUERY_CHARS] + q = _FTS5_SPECIAL_RE.sub(" ", q) + if "%" in q: + q = q.replace("%", " ") + q = re.sub(r"\*+", "*", q) + q = re.sub(r"(^|\s)\*", r"\1", q) + q = re.sub(r"(?i)^(AND|OR|NOT)\b\s*", "", q.strip()) + q = re.sub(r"(?i)\s+(AND|OR|NOT)\s*$", "", q.strip()) + q = re.sub(r"\b(\w+(?:[._-]\w+)+)\b", r'"\1"', q) + return q.strip() + + +def _fts_available(conn: sqlite3.Connection) -> bool: + try: + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' " + "AND name = 'messages_fts' LIMIT 1" + ).fetchone() + return row is not None + except sqlite3.Error: + return False + + +def _scope_clauses( + scope: str, chat_id: Optional[str], thread_id: Optional[str] +) -> Tuple[List[str], List[Any]]: + """Build the scope WHERE clauses + params (empty for scope='all').""" + clauses: List[str] = [] + params: List[Any] = [] + if scope == "chat" and chat_id: + clauses.append("s.chat_id = ?") + params.append(chat_id) + if thread_id: + clauses.append("s.thread_id = ?") + params.append(thread_id) + return clauses, params + + +def _row_to_hit(row: sqlite3.Row) -> Dict[str, Any]: + ts = row["timestamp"] + try: + ts_ms = int(float(ts) * 1000) + except (TypeError, ValueError): + ts_ms = 0 + return { + "message_id": str(row["id"]), + "chat_id": row["chat_id"], + "thread_id": row["thread_id"], + "role": row["role"], + "snippet": row["snippet"] or "", + "ts": ts_ms, + } + + +def _fts_query( + conn: sqlite3.Connection, + query: str, + scope: str, + chat_id: Optional[str], + thread_id: Optional[str], + limit: int, +) -> List[Dict[str, Any]]: + where = ["messages_fts MATCH ?", "(m.active = 1 OR m.compacted = 1)"] + params: List[Any] = [query] + scope_clauses, scope_params = _scope_clauses(scope, chat_id, thread_id) + where.extend(scope_clauses) + params.extend(scope_params) + params.extend([limit]) + sql = f""" + SELECT + m.id, + m.role, + snippet(messages_fts, -1, '>>>', '<<<', '...', 40) AS snippet, + m.timestamp, + s.chat_id, + s.thread_id + FROM messages_fts + JOIN messages m ON m.id = messages_fts.rowid + JOIN sessions s ON s.id = m.session_id + WHERE {' AND '.join(where)} + ORDER BY rank + LIMIT ? + """ + rows = conn.execute(sql, params).fetchall() + return [_row_to_hit(r) for r in rows] + + +def _like_query( + conn: sqlite3.Connection, + query: str, + scope: str, + chat_id: Optional[str], + thread_id: Optional[str], + limit: int, +) -> List[Dict[str, Any]]: + """Substring fallback when FTS5 is unavailable.""" + # First plain word of the query is the LIKE needle (best-effort). + needle = re.split(r"\s+", query.strip(), maxsplit=1)[0].strip('"') + if not needle: + return [] + like = f"%{needle}%" + where = ["(m.active = 1 OR m.compacted = 1)", "m.content LIKE ?"] + params: List[Any] = [like] + scope_clauses, scope_params = _scope_clauses(scope, chat_id, thread_id) + where.extend(scope_clauses) + params.extend(scope_params) + params.extend([limit]) + sql = f""" + SELECT + m.id, + m.role, + substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet, + m.timestamp, + s.chat_id, + s.thread_id + FROM messages m + JOIN sessions s ON s.id = m.session_id + WHERE {' AND '.join(where)} + ORDER BY m.timestamp DESC + LIMIT ? + """ + # The needle appears twice (LIKE + instr); params order: like, scope..., needle, limit + full_params = [like, *scope_params, needle, limit] + rows = conn.execute(sql, full_params).fetchall() + return [_row_to_hit(r) for r in rows] + + +def search( + db_path: Path, + query: str, + scope: str = "all", + chat_id: Optional[str] = None, + thread_id: Optional[str] = None, + limit: int = DEFAULT_LIMIT, +) -> List[Dict[str, Any]]: + """Run a scoped search over the session store. Returns a list of hits. + + Never raises: any DB/FTS error yields an empty result (the caller sends an + empty ``search.results``). + """ + sanitized = _sanitize(query) + if not sanitized: + return [] + db_path = Path(db_path) + if not db_path.exists(): + return [] + limit = max(1, min(int(limit or DEFAULT_LIMIT), MAX_LIMIT)) + scope = (scope or "all").strip().lower() + if scope not in ("all", "chat"): + scope = "all" + + try: + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + except sqlite3.Error as e: + logger.warning("android search: open failed: %s", e) + return [] + conn.row_factory = sqlite3.Row + try: + if _fts_available(conn): + try: + return _fts_query(conn, sanitized, scope, chat_id, thread_id, limit) + except sqlite3.Error as e: + logger.debug("android search: FTS5 failed, using LIKE: %s", e) + return _like_query(conn, sanitized, scope, chat_id, thread_id, limit) + except sqlite3.Error as e: + logger.warning("android search: query failed: %s", e) + return [] + finally: + try: + conn.close() + except Exception: + pass \ No newline at end of file diff --git a/gateway-plugin/ws_server.py b/gateway-plugin/ws_server.py index 0fe14af..d7b88f5 100644 --- a/gateway-plugin/ws_server.py +++ b/gateway-plugin/ws_server.py @@ -240,7 +240,7 @@ class WsServer: ack = protocol.hello_ack( server_caps=self._adapter.server_caps(), - sync_cursor=0, # outbox lands in M3; cursor starts at 0 + sync_cursor=self._adapter._outbox.latest_cursor(), channels=self._adapter.channel_list(), ) try: @@ -276,6 +276,20 @@ class WsServer: await self._send_quiet(ws, protocol.pong(ts if isinstance(ts, int) else None)) elif frame.type == protocol.TYPE_MESSAGE_SEND: await self._adapter.on_message_send(frame, device_id) + elif frame.type == protocol.TYPE_CHANNEL_CREATE: + await self._adapter.on_channel_create(frame, device_id) + elif frame.type == protocol.TYPE_CHANNEL_RENAME: + await self._adapter.on_channel_rename(frame, device_id) + elif frame.type == protocol.TYPE_CHANNEL_SET_DEFAULT: + await self._adapter.on_channel_set_default(frame, device_id) + elif frame.type == protocol.TYPE_CHANNEL_DELETE: + await self._adapter.on_channel_delete(frame, device_id) + elif frame.type == protocol.TYPE_CHANNEL_LIST: + await self._adapter.on_channel_list(frame, device_id) + elif frame.type == protocol.TYPE_SEARCH: + 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 == "fcm.register": fcm_token = frame.payload.get("fcm_token") ntfy_topic = frame.payload.get("ntfy_topic")