From 613d01c1b6fe0709697483f50822f2f79d796088 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:43:04 +0300 Subject: [PATCH 1/4] fix(#1163): avoid streaming JSON tree copies --- .../m57/hermescontrol/data/ws/EventParser.kt | 17 +++- .../hermescontrol/data/ws/HermesWsClient.kt | 9 +-- .../hermescontrol/data/ws/JsonRpcModels.kt | 7 ++ .../data/ws/HermesWsClientTest.kt | 41 ++++++++++ .../data/ws/StreamingEventParserTest.kt | 78 +++++++++++++++++++ 5 files changed, 145 insertions(+), 7 deletions(-) create mode 100644 app/src/test/java/com/m57/hermescontrol/data/ws/StreamingEventParserTest.kt 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)") + }, + ) +} From d0b4db53f66e3d770a824eed5315413cab9947d0 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:50:13 +0300 Subject: [PATCH 2/4] test(#1163): verify streaming parser on Android emulator --- .../data/ws/StreamingEventDeviceTest.kt | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt diff --git a/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt b/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt new file mode 100644 index 00000000..b16f20cc --- /dev/null +++ b/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt @@ -0,0 +1,76 @@ +package com.m57.hermescontrol.data.ws + +import androidx.compose.ui.test.assertIsDisplayed +import androidx.compose.ui.test.junit4.createComposeRule +import androidx.compose.ui.test.onNodeWithTag +import androidx.compose.ui.test.onNodeWithText +import androidx.compose.ui.test.performClick +import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.m57.hermescontrol.theme.HermesControlTheme +import com.m57.hermescontrol.ui.chat.components.ReasoningCard +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.Rule +import org.junit.Test +import org.junit.runner.RunWith + +/** Issue #1163: execute the wire parser on ART, without backend credentials or model calls. */ +@RunWith(AndroidJUnit4::class) +class StreamingEventDeviceTest { + @get:Rule + val compose = createComposeRule() + + @Test + fun streamingBurstPreservesEveryChunkAndSessionOnDevice() { + val types = listOf("message.token", "message.delta", "thinking.delta", "reasoning.delta") + for (type in types) { + val actual = StringBuilder() + val expected = StringBuilder() + repeat(1000) { seq -> + val frame = + """{"method":"event","params":{"type":"$type","seq":$seq,""" + + """"payload":{"session_id":"device-test","text":"chunk-$seq 🌙\n"}}}""" + val response = Json.decodeFromString(frame) + val event = EventParser.parse(response, frame) + val (token, session) = + when (event) { + is WsEvent.MessageToken -> event.token to event.sessionId + is WsEvent.ThinkingDelta -> event.token to event.sessionId + is WsEvent.ReasoningDelta -> event.token to event.sessionId + else -> error("Unexpected streaming event: $event") + } + assertEquals("device-test", session) + actual.append(token) + expected.append("chunk-$seq 🌙\n") + } + assertEquals(type, expected.toString(), actual.toString()) + } + } + + @Test + fun reasoningFastPathAvoidsContainerTraversalAndRendersOnDevice() { + val fields = + mapOf( + "type" to JsonPrimitive("reasoning.delta"), + "payload" to JsonObject(mapOf("text" to JsonPrimitive("Device streaming verified"))), + ) + val params = + JsonObject( + object : Map by fields { + override val entries: Set> + get() = error("Streaming parser traversed the params tree") + }, + ) + val event = EventParser.parse(JsonRpcResponse(method = "event", params = params)) as WsEvent.ReasoningDelta + compose.setContent { + HermesControlTheme { + ReasoningCard(reasoningText = event.token) + } + } + compose.onNodeWithTag("reasoning_card").performClick() + compose.onNodeWithText("Device streaming verified").assertIsDisplayed() + } +} From ac180ba5dfac61209f89d0bd6f6cff574a393c1c Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Thu, 17 Sep 2026 16:15:11 +0300 Subject: [PATCH 3/4] test(#1163): cover nested payloads and escaped streaming text --- .../m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt b/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt index b16f20cc..27a1073f 100644 --- a/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt +++ b/app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt @@ -32,7 +32,8 @@ class StreamingEventDeviceTest { repeat(1000) { seq -> val frame = """{"method":"event","params":{"type":"$type","seq":$seq,""" + - """"payload":{"session_id":"device-test","text":"chunk-$seq 🌙\n"}}}""" + """"payload":{"session_id":"device-test","text":"chunk-$seq ../ \uD83C\uDF19\n",""" + + """"extra":{"nested":[{"text":"ignored","path":"../ignored"}]}}}}""" val response = Json.decodeFromString(frame) val event = EventParser.parse(response, frame) val (token, session) = @@ -44,7 +45,7 @@ class StreamingEventDeviceTest { } assertEquals("device-test", session) actual.append(token) - expected.append("chunk-$seq 🌙\n") + expected.append("chunk-$seq ../ 🌙\n") } assertEquals(type, expected.toString(), actual.toString()) } From 287f3c1561d4d4d895c848bc28ba4bcc328ab24b Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Thu, 17 Sep 2026 16:32:57 +0300 Subject: [PATCH 4/4] test(#1163): subscribe before sending dedup frames --- .../java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 058abae9..d933bf9f 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 @@ -1671,7 +1671,10 @@ class HermesWsClientTest { val tokenALatch = CountDownLatch(1) val tokenBLatch = CountDownLatch(1) val collectorJob = - kotlinx.coroutines.CoroutineScope(kotlinx.coroutines.Dispatchers.IO).launch { + // Subscribe before sending A: this SharedFlow does not replay missed tokens (#1163). + kotlinx.coroutines.CoroutineScope(kotlinx.coroutines.Dispatchers.IO).launch( + start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED, + ) { HermesWsClient.events.collect { event -> if (event is WsEvent.MessageToken) { receivedTokens.add(event.token)