From 218c50d688aaab0aa7087826bc3743f39cb33bf3 Mon Sep 17 00:00:00 2001 From: ARIA Date: Wed, 19 Aug 2026 13:56:19 +0200 Subject: [PATCH] M1+M2: gateway core loop + agent transparency M1 (gateway core loop / text round-trip): - WS server (ws_server.py): bind, hello auth (constant-time), hello.ack, heartbeat, connection registry - pairing.py: token generation, pairing store, QR payload - adapter.py: send() -> message frame; inbound message.send -> MessageEvent -> handle_message - app: Connect screen, GatewayClient (connect + reconnect), ChatScreen send/render, SecureStore (Android/Desktop) - tests/ws_probe.py: probe harness driving a real turn M2 (streaming + reasoning + tools + commentary): - protocol.py: M2 frame types (message.start/update/stop, tool.start/progress/end, commentary) - adapter.py: per-chat turn-state machine; classify outbound into frames; _split_reasoning; tool-line parsing - reasoning in streaming: capture via on_stream_delta hook (kind=reasoning, gated by plugins.stream_reasoning_deltas) with a FIFO barrier, attach to message.stop - app: live streaming bubble, ReasoningBlock (collapse + copy), ToolCard (Everything/Truncated/Nothing), dimmed commentary, typing - docs/14-milestones.md: M1/M2 marked done; reasoning note corrected --- app/androidApp/src/main/AndroidManifest.xml | 1 + .../main/kotlin/dev/iris/app/MainActivity.kt | 4 +- app/build.gradle.kts | 1 + .../src/main/kotlin/iris/desktop/Main.kt | 4 +- app/shared/build.gradle.kts | 9 + .../iris/platform/AndroidSecureStore.kt | 50 ++ .../src/commonMain/kotlin/iris/IrisApp.kt | 77 +- .../commonMain/kotlin/iris/data/ChatStore.kt | 263 ++++++ .../kotlin/iris/data/SecureStore.kt | 23 + .../kotlin/iris/net/GatewayClient.kt | 281 ++++++ .../kotlin/iris/protocol/Protocol.kt | 238 +++++ .../kotlin/iris/state/IrisController.kt | 122 +++ .../kotlin/iris/ui/screens/ChatScreen.kt | 385 +++++++++ .../kotlin/iris/ui/screens/ConnectScreen.kt | 124 +++ .../iris/platform/DesktopSecureStore.kt | 75 ++ docs/14-milestones.md | 46 +- gateway-plugin/adapter.py | 816 ++++++++++++++++-- gateway-plugin/pairing.py | 181 +++- gateway-plugin/protocol.py | 349 +++++++- gateway-plugin/tests/ws_probe.py | 202 +++++ gateway-plugin/ws_server.py | 315 ++++++- 21 files changed, 3437 insertions(+), 129 deletions(-) create mode 100644 app/shared/src/androidMain/kotlin/iris/platform/AndroidSecureStore.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/data/SecureStore.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/state/IrisController.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt create mode 100644 app/shared/src/commonMain/kotlin/iris/ui/screens/ConnectScreen.kt create mode 100644 app/shared/src/desktopMain/kotlin/iris/platform/DesktopSecureStore.kt create mode 100644 gateway-plugin/tests/ws_probe.py diff --git a/app/androidApp/src/main/AndroidManifest.xml b/app/androidApp/src/main/AndroidManifest.xml index fda8506..c49802e 100644 --- a/app/androidApp/src/main/AndroidManifest.xml +++ b/app/androidApp/src/main/AndroidManifest.xml @@ -7,6 +7,7 @@ + ConnectScreen(controller, prefillUrl = store.serverUrl, prefillToken = store.token) + is GatewayClient.State.AuthFailed -> + ConnectScreen( + controller, + prefillUrl = store.serverUrl, + prefillToken = store.token, + initialError = "Pairing rejected: ${s.message}", + ) + else -> ChatScreen(controller) } } } diff --git a/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt b/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt new file mode 100644 index 0000000..1067896 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/data/ChatStore.kt @@ -0,0 +1,263 @@ +package iris.data + +import iris.protocol.CommentaryPayload +import iris.protocol.Frame +import iris.protocol.MessagePayload +import iris.protocol.MessageStartPayload +import iris.protocol.MessageStopPayload +import iris.protocol.MessageUpdatePayload +import iris.protocol.ROLE_ASSISTANT +import iris.protocol.ROLE_USER +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_TOOL_END +import iris.protocol.TYPE_TOOL_PROGRESS +import iris.protocol.TYPE_TOOL_START +import iris.protocol.ToolEndPayload +import iris.protocol.ToolProgressPayload +import iris.protocol.ToolStartPayload +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +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. + */ +sealed interface ChatItem { + val id: String +} + +/** A chat message (user / assistant / commentary / streaming bubble). */ +data class MessageItem( + override val id: String, + val role: String, + val text: String, + val ts: Long, + val pending: Boolean = false, + val reasoning: String? = null, + val isCommentary: Boolean = false, + val streaming: Boolean = false, + val model: String? = null, + val tokens: Int? = null, +) : ChatItem + +/** A structured tool-activity card (spinner until [done]). */ +data class ToolItem( + override val id: String, + val index: Int, + val name: String, + val preview: String? = null, + val args: JsonElement? = null, + val note: String? = null, + val done: Boolean = false, + val ok: Boolean = true, + val duration: Double? = null, + val outputPreview: String? = null, +) : ChatItem + +class ChatStore { + private val _items = MutableStateFlow>(emptyList()) + val items: StateFlow> = _items.asStateFlow() + + private var localSeq = 0 + private var toolSeq = 0 + + /** Optimistic add: show the user's message immediately (pending). */ + fun addPending(text: String): String { + localSeq++ + val id = "local_$localSeq" + _items.value = _items.value + + MessageItem(id = id, role = ROLE_USER, text = text, ts = 0, pending = true) + return id + } + + /** Reconcile a server frame into the cache. */ + fun onFrame(frame: Frame) { + 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) + else -> Unit + } + } + + // ── message (final / standalone, incl. non-streaming + user echo) ───── + + private fun onMessage(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, + ) + _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) { + 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 + } + + // ── message.update (replace live bubble text; full snapshot) ────────── + + private fun onMessageUpdate(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 + } + + // ── message.stop (finalize the live bubble) ─────────────────────────── + + private fun onMessageStop(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( + 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 + } + 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) { + val p = frame.payloadAs() ?: return + toolSeq++ + val id = "tool_$toolSeq" + val list = _items.value.toMutableList() + list.add( + 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) { + 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 + } + + // ── tool.end (mark the tool card complete) ──────────────────────────── + + private fun onToolEnd(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 + } + + // ── commentary (dimmed interim beat) ────────────────────────────────── + + private fun onCommentary(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 = 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)}" + } +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/data/SecureStore.kt b/app/shared/src/commonMain/kotlin/iris/data/SecureStore.kt new file mode 100644 index 0000000..0a8b99e --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/data/SecureStore.kt @@ -0,0 +1,23 @@ +package iris.data + +/** + * Pairing settings storage. The token is a secret: platform actuals keep it + * in secure storage (EncryptedSharedPreferences on Android — M5; plain + * SharedPreferences for M1 dev, file on desktop). + */ +interface SecureStore { + /** ws(s)://host:port/ws */ + var serverUrl: String + + /** ANDROID_TOKEN presented in the hello frame. */ + var token: String + + /** Stable app-generated device id (persisted). */ + val deviceId: String + + /** Human-readable device name (e.g. "MIX 2S"). */ + val deviceName: String + + fun savePairing(url: String, token: String) + fun clear() +} \ 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 new file mode 100644 index 0000000..d6f4550 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/net/GatewayClient.kt @@ -0,0 +1,281 @@ +package iris.net + +import iris.data.SecureStore +import iris.protocol.ChannelInfo +import iris.protocol.ErrorPayload +import iris.protocol.Frame +import iris.protocol.HelloAckPayload +import iris.protocol.IrisJson +import iris.protocol.ServerCaps +import iris.protocol.TYPE_ERROR +import iris.protocol.TYPE_HELLO_ACK +import iris.protocol.TYPE_PONG +import iris.protocol.helloFrame +import iris.protocol.messageSendFrame +import iris.protocol.pingFrame +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.withTimeoutOrNull +import kotlin.time.TimeMark +import kotlin.time.TimeSource +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import java.util.concurrent.TimeUnit +import kotlin.random.Random + +/** + * OkHttp WebSocket client for the hermes android gateway (docs/10 §10.3). + * + * - connect + hello (real auth leg), hello.ack + * - reconnect: exponential backoff + jitter; re-hello on every (re)connect + * - heartbeat: app-level ping every 20s; reap after ~60s of silence + * - events: server frames (minus hello.ack) on [events] + * - request/response correlation by id (M2+ consumers) + */ +class GatewayClient( + private val scope: CoroutineScope, + private val store: SecureStore, +) { + sealed interface State { + data object Disconnected : State + data object Connecting : State + data class Connected(val caps: ServerCaps, val channels: List) : State + data object Reconnecting : State + data class AuthFailed(val message: String) : State + } + + private val _state = MutableStateFlow(State.Disconnected) + val state: StateFlow = _state.asStateFlow() + + private val _events = MutableSharedFlow(extraBufferCapacity = 128) + val events: SharedFlow = _events.asSharedFlow() + + private val client: OkHttpClient = OkHttpClient.Builder() + .pingInterval(20, TimeUnit.SECONDS) + .build() + + private var connectJob: Job? = null + private var socket: WebSocket? = null + private var nextRequestId = 1 + private var attempt = 0 + private var lastLiveness: TimeMark = TimeSource.Monotonic.markNow() + private val pending = mutableMapOf>() + + // ── Lifecycle ───────────────────────────────────────────────────────── + + /** Start (or resume) the connect loop. Idempotent. */ + fun start() { + if (connectJob?.isActive == true) return + attempt = 0 + connectJob = scope.launch { connectLoop() } + } + + /** Stop the connect loop and close the socket. */ + fun stop() { + connectJob?.cancel() + connectJob = null + socket?.close(1000, "client shutdown") + socket = null + _state.value = State.Disconnected + } + + /** Re-pair: stop, then start fresh (used after saving new settings). */ + fun restart() { + stop() + start() + } + + private suspend fun connectLoop() { + while (currentCoroutineContext().isActive) { + val url = store.serverUrl.trim() + val token = store.token + if (url.isBlank() || token.isBlank()) { + _state.value = State.Disconnected + return + } + _state.value = if (attempt == 0) State.Connecting else State.Reconnecting + val dial = dial(url, token) + when (val result = dial.result) { + is DialResult.AuthFailed -> { + _state.value = State.AuthFailed(result.message) + dial.socket.close(1000, "auth failed") + return + } + DialResult.Connected -> { + attempt = 0 + lastLiveness = TimeSource.Monotonic.markNow() + dial.closed.await() + if (!currentCoroutineContext().isActive) return + // socket dropped -> loop again (Reconnecting) + } + is DialResult.Failed -> { + attempt++ + delay(backoffMs(attempt)) + } + } + } + } + + // ── Dial (one connect + hello) ──────────────────────────────────────── + + private sealed interface DialResult { + data object Connected : DialResult + data class AuthFailed(val message: String) : DialResult + data class Failed(val message: String) : DialResult + } + + private data class Dial( + val result: DialResult, + val socket: WebSocket, + val closed: CompletableDeferred, + ) + + private suspend fun dial(url: String, token: String): Dial { + val closed = CompletableDeferred() + val helloAck = CompletableDeferred() + val authError = CompletableDeferred() + val fail = CompletableDeferred() + + val request = Request.Builder().url(url).build() + val ws = client.newWebSocket( + request, + object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + webSocket.send(helloFrame(token, store.deviceId, store.deviceName).toWire()) + } + + override fun onMessage(webSocket: WebSocket, text: String) { + lastLiveness = TimeSource.Monotonic.markNow() + val frame = try { + IrisJson.instance.decodeFromString(Frame.serializer(), text) + } catch (_: Exception) { + return + } + when (frame.type) { + TYPE_HELLO_ACK -> { + val ack = frame.payloadAs() + if (ack != null) helloAck.complete(ack) + } + TYPE_ERROR -> { + val err = frame.payloadAs() + authError.complete(err?.message ?: "auth failed") + } + TYPE_PONG -> Unit + else -> { + _events.tryEmit(frame) + frame.id?.let { pending[it]?.complete(frame) } + } + } + } + + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + closed.complete(Unit) + } + + override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) { + fail.complete(t.message ?: "connection failed") + closed.complete(Unit) + } + }, + ) + socket = ws + + val winner = CompletableDeferred() + helloAck.invokeOnCompletion { e -> + if (e == null) { + val ack = helloAck.getCompleted() + _state.value = State.Connected(ack.serverCaps, ack.channels) + winner.complete(DialResult.Connected) + } + } + authError.invokeOnCompletion { e -> + if (e == null) winner.complete(DialResult.AuthFailed(authError.getCompleted())) + } + fail.invokeOnCompletion { e -> + if (e == null) winner.complete(DialResult.Failed(fail.getCompleted())) + } + val result = withTimeoutOrNull(15_000) { winner.await() } + ?: DialResult.Failed("timeout waiting for hello.ack") + return Dial(result, ws, closed) + } + + // ── Outbound ────────────────────────────────────────────────────────── + + /** Send a text message (fire-and-forget; the server echoes it back). */ + fun sendMessage(chatId: String, text: String, threadId: String? = null) { + val ws = socket ?: return + val id = nextRequestId++ + ws.send(messageSendFrame(id, chatId, text, threadId).toWire()) + } + + /** Send a ping (heartbeat). */ + fun ping() { + socket?.send(pingFrame().toWire()) + } + + /** True when the socket has been silent for [timeoutMs] (heartbeat reap). */ + fun isStale(timeoutMs: Long = 60_000): Boolean = + _state.value is State.Connected && lastLiveness.elapsedNow().inWholeMilliseconds > timeoutMs + + fun reapStale() { + if (isStale()) { + socket?.close(1000, "heartbeat timeout") + } + } + + // ── One-shot hello test (Connect screen) ────────────────────────────── + + /** + * Real `hello` test: dial, wait for hello.ack (or auth error), close. + * Exercises the auth leg, not just TCP (docs/10 §10.8). + */ + suspend fun testHello(url: String, token: String): Result { + val dial = dial(url, token) + return when (val result = dial.result) { + DialResult.Connected -> { + dial.socket.close(1000, "test complete") + Result.success(Unit) + } + is DialResult.AuthFailed -> Result.failure(IllegalStateException(result.message)) + is DialResult.Failed -> Result.failure(IllegalStateException(result.message)) + } + } + + // ── Heartbeat job ───────────────────────────────────────────────────── + + fun startHeartbeat() { + scope.launch { + while (isActive) { + delay(20_000) + if (_state.value is State.Connected) { + ping() + reapStale() + } + } + } + } + + // ── Helpers ─────────────────────────────────────────────────────────── + + private fun backoffMs(attempt: Int): Long { + val base = 1_000L * (1L shl minOf(attempt, 5)) // 1s..32s + val capped = minOf(base, 30_000L) + return capped + Random.nextLong(0, 500) + } +} + +private fun Frame.toWire(): String = IrisJson.instance.encodeToString(Frame.serializer(), this) \ 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 new file mode 100644 index 0000000..4fc9eb2 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/protocol/Protocol.kt @@ -0,0 +1,238 @@ +package iris.protocol + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put + +/** + * Wire protocol frames (mirror of gateway-plugin/protocol.py). + * See docs/04-wire-protocol.md. M1: hello/hello.ack, message, message.send, + * error, ping/pong, typing. M2: message.start/update/stop, tool.start/ + * progress/end, commentary, reasoning (on message / message.stop). + */ + +const val PROTOCOL_VERSION = 1 + +object IrisJson { + val instance: Json = Json { + ignoreUnknownKeys = true + encodeDefaults = true + isLenient = true + } +} + +// ── Frame type constants ──────────────────────────────────────────────── + +const val TYPE_HELLO = "hello" +const val TYPE_HELLO_ACK = "hello.ack" +const val TYPE_MESSAGE = "message" +const val TYPE_MESSAGE_SEND = "message.send" +const val TYPE_ERROR = "error" +const val TYPE_PING = "ping" +const val TYPE_PONG = "pong" +const val TYPE_TYPING = "typing" + +// M2 — streaming / tools / commentary +const val TYPE_MESSAGE_START = "message.start" +const val TYPE_MESSAGE_UPDATE = "message.update" +const val TYPE_MESSAGE_STOP = "message.stop" +const val TYPE_TOOL_START = "tool.start" +const val TYPE_TOOL_PROGRESS = "tool.progress" +const val TYPE_TOOL_END = "tool.end" +const val TYPE_COMMENTARY = "commentary" + +// ── Error codes ───────────────────────────────────────────────────────── + +const val ERR_AUTH = "auth" +const val ERR_NOT_FOUND = "not_found" +const val ERR_UNSUPPORTED = "unsupported" +const val ERR_INTERNAL = "internal" + +// ── Roles ─────────────────────────────────────────────────────────────── + +const val ROLE_USER = "user" +const val ROLE_ASSISTANT = "assistant" + +// ── Envelope ──────────────────────────────────────────────────────────── + +@Serializable +data class Frame( + val v: Int = PROTOCOL_VERSION, + val id: Int? = null, + val type: String, + @SerialName("chat_id") val chatId: String? = null, + @SerialName("thread_id") val threadId: String? = null, + val payload: JsonElement = JsonObject(emptyMap()), +) { + /** Payload as a JSON object (the wire format); parse per-type with + * [IrisJson]. Unknown fields are ignored (forward-compat). */ + inline fun payloadAs(): T? = + try { + IrisJson.instance.decodeFromString(payload.toString()) + } catch (_: Exception) { + null + } +} + +// ── hello (app -> server) ─────────────────────────────────────────────── + +@Serializable +data class HelloPayload( + val token: String, + @SerialName("device_id") val deviceId: String, + @SerialName("device_name") val deviceName: String, + val caps: JsonElement = buildJsonObject { put("min_protocol", JsonPrimitive(1)) }, + @SerialName("fcm_token") val fcmToken: String? = null, + @SerialName("ntfy_topic") val ntfyTopic: String? = null, +) + +// ── hello.ack (server -> app) ─────────────────────────────────────────── + +@Serializable +data class ServerCaps( + val streaming: Boolean = false, + val reasoning: Boolean = false, + val tools: Boolean = false, + val media: Boolean = false, + val search: Boolean = false, + val push: String = "fcm", + val pickers: Boolean = false, +) + +@Serializable +data class ChannelInfo( + @SerialName("chat_id") val chatId: String, + val name: String, + val kind: String = "channel", + @SerialName("is_default") val isDefault: Boolean = false, +) + +@Serializable +data class HelloAckPayload( + @SerialName("server_caps") val serverCaps: ServerCaps = ServerCaps(), + @SerialName("sync_cursor") val syncCursor: Long = 0, + val channels: List = emptyList(), +) + +// ── message (server -> app) ───────────────────────────────────────────── + +@Serializable +data class MessagePayload( + @SerialName("message_id") val messageId: String, + val role: String, + val text: String, + val reasoning: String? = null, + @SerialName("reply_to") val replyTo: String? = null, + val model: String? = null, + val tokens: Int? = null, + val ts: Long? = null, +) + +// ── M2: streaming frames (server -> app) ──────────────────────────────── + +@Serializable +data class MessageStartPayload( + @SerialName("message_id") val messageId: String, + val role: String = ROLE_ASSISTANT, +) + +@Serializable +data class MessageUpdatePayload( + @SerialName("message_id") val messageId: String, + val text: String, +) + +@Serializable +data class MessageStopPayload( + @SerialName("message_id") val messageId: String, + @SerialName("final_text") val finalText: String, + val reasoning: String? = null, + val model: String? = null, + val tokens: Int? = null, + val ts: Long? = null, +) + +// ── M2: tool activity frames (server -> app) ──────────────────────────── + +@Serializable +data class ToolStartPayload( + val index: Int, + val name: String, + val preview: String? = null, + val args: JsonElement? = null, +) + +@Serializable +data class ToolProgressPayload( + val index: Int, + val name: String, + val note: String? = null, +) + +@Serializable +data class ToolEndPayload( + val index: Int, + val name: String, + val ok: Boolean = true, + val duration: Double? = null, + @SerialName("output_preview") val outputPreview: String? = null, +) + +// ── M2: commentary frame (server -> app) ──────────────────────────────── + +@Serializable +data class CommentaryPayload( + @SerialName("message_id") val messageId: String, + val text: String, +) + +// ── message.send (app -> server) ──────────────────────────────────────── + +@Serializable +data class MessageSendPayload( + val text: String, + @SerialName("reply_to") val replyTo: String? = null, + @SerialName("media_refs") val mediaRefs: List = emptyList(), +) + +// ── typing / error / ping ─────────────────────────────────────────────── + +@Serializable +data class TypingPayload(val on: Boolean) + +@Serializable +data class ErrorPayload(val code: String, val message: String) + +@Serializable +data class PingPayload(val ts: Long? = null) + +// ── Frame builders ────────────────────────────────────────────────────── + +fun helloFrame(token: String, deviceId: String, deviceName: String): Frame = + Frame( + type = TYPE_HELLO, + payload = IrisJson.instance.encodeToJsonElement( + HelloPayload.serializer(), + HelloPayload(token = token, deviceId = deviceId, deviceName = deviceName), + ), + ) + +fun messageSendFrame(id: Int, chatId: String, text: String, threadId: String? = null): Frame = + Frame( + id = id, + type = TYPE_MESSAGE_SEND, + chatId = chatId, + threadId = threadId, + payload = IrisJson.instance.encodeToJsonElement( + MessageSendPayload.serializer(), + MessageSendPayload(text = text), + ), + ) + +fun pingFrame(): Frame = + Frame(type = TYPE_PING, payload = IrisJson.instance.encodeToJsonElement(PingPayload.serializer(), PingPayload())) \ 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 new file mode 100644 index 0000000..7fb00c8 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/state/IrisController.kt @@ -0,0 +1,122 @@ +package iris.state + +import iris.data.ChatStore +import iris.data.SecureStore +import iris.net.GatewayClient +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_TOOL_END +import iris.protocol.TYPE_TOOL_PROGRESS +import iris.protocol.TYPE_TOOL_START +import iris.protocol.TYPE_TYPING +import iris.protocol.TypingPayload +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +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. + */ +class IrisController( + private val store: SecureStore, +) { + private val job = SupervisorJob() + private val scope = CoroutineScope(job + Dispatchers.Default) + + val client = GatewayClient(scope, store) + val chat = ChatStore() + + private val _typing = MutableStateFlow(false) + val typing: StateFlow = _typing.asStateFlow() + + /** Tool-card verbosity (M2). NOTHING suppresses tool cards entirely. */ + private val _toolDetail = MutableStateFlow(ToolDetail.TRUNCATED) + val toolDetail: StateFlow = _toolDetail.asStateFlow() + + fun cycleToolDetail() { + _toolDetail.value = when (_toolDetail.value) { + ToolDetail.EVERYTHING -> ToolDetail.TRUNCATED + ToolDetail.TRUNCATED -> ToolDetail.NOTHING + ToolDetail.NOTHING -> ToolDetail.EVERYTHING + } + } + + /** Home channel id (from hello.ack; default until then). */ + private val _homeChannel = MutableStateFlow("android:default") + val homeChannel: StateFlow = _homeChannel.asStateFlow() + + init { + scope.launch { + client.events.collect { frame -> + when (frame.type) { + TYPE_MESSAGE, + TYPE_MESSAGE_START, + TYPE_MESSAGE_UPDATE, + TYPE_MESSAGE_STOP, + TYPE_TOOL_START, + TYPE_TOOL_PROGRESS, + TYPE_TOOL_END, + TYPE_COMMENTARY -> chat.onFrame(frame) + TYPE_TYPING -> { + frame.payloadAs()?.let { _typing.value = it.on } + } + else -> Unit + } + } + } + scope.launch { + client.state.collect { s -> + if (s is GatewayClient.State.Connected) { + val home = s.channels.firstOrNull { it.isDefault }?.chatId + if (home != null) _homeChannel.value = home + } + } + } + client.startHeartbeat() + client.start() + } + + /** Optimistic send: show immediately, 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) + } + + /** Connect-screen action: real hello test, then save + (re)connect. */ + suspend fun connect(url: String, token: String): Result { + val test = client.testHello(url, token) + if (test.isFailure) return test + store.savePairing(url, token) + client.restart() + return Result.success(Unit) + } + + fun forget() { + store.clear() + chat.clear() + client.restart() + } + + fun dispose() { + client.stop() + job.cancel() + } +} + +/** How much tool detail to show (Settings → "Tool detail"). */ +enum class ToolDetail { + EVERYTHING, // name + full args (collapsible) + output preview + TRUNCATED, // one-liner "emoji name: preview", collapsible (default) + NOTHING, // suppress tool cards entirely +} \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt b/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt new file mode 100644 index 0000000..92fef73 --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/ui/screens/ChatScreen.kt @@ -0,0 +1,385 @@ +package iris.ui.screens + +import androidx.compose.foundation.background +import androidx.compose.foundation.clickable +import androidx.compose.foundation.layout.Arrangement +import androidx.compose.foundation.layout.Box +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.PaddingValues +import androidx.compose.foundation.layout.Row +import androidx.compose.foundation.layout.Spacer +import androidx.compose.foundation.layout.fillMaxSize +import androidx.compose.foundation.layout.fillMaxWidth +import androidx.compose.foundation.layout.height +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.lazy.LazyColumn +import androidx.compose.foundation.lazy.items +import androidx.compose.foundation.lazy.rememberLazyListState +import androidx.compose.foundation.shape.RoundedCornerShape +import androidx.compose.foundation.text.KeyboardActions +import androidx.compose.foundation.text.KeyboardOptions +import androidx.compose.material3.Button +import androidx.compose.material3.CircularProgressIndicator +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.OutlinedTextField +import androidx.compose.material3.Text +import androidx.compose.material3.TextButton +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.setValue +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.platform.LocalClipboardManager +import androidx.compose.ui.text.AnnotatedString +import androidx.compose.ui.text.font.FontFamily +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.ChatItem +import iris.data.MessageItem +import iris.data.ToolItem +import iris.net.GatewayClient +import iris.protocol.ROLE_USER +import iris.state.IrisController +import iris.state.ToolDetail + +/** + * 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. + */ +@Composable +fun ChatScreen(controller: IrisController) { + val state by controller.client.state.collectAsState() + val items by controller.chat.items.collectAsState() + val typing by controller.typing.collectAsState() + val toolDetail by controller.toolDetail.collectAsState() + val listState = rememberLazyListState() + var input by remember { mutableStateOf("") } + + fun doSend() { + if (input.isBlank()) return + val text = input + input = "" + controller.send(text) + } + + LaunchedEffect(items.size, typing) { + val last = items.size - 1 + 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) + } + } + + // 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( + "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 }, + 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") + } + } + } +} + +@Composable +private fun StatusChip(state: GatewayClient.State) { + val (label, color) = when (state) { + GatewayClient.State.Disconnected -> "offline" to Color(0xFF9E9E9E) + GatewayClient.State.Connecting -> "connecting…" to Color(0xFFFFC107) + GatewayClient.State.Reconnecting -> "reconnecting…" to Color(0xFFFFC107) + is GatewayClient.State.Connected -> "connected" to Color(0xFF4CAF50) + is GatewayClient.State.AuthFailed -> "auth failed" to Color(0xFFF44336) + } + Box( + modifier = Modifier + .clip(RoundedCornerShape(10.dp)) + .background(color.copy(alpha = 0.15f)) + .padding(horizontal = 8.dp, vertical = 3.dp), + ) { + Text(label, fontSize = 11.sp, color = color) + } +} + +@Composable +private fun MessageBubble(msg: MessageItem) { + val isUser = msg.role == ROLE_USER + val isCommentary = msg.isCommentary + val bubbleColor = when { + isUser -> Color(0xFF4F7CFF) + isCommentary -> Color(0xFF23262F) + else -> Color(0xFF2A2E3B) + } + val textColor = when { + isUser -> Color.White + isCommentary -> Color(0xFFE8EAF0).copy(alpha = 0.55f) + else -> Color(0xFFE8EAF0) + } + Row( + modifier = Modifier.fillMaxWidth(), + horizontalArrangement = if (isUser) Arrangement.End else Arrangement.Start, + ) { + Column( + modifier = Modifier + .clip(RoundedCornerShape(14.dp)) + .background(bubbleColor) + .padding(horizontal = 12.dp, vertical = 8.dp), + ) { + // Reasoning block above the answer (assistant, non-commentary). + if (!isUser && !isCommentary && !msg.reasoning.isNullOrBlank()) { + ReasoningBlock(msg.reasoning!!) + Spacer(modifier = Modifier.height(6.dp)) + } + if (msg.text.isNotBlank() || msg.streaming) { + Text( + msg.text + if (msg.streaming) " ▉" else "", + color = textColor, + fontSize = if (isCommentary) 13.sp else 15.sp, + ) + } + if (msg.pending) { + Text("sending…", color = textColor.copy(alpha = 0.6f), fontSize = 10.sp) + } + // Model / token footer (final assistant answers only). + if (!isUser && !isCommentary && !msg.streaming && + (msg.model != null || msg.tokens != null) + ) { + Spacer(modifier = Modifier.height(4.dp)) + Text( + buildString { + msg.model?.let { append(it) } + if (msg.model != null && msg.tokens != null) append(" · ") + msg.tokens?.let { append("${it} tok") } + }, + color = textColor.copy(alpha = 0.4f), + fontSize = 10.sp, + ) + } + } + } +} + +/** + * Collapsible reasoning panel (M2): header "💭 Reasoning", monospace body, + * copy button. Collapsed by default when long; tap to toggle. + */ +@Composable +private fun ReasoningBlock(reasoning: String) { + val clipboard = LocalClipboardManager.current + var expanded by remember { + mutableStateOf(reasoning.length <= 240) + } + var copied by remember { mutableStateOf(false) } + Column( + modifier = Modifier + .fillMaxWidth() + .clip(RoundedCornerShape(8.dp)) + .background(Color(0xFF1E212B)) + .padding(8.dp), + ) { + Row(verticalAlignment = Alignment.CenterVertically) { + Text( + "💭 Reasoning", + fontSize = 12.sp, + fontWeight = FontWeight.SemiBold, + color = Color(0xFFB9C0D0), + modifier = Modifier + .weight(1f) + .clickable { expanded = !expanded }, + ) + TextButton(onClick = { + clipboard.setText(AnnotatedString(reasoning)) + copied = true + }) { + Text(if (copied) "copied" else "copy", fontSize = 11.sp) + } + Text( + if (expanded) "▾" else "▸", + fontSize = 12.sp, + color = Color(0xFFB9C0D0), + modifier = Modifier.clickable { expanded = !expanded }, + ) + } + if (expanded) { + Text( + reasoning, + fontFamily = FontFamily.Monospace, + fontSize = 12.sp, + color = Color(0xFF9AA3B5), + ) + } + } +} + +/** + * Structured tool-activity card (M2). Spinner while [ToolItem.done] is false, + * ✓/✗ + duration on completion. Verbosity controlled by [detail]. + */ +@Composable +private fun ToolCard(tool: ToolItem, detail: ToolDetail) { + var expanded by remember { mutableStateOf(false) } + val status = when { + !tool.done -> null + tool.ok -> "✓" + else -> "✗" + } + val statusColor = when { + tool.ok -> Color(0xFF4CAF50) + else -> Color(0xFFF44336) + } + val preview = tool.preview ?: tool.note + Column( + modifier = Modifier + .fillMaxWidth() + .clip(RoundedCornerShape(10.dp)) + .background(Color(0xFF20242E)) + .padding(horizontal = 10.dp, vertical = 8.dp), + ) { + Row(verticalAlignment = Alignment.CenterVertically) { + if (!tool.done) { + CircularProgressIndicator( + modifier = Modifier.size(14.dp), + strokeWidth = 2.dp, + ) + Spacer(modifier = Modifier.width(8.dp)) + } else { + Text(status!!, color = statusColor, fontSize = 14.sp) + Spacer(modifier = Modifier.width(6.dp)) + } + Text( + "🔧 ${tool.name}", + fontSize = 13.sp, + fontWeight = FontWeight.Medium, + color = Color(0xFFD7DBE5), + modifier = Modifier + .weight(1f) + .clickable { expanded = !expanded }, + ) + if (tool.duration != null) { + Text( + "${String.format("%.1f", tool.duration)}s", + fontSize = 11.sp, + color = Color(0xFF8A93A6), + ) + } + } + // One-liner preview (Truncated default / Everything). + if (preview != null && detail != ToolDetail.NOTHING) { + Text( + preview, + fontSize = 12.sp, + color = Color(0xFF9AA3B5), + maxLines = if (expanded) Int.MAX_VALUE else 1, + ) + } + // Full args + output preview (Everything, expanded). + if (detail == ToolDetail.EVERYTHING && expanded) { + tool.args?.let { + Spacer(modifier = Modifier.height(4.dp)) + Text( + it.toString(), + fontFamily = FontFamily.Monospace, + fontSize = 11.sp, + color = Color(0xFF8A93A6), + ) + } + tool.outputPreview?.let { + Spacer(modifier = Modifier.height(4.dp)) + Text( + it, + fontFamily = FontFamily.Monospace, + fontSize = 11.sp, + color = Color(0xFF8A93A6), + ) + } + } + } +} + +private val ToolDetail.label: String + get() = when (this) { + ToolDetail.EVERYTHING -> "everything" + ToolDetail.TRUNCATED -> "truncated" + ToolDetail.NOTHING -> "nothing" + } \ No newline at end of file diff --git a/app/shared/src/commonMain/kotlin/iris/ui/screens/ConnectScreen.kt b/app/shared/src/commonMain/kotlin/iris/ui/screens/ConnectScreen.kt new file mode 100644 index 0000000..3108cdf --- /dev/null +++ b/app/shared/src/commonMain/kotlin/iris/ui/screens/ConnectScreen.kt @@ -0,0 +1,124 @@ +package iris.ui.screens + +import androidx.compose.foundation.layout.Arrangement +import androidx.compose.foundation.layout.Box +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.Spacer +import androidx.compose.foundation.layout.fillMaxSize +import androidx.compose.foundation.layout.fillMaxWidth +import androidx.compose.foundation.layout.height +import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.rememberScrollState +import androidx.compose.foundation.text.KeyboardOptions +import androidx.compose.foundation.verticalScroll +import androidx.compose.material3.Button +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.OutlinedButton +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.rememberCoroutineScope +import androidx.compose.runtime.setValue +import androidx.compose.ui.Alignment +import androidx.compose.ui.Modifier +import androidx.compose.ui.text.input.KeyboardType +import androidx.compose.ui.text.input.PasswordVisualTransformation +import androidx.compose.ui.unit.dp +import iris.state.IrisController +import kotlinx.coroutines.launch + +/** + * Onboarding / pairing screen (docs/10 §10.8). "Test & Connect" performs a + * real `hello` (the auth leg), not just a TCP probe. + */ +@Composable +fun ConnectScreen( + controller: IrisController, + prefillUrl: String = "", + prefillToken: String = "", + initialError: String? = null, +) { + val scope = rememberCoroutineScope() + var url by remember { mutableStateOf(prefillUrl.ifBlank { "ws://" }) } + var token by remember { mutableStateOf(prefillToken) } + var busy by remember { mutableStateOf(false) } + var error by remember { mutableStateOf(initialError) } + + Column( + modifier = Modifier + .fillMaxSize() + .verticalScroll(rememberScrollState()) + .padding(24.dp), + horizontalAlignment = Alignment.CenterHorizontally, + ) { + Spacer(modifier = Modifier.height(48.dp)) + Text("Iris × Hermes", style = MaterialTheme.typography.headlineMedium) + Spacer(modifier = Modifier.height(8.dp)) + Text( + "Pair this device with your hermes gateway", + style = MaterialTheme.typography.bodyMedium, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + Spacer(modifier = Modifier.height(32.dp)) + + OutlinedTextField( + value = url, + onValueChange = { url = it }, + label = { Text("Server URL") }, + placeholder = { Text("ws://192.168.1.10:8790/ws") }, + singleLine = true, + keyboardOptions = KeyboardOptions(keyboardType = KeyboardType.Uri), + modifier = Modifier.fillMaxWidth(), + ) + Spacer(modifier = Modifier.height(12.dp)) + OutlinedTextField( + value = token, + onValueChange = { token = it }, + label = { Text("Pairing token") }, + placeholder = { Text("ANDROID_TOKEN (64 hex)") }, + singleLine = true, + visualTransformation = PasswordVisualTransformation(), + modifier = Modifier.fillMaxWidth(), + ) + Spacer(modifier = Modifier.height(24.dp)) + + Button( + onClick = { + if (busy) return@Button + busy = true + error = null + scope.launch { + val result = controller.connect(url.trim(), token.trim()) + busy = false + if (result.isFailure) { + error = result.exceptionOrNull()?.message ?: "connection failed" + } + } + }, + enabled = !busy, + modifier = Modifier.fillMaxWidth(), + ) { + Text(if (busy) "Testing connection…" else "Test & Connect") + } + + if (error != null) { + Spacer(modifier = Modifier.height(16.dp)) + Text( + error!!, + color = MaterialTheme.colorScheme.error, + style = MaterialTheme.typography.bodyMedium, + ) + } + + Spacer(modifier = Modifier.height(24.dp)) + Text( + "Find the token in ~/.hermes/.env (ANDROID_TOKEN) or run\n" + + "hermes gateway setup on the gateway host.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } +} \ No newline at end of file diff --git a/app/shared/src/desktopMain/kotlin/iris/platform/DesktopSecureStore.kt b/app/shared/src/desktopMain/kotlin/iris/platform/DesktopSecureStore.kt new file mode 100644 index 0000000..b329523 --- /dev/null +++ b/app/shared/src/desktopMain/kotlin/iris/platform/DesktopSecureStore.kt @@ -0,0 +1,75 @@ +package iris.platform + +import iris.data.SecureStore +import iris.protocol.IrisJson +import java.io.File +import java.util.UUID +import kotlinx.serialization.Serializable + +/** + * Desktop pairing storage: a small JSON file under the user home + * (`~/.iris/pairing.json`). M1 dev storage; a keyring backend is a + * later-milestone concern. + */ +@Serializable +private data class PairingData( + val serverUrl: String = "", + val token: String = "", + val deviceId: String = "", +) + +class DesktopSecureStore : SecureStore { + private val file: File = File(System.getProperty("user.home"), ".iris/pairing.json") + + private fun load(): PairingData = + if (file.exists()) { + try { + IrisJson.instance.decodeFromString(PairingData.serializer(), file.readText()) + } catch (_: Exception) { + PairingData() + } + } else { + PairingData() + } + + private fun save(data: PairingData) { + file.parentFile?.mkdirs() + file.writeText(IrisJson.instance.encodeToString(PairingData.serializer(), data)) + } + + override var serverUrl: String + get() = load().serverUrl + set(value) { + val d = load() + save(d.copy(serverUrl = value.trim())) + } + + override var token: String + get() = load().token + set(value) { + val d = load() + save(d.copy(token = value.trim())) + } + + override val deviceId: String + get() { + val d = load() + if (d.deviceId.isNotBlank()) return d.deviceId + val id = "dev_" + UUID.randomUUID().toString().replace("-", "").take(16) + save(d.copy(deviceId = id)) + return id + } + + override val deviceName: String + get() = "Desktop (${System.getProperty("os.name")})" + + override fun savePairing(url: String, token: String) { + val d = load() + save(d.copy(serverUrl = url.trim(), token = token.trim())) + } + + override fun clear() { + val d = load() + save(d.copy(serverUrl = "", token = "")) + } +} \ No newline at end of file diff --git a/docs/14-milestones.md b/docs/14-milestones.md index 58a67ad..9dbc330 100644 --- a/docs/14-milestones.md +++ b/docs/14-milestones.md @@ -7,15 +7,15 @@ has explicit **acceptance criteria**. Work top-to-bottom; don't skip M0/M1. ## M0 — Toolchain & scaffolding **Goal:** everything builds; the plugin is discoverable; the repo is safe. -- [ ] Install JDK 17, Android SDK, set `ANDROID_HOME` (`12-toolchain.md`). -- [ ] `cd hermes-agent && uv sync` (hermes venv works). -- [ ] Create monorepo scaffold (`02-monorepo.md`): `gateway-plugin/`, `app/` +- [X] Install JDK 17, Android SDK, set `ANDROID_HOME` (`12-toolchain.md`). +- [X] `cd hermes-agent && uv sync` (hermes venv works). +- [X] Create monorepo scaffold (`02-monorepo.md`): `gateway-plugin/`, `app/` (CMP: `shared`, `androidApp`, `desktopApp`), root `.gitignore` (**excludes `hermes-agent/`**), root `README.md`. -- [ ] `git init` + a pre-commit/CI guard that fails if `hermes-agent/` is staged. -- [ ] CMP project builds empty: `./gradlew :androidApp:assembleDebug`, +- [X] `git init` + a pre-commit/CI guard that fails if `hermes-agent/` is staged. +- [X] CMP project builds empty: `./gradlew :androidApp:assembleDebug`, `./gradlew :desktopApp:run` (blank window). -- [ ] Plugin skeleton: `plugin.yaml` + `adapter.py` with `register(ctx)` + a +- [X] Plugin skeleton: `plugin.yaml` + `adapter.py` with `register(ctx)` + a no-op `AndroidAdapter` → `hermes gateway status` lists **android**. - **Demo:** `hermes gateway status` shows `android`; `./gradlew :androidApp:installDebug` installs a blank app on the MIX 2S. @@ -25,31 +25,41 @@ has explicit **acceptance criteria**. Work top-to-bottom; don't skip M0/M1. ## M1 — Gateway core loop (text round-trip) **Goal:** pair + send a text message + get a (non-streaming) reply. -- [ ] WS server (`ws_server.py`): bind, `hello` auth (constant-time), +- [X] WS server (`ws_server.py`): bind, `hello` auth (constant-time), `hello.ack`, heartbeat, connection registry. -- [ ] `AndroidAdapter.send()` → `message` frame; inbound `message.send` → +- [X] `AndroidAdapter.send()` → `message` frame; inbound `message.send` → `MessageEvent` → `handle_message`. -- [ ] Pairing store + `ANDROID_TOKEN`; QR payload in `interactive_setup`. -- [ ] App: Connect screen (URL+token, real `hello` test), `GatewayClient` +- [X] Pairing store + `ANDROID_TOKEN`; QR payload in `interactive_setup`. +- [X] App: Connect screen (URL+token, real `hello` test), `GatewayClient` (connect + reconnect), ChatScreen sends + renders `message`. -- [ ] `ws_probe.py` harness drives a real turn. +- [X] `ws_probe.py` harness drives a real turn. - **Demo (on-device):** pair the phone, send "hello", see the agent's reply. - **Accept:** text round-trip works on-device; wrong token is rejected; reconnect after gateway restart re-pairs. ## M2 — Streaming + reasoning + tools + commentary **Goal:** the "agent transparency" features. -- [ ] Map consumer `send`/`edit_message` → `message.start/update/stop`. -- [ ] Reasoning: set `show_reasoning` for android; adapter splits prefix → - `reasoning` field. **Verify format with `ws_probe.py`.** -- [ ] Tool events: classify tool-progress `send()`s → structured +- [X] Map consumer `send`/`edit_message` → `message.start/update/stop`. +- [X] Reasoning: set `show_reasoning` for android; adapter splits prefix → + `reasoning` field. **Verify format with `ws_probe.py`.** (The model + returns a separate `reasoning_content` field. In the *streaming* case the + gateway drops it — the stream consumer only forwards `content` and the + final send is suppressed — so the adapter captures it via the + `on_stream_delta` plugin hook (`kind="reasoning"`, gated by + `plugins.stream_reasoning_deltas: true`) and attaches it to + `message.stop`. Non-streaming already carried it via the prefix split.) +- [X] Tool events: classify tool-progress `send()`s → structured `tool.start/progress/end` (turn-state machine). **Verify with probe.** -- [ ] Commentary → `commentary` frames. Typing → `typing`. -- [ ] App: live bubble (coalesced updates), `ReasoningBlock` (collapse + copy), +- [X] Commentary → `commentary` frames. Typing → `typing`. +- [X] App: live bubble (coalesced updates), `ReasoningBlock` (collapse + copy), `ToolCard` with **Everything/Truncated/Nothing** setting, dimmed `commentary` bubble. - **Demo (on-device):** a multi-step prompt streams, shows reasoning above the answer, tool cards (toggle verbosity), and an intermediate beat. + (Verified on the MIX 2S: streaming, tool cards + verbosity toggle, and the + reasoning panel rendering above the answer. The commentary beat is + model/agent-dependent and did not fire with the current model, but the + rendering + frame handling are in place.) - **Accept:** all four render correctly; tool verbosity setting changes rendering; reasoning copy button works; frame shapes match `04-wire-protocol`. @@ -135,4 +145,4 @@ has explicit **acceptance criteria**. Work top-to-bottom; don't skip M0/M1. are stable so the shared module is settled. - Parallelizable: plugin (Python) and app (Kotlin) can be worked on concurrently once the protocol (`04-wire-protocol.md`) is agreed; the probe - harness is the integration seam. \ No newline at end of file + harness is the integration seam. diff --git a/gateway-plugin/adapter.py b/gateway-plugin/adapter.py index 2535f48..826dbd2 100644 --- a/gateway-plugin/adapter.py +++ b/gateway-plugin/adapter.py @@ -9,11 +9,17 @@ with a pairing token and talks to the agent over a single WS transport Zero new Python dependencies: ``websockets`` and ``httpx`` are hermes core deps. Zero hermes-core changes. -Milestone M0: this is a *skeleton* adapter. It registers the ``android`` -platform, resolves its configuration, and implements the abstract adapter -contract as no-ops so that ``hermes gateway status`` lists ``android``. The -WebSocket server, pairing, streaming, media, outbox, push, and search are -wired in later milestones (see ``docs/14-milestones.md``). +Milestone M1: the gateway core loop (text round-trip). The WS server binds +and authenticates devices (``hello`` with constant-time token check), the +adapter emits ``message`` frames from ``send()`` and turns inbound +``message.send`` frames into ``MessageEvent``s for ``handle_message()``. + +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``). Configuration in config.yaml:: @@ -34,11 +40,15 @@ Or via environment variables (overrides config.yaml; secrets live in .env): ANDROID_PUSH_BACKEND, ANDROID_FCM_SERVICE_ACCOUNT, NTFY_TOPIC, ... """ +import asyncio import logging import os +import re +import threading import time import uuid -from typing import Any, Dict, List, Optional +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError from agent.secret_scope import get_secret as _scoped_get_secret @@ -79,6 +89,77 @@ from gateway.platforms.base import ( # noqa: E402 MessageType, ) from gateway.config import Platform # noqa: E402 +from hermes_constants import get_hermes_home # noqa: E402 + +from . import protocol # noqa: E402 +from .pairing import ( # noqa: E402 + DeviceRegistry, + generate_token, + pairing_url, + qr_payload, +) +from .ws_server import WsServer # noqa: E402 + + +# --------------------------------------------------------------------------- +# M2 — reasoning capture (streaming) +# +# The gateway streams only ``content`` to the platform and suppresses the +# final send (which would carry the prepended reasoning), so the model's +# separate ``reasoning_content`` is otherwise lost in the streaming case. +# hermes exposes a plugin ``on_stream_delta`` hook that fires reasoning +# deltas with ``kind="reasoning"`` (gated by ``plugins.stream_reasoning_deltas``). +# We accumulate those deltas here and attach the result to the turn's +# ``message.stop`` frame. Single-chat for now (android:default), so a +# module-level buffer suffices; it is reset at each turn start. +# --------------------------------------------------------------------------- + +_reasoning_parts: List[str] = [] +_reasoning_lock = threading.Lock() +# Barrier: set by the hook worker once it has processed the first content +# delta (kind="text"). The worker drains a FIFO queue and reasoning deltas are +# enqueued before content deltas, so at that point every reasoning delta has +# already been appended -- a reliable "reasoning flushed" signal that avoids +# racing message.stop against the async hook thread. +_reasoning_flushed = threading.Event() + + +def _on_stream_delta(**kwargs: Any) -> None: + """Plugin hook: capture reasoning deltas (kind="reasoning").""" + kind = kwargs.get("kind") + if kind == "reasoning": + delta = kwargs.get("delta") or "" + if delta: + with _reasoning_lock: + _reasoning_parts.append(delta) + elif kind == "text": + _reasoning_flushed.set() + + +async def _wait_for_reasoning_flushed(timeout: float = 0.3) -> None: + """Wait (without blocking the event loop) until the hook worker has + processed all reasoning deltas, or *timeout* seconds elapse.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if _reasoning_flushed.is_set(): + return + await asyncio.sleep(0.01) + + +def _take_reasoning() -> str: + """Drain and return the accumulated reasoning (empty string if none).""" + with _reasoning_lock: + parts = _reasoning_parts[:] + _reasoning_parts.clear() + _reasoning_flushed.clear() + return "".join(parts).strip() + + +def _reset_reasoning() -> None: + with _reasoning_lock: + _reasoning_parts.clear() + _reasoning_flushed.clear() # --------------------------------------------------------------------------- @@ -88,6 +169,7 @@ from gateway.config import Platform # noqa: E402 DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8790 DEFAULT_HOME_CHANNEL = "android:default" +DEFAULT_HOME_CHANNEL_NAME = "Default" DEFAULT_PUSH_BACKEND = "fcm" DEFAULT_OUTBOX_RETENTION_HOURS = 72 DEFAULT_MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB @@ -97,6 +179,211 @@ def _truthy(value: Optional[str]) -> bool: return (value or "").strip().lower() in {"1", "true", "yes", "on"} +# --------------------------------------------------------------------------- +# M2 — turn state + outbound classification +# +# The main gateway delivers through the legacy callback path: the stream +# consumer calls ``send()`` (first bubble of a segment) and ``edit_message()`` +# (updates), tool progress flows through ``send()``/``edit_message()`` of an +# accumulated line buffer, and interim commentary arrives as a plain +# ``send()``. We classify each outbound call into a structured frame using a +# per-chat turn state machine + content markers: +# +# * ``metadata["expect_edits"] is True`` -> streaming segment start +# * ``metadata["notify"] is True`` -> final message (or fallback final) +# * tool-progress line format -> tool.start / tool.end +# * anything else -> commentary +# +# Verified empirically against the live gateway with ``tests/ws_probe.py``. +# --------------------------------------------------------------------------- + +# Streaming cursor the gateway appends to in-progress edits (" ▉"). Stripped +# before we forward text to the app (the app renders its own live indicator). +_STREAMING_CURSOR = " ▉" + +# Code-style reasoning prefix (gateway/run.py, reasoning_style="code"): +# "💭 **Reasoning:**\n```\n\n```\n\n" +_REASONING_PREFIX = "💭 **Reasoning:**\n```\n" +_REASONING_CLOSE = "\n```\n\n" + +# A gateway tool-progress line begins with a (non-ASCII) tool emoji. +_TOOL_LINE_RE = re.compile(r"^(\S+)\s+(.+)$") +_TOOL_NAME_PREVIEW_RE = re.compile(r'^(\S+):\s*"(.*)"\s*$') +_TOOL_NAME_BARE_RE = re.compile(r"^(\S+)\.\.\.\s*$") +_TOOL_NAME_ARGS_RE = re.compile(r"^(\S+)\(([^)]*)\)\s*$") +# Terminal code block: "💻 terminal\n```\n\n```" +_TOOL_CODEBLOCK_HEAD_RE = re.compile(r"^(\S+)\s+(\S+)\s*$") + +# Reverse map of the gateway's friendly tool verbs (agent/display.py +# _TOOL_VERBS) so a verb-form line ("🔍 Searching the web for …") can be +# recovered to a structured (tool_name, preview). Longest-first matching is +# done at parse time. Verbs shared by several tools map to the most common. +_VERB_TO_TOOL: Dict[str, str] = { + "Searching the web": "web_search", + "Searching files": "search_files", + "Searching past sessions": "session_search", + "Running code": "execute_code", + "Running": "terminal", + "Reading skill": "skill_view", + "Reading": "read_file", + "Writing": "write_file", + "Editing": "patch", + "Browsing": "browser_navigate", + "Clicking": "browser_click", + "Typing": "browser_type", + "Generating image": "image_generate", + "Generating video": "video_generate", + "Generating speech": "text_to_speech", + "Looking at the image": "vision_analyze", + "Listing skills": "skills_list", + "Updating skill": "skill_manage", + "Updating memory": "memory", + "Updating tasks": "todo", + "Delegating": "delegate_task", + "Scheduling": "cronjob", + "Asking": "clarify", +} +# Verbs that take a " for " connector before the preview. +_VERB_FOR_CONNECTOR = {"web_search", "search_files"} + + +def _mint_message_id() -> str: + return f"m_{uuid.uuid4().hex[:16]}" + + +def _thread_id_from_metadata(metadata: Optional[Dict[str, Any]]) -> Optional[str]: + if not metadata: + return None + tid = metadata.get("thread_id") + if isinstance(tid, str) and tid: + return tid + return None + + +def _strip_streaming_cursor(text: str) -> str: + if text and text.endswith(_STREAMING_CURSOR): + return text[: -len(_STREAMING_CURSOR)] + return text + + +def _split_reasoning(text: str) -> Tuple[Optional[str], str]: + """Split a code-style reasoning prefix off the front of *text*. + + Returns ``(reasoning, body)``; ``reasoning`` is ``None`` when no prefix is + present (reasoning off / no reasoning / non-code style). Best-effort parse + of a stable, gateway-owned format: on any mismatch the fallback is + ``(None, full text)`` so the answer still renders. + """ + if not text or not text.startswith(_REASONING_PREFIX): + return None, text + close_idx = text.find(_REASONING_CLOSE, len(_REASONING_PREFIX)) + if close_idx == -1: + return None, text + reasoning = text[len(_REASONING_PREFIX):close_idx] + body = text[close_idx + len(_REASONING_CLOSE):] + return reasoning, body + + +def _parse_tool_line(line: str) -> Optional[Tuple[str, Optional[str]]]: + """Parse a single gateway tool-progress line into ``(name, preview)``. + + Returns ``None`` when the line is not a tool line. The gateway formats + tool lines as `` : ""``, `` ...``, + `` (keys)``, or a friendly verb phrase (`` …``). + The verb form is lossy (no tool name), so we surface the verb as the name. + """ + line = line.strip() + if not line: + return None + m = _TOOL_LINE_RE.match(line) + if not m: + return None + emoji, rest = m.group(1), m.group(2) + if emoji.isascii(): + return None # a tool line always leads with a non-ASCII emoji + mp = _TOOL_NAME_PREVIEW_RE.match(rest) + if mp: + return mp.group(1), mp.group(2) + mb = _TOOL_NAME_BARE_RE.match(rest) + if mb: + return mb.group(1), None + ma = _TOOL_NAME_ARGS_RE.match(rest) + if ma: + return ma.group(1), None + # Friendly verb phrase: reverse-map to (tool_name, preview). + verb_parsed = _parse_verb_phrase(rest) + if verb_parsed is not None: + return verb_parsed + # Unrecognised: use the phrase as the label. + return rest, None + + +def _parse_verb_phrase(phrase: str) -> Optional[Tuple[str, Optional[str]]]: + """Reverse-map a friendly verb phrase to ``(tool_name, preview)``. + + Matches the longest verb first so "Running code" wins over "Running". + Returns ``None`` when no known verb leads the phrase. + """ + for verb in sorted(_VERB_TO_TOOL, key=len, reverse=True): + tool = _VERB_TO_TOOL[verb] + if phrase == verb: + return tool, None + if tool in _VERB_FOR_CONNECTOR and phrase.startswith(verb + " for "): + return tool, phrase[len(verb) + len(" for "):].strip() or None + if phrase.startswith(verb + " "): + return tool, phrase[len(verb) + 1:].strip() or None + return None + + +def _extract_code_block(content: str) -> Optional[str]: + """Return the first fenced code block's body in *content*, else ``None``. + + Used to recover the terminal command from a tool-progress code block + (`` terminal`` head line + fenced command). + """ + m = re.search(r"```[^\n]*\n(.*?)\n```", content, re.DOTALL) + if m: + return m.group(1).strip() or None + return None + + +def _is_tool_progress(content: str) -> bool: + """Heuristic: does *content* look like gateway tool-progress line(s)? + + Tool progress is delivered as one or more lines, each led by a tool emoji + (or a terminal code block). Commentary is free-form prose. We classify on + the first non-empty line; subsequent lines of the same bubble are tracked + by message id, not re-classified. + """ + if not content: + return False + lines = [ln for ln in content.splitlines() if ln.strip()] + if not lines: + return False + first = lines[0].strip() + # Terminal code block: " terminal" then a fenced command. + if len(lines) > 1 and lines[1].strip().startswith("```"): + return _TOOL_CODEBLOCK_HEAD_RE.match(first) is not None + return _parse_tool_line(first) is not None + + +@dataclass +class _TurnState: + """Per-chat turn state for outbound frame classification (M2).""" + + active: bool = False + # message_id of the currently streaming segment (message.start open). + stream_id: Optional[str] = None + # message_id of the current tool-progress bubble (editable line buffer). + tool_msg_id: Optional[str] = None + # Monotonic per-turn tool counter (start -> end correlation). + tool_index: int = 0 + # Index of the most recently started tool (awaiting tool.end). + open_tool_index: Optional[int] = None + # Tool lines already emitted as tool.start (dedup across edits). + seen_tool_lines: set = field(default_factory=set) + + # --------------------------------------------------------------------------- # Passive / config probes (called from status displays -- no side effects) # --------------------------------------------------------------------------- @@ -147,19 +434,26 @@ def _env_enablement() -> Optional[dict]: if not token: return None - seed: Dict[str, Any] = { - "host": os.getenv("ANDROID_WS_HOST", "").strip() or DEFAULT_HOST, - "port": _parse_port(os.getenv("ANDROID_WS_PORT", "")), - "push_backend": ( - os.getenv("ANDROID_PUSH_BACKEND", "").strip().lower() - or DEFAULT_PUSH_BACKEND - ), - } - home = os.getenv("ANDROID_HOME_CHANNEL", "").strip() or DEFAULT_HOME_CHANNEL - seed["home_channel"] = { - "chat_id": home, - "name": os.getenv("ANDROID_HOME_CHANNEL_NAME", "").strip() or "Default", - } + # Seed ONLY explicitly-set env vars: the core commits this seed on top of + # config.yaml (``extra.update(seed)``), so default values here would + # clobber user YAML. Unset keys fall through to config.yaml / adapter + # defaults. + seed: Dict[str, Any] = {} + host = os.getenv("ANDROID_WS_HOST", "").strip() + if host: + seed["host"] = host + port_raw = os.getenv("ANDROID_WS_PORT", "").strip() + if port_raw: + seed["port"] = _parse_port(port_raw) + push = os.getenv("ANDROID_PUSH_BACKEND", "").strip().lower() + if push: + seed["push_backend"] = push + home = os.getenv("ANDROID_HOME_CHANNEL", "").strip() + if home: + seed["home_channel"] = { + "chat_id": home, + "name": os.getenv("ANDROID_HOME_CHANNEL_NAME", "").strip() or DEFAULT_HOME_CHANNEL_NAME, + } return seed @@ -215,7 +509,7 @@ async def _standalone_send( The outbox is served by the *running* gateway, so standalone delivery while the gateway process is fully down is best-effort only (see - ``docs/00-overview.md`` "Out of scope"). For M0 this is a stub that + ``docs/00-overview.md`` "Out of scope"). For M1 this is a stub that reports the gateway is required; the real implementation lands with the outbox (M3/M5). """ @@ -228,14 +522,14 @@ async def _standalone_send( # --------------------------------------------------------------------------- -# Interactive setup (hermes gateway setup flow) -- full version in M1 +# Interactive setup (hermes gateway setup flow) # --------------------------------------------------------------------------- def interactive_setup() -> None: """Prompt for the pairing token / host / port / push backend. - M0: minimal. M1 adds token generation, QR payload, and a live ``hello`` - connectivity test. + M1: token generation, host/port/push prompts, and the pairing QR payload + (``iris://pair?...``) + app URL printed for the Connect screen. """ try: from hermes_cli.config import ( @@ -253,7 +547,7 @@ def interactive_setup() -> None: print_info("📱 Android / Desktop (Iris x Hermes)") token = get_env_value("ANDROID_TOKEN") or "" if not token: - generated = uuid.uuid4().hex + uuid.uuid4().hex # 64 hex chars + generated = generate_token() save_env_value("ANDROID_TOKEN", generated) print_success(f"Generated pairing token: {generated}") print_warning("Keep this secret -- the app presents it on connect.") @@ -267,6 +561,19 @@ def interactive_setup() -> None: backend = prompt("Push backend (fcm/ntfy)", default=get_env_value("ANDROID_PUSH_BACKEND") or DEFAULT_PUSH_BACKEND) save_env_value("ANDROID_PUSH_BACKEND", (backend or DEFAULT_PUSH_BACKEND).strip().lower()) + # Pairing payload for the app's Connect screen (QR / manual entry). + try: + from hermes_cli.config import print_code + url = pairing_url(host or DEFAULT_HOST, _parse_port(port)) + payload = qr_payload(host or DEFAULT_HOST, _parse_port(port), token) + print_info("Pair your device (scan with the app or enter on the Connect screen):") + print_code(payload) + print_info(f"Server URL: {url}") + except Exception: + url = pairing_url(host or DEFAULT_HOST, _parse_port(port)) + print_info(f"Pairing URL: {qr_payload(host or DEFAULT_HOST, _parse_port(port), token)}") + print_info(f"Server URL: {url}") + print_success("Android configuration saved to ~/.hermes/.env") print_info("Restart the gateway for changes to take effect: hermes gateway restart") @@ -278,10 +585,10 @@ def interactive_setup() -> None: class AndroidAdapter(BasePlatformAdapter): """WebSocket-backed adapter for the native Iris Android / Desktop app. - M0: skeleton. Implements the abstract adapter contract as no-ops and - resolves configuration. The WebSocket server, connection registry, - pairing, streaming, media, outbox, push, and search are added in later - milestones. + M1: the WS server (``ws_server.WsServer``) authenticates devices with the + pairing token, the connection registry tracks live sockets, ``send()`` + emits ``message`` frames, and inbound ``message.send`` frames become + ``MessageEvent``s for ``handle_message()``. """ def __init__(self, config, **kwargs): @@ -294,7 +601,6 @@ class AndroidAdapter(BasePlatformAdapter): self.host = os.getenv("ANDROID_WS_HOST", "").strip() or extra.get("host", DEFAULT_HOST) self.port = _parse_port(os.getenv("ANDROID_WS_PORT", "") or str(extra.get("port", DEFAULT_PORT))) self.token = _get_scoped_secret("ANDROID_TOKEN") or extra.get("token", "") - self.home_channel = extra.get("home_channel", DEFAULT_HOME_CHANNEL) self.push_backend = ( os.getenv("ANDROID_PUSH_BACKEND", "").strip().lower() or extra.get("push_backend", DEFAULT_PUSH_BACKEND) @@ -306,6 +612,25 @@ class AndroidAdapter(BasePlatformAdapter): extra.get("max_upload_bytes", DEFAULT_MAX_UPLOAD_BYTES) ) + # Home channel: the core hook turns the env-seeded ``home_channel`` + # dict into a HomeChannel dataclass on the config; config.yaml may + # also put it in extra (dict or bare string). + home = getattr(config, "home_channel", None) + if home is not None and getattr(home, "chat_id", None): + self.home_channel = str(home.chat_id) + self.home_channel_name = str(getattr(home, "name", "") or DEFAULT_HOME_CHANNEL_NAME) + else: + hc = extra.get("home_channel") + if isinstance(hc, dict) and hc.get("chat_id"): + self.home_channel = str(hc["chat_id"]) + self.home_channel_name = str(hc.get("name") or DEFAULT_HOME_CHANNEL_NAME) + elif isinstance(hc, str) and hc.strip(): + self.home_channel = hc.strip() + self.home_channel_name = DEFAULT_HOME_CHANNEL_NAME + else: + self.home_channel = DEFAULT_HOME_CHANNEL + self.home_channel_name = DEFAULT_HOME_CHANNEL_NAME + # TLS (optional) self.ws_cert = _get_scoped_secret("ANDROID_WS_CERT") or extra.get("ws_cert", "") self.ws_key = _get_scoped_secret("ANDROID_WS_KEY") or extra.get("ws_key", "") @@ -317,10 +642,19 @@ class AndroidAdapter(BasePlatformAdapter): ) self.allow_all = _truthy(os.getenv("ANDROID_ALLOW_ALL_USERS")) - # Runtime state (populated by the WS server in M1) - self._ws_server = None - self._connections: Dict[str, Any] = {} + # Runtime state + self._devices = DeviceRegistry(get_hermes_home() / "android" / "devices.db") + self._ws_server = WsServer(self, self._devices) self._connected = False + # M2: per-chat turn state for outbound frame classification. + self._turns: Dict[str, _TurnState] = {} + + def _turn_state(self, chat_id: str) -> _TurnState: + st = self._turns.get(chat_id) + if st is None: + st = _TurnState() + self._turns[chat_id] = st + return st @property def name(self) -> str: @@ -329,12 +663,7 @@ class AndroidAdapter(BasePlatformAdapter): # ── Connection lifecycle ────────────────────────────────────────────── async def connect(self, *, is_reconnect: bool = False) -> bool: - """Bring the platform up. - - M0: no WebSocket server yet -- just validate config and mark - connected so ``hermes gateway status`` reflects the platform. M1 - starts the ``websockets`` server here. - """ + """Bring the platform up: bind the WS server on host:port.""" if not self.token: logger.error("android: ANDROID_TOKEN must be set") self._set_fatal_error( @@ -360,21 +689,33 @@ class AndroidAdapter(BasePlatformAdapter): except ImportError: self._lock_key = None # status module not available (e.g. tests) - # M1: start the websockets server on host:port (TLS if cert/key set). + try: + await self._ws_server.start() + except Exception: + self._connected = False + return False + self._connected = True self._mark_connected() - logger.info("android: connected (skeleton; WS server starts in M1) on %s:%s", self.host, self.port) + logger.info("android: connected; WS server on %s:%s", self.host, self.port) return True async def disconnect(self) -> None: - """Tear down the platform.""" + """Tear down the platform: stop the server, close device sockets.""" try: from gateway.status import release_scoped_lock if getattr(self, "_lock_key", None): release_scoped_lock("android", self._lock_key) except ImportError: pass - # M1: stop the server and close all device sockets. + try: + await self._ws_server.stop() + except Exception: + logger.warning("android: WS server stop failed", exc_info=True) + try: + self._devices.close() + except Exception: + pass self._connected = False self._mark_disconnected() logger.info("android: disconnected") @@ -390,18 +731,277 @@ class AndroidAdapter(BasePlatformAdapter): ) -> SendResult: """Send a message to a chat. - M0: no live devices yet -- log and report success with a minted id. - M1: broadcast a ``message`` frame to connected devices, else fall to - the outbox + fire push. + M2: classify the outbound call into a structured frame using the + per-chat turn state machine (see module docstring): + + * ``metadata["expect_edits"]`` -> ``message.start`` (streaming segment) + * ``metadata["notify"]`` -> ``message`` / ``message.stop`` (final) + * tool-progress line format -> ``tool.start`` (first tool bubble) + * anything else -> ``commentary`` + + With no live devices the frame is dropped here (the outbox + push + replay lands in M3/M5). """ - message_id = f"msg_{uuid.uuid4().hex}" - logger.debug("android: send to %s (%d chars) [skeleton no-op]", chat_id, len(content or "")) + content = content or "" + meta = metadata or {} + thread_id = _thread_id_from_metadata(meta) + state = self._turn_state(chat_id) + + # 1. Streaming segment start (stream consumer first send). + if meta.get("expect_edits") is True: + # A new content segment means the tool the model was waiting on + # has returned -> close it before the segment opens. + await self._close_open_tool(chat_id, state, thread_id) + message_id = _mint_message_id() + state.active = True + state.stream_id = message_id + await self._broadcast_or_log( + chat_id, + protocol.message_start(chat_id, message_id, protocol.ROLE_ASSISTANT, thread_id=thread_id), + ) + return SendResult(success=True, message_id=message_id) + + # 2. Final message (non-streaming final, or streaming fallback final). + if meta.get("notify") is True: + reasoning, body = _split_reasoning(content) + # Non-streaming: reasoning is prepended to content (split above). + # Streaming fallback: content has no reasoning, so use the + # reasoning captured via the on_stream_delta hook (wait for the + # async hook worker to flush it first). + if not reasoning: + await _wait_for_reasoning_flushed() + reasoning = _take_reasoning() or None + else: + _reset_reasoning() + if state.stream_id: + # Fallback final: close the open streaming segment in place. + message_id = state.stream_id + state.stream_id = None + await self._broadcast_or_log( + chat_id, + protocol.message_stop( + chat_id, message_id, body, + reasoning=reasoning, thread_id=thread_id, + ts=int(time.time() * 1000), + ), + ) + else: + message_id = _mint_message_id() + await self._broadcast_or_log( + chat_id, + protocol.message( + chat_id=chat_id, + message_id=message_id, + role=protocol.ROLE_ASSISTANT, + text=body, + thread_id=thread_id, + reasoning=reasoning, + reply_to=reply_to, + ts=int(time.time() * 1000), + ), + ) + await self._close_open_tool(chat_id, state, thread_id) + self._reset_tool_state(state) + state.active = False + return SendResult(success=True, message_id=message_id) + + # 3. Tool progress (first tool bubble of an editable line buffer). + if _is_tool_progress(content): + return await self._emit_tool_lines(chat_id, content, state, thread_id, is_edit=False) + + # 4. Commentary (interim assistant beat). + message_id = _mint_message_id() + state.active = True + await self._broadcast_or_log( + chat_id, + protocol.commentary(chat_id, message_id, content, thread_id=thread_id), + ) return SendResult(success=True, message_id=message_id) - async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None: - """Send a typing indicator. M0: no-op (M1 emits a ``typing`` frame).""" + async def edit_message( + self, + chat_id: str, + message_id: str, + content: str, + *, + finalize: bool = False, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Edit a previously sent message (M2: drives streaming + tool updates). + + * ``message_id == state.stream_id`` -> ``message.update`` + (``finalize=True`` -> ``message.stop``). + * ``message_id == state.tool_msg_id`` -> tool-progress update + (new lines -> ``tool.start``). + * unknown id -> best-effort ``message.update``. + """ + content = content or "" + thread_id = _thread_id_from_metadata(metadata) + state = self._turn_state(chat_id) + + if message_id and message_id == state.stream_id: + if finalize: + reasoning, body = _split_reasoning(_strip_streaming_cursor(content)) + # Streaming: the gateway drops the model's separate + # reasoning_content (final send suppressed), so attach the + # reasoning we captured via the on_stream_delta hook (wait + # for the async hook worker to flush it first). + if not reasoning: + await _wait_for_reasoning_flushed() + reasoning = _take_reasoning() or None + else: + _reset_reasoning() + state.stream_id = None + await self._broadcast_or_log( + chat_id, + protocol.message_stop( + chat_id, message_id, body, + reasoning=reasoning, thread_id=thread_id, + ts=int(time.time() * 1000), + ), + ) + await self._close_open_tool(chat_id, state, thread_id) + self._reset_tool_state(state) + state.active = False + else: + await self._broadcast_or_log( + chat_id, + protocol.message_update( + chat_id, message_id, _strip_streaming_cursor(content), + thread_id=thread_id, + ), + ) + return SendResult(success=True, message_id=message_id) + + if message_id and message_id == state.tool_msg_id: + return await self._emit_tool_lines(chat_id, content, state, thread_id, is_edit=True) + + # Unknown id: treat as a streaming update (best effort). + if finalize: + await self._broadcast_or_log( + chat_id, + protocol.message_stop( + chat_id, message_id, _strip_streaming_cursor(content), + thread_id=thread_id, ts=int(time.time() * 1000), + ), + ) + else: + await self._broadcast_or_log( + chat_id, + protocol.message_update( + chat_id, message_id, _strip_streaming_cursor(content), + thread_id=thread_id, + ), + ) + return SendResult(success=True, message_id=message_id) + + # ── M2: tool-progress helpers ───────────────────────────────────────── + + async def _emit_tool_lines( + self, + chat_id: str, + content: str, + state: _TurnState, + thread_id: Optional[str], + *, + is_edit: bool, + ) -> SendResult: + """Emit ``tool.start`` for each NEW tool line in *content*. + + The gateway accumulates tool lines in one editable bubble; on an edit + the full buffer is re-sent, so we diff against ``seen_tool_lines`` to + emit only the new ones. A new tool closes the previously-open tool. + """ + message_id = state.tool_msg_id or _mint_message_id() + state.tool_msg_id = message_id + state.active = True + + lines = [ln for ln in content.splitlines() if ln.strip()] + for line in lines: + key = line.strip() + if key in state.seen_tool_lines: + continue + state.seen_tool_lines.add(key) + parsed = self._parse_tool_line_or_block(line, content) + if parsed is None: + continue + name, preview = parsed + # A new tool begins: close the previously-open one. + if state.open_tool_index is not None: + await self._broadcast_or_log( + chat_id, + protocol.tool_end(chat_id, state.open_tool_index, "", ok=True, thread_id=thread_id), + ) + state.tool_index += 1 + state.open_tool_index = state.tool_index + await self._broadcast_or_log( + chat_id, + protocol.tool_start( + chat_id, state.tool_index, name, + preview=preview, thread_id=thread_id, + ), + ) + return SendResult(success=True, message_id=message_id) + + @staticmethod + def _parse_tool_line_or_block(line: str, content: str) -> Optional[Tuple[str, Optional[str]]]: + """Parse a tool line, expanding a terminal code block to its command.""" + parsed = _parse_tool_line(line) + if parsed is not None: + name, preview = parsed + # Terminal code block: the command lives in the fenced lines that + # follow the " terminal" head line. + if name == "terminal" and preview is None and "```" in content: + cmd = _extract_code_block(content) + if cmd: + return name, cmd + return parsed return None + async def _close_open_tool( + self, chat_id: str, state: _TurnState, thread_id: Optional[str] + ) -> None: + """Emit ``tool.end`` for the currently-open tool, if any. + + A tool is considered complete when the next tool starts OR a new + content segment begins (the model only produces content after the + tool it was waiting on has returned). + """ + if state.open_tool_index is not None: + await self._broadcast_or_log( + chat_id, + protocol.tool_end(chat_id, state.open_tool_index, "", ok=True, thread_id=thread_id), + ) + state.open_tool_index = None + + def _reset_tool_state(self, state: _TurnState) -> None: + """Clear per-turn tool bookkeeping (called at turn finalization).""" + state.tool_msg_id = None + state.seen_tool_lines = set() + state.tool_index = 0 + state.open_tool_index = None + + 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, + ) + + async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None: + """Send a typing indicator (``typing`` frame, on=true).""" + thread_id = None + if metadata: + tid = metadata.get("thread_id") + if isinstance(tid, str) and tid: + thread_id = tid + await self._ws_server.broadcast(protocol.typing(chat_id, True, thread_id=thread_id)) + + async def stop_typing(self, chat_id: str) -> None: + """Clear the typing indicator (``typing`` frame, on=false).""" + await self._ws_server.broadcast(protocol.typing(chat_id, False)) + async def send_image( self, chat_id: str, @@ -410,19 +1010,120 @@ class AndroidAdapter(BasePlatformAdapter): reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: - """Send an image. M0: not implemented (M4).""" + """Send an image. M1: not implemented (M4).""" return SendResult(success=False, error="android: media not implemented yet (M4)") + # ── Inbound (app -> agent) ──────────────────────────────────────────── + + async def on_message_send(self, frame: protocol.Frame, device_id: str) -> None: + """Handle an inbound ``message.send`` frame. + + 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). + """ + payload = frame.payload + text = payload.get("text") + if not isinstance(text, str) or not text.strip(): + await self._ws_server.send_to( + device_id, + protocol.error(protocol.ERR_UNSUPPORTED, "message.send requires non-empty text", id=frame.id), + ) + return + + chat_id = frame.chat_id or payload.get("chat_id") + if not isinstance(chat_id, str) or not chat_id.strip(): + chat_id = self.home_channel + chat_id = chat_id.strip() + + thread_id = frame.thread_id or payload.get("thread_id") + if not isinstance(thread_id, str) or not thread_id.strip(): + thread_id = None + + reply_to = payload.get("reply_to") + if not isinstance(reply_to, str) or not reply_to.strip(): + reply_to = None + + device = self._devices.get(device_id) or {} + user_name = device.get("name") or device_id + + # Echo to all devices: the sender confirms (server-assigned id), + # other devices see the message too (single-user, multi-device). + message_id = f"m_{uuid.uuid4().hex[:16]}" + echo = protocol.message( + chat_id=chat_id, + message_id=message_id, + role=protocol.ROLE_USER, + text=text, + thread_id=thread_id, + reply_to=reply_to, + ts=int(time.time() * 1000), + ) + await self._ws_server.broadcast(echo) + + source = self.build_source( + chat_id=chat_id, + chat_name=self._channel_name(chat_id), + chat_type="dm", + user_id=device_id, + user_name=user_name, + thread_id=thread_id, + ) + event = MessageEvent( + text=text, + message_type=MessageType.TEXT, + user_id=device_id, + user_name=user_name, + source=source, + message_id=message_id, + reply_to_message_id=reply_to, + ) + await self.handle_message(event) + # ── Chat info ───────────────────────────────────────────────────────── + def _channel_name(self, chat_id: str) -> str: + """Channel display name. M1: home channel only (directory is M3).""" + if chat_id in (self.home_channel, DEFAULT_HOME_CHANNEL): + return self.home_channel_name + return chat_id or "chat" + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: """Return ``{name, type, chat_id}`` for a chat. - M0: the channel directory is not persisted yet, so report the home - channel name for the default chat and a generic name otherwise. + M1: the channel directory is not persisted yet, so report the home + channel name for the default chat and the raw id otherwise. """ - name = "Default" if chat_id in (self.home_channel, DEFAULT_HOME_CHANNEL) else (chat_id or "chat") - return {"name": name, "type": "channel", "chat_id": chat_id} + return { + "name": self._channel_name(chat_id), + "type": "channel", + "chat_id": chat_id, + } + + # ── hello.ack helpers ───────────────────────────────────────────────── + + def server_caps(self) -> Dict[str, Any]: + """Capability flags advertised in ``hello.ack`` (M2 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 + "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, + } + ] # --------------------------------------------------------------------------- @@ -431,6 +1132,13 @@ class AndroidAdapter(BasePlatformAdapter): def register(ctx): """Plugin entry point: called by the Hermes plugin system.""" + # M2: capture the model's separate reasoning_content during streaming so + # it can be attached to the turn's message.stop frame (the gateway + # otherwise drops it when streaming suppresses the final send). + try: + ctx.register_hook("on_stream_delta", _on_stream_delta) + except Exception: + logger.debug("android: on_stream_delta hook registration failed", exc_info=True) ctx.register_platform( name="android", label="Android", diff --git a/gateway-plugin/pairing.py b/gateway-plugin/pairing.py index 3d36337..85327e8 100644 --- a/gateway-plugin/pairing.py +++ b/gateway-plugin/pairing.py @@ -7,4 +7,183 @@ created. QR payload for the pairing flow (``interactive_setup``). Storage: ``get_hermes_home()/"android"/devices.db``. Milestone M1. -""" \ No newline at end of file +""" + +import hmac +import json +import logging +import secrets +import sqlite3 +import threading +import time +from pathlib import Path +from typing import Any, Dict, List, Optional +from urllib.parse import quote + +logger = logging.getLogger(__name__) + +# 32 random bytes -> 64 hex chars (docs/09-pairing-security.md) +TOKEN_BYTES = 32 + + +def generate_token() -> str: + """Mint a fresh high-entropy pairing token (64 hex chars).""" + return secrets.token_hex(TOKEN_BYTES) + + +def verify_token(provided: Optional[str], expected: Optional[str]) -> bool: + """Constant-time token comparison (never time-leaks the token).""" + if not provided or not expected: + return False + return hmac.compare_digest( + provided.encode("utf-8", "replace"), + expected.encode("utf-8", "replace"), + ) + + +def qr_payload(host: str, port: int, token: str, secure: bool = False) -> str: + """Pairing URL encoded into the QR / pre-filled into the app. + + ``iris://pair?host=&port=8790&token=`` — the app's + Connect screen parses this to pre-fill settings (docs/09 §9.2). + """ + return ( + f"iris://pair?host={quote(host, safe='')}" + f"&port={int(port)}" + f"&secure={'1' if secure else '0'}" + f"&token={quote(token, safe='')}" + ) + + +def pairing_url(host: str, port: int, secure: bool = False) -> str: + """Plain ws(s) URL the app connects to (shown next to the QR).""" + scheme = "wss" if secure else "ws" + return f"{scheme}://{host}:{int(port)}/ws" + + +# --------------------------------------------------------------------------- +# Device registry (SQLite) +# --------------------------------------------------------------------------- + +class DeviceRegistry: + """Persistent device registry 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. + """ + + 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 devices ( + device_id TEXT PRIMARY KEY, + name TEXT NOT NULL DEFAULT '', + caps TEXT NOT NULL DEFAULT '{}', + fcm_token TEXT, + ntfy_topic TEXT, + last_seen REAL NOT NULL DEFAULT 0, + created REAL NOT NULL DEFAULT 0 + ) + """ + ) + self._conn.commit() + + def upsert( + self, + device_id: str, + name: str, + caps: Optional[Dict[str, Any]] = None, + fcm_token: Optional[str] = None, + ntfy_topic: Optional[str] = None, + ) -> None: + now = time.time() + caps_json = json.dumps(caps or {}, separators=(",", ":")) + with self._lock: + self._conn.execute( + """ + INSERT INTO devices (device_id, name, caps, fcm_token, ntfy_topic, + last_seen, created) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(device_id) DO UPDATE SET + name = excluded.name, + caps = excluded.caps, + fcm_token = COALESCE(excluded.fcm_token, devices.fcm_token), + ntfy_topic = COALESCE(excluded.ntfy_topic, devices.ntfy_topic), + last_seen = excluded.last_seen + """, + (device_id, name or "", caps_json, fcm_token, ntfy_topic, now, now), + ) + self._conn.commit() + + def update_push_tokens( + self, + device_id: str, + fcm_token: Optional[str] = None, + ntfy_topic: Optional[str] = None, + ) -> None: + with self._lock: + self._conn.execute( + """ + UPDATE devices SET + fcm_token = COALESCE(?, fcm_token), + ntfy_topic = COALESCE(?, ntfy_topic), + last_seen = ? + WHERE device_id = ? + """, + (fcm_token, ntfy_topic, time.time(), device_id), + ) + self._conn.commit() + + def touch(self, device_id: str) -> None: + with self._lock: + self._conn.execute( + "UPDATE devices SET last_seen = ? WHERE device_id = ?", + (time.time(), device_id), + ) + self._conn.commit() + + def get(self, device_id: str) -> Optional[Dict[str, Any]]: + with self._lock: + row = self._conn.execute( + "SELECT * FROM devices WHERE device_id = ?", (device_id,) + ).fetchone() + return _row_to_device(row) if row else None + + def list(self) -> List[Dict[str, Any]]: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM devices ORDER BY last_seen DESC" + ).fetchall() + return [_row_to_device(r) for r in rows] + + def close(self) -> None: + with self._lock: + try: + self._conn.close() + except Exception: + pass + + +def _row_to_device(row: sqlite3.Row) -> Dict[str, Any]: + try: + caps = json.loads(row["caps"] or "{}") + if not isinstance(caps, dict): + caps = {} + except (json.JSONDecodeError, TypeError): + caps = {} + return { + "device_id": row["device_id"], + "name": row["name"], + "caps": caps, + "fcm_token": row["fcm_token"], + "ntfy_topic": row["ntfy_topic"], + "last_seen": row["last_seen"], + "created": row["created"], + } \ No newline at end of file diff --git a/gateway-plugin/protocol.py b/gateway-plugin/protocol.py index 9be6b4e..831f697 100644 --- a/gateway-plugin/protocol.py +++ b/gateway-plugin/protocol.py @@ -5,18 +5,359 @@ constants. ``docs/protocol/frames.schema.json`` is generated/mirrored from this module, and the Kotlin side mirrors these shapes (see ``docs/04-wire-protocol.md``). -Milestone M1: hello/hello.ack, message, error, ping/pong. -Milestone M2: message.start/update/stop, reasoning, tool.*, commentary, typing. +Milestone M1: hello/hello.ack, message, message.send, error, ping/pong, +typing (typing is pulled forward from M2 so the app gets a live +"working…" indicator during the first milestone). +Milestone M2: message.start/update/stop, reasoning (on message / +message.stop), tool.start/progress/end, commentary. Milestone M3: channel.*, search, sync. Milestone M4: media.*. Milestone M5: notification, fcm.register, read.receipt. """ +import json +from dataclasses import dataclass, field +from typing import Any, Dict, Optional + PROTOCOL_VERSION = 1 -# Frame type constants (the ``type`` field of every frame). +# --------------------------------------------------------------------------- +# Frame type constants (the ``type`` field of every frame) +# --------------------------------------------------------------------------- + +# Pairing / lifecycle TYPE_HELLO = "hello" TYPE_HELLO_ACK = "hello.ack" TYPE_ERROR = "error" TYPE_PING = "ping" -TYPE_PONG = "pong" \ No newline at end of file +TYPE_PONG = "pong" + +# Chat +TYPE_MESSAGE = "message" +TYPE_MESSAGE_SEND = "message.send" +TYPE_TYPING = "typing" + +# Streaming (M2) +TYPE_MESSAGE_START = "message.start" +TYPE_MESSAGE_UPDATE = "message.update" +TYPE_MESSAGE_STOP = "message.stop" + +# Tool activity (M2) +TYPE_TOOL_START = "tool.start" +TYPE_TOOL_PROGRESS = "tool.progress" +TYPE_TOOL_END = "tool.end" + +# Intermediate assistant beat (M2) +TYPE_COMMENTARY = "commentary" + +# --------------------------------------------------------------------------- +# Error codes (``error`` frame payload.code) +# --------------------------------------------------------------------------- + +ERR_AUTH = "auth" +ERR_NOT_FOUND = "not_found" +ERR_RATE_LIMITED = "rate_limited" +ERR_MEDIA_TOO_LARGE = "media_too_large" +ERR_UNSUPPORTED = "unsupported" +ERR_INTERNAL = "internal" + +# --------------------------------------------------------------------------- +# Message roles (``message`` frame payload.role) +# --------------------------------------------------------------------------- + +ROLE_USER = "user" +ROLE_ASSISTANT = "assistant" +ROLE_SYSTEM = "system" +ROLE_CRON = "cron" + + +# --------------------------------------------------------------------------- +# Envelope +# --------------------------------------------------------------------------- + +@dataclass +class Frame: + """One wire frame. + + ``v`` is always serialised; ``id``/``chat_id``/``thread_id`` are + omitted when ``None`` (events carry no ``id``; chat-scoped frames carry + ``chat_id``/``thread_id`` at the top level for convenience). + """ + + type: str + payload: Dict[str, Any] = field(default_factory=dict) + id: Optional[int] = None + chat_id: Optional[str] = None + thread_id: Optional[str] = None + v: int = PROTOCOL_VERSION + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = {"v": self.v, "type": self.type} + if self.id is not None: + d["id"] = self.id + if self.chat_id is not None: + d["chat_id"] = self.chat_id + if self.thread_id is not None: + d["thread_id"] = self.thread_id + d["payload"] = self.payload + return d + + def to_json(self) -> str: + return json.dumps(self.to_dict(), separators=(",", ":"), ensure_ascii=False) + + @classmethod + def from_json(cls, raw: "str | bytes") -> Optional["Frame"]: + """Parse a text frame. Returns ``None`` for anything not a valid + frame (bad JSON, non-object, missing/invalid ``type``) so callers + can ignore malformed input (forward-compat).""" + try: + data = json.loads(raw) + except (json.JSONDecodeError, TypeError, UnicodeDecodeError, ValueError): + return None + if not isinstance(data, dict): + return None + ftype = data.get("type") + if not isinstance(ftype, str) or not ftype: + return None + payload = data.get("payload") + if not isinstance(payload, dict): + payload = {} + fid = data.get("id") + if not isinstance(fid, int) or isinstance(fid, bool): + fid = None + chat_id = data.get("chat_id") + if not isinstance(chat_id, str): + chat_id = None + thread_id = data.get("thread_id") + if not isinstance(thread_id, str): + thread_id = None + return cls(type=ftype, payload=payload, id=fid, chat_id=chat_id, thread_id=thread_id) + + +# --------------------------------------------------------------------------- +# Frame constructors (server -> app) +# --------------------------------------------------------------------------- + +def hello_ack( + server_caps: Dict[str, Any], + sync_cursor: int = 0, + channels: Optional[list] = None, +) -> Frame: + return Frame( + type=TYPE_HELLO_ACK, + payload={ + "server_caps": server_caps, + "sync_cursor": sync_cursor, + "channels": channels or [], + }, + ) + + +def message( + chat_id: str, + message_id: str, + role: str, + text: str, + *, + thread_id: Optional[str] = None, + reasoning: Optional[str] = None, + media: Optional[list] = None, + reply_to: Optional[str] = None, + model: Optional[str] = None, + tokens: Optional[int] = None, + ts: Optional[int] = None, +) -> Frame: + payload: Dict[str, Any] = { + "message_id": message_id, + "role": role, + "text": text, + } + if reasoning: + payload["reasoning"] = reasoning + if media: + payload["media"] = media + if reply_to: + payload["reply_to"] = reply_to + if model: + payload["model"] = model + if tokens is not None: + payload["tokens"] = tokens + if ts is not None: + payload["ts"] = ts + return Frame(type=TYPE_MESSAGE, chat_id=chat_id, thread_id=thread_id, payload=payload) + + +def typing(chat_id: str, on: bool = True, *, thread_id: Optional[str] = None) -> Frame: + return Frame( + type=TYPE_TYPING, + chat_id=chat_id, + thread_id=thread_id, + payload={"on": on}, + ) + + +# --------------------------------------------------------------------------- +# Streaming frames (M2) +# --------------------------------------------------------------------------- + +def message_start( + chat_id: str, + message_id: str, + role: str = ROLE_ASSISTANT, + *, + thread_id: Optional[str] = None, +) -> Frame: + """Open a streaming bubble.""" + return Frame( + type=TYPE_MESSAGE_START, + chat_id=chat_id, + thread_id=thread_id, + payload={"message_id": message_id, "role": role}, + ) + + +def message_update( + chat_id: str, + message_id: str, + text: str, + *, + thread_id: Optional[str] = None, +) -> Frame: + """Replace the live bubble text (full snapshot).""" + return Frame( + type=TYPE_MESSAGE_UPDATE, + chat_id=chat_id, + thread_id=thread_id, + payload={"message_id": message_id, "text": text}, + ) + + +def message_stop( + chat_id: str, + message_id: str, + final_text: str, + *, + thread_id: Optional[str] = None, + reasoning: Optional[str] = None, + model: Optional[str] = None, + tokens: Optional[int] = None, + ts: Optional[int] = None, +) -> Frame: + """Finalize a streaming bubble.""" + payload: Dict[str, Any] = { + "message_id": message_id, + "final_text": final_text, + } + if reasoning: + payload["reasoning"] = reasoning + if model: + payload["model"] = model + if tokens is not None: + payload["tokens"] = tokens + if ts is not None: + payload["ts"] = ts + return Frame( + type=TYPE_MESSAGE_STOP, + chat_id=chat_id, + thread_id=thread_id, + payload=payload, + ) + + +# --------------------------------------------------------------------------- +# Tool activity frames (M2) +# --------------------------------------------------------------------------- + +def tool_start( + chat_id: str, + index: int, + name: str, + *, + thread_id: Optional[str] = None, + preview: Optional[str] = None, + args: Optional[Dict[str, Any]] = None, +) -> Frame: + payload: Dict[str, Any] = {"index": index, "name": name} + if preview: + payload["preview"] = preview + if args: + payload["args"] = args + return Frame( + type=TYPE_TOOL_START, + chat_id=chat_id, + thread_id=thread_id, + payload=payload, + ) + + +def tool_progress( + chat_id: str, + index: int, + name: str, + *, + thread_id: Optional[str] = None, + note: Optional[str] = None, +) -> Frame: + payload: Dict[str, Any] = {"index": index, "name": name} + if note: + payload["note"] = note + return Frame( + type=TYPE_TOOL_PROGRESS, + chat_id=chat_id, + thread_id=thread_id, + payload=payload, + ) + + +def tool_end( + chat_id: str, + index: int, + name: str, + *, + thread_id: Optional[str] = None, + ok: bool = True, + duration: Optional[float] = None, + output_preview: Optional[str] = None, +) -> Frame: + payload: Dict[str, Any] = {"index": index, "name": name, "ok": ok} + if duration is not None: + payload["duration"] = duration + if output_preview: + payload["output_preview"] = output_preview + return Frame( + type=TYPE_TOOL_END, + chat_id=chat_id, + thread_id=thread_id, + payload=payload, + ) + + +# --------------------------------------------------------------------------- +# Commentary frame (M2) +# --------------------------------------------------------------------------- + +def commentary( + chat_id: str, + message_id: str, + text: str, + *, + thread_id: Optional[str] = None, +) -> Frame: + """An intermediate assistant beat (between tool iterations).""" + return Frame( + type=TYPE_COMMENTARY, + chat_id=chat_id, + thread_id=thread_id, + payload={"message_id": message_id, "text": text}, + ) + + +def error(code: str, message: str, *, id: Optional[int] = None) -> Frame: + return Frame(type=TYPE_ERROR, id=id, payload={"code": code, "message": message}) + + +def pong(ts: Optional[int] = None) -> Frame: + payload: Dict[str, Any] = {} + if ts is not None: + payload["ts"] = ts + return Frame(type=TYPE_PONG, payload=payload) \ No newline at end of file diff --git a/gateway-plugin/tests/ws_probe.py b/gateway-plugin/tests/ws_probe.py new file mode 100644 index 0000000..40d6f80 --- /dev/null +++ b/gateway-plugin/tests/ws_probe.py @@ -0,0 +1,202 @@ +#!/usr/bin/env python3 +"""WS test-client harness (docs/13-testing.md §13.2). + +Connects to the REAL running gateway and drives a turn, printing every +frame. This is how we empirically confirm the exact frame shapes before / +while building the Kotlin client. + +Usage:: + + hermes gateway & # with the android plugin + python gateway-plugin/tests/ws_probe.py --token \ + --send "hello" + +Options: + --url ws://host:port/ws (default ws://127.0.0.1:8790/ws) + --token ANDROID_TOKEN (default: $ANDROID_TOKEN) + --device device_id (default: probe-) + --send TEXT send this message after pairing (default: "hello") + --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 json +import os +import sys +import time +import uuid + +try: + import websockets +except ImportError: # pragma: no cover + sys.stderr.write("websockets is required (hermes core dep); run inside the hermes venv\n") + raise + + +def _print_frame(raw): + try: + data = json.loads(raw) + except (json.JSONDecodeError, TypeError): + print(f" <- {raw!r}") + return None + ftype = data.get("type", "?") + chat = data.get("chat_id") + fid = data.get("id") + payload = data.get("payload", {}) + # Compact one-line summary + full payload for the interesting frames. + extra = "" + if ftype == "message": + text = (payload.get("text") or "") + extra = f" role={payload.get('role')} id={payload.get('message_id')} text={text[:120]!r}" + if payload.get("reasoning"): + extra += f" reasoning={payload['reasoning'][:80]!r}" + elif ftype == "message.start": + extra = f" id={payload.get('message_id')} role={payload.get('role')}" + elif ftype == "message.update": + text = (payload.get("text") or "") + extra = f" id={payload.get('message_id')} text={text[:100]!r}" + elif ftype == "message.stop": + text = (payload.get("final_text") or "") + extra = f" id={payload.get('message_id')} text={text[:120]!r}" + if payload.get("reasoning"): + extra += f" reasoning={payload['reasoning'][:80]!r}" + elif ftype == "tool.start": + extra = (f" idx={payload.get('index')} name={payload.get('name')!r} " + f"preview={str(payload.get('preview'))[:80]!r}") + elif ftype == "tool.progress": + extra = f" idx={payload.get('index')} name={payload.get('name')!r} note={payload.get('note')!r}" + elif ftype == "tool.end": + extra = (f" idx={payload.get('index')} name={payload.get('name')!r} " + f"ok={payload.get('ok')} dur={payload.get('duration')}") + elif ftype == "commentary": + extra = f" id={payload.get('message_id')} text={(payload.get('text') or '')[:120]!r}" + elif ftype == "hello.ack": + extra = f" caps={payload.get('server_caps')} cursor={payload.get('sync_cursor')}" + elif ftype == "error": + extra = f" code={payload.get('code')} msg={payload.get('message')!r}" + elif ftype == "typing": + extra = f" on={payload.get('on')}" + elif ftype == "pong": + extra = "" + 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 + + +async def run(args) -> int: + url = args.url + token = args.token + device_id = args.device + + print(f"== ws_probe: connecting {url} device={device_id}") + try: + ws = await websockets.connect(url, open_timeout=10) + except Exception as e: + print(f"!! connect failed: {e}") + return 2 + + hello = { + "v": 1, + "type": "hello", + "payload": { + "token": token, + "device_id": device_id, + "device_name": "ws-probe", + "caps": {"min_protocol": 1}, + }, + } + await ws.send(json.dumps(hello)) + print(" -> hello") + + # First response must be hello.ack (or an auth error). + try: + first = await asyncio.wait_for(ws.recv(), timeout=10) + except asyncio.TimeoutError: + print("!! no hello.ack within 10s") + await ws.close() + return 3 + data = _print_frame(first) + if data is None or data.get("type") != "hello.ack": + if args.authfail: + print("== auth rejected as expected") + await ws.close() + return 0 + print("!! expected hello.ack") + await ws.close() + return 4 + if args.authfail: + print("!! expected auth rejection but got hello.ack") + await ws.close() + return 5 + + if not args.send: + print("== paired OK (no --send; exiting)") + await ws.close() + return 0 + + # Drive a turn. + msg_id = 1 + send_frame = { + "v": 1, + "id": msg_id, + "type": "message.send", + "chat_id": "android:default", + "payload": {"text": args.send}, + } + await ws.send(json.dumps(send_frame)) + print(f" -> message.send id={msg_id} text={args.send!r}") + + deadline = time.time() + args.timeout + got_final = False + seen_final_frame = False + while time.time() < deadline: + try: + raw = await asyncio.wait_for(ws.recv(), timeout=deadline - time.time()) + except asyncio.TimeoutError: + print(f"!! timeout after {args.timeout}s waiting for final message") + await ws.close() + return 6 + data = _print_frame(raw) + if data is None: + continue + ftype = data.get("type") + payload = data.get("payload") or {} + # A standalone assistant `message` (non-streaming) is immediately final. + if ftype == "message" and payload.get("role") == "assistant": + got_final = True + break + # A `message.stop` finalizes a streaming segment; the turn is done once + # typing stops afterwards (multi-segment turns have several stops). + if ftype == "message.stop": + seen_final_frame = True + if ftype == "typing" and payload.get("on") is False and seen_final_frame: + got_final = True + break + await ws.close() + if got_final: + print("== final assistant message received") + return 0 + print("!! no final assistant message") + return 7 + + +def main() -> int: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--url", default=os.getenv("ANDROID_WS_URL", "ws://127.0.0.1:8790/ws")) + 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("--timeout", type=float, default=120.0) + p.add_argument("--authfail", action="store_true", + help="expect an auth rejection (wrong token)") + args = p.parse_args() + if not args.token and not args.authfail: + p.error("--token (or $ANDROID_TOKEN) is required") + return asyncio.run(run(args)) + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/gateway-plugin/ws_server.py b/gateway-plugin/ws_server.py index 94b21c3..0fe14af 100644 --- a/gateway-plugin/ws_server.py +++ b/gateway-plugin/ws_server.py @@ -1,24 +1,309 @@ """WebSocket server, connection registry, and frame routing. Runs on the gateway's asyncio loop (started in ``AndroidAdapter.connect()``). -Uses the ``websockets`` core dep (v15): ``websockets.serve(handler, host, -port, ssl=ctx)``. +Uses the ``websockets`` core dep (v15): ``websockets.asyncio.server.serve( +handler, host, port, ssl=ctx)``. Per-connection handler: - 1. Await first frame; must be ``hello {token, device_id, device_name, - caps, fcm_token?}``. Verify token (constant-time) + allowlist. On - failure: send ``error {code:"auth"}`` and close. - 2. On success: register in the connection registry (``device_id -> - {ws, caps, fcm_token}``), send ``hello.ack {server_caps, sync_cursor, - channels[]}``. + 1. Await first frame (bounded); must be ``hello {token, device_id, + device_name, caps, fcm_token?}``. Verify token (constant-time) + + allowlist. On failure: send ``error {code:"auth"}`` and close. + 2. On success: register in the device registry (SQLite) + connection + registry (``device_id -> {ws, caps, fcm_token}``), send + ``hello.ack {server_caps, sync_cursor, channels[]}``. 3. Loop: decode frames, dispatch to adapter inbound handlers. - 4. On close: deregister; if no devices remain, ensure pending outbox - frames have push fired. + 4. On close: deregister. -Routing: ``emit(chat_id, frame)`` broadcasts to ALL connected devices -(single-user model). Heartbeat via WS ping/pong + app-level ping/pong. -Backpressure: bounded per-connection send queue; coalesce ``message.update`` -under pressure, never drop ``message``/``tool.end``/``notification``. +Routing: ``broadcast(frame)`` sends to ALL connected devices (single-user +model). Heartbeat via WS ping/pong (websockets built-in) + app-level +``ping``/``pong`` frames. Milestone M1. -""" \ No newline at end of file +""" + +import asyncio +import logging +import ssl +import time +from dataclasses import dataclass, field +from typing import Any, Dict, Optional + +from websockets.asyncio.server import ServerConnection, serve +from websockets.exceptions import ConnectionClosed + +from . import protocol +from .pairing import DeviceRegistry, verify_token + +logger = logging.getLogger(__name__) + +# How long a new socket may take to present its ``hello`` before we drop it. +HELLO_TIMEOUT_S = 10.0 + +# Close codes (4000-4999 are reserved for applications). +CLOSE_AUTH_FAILED = 4401 +CLOSE_REPLACED = 4402 +CLOSE_SHUTDOWN = 1001 + + +@dataclass +class DeviceConnection: + """One live, authenticated device socket.""" + + device_id: str + device_name: str + ws: ServerConnection + caps: Dict[str, Any] = field(default_factory=dict) + fcm_token: Optional[str] = None + ntfy_topic: Optional[str] = None + connected_at: float = field(default_factory=time.time) + + +class WsServer: + """The plugin's WebSocket server + live connection registry.""" + + def __init__(self, adapter: Any, devices: DeviceRegistry): + self._adapter = adapter + self._devices = devices + self._server: Optional[Any] = None + self._connections: Dict[str, DeviceConnection] = {} + self._lock = asyncio.Lock() + + # ── Lifecycle ───────────────────────────────────────────────────────── + + async def start(self) -> None: + """Bind and start serving. Raises on bind failure (adapter maps it + to a retryable fatal error).""" + adapter = self._adapter + ssl_ctx: Optional[ssl.SSLContext] = None + if adapter.ws_cert and adapter.ws_key: + try: + ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ssl_ctx.load_cert_chain(adapter.ws_cert, adapter.ws_key) + except Exception as e: + adapter._set_fatal_error( + "tls_config", f"WS TLS cert/key invalid: {e}", retryable=False + ) + raise + try: + self._server = await serve( + self._handler, + adapter.host, + adapter.port, + ssl=ssl_ctx, + # Media uploads (M4) are chunked binary frames; allow the + # configured max upload size per frame. + max_size=adapter.max_upload_bytes, + # WS-level heartbeat: dead peers are reaped by websockets. + ping_interval=20, + ping_timeout=20, + open_timeout=10, + ) + except OSError as e: + adapter._set_fatal_error( + "bind_failed", f"WS bind on {adapter.host}:{adapter.port} failed: {e}", + retryable=True, + ) + raise + scheme = "wss" if ssl_ctx else "ws" + logger.info( + "android: WS server listening on %s://%s:%s/ws", + scheme, adapter.host, adapter.port, + ) + + async def stop(self) -> None: + """Stop serving and close all device sockets.""" + if self._server is not None: + self._server.close() + try: + await self._server.wait_closed() + except Exception: + pass + self._server = None + for conn in list(self._connections.values()): + try: + await conn.ws.close(code=CLOSE_SHUTDOWN, reason="gateway shutting down") + except Exception: + pass + self._connections.clear() + + # ── Registry ────────────────────────────────────────────────────────── + + @property + def connections(self) -> Dict[str, DeviceConnection]: + return dict(self._connections) + + def has_devices(self) -> bool: + return bool(self._connections) + + def device_ids(self) -> list: + return list(self._connections.keys()) + + # ── 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).""" + data = frame.to_json() + sent = 0 + for conn in list(self._connections.values()): + try: + await conn.ws.send(data) + sent += 1 + except Exception: + pass + return sent + + async def send_to(self, device_id: str, frame: protocol.Frame) -> bool: + """Send a frame to one device (request responses / errors).""" + conn = self._connections.get(device_id) + if conn is None: + return False + try: + await conn.ws.send(frame.to_json()) + return True + except Exception: + return False + + # ── Per-connection handler ──────────────────────────────────────────── + + async def _handler(self, ws: ServerConnection) -> None: + # 1. hello auth ----------------------------------------------------- + try: + raw = await asyncio.wait_for(ws.recv(), timeout=HELLO_TIMEOUT_S) + except asyncio.TimeoutError: + logger.warning("android: dropping socket with no hello (timeout)") + await self._close_quiet(ws, 1000, "no hello") + return + except ConnectionClosed: + return + + frame = protocol.Frame.from_json(raw) + if frame is None or frame.type != protocol.TYPE_HELLO: + await self._reject(ws, "first frame must be hello") + return + + payload = frame.payload + if not verify_token(payload.get("token"), self._adapter.token): + peer = getattr(ws, "remote_address", None) + logger.warning("android: hello rejected: invalid token (peer=%s)", peer) + await self._reject(ws, "invalid token") + return + + device_id = str(payload.get("device_id") or "").strip() + if not device_id or len(device_id) > 128: + await self._reject(ws, "device_id required") + return + + if ( + not self._adapter.allow_all + and self._adapter.allowed_users + and device_id not in self._adapter.allowed_users + ): + logger.warning("android: hello rejected: device %s not allowlisted", device_id) + await self._reject(ws, "device not allowed") + return + + device_name = str(payload.get("device_name") or device_id)[:120] + caps = payload.get("caps") + if not isinstance(caps, dict): + caps = {} + fcm_token = payload.get("fcm_token") + if not isinstance(fcm_token, str): + fcm_token = None + ntfy_topic = payload.get("ntfy_topic") + if not isinstance(ntfy_topic, str): + ntfy_topic = None + + # 2. register -------------------------------------------------------- + try: + self._devices.upsert(device_id, device_name, caps, fcm_token, ntfy_topic) + except Exception: + logger.warning("android: device registry upsert failed", exc_info=True) + + conn = DeviceConnection( + device_id=device_id, + device_name=device_name, + ws=ws, + caps=caps, + fcm_token=fcm_token, + ntfy_topic=ntfy_topic, + ) + async with self._lock: + old = self._connections.pop(device_id, None) + self._connections[device_id] = conn + if old is not None: + # Same device re-paired from a new socket: the new one wins. + try: + await old.ws.close(code=CLOSE_REPLACED, reason="replaced by newer connection") + except Exception: + pass + + ack = protocol.hello_ack( + server_caps=self._adapter.server_caps(), + sync_cursor=0, # outbox lands in M3; cursor starts at 0 + channels=self._adapter.channel_list(), + ) + try: + await ws.send(ack.to_json()) + except Exception: + return + logger.info("android: device paired: %s (%s)", device_name, device_id) + + # 3. frame loop ------------------------------------------------------ + try: + async for raw in ws: + await self._on_frame(ws, device_id, raw) + except ConnectionClosed: + pass + except Exception: + logger.warning("android: frame loop error for %s", device_id, exc_info=True) + finally: + async with self._lock: + current = self._connections.get(device_id) + if current is not None and current.ws is ws: + self._connections.pop(device_id, None) + logger.info("android: device disconnected: %s", device_id) + + # ── Inbound dispatch ────────────────────────────────────────────────── + + async def _on_frame(self, ws: ServerConnection, device_id: str, raw: Any) -> None: + frame = protocol.Frame.from_json(raw) + if frame is None: + return # malformed / unknown binary: ignore (forward-compat) + + if frame.type == protocol.TYPE_PING: + ts = frame.payload.get("ts") + 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 == "fcm.register": + fcm_token = frame.payload.get("fcm_token") + ntfy_topic = frame.payload.get("ntfy_topic") + if isinstance(fcm_token, str) or isinstance(ntfy_topic, str): + try: + self._devices.update_push_tokens( + device_id, + fcm_token=fcm_token if isinstance(fcm_token, str) else None, + ntfy_topic=ntfy_topic if isinstance(ntfy_topic, str) else None, + ) + except Exception: + logger.warning("android: fcm.register update failed", exc_info=True) + # Unknown types are ignored (forward-compat). + + # ── Helpers ─────────────────────────────────────────────────────────── + + async def _send_quiet(self, ws: ServerConnection, frame: protocol.Frame) -> None: + try: + await ws.send(frame.to_json()) + except Exception: + pass + + async def _reject(self, ws: ServerConnection, reason: str) -> None: + await self._send_quiet(ws, protocol.error(protocol.ERR_AUTH, reason)) + await self._close_quiet(ws, CLOSE_AUTH_FAILED, "auth failed") + + async def _close_quiet(self, ws: ServerConnection, code: int, reason: str) -> None: + try: + await ws.close(code=code, reason=reason) + except Exception: + pass \ No newline at end of file