diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt index 3306955b..c278711e 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/EventParser.kt @@ -2,6 +2,7 @@ package com.m57.hermescontrol.data.ws import android.util.Log import com.m57.hermescontrol.ui.chat.extractTodosFromMap +import kotlinx.serialization.json.JsonObject /** * Converts raw [JsonRpcResponse] objects into typed [WsEvent] instances. @@ -38,8 +39,22 @@ object EventParser { } // ── Notification / event (no id, has method) ───────────────────── + val jsonParams = response.params ?: return WsEvent.Unknown(rawJson) + // Issue #1163: token events need only scalar lookups, never recursive map/list copies. + val eventType = jsonParams["type"].eventStringOrNull() + if (eventType == "message.token" || eventType == "message.delta" || + eventType == "thinking.delta" || eventType == "reasoning.delta" + ) { + val sessionId = jsonParams.eventSessionId() + val token = (jsonParams["payload"] as? JsonObject)?.get("text").eventStringOrNull() ?: "" + return when (eventType) { + "thinking.delta" -> WsEvent.ThinkingDelta(token, sessionId) + "reasoning.delta" -> WsEvent.ReasoningDelta(token, sessionId) + else -> WsEvent.MessageToken(token, sessionId) + } + } @Suppress("UNCHECKED_CAST") - val params = response.params?.toAny() as? Map ?: return WsEvent.Unknown(rawJson) + val params = jsonParams.toAny() as Map return parseParams(params, rawJson) } diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt index 65d3ecc8..883ffd3a 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt @@ -1316,12 +1316,9 @@ object HermesWsClient { } if (rpc.id == null) { - @Suppress("UNCHECKED_CAST") - val params = rpc.params?.toAny() as? Map - val sid = - (params?.get("session_id") as? String) - ?: ((params?.get("payload") as? Map<*, *>)?.get("session_id") as? String) - val seq = (params?.get("seq") as? Number)?.toInt() + // Issue #1163: inspect scalars, not a throwaway copy of the whole params tree. + val sid = rpc.params?.eventSessionId() + val seq = ((rpc.params?.get("seq") as? JsonPrimitive)?.toAny() as? Number)?.toInt() if (!sid.isNullOrBlank() && seq != null) { val prev = lastSeenSeq[sid] ?: 0 diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt index 796df16c..489ac929 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt @@ -102,6 +102,13 @@ fun Any?.toJsonElement(): JsonElement = } } +// Issue #1163: scalar conversion preserves toAny() type semantics without copying JSON containers. +internal fun JsonElement?.eventStringOrNull(): String? = (this as? JsonPrimitive)?.toAny() as? String + +internal fun JsonObject.eventSessionId(): String? = + get("session_id").eventStringOrNull() + ?: (get("payload") as? JsonObject)?.get("session_id").eventStringOrNull() + fun JsonElement.toAny(): Any? = when (this) { is JsonNull -> { diff --git a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt index dff70f1a..058abae9 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt @@ -1826,6 +1826,47 @@ class HermesWsClientTest { collectorJob.cancel() } + @Test + fun testStreamingDedupPreservesSessionFallbackAndNumericSequenceSemantics() { + // Issue #1163: exercise the real listener, including the early duplicate return. + val socket = mockk(relaxed = true) + val intentionalClose = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalClose.isAccessible = true + (intentionalClose.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + val generation = HermesWsClient::class.java.getDeclaredField("connectionGeneration") + generation.isAccessible = true + val constructor = + Class + .forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl") + .declaredConstructors + .single() + constructor.isAccessible = true + val listener = + constructor.newInstance( + (generation.get(HermesWsClient) as AtomicInteger).get(), + ) as WebSocketListener + // Directly install the reflection-built listener so onMessage runs even + // though this test never performed a real OkHttp connect. + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, socket) + mockkObject(EventParser) + + val seqValues = listOf("6", "6.9", "4294967302", "\"6\"", "null", "true", "{}", "[]") + for ((index, seq) in seqValues.withIndex()) { + val sid = "stream-$index" + HermesWsClient.setSeqWatermark(sid, 5) + val text = + """{"method":"event","params":{"type":"message.token","session_id":false,"seq":$seq,""" + + """"payload":{"session_id":"$sid","text":"chunk"}}}""" + listener.onMessage(socket, text) + listener.onMessage(socket, text) + val numeric = index < 3 + assertEquals(if (numeric) 6 else 5, HermesWsClient.getSeqWatermarks()[sid]) + verify(exactly = if (numeric) 1 else 2) { EventParser.parse(any(), text) } + } + } + @Test fun testTerminalCloseCode4403SetsAuthExpired() { val socket = mockk(relaxed = true) diff --git a/app/src/test/java/com/m57/hermescontrol/data/ws/StreamingEventParserTest.kt b/app/src/test/java/com/m57/hermescontrol/data/ws/StreamingEventParserTest.kt new file mode 100644 index 00000000..d33bb09e --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/data/ws/StreamingEventParserTest.kt @@ -0,0 +1,78 @@ +package com.m57.hermescontrol.data.ws + +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import org.junit.Assert.assertEquals +import org.junit.Test + +class StreamingEventParserTest { + private val types = listOf("message.token", "message.delta", "thinking.delta", "reasoning.delta") + + @Test + fun `streaming parsing matches map parser for wire field variants`() { + val sessions = listOf("\"outer\"", "\"\"", "null", "42", "true", "{}", "[]") + val payloads = + listOf( + "null", + "[]", + "42", + "{}", + """{"session_id":"inner","text":"hello 🌙\n"}""", + """{"session_id":null,"text":null}""", + """{"session_id":false,"text":123}""", + """{"session_id":[],"text":true}""", + """{"session_id":{},"text":{}}""", + """{"text":[]}""", + ) + for (type in types) { + for (session in sessions + null) { + for (payload in payloads + null) { + val fields = mutableListOf("\"type\":\"$type\"") + if (session != null) fields.add("\"session_id\":$session") + if (payload != null) fields.add("\"payload\":$payload") + val params = Json.parseToJsonElement("{${fields.joinToString()}}") as JsonObject + + @Suppress("UNCHECKED_CAST") + val legacy = EventParser.parseParams(params.toAny() as Map) + assertEquals( + params.toString(), + legacy, + EventParser.parse(JsonRpcResponse(method = "event", params = params)), + ) + } + } + } + } + + @Test + fun `streaming parsing never traverses params or payload maps`() { + for (type in types) { + val payload = lookupOnlyObject(mapOf("text" to JsonPrimitive("chunk"))) + val params = + lookupOnlyObject( + mapOf( + "type" to JsonPrimitive(type), + "session_id" to JsonPrimitive("session"), + "payload" to payload, + ), + ) + val expected = + when (type) { + "thinking.delta" -> WsEvent.ThinkingDelta("chunk", "session") + "reasoning.delta" -> WsEvent.ReasoningDelta("chunk", "session") + else -> WsEvent.MessageToken("chunk", "session") + } + assertEquals(expected, EventParser.parse(JsonRpcResponse(method = "event", params = params))) + } + } + + private fun lookupOnlyObject(fields: Map): JsonObject = + JsonObject( + object : Map by fields { + override val entries: Set> + get() = error("Streaming path must look up fields, not traverse the JSON tree (#1163)") + }, + ) +}