Merge pull request #1166 from Hy4ri/fix/1163-streaming-json-allocations

fix(#1163): avoid streaming JSON tree copies

app/src/androidTest/java/com/m57/hermescontrol/data/ws/StreamingEventDeviceTest.kt

@@ -0,0 +1,77 @@
+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 ../ \uD83C\uDF19\n",""" +
+                        """"extra":{"nested":[{"text":"ignored","path":"../ignored"}]}}}}"""
+                val response = Json.decodeFromString<JsonRpcResponse>(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<String, JsonElement>(
+                "type" to JsonPrimitive("reasoning.delta"),
+                "payload" to JsonObject(mapOf("text" to JsonPrimitive("Device streaming verified"))),
+            )
+        val params =
+            JsonObject(
+                object : Map<String, JsonElement> by fields {
+                    override val entries: Set<Map.Entry<String, JsonElement>>
+                        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()
+    }
+}

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<String, Any?> ?: return WsEvent.Unknown(rawJson)
+        val params = jsonParams.toAny() as Map<String, Any?>
         return parseParams(params, rawJson)
     }
 

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<String, Any?>
-                        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

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 -> {

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)
@@ -1826,6 +1829,47 @@ class HermesWsClientTest {
         collectorJob.cancel()
     }
 
+    @Test
+    fun testStreamingDedupPreservesSessionFallbackAndNumericSequenceSemantics() {
+        // Issue #1163: exercise the real listener, including the early duplicate return.
+        val socket = mockk<WebSocket>(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<WebSocket>(relaxed = true)

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<String, Any?>)
+                    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<String, JsonElement>): JsonObject =
+        JsonObject(
+            object : Map<String, JsonElement> by fields {
+                override val entries: Set<Map.Entry<String, JsonElement>>
+                    get() = error("Streaming path must look up fields, not traverse the JSON tree (#1163)")
+            },
+        )
+}