fix(#1163): avoid streaming JSON tree copies
This commit is contained in:
@@ -2,6 +2,7 @@ package com.m57.hermescontrol.data.ws
|
|||||||
|
|
||||||
import android.util.Log
|
import android.util.Log
|
||||||
import com.m57.hermescontrol.ui.chat.extractTodosFromMap
|
import com.m57.hermescontrol.ui.chat.extractTodosFromMap
|
||||||
|
import kotlinx.serialization.json.JsonObject
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Converts raw [JsonRpcResponse] objects into typed [WsEvent] instances.
|
* Converts raw [JsonRpcResponse] objects into typed [WsEvent] instances.
|
||||||
@@ -38,8 +39,22 @@ object EventParser {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ── Notification / event (no id, has method) ─────────────────────
|
// ── 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")
|
@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)
|
return parseParams(params, rawJson)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1316,12 +1316,9 @@ object HermesWsClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (rpc.id == null) {
|
if (rpc.id == null) {
|
||||||
@Suppress("UNCHECKED_CAST")
|
// Issue #1163: inspect scalars, not a throwaway copy of the whole params tree.
|
||||||
val params = rpc.params?.toAny() as? Map<String, Any?>
|
val sid = rpc.params?.eventSessionId()
|
||||||
val sid =
|
val seq = ((rpc.params?.get("seq") as? JsonPrimitive)?.toAny() as? Number)?.toInt()
|
||||||
(params?.get("session_id") as? String)
|
|
||||||
?: ((params?.get("payload") as? Map<*, *>)?.get("session_id") as? String)
|
|
||||||
val seq = (params?.get("seq") as? Number)?.toInt()
|
|
||||||
|
|
||||||
if (!sid.isNullOrBlank() && seq != null) {
|
if (!sid.isNullOrBlank() && seq != null) {
|
||||||
val prev = lastSeenSeq[sid] ?: 0
|
val prev = lastSeenSeq[sid] ?: 0
|
||||||
|
|||||||
@@ -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? =
|
fun JsonElement.toAny(): Any? =
|
||||||
when (this) {
|
when (this) {
|
||||||
is JsonNull -> {
|
is JsonNull -> {
|
||||||
|
|||||||
@@ -1826,6 +1826,47 @@ class HermesWsClientTest {
|
|||||||
collectorJob.cancel()
|
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
|
@Test
|
||||||
fun testTerminalCloseCode4403SetsAuthExpired() {
|
fun testTerminalCloseCode4403SetsAuthExpired() {
|
||||||
val socket = mockk<WebSocket>(relaxed = true)
|
val socket = mockk<WebSocket>(relaxed = true)
|
||||||
|
|||||||
@@ -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)")
|
||||||
|
},
|
||||||
|
)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user