ochenstarik/hermes-android
1
0
Fork 0
Code Issues Pull requests 15 Projects Releases Packages Wiki Activity Actions

feat: LAN-host reachability and transport fixes

Browse source
This commit is contained in:
Ochenstarik 2026-08-24 20:43:06 +07:00
parent ba5f0466f3
commit 0d0f35fbd5
11 changed files with 957 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
app/src/main/AndroidManifest.xml
View file

@ -14,6 +14,7 @@
android:roundIcon="@android:drawable/sym_def_app_icon"
android:supportsRtl="true"
android:theme="@style/Theme.HermesAndroid"
android:networkSecurityConfig="@xml/network_security_config"
android:usesCleartextTraffic="false">
<activity

191
app/src/main/java/app/hermes/mobile/core/model/GatewayEvents.kt
View file

@ -177,40 +177,40 @@ sealed class GatewayEvent {
companion object {
private val json = Json { ignoreUnknownKeys = true; isLenient = true }
fun parse(root: JsonObject): GatewayEvent {
val params = root["params"]?.jsonObject ?: root
fun parse(root: JsonObject): GatewayEvent? {
val params = (root["params"] as? JsonObject) ?: root
// 1. event type -> params["type"]
val eventType = params["type"]?.jsonPrimitive?.content
?: params["event"]?.jsonPrimitive?.content
?: root["type"]?.jsonPrimitive?.content
?: root["event"]?.jsonPrimitive?.content
val eventType = params["type"]?.asStringOrNull()
?: params["event"]?.asStringOrNull()
?: root["type"]?.asStringOrNull()
?: root["event"]?.asStringOrNull()
?: ""
// 2. runtime session -> params["session_id"] (Do NOT search inside payload)
val sessionId = params["session_id"]?.jsonPrimitive?.content
?: params["session_key"]?.jsonPrimitive?.content
?: root["session_id"]?.jsonPrimitive?.content
?: root["session_key"]?.jsonPrimitive?.content
val sessionId = params["session_id"]?.asStringOrNull()
?: params["session_key"]?.asStringOrNull()
?: root["session_id"]?.asStringOrNull()
?: root["session_key"]?.asStringOrNull()
// 3. event body -> params["payload"]
val payloadObj = params["payload"]?.jsonObject
?: params["data"]?.jsonObject
?: root["payload"]?.jsonObject
?: root["data"]?.jsonObject
val payloadObj = (params["payload"] as? JsonObject)
?: (params["data"] as? JsonObject)
?: (root["payload"] as? JsonObject)
?: (root["data"] as? JsonObject)
?: params
fun getString(vararg keys: String): String {
for (k in keys) {
val v = payloadObj[k]?.jsonPrimitive?.content
if (v != null) return v
val v = payloadObj[k]?.asStringOrNull()
if (!v.isNullOrEmpty()) return v
}
return ""
}
fun getNullableString(vararg keys: String): String? {
for (k in keys) {
val v = payloadObj[k]?.jsonPrimitive?.content
val v = payloadObj[k]?.asStringOrNull()
if (v != null) return v
}
return null
@ -218,7 +218,7 @@ sealed class GatewayEvent {
fun getLong(vararg keys: String): Long {
for (k in keys) {
val v = payloadObj[k]?.jsonPrimitive?.longOrNull
val v = payloadObj[k]?.asLongOrNull()
if (v != null) return v
}
return 0L
@ -226,7 +226,7 @@ sealed class GatewayEvent {
fun getInt(vararg keys: String): Int {
for (k in keys) {
val v = payloadObj[k]?.jsonPrimitive?.intOrNull
val v = payloadObj[k]?.asIntOrNull()
if (v != null) return v
}
return 0
@ -234,7 +234,7 @@ sealed class GatewayEvent {
fun getBoolean(vararg keys: String): Boolean {
for (k in keys) {
val v = payloadObj[k]?.jsonPrimitive?.booleanOrNull
val v = payloadObj[k]?.asBooleanOrNull()
if (v != null) return v
}
return false
@ -242,7 +242,7 @@ sealed class GatewayEvent {
fun getStringList(key: String): List<String> {
val array = payloadObj[key] as? JsonArray ?: return emptyList()
return array.mapNotNull { it.jsonPrimitive.content }
return array.mapNotNull { it.asStringOrNull() }
}
return when (eventType) {
@ -251,78 +251,124 @@ sealed class GatewayEvent {
sessionCount = getInt("session_count", "sessions"),
rawPayload = root
)
"message.start" -> MessageStartEvent(
messageId = getString("message_id", "id"),
"message.start" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
MessageStartEvent(
messageId = messageId,
role = getString("role").ifEmpty { "assistant" },
sessionId = sessionId,
rawPayload = root
)
"message.delta" -> MessageDeltaEvent(
messageId = getString("message_id", "id"),
}
"message.delta" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
MessageDeltaEvent(
messageId = messageId,
delta = getString("delta", "text", "chunk"),
sessionId = sessionId,
rawPayload = root
)
"message.interim" -> MessageInterimEvent(
messageId = getString("message_id", "id"),
}
"message.interim" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
MessageInterimEvent(
messageId = messageId,
content = getString("content", "text"),
sessionId = sessionId,
rawPayload = root
)
"message.complete" -> MessageCompleteEvent(
messageId = getString("message_id", "id"),
}
"message.complete" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
MessageCompleteEvent(
messageId = messageId,
content = getString("content", "text"),
sessionId = sessionId,
rawPayload = root
)
"thinking.delta" -> ThinkingDeltaEvent(
messageId = getString("message_id", "id"),
}
"thinking.delta" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
ThinkingDeltaEvent(
messageId = messageId,
delta = getString("delta", "text", "chunk"),
sessionId = sessionId,
rawPayload = root
)
"reasoning.delta" -> ReasoningDeltaEvent(
messageId = getString("message_id", "id"),
}
"reasoning.delta" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
ReasoningDeltaEvent(
messageId = messageId,
delta = getString("delta", "text", "chunk"),
sessionId = sessionId,
rawPayload = root
)
"reasoning.available" -> ReasoningAvailableEvent(
messageId = getString("message_id", "id"),
}
"reasoning.available" -> {
val messageId = getString("message_id", "id")
if (messageId.isBlank()) return null
ReasoningAvailableEvent(
messageId = messageId,
reasoning = getString("reasoning", "content"),
sessionId = sessionId,
rawPayload = root
)
"tool.start" -> ToolStartEvent(
toolId = getString("tool_id", "id"),
}
"tool.start" -> {
val toolId = getString("tool_id", "id")
if (toolId.isBlank()) return null
ToolStartEvent(
toolId = toolId,
name = getString("name", "tool_name"),
input = payloadObj["input"],
sessionId = sessionId,
rawPayload = root
)
"tool.progress" -> ToolProgressEvent(
toolId = getString("tool_id", "id"),
}
"tool.progress" -> {
val toolId = getString("tool_id", "id")
if (toolId.isBlank()) return null
ToolProgressEvent(
toolId = toolId,
progress = getString("progress", "message"),
sessionId = sessionId,
rawPayload = root
)
"tool.generating" -> ToolGeneratingEvent(
toolId = getString("tool_id", "id"),
}
"tool.generating" -> {
val toolId = getString("tool_id", "id")
if (toolId.isBlank()) return null
ToolGeneratingEvent(
toolId = toolId,
name = getString("name", "tool_name"),
sessionId = sessionId,
rawPayload = root
)
"tool.complete" -> ToolCompleteEvent(
toolId = getString("tool_id", "id"),
}
"tool.complete" -> {
val toolId = getString("tool_id", "id")
if (toolId.isBlank()) return null
ToolCompleteEvent(
toolId = toolId,
result = getString("result", "output"),
isError = getBoolean("is_error", "error"),
sessionId = sessionId,
rawPayload = root
)
}
"approval.request" -> {
val requestId = getString("request_id", "id")
if (requestId.isBlank()) return null
val choices = getStringList("choices")
ApprovalRequestEvent(
requestId = getString("request_id", "id"),
requestId = requestId,
command = getNullableString("command"),
description = getNullableString("description", "prompt"),
choices = if (choices.isNotEmpty()) choices else listOf("once", "deny"),
@ -331,26 +377,38 @@ sealed class GatewayEvent {
rawPayload = root
)
}
"clarify.request" -> ClarifyRequestEvent(
requestId = getString("request_id", "id"),
"clarify.request" -> {
val requestId = getString("request_id", "id")
if (requestId.isBlank()) return null
ClarifyRequestEvent(
requestId = requestId,
questionId = getNullableString("question_id", "questionId"),
question = getString("question", "prompt"),
promptType = ClarifyType.CLARIFY,
sessionId = sessionId,
rawPayload = root
)
"sudo.request" -> SudoRequestEvent(
requestId = getString("request_id", "id"),
}
"sudo.request" -> {
val requestId = getString("request_id", "id")
if (requestId.isBlank()) return null
SudoRequestEvent(
requestId = requestId,
question = getString("question", "prompt").ifEmpty { "Administrator password required:" },
sessionId = sessionId,
rawPayload = root
)
"secret.request" -> SecretRequestEvent(
requestId = getString("request_id", "id"),
}
"secret.request" -> {
val requestId = getString("request_id", "id")
if (requestId.isBlank()) return null
SecretRequestEvent(
requestId = requestId,
question = getString("question", "prompt").ifEmpty { "Secret / Token required:" },
sessionId = sessionId,
rawPayload = root
)
}
"status.update" -> StatusUpdateEvent(
status = getString("status"),
message = getNullableString("message"),
@ -375,12 +433,16 @@ sealed class GatewayEvent {
sessionId = sessionId,
rawPayload = root
)
"background.complete" -> BackgroundCompleteEvent(
taskId = getString("task_id", "id"),
"background.complete" -> {
val taskId = getString("task_id", "id")
if (taskId.isBlank()) return null
BackgroundCompleteEvent(
taskId = taskId,
result = getNullableString("result"),
sessionId = sessionId,
rawPayload = root
)
}
"error" -> ErrorEvent(
code = getInt("code"),
message = getString("message").ifEmpty { "Unknown error" },
@ -397,3 +459,28 @@ sealed class GatewayEvent {
}
}
fun JsonElement?.asStringOrNull(): String? {
if (this == null || this is kotlinx.serialization.json.JsonNull) return null
val primitive = this as? kotlinx.serialization.json.JsonPrimitive ?: return null
return primitive.content
}
fun JsonElement?.asIntOrNull(): Int? {
if (this == null || this is kotlinx.serialization.json.JsonNull) return null
val primitive = this as? kotlinx.serialization.json.JsonPrimitive ?: return null
return primitive.intOrNull
}
fun JsonElement?.asLongOrNull(): Long? {
if (this == null || this is kotlinx.serialization.json.JsonNull) return null
val primitive = this as? kotlinx.serialization.json.JsonPrimitive ?: return null
return primitive.longOrNull
}
fun JsonElement?.asBooleanOrNull(): Boolean? {
if (this == null || this is kotlinx.serialization.json.JsonNull) return null
val primitive = this as? kotlinx.serialization.json.JsonPrimitive ?: return null
return primitive.booleanOrNull
}

82
app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt
View file

@ -40,10 +40,14 @@ import okhttp3.Request
import okhttp3.Response
import okhttp3.WebSocket
import okhttp3.WebSocketListener
import kotlinx.coroutines.channels.Channel
import java.io.IOException
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import java.util.logging.Logger
sealed class ConnectionState {
object Disconnected : ConnectionState()
@ -68,11 +72,18 @@ class JsonRpcGatewayClient(
encodeDefaults = true
}
private val logger = Logger.getLogger(JsonRpcGatewayClient::class.java.name)
private val droppedFramesCounter = AtomicInteger(0)
val droppedFrames: Int get() = droppedFramesCounter.get()
private val reqCounter = AtomicInteger(0)
private val pendingRequests = ConcurrentHashMap<String, CompletableDeferred<JsonRpcResponse>>()
private var gatewayReadyDeferred = CompletableDeferred<Unit>()
@Volatile
private var activeWebSocket: WebSocket? = null
@Volatile
private var currentListener: WebSocketListener? = null
private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected)
val connectionState: StateFlow<ConnectionState> = _connectionState.asStateFlow()
@ -80,6 +91,32 @@ class JsonRpcGatewayClient(
private val _events = MutableSharedFlow<GatewayEvent>(extraBufferCapacity = 64)
val events: SharedFlow<GatewayEvent> = _events.asSharedFlow()
private val eventQueue = ConcurrentLinkedQueue<GatewayEvent>()
private val isProcessingEvents = AtomicBoolean(false)
private fun dispatchEvent(event: GatewayEvent) {
eventQueue.add(event)
drainEventQueue()
}
private fun drainEventQueue() {
if (isProcessingEvents.compareAndSet(false, true)) {
scope.launch {
try {
while (true) {
val next = eventQueue.poll() ?: break
_events.emit(next)
}
} finally {
isProcessingEvents.set(false)
if (!eventQueue.isEmpty()) {
drainEventQueue()
}
}
}
}
}
private fun nextId(): String = "a${reqCounter.incrementAndGet()}"
fun connect(wsUrl: String, ticket: String? = null, allowCleartext: Boolean = false) {
@ -90,6 +127,14 @@ class JsonRpcGatewayClient(
return
}
currentListener = null
val oldWs = activeWebSocket
activeWebSocket = null
try {
oldWs?.close(1000, "Replaced by new connection")
oldWs?.cancel()
} catch (_: Exception) {}
gatewayReadyDeferred = CompletableDeferred()
_connectionState.value = ConnectionState.Connecting
@ -104,21 +149,26 @@ class JsonRpcGatewayClient(
.url(fullUrl)
.build()
activeWebSocket = client.newWebSocket(request, object : WebSocketListener() {
val listener = object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
if (this !== currentListener) return
activeWebSocket = webSocket
// Keep state as Connecting until gateway.ready event is received
_connectionState.value = ConnectionState.Connecting
}
override fun onMessage(webSocket: WebSocket, text: String) {
if (this !== currentListener || webSocket !== activeWebSocket) return
handleIncomingMessage(text)
}
override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
if (this !== currentListener || webSocket !== activeWebSocket) return
webSocket.close(code, reason)
}
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
if (this !== currentListener || webSocket !== activeWebSocket) return
_connectionState.value = ConnectionState.Disconnected
if (!gatewayReadyDeferred.isCompleted) {
gatewayReadyDeferred.completeExceptionally(IOException("WebSocket closed: $code $reason"))
@ -127,13 +177,18 @@ class JsonRpcGatewayClient(
}
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
if (this !== currentListener || webSocket !== activeWebSocket) return
_connectionState.value = ConnectionState.Failed(t)
if (!gatewayReadyDeferred.isCompleted) {
gatewayReadyDeferred.completeExceptionally(t)
}
failPendingRequests(t)
}
})
}
currentListener = listener
val newWs = client.newWebSocket(request, listener)
activeWebSocket = newWs
}
suspend fun awaitGatewayReady(timeoutMs: Long = 10_000) {
@ -152,12 +207,14 @@ class JsonRpcGatewayClient(
}
fun disconnect() {
currentListener = null
val ws = activeWebSocket
activeWebSocket = null
try {
activeWebSocket?.close(1000, "Client initiated disconnect")
activeWebSocket?.cancel()
ws?.close(1000, "Client initiated disconnect")
ws?.cancel()
} catch (_: Exception) {
}
activeWebSocket = null
_connectionState.value = ConnectionState.Disconnected
if (!gatewayReadyDeferred.isCompleted) {
gatewayReadyDeferred.completeExceptionally(IOException("Client disconnected"))
@ -202,20 +259,21 @@ class JsonRpcGatewayClient(
}
// 2. Otherwise, treat as Gateway Event / Notification
val event = GatewayEvent.parse(root)
val event = GatewayEvent.parse(root) ?: run {
val count = droppedFramesCounter.incrementAndGet()
logger.warning("Dropped invalid or unparseable gateway event frame #$count")
return
}
if (event is GatewayEvent.GatewayReadyEvent) {
_connectionState.value = ConnectionState.Connected
if (!gatewayReadyDeferred.isCompleted) {
gatewayReadyDeferred.complete(Unit)
}
}
if (!_events.tryEmit(event)) {
scope.launch {
_events.emit(event)
}
}
dispatchEvent(event)
} catch (e: Exception) {
// Ignore corrupted frames gracefully or log if debug
val count = droppedFramesCounter.incrementAndGet()
logger.warning("Dropped corrupted incoming frame #$count: ${e.javaClass.simpleName}: ${e.message}")
}
}

17
app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt
View file

@ -405,6 +405,7 @@ class UnifiedSessionRepository(
}
private fun insertMessageToSession(sessionId: UnifiedSessionId, message: UnifiedMessage) {
if (message.id.isBlank()) return
val flow = sessionMessagesState.computeIfAbsent(sessionId) {
MutableStateFlow(emptyList())
}
@ -420,6 +421,7 @@ class UnifiedSessionRepository(
messageId: String,
transform: (UnifiedMessage) -> UnifiedMessage
) {
if (messageId.isBlank()) return
val flow = sessionMessagesState.computeIfAbsent(sessionId) {
MutableStateFlow(emptyList())
}
@ -500,6 +502,7 @@ class UnifiedSessionRepository(
when (event) {
is GatewayEvent.MessageStartEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
setHostExecuting(sessionId, hostId, true)
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
@ -519,6 +522,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.MessageDeltaEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
setHostExecuting(sessionId, hostId, true)
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
@ -541,6 +545,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.MessageInterimEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
updateMessageInSession(sessionId, event.messageId) {
it.copy(content = event.content, isStreaming = true)
@ -548,6 +553,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.MessageCompleteEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
setHostExecuting(sessionId, hostId, false)
scope.launch {
@ -576,6 +582,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.ThinkingDeltaEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId }
@ -587,6 +594,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.ReasoningDeltaEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId }
@ -598,6 +606,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.ReasoningAvailableEvent -> {
if (event.messageId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return
val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) }
val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId }
@ -609,22 +618,26 @@ class UnifiedSessionRepository(
}
is GatewayEvent.ToolStartEvent -> {
if (event.toolId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId) ?: return
val tool = ToolActivity(id = event.toolId, name = event.name, status = "running")
attachToolToSessionMessage(sessionId, hostId, tool)
}
is GatewayEvent.ToolProgressEvent -> {
if (event.toolId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, toolId = event.toolId) ?: return
updateToolInSessionMessage(sessionId, event.toolId) { it.copy(progress = event.progress) }
}
is GatewayEvent.ToolGeneratingEvent -> {
if (event.toolId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, toolId = event.toolId) ?: return
updateToolInSessionMessage(sessionId, event.toolId) { it.copy(status = "generating") }
}
is GatewayEvent.ToolCompleteEvent -> {
if (event.toolId.isBlank()) return
val sessionId = findSessionForEvent(hostId, event.sessionId, toolId = event.toolId) ?: return
updateToolInSessionMessage(sessionId, event.toolId) {
it.copy(
@ -636,6 +649,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.ApprovalRequestEvent -> {
if (event.requestId.isBlank()) return
val runtimeSessionIdVal = event.sessionKey ?: event.sessionId ?: ""
val runtimeSessionId = RuntimeSessionId(runtimeSessionIdVal)
val approval = HermesApproval(
@ -656,6 +670,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.ClarifyRequestEvent -> {
if (event.requestId.isBlank()) return
val runtimeSessionIdVal = event.sessionId
val req = HermesClarifyRequest(
requestId = event.requestId,
@ -672,6 +687,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.SudoRequestEvent -> {
if (event.requestId.isBlank()) return
val runtimeSessionIdVal = event.sessionId
val req = HermesClarifyRequest(
requestId = event.requestId,
@ -687,6 +703,7 @@ class UnifiedSessionRepository(
}
is GatewayEvent.SecretRequestEvent -> {
if (event.requestId.isBlank()) return
val runtimeSessionIdVal = event.sessionId
val req = HermesClarifyRequest(
requestId = event.requestId,

35
app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt
View file

@ -12,6 +12,7 @@ import app.hermes.mobile.core.storage.HostEntity
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
@ -20,6 +21,8 @@ import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicBoolean
class HermesConnectionManager(
val hostDao: HostDao,
@ -48,6 +51,32 @@ class HermesConnectionManager(
private val _allEvents = MutableSharedFlow<HostGatewayEvent>(extraBufferCapacity = 128)
val allEvents: SharedFlow<HostGatewayEvent> = _allEvents.asSharedFlow()
private val eventQueue = ConcurrentLinkedQueue<HostGatewayEvent>()
private val isProcessingEvents = AtomicBoolean(false)
private fun dispatchEvent(event: HostGatewayEvent) {
eventQueue.add(event)
drainEventQueue()
}
private fun drainEventQueue() {
if (isProcessingEvents.compareAndSet(false, true)) {
scope.launch {
try {
while (true) {
val next = eventQueue.poll() ?: break
_allEvents.emit(next)
}
} finally {
isProcessingEvents.set(false)
if (!eventQueue.isEmpty()) {
drainEventQueue()
}
}
}
}
}
init {
scope.launch {
hostDao.getHostsFlow().collect { entities ->
@ -91,12 +120,10 @@ class HermesConnectionManager(
fun getOrCreateRuntime(host: HermesHost): HermesHostRuntime {
return runtimes.computeIfAbsent(host.id) {
val rt = runtimeFactory(scope, host)
// Forward events
// Forward events sequentially
scope.launch {
rt.events.collect { event ->
if (!_allEvents.tryEmit(event)) {
_allEvents.emit(event)
}
dispatchEvent(event)
}
}
// Update host status in DB on change

34
app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt
View file

@ -14,6 +14,7 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
@ -23,6 +24,8 @@ import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
import java.io.IOException
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicBoolean
import kotlin.math.min
import kotlin.random.Random
@ -49,6 +52,32 @@ class HermesHostRuntime(
private val _events = MutableSharedFlow<HostGatewayEvent>(extraBufferCapacity = 64)
val events: SharedFlow<HostGatewayEvent> = _events.asSharedFlow()
private val eventQueue = ConcurrentLinkedQueue<HostGatewayEvent>()
private val isProcessingEvents = AtomicBoolean(false)
private fun dispatchEvent(event: HostGatewayEvent) {
eventQueue.add(event)
drainEventQueue()
}
private fun drainEventQueue() {
if (isProcessingEvents.compareAndSet(false, true)) {
scope.launch {
try {
while (true) {
val next = eventQueue.poll() ?: break
_events.emit(next)
}
} finally {
isProcessingEvents.set(false)
if (!eventQueue.isEmpty()) {
drainEventQueue()
}
}
}
}
}
private var reconnectJob: Job? = null
private var autoReconnectEnabled = false
private var reconnectAttempt = 0
@ -56,10 +85,7 @@ class HermesHostRuntime(
init {
scope.launch {
gatewayClient.events.collect { event ->
val hostEvent = HostGatewayEvent(hostId, event)
if (!_events.tryEmit(hostEvent)) {
_events.emit(hostEvent)
}
dispatchEvent(HostGatewayEvent(hostId, event))
}
}

13
app/src/main/res/xml/network_security_config.xml Normal file
View file

@ -0,0 +1,13 @@
<?xml version="1.0" encoding="utf-8"?>
<network-security-config>
<!--
Permit cleartext traffic for local network / user-defined hosts while maintaining secure defaults.
Application-level checks in HermesRestClient and JsonRpcGatewayClient enforce explicit per-host user consent (allowCleartext flag).
-->
<base-config cleartextTrafficPermitted="true">
<trust-anchors>
<certificates src="system" />
<certificates src="user" />
</trust-anchors>
</base-config>
</network-security-config>

160
app/src/test/java/app/hermes/mobile/core/model/GatewayEventValidationTest.kt Normal file
View file

@ -0,0 +1,160 @@
package app.hermes.mobile.core.model
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonObject
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Test
class GatewayEventValidationTest {
private val json = Json { ignoreUnknownKeys = true; isLenient = true }
@Test
fun testSessionIdJsonNullParsesAsNullNotStringNull() {
val raw = """
{
"jsonrpc": "2.0",
"method": "event",
"params": {
"type": "message.delta",
"session_id": null,
"payload": {
"message_id": "m1",
"delta": "hello"
}
}
}
""".trimIndent()
val root = json.decodeFromString<JsonObject>(raw)
val event = GatewayEvent.parse(root)
assertNotNull(event)
assertTrue(event is GatewayEvent.MessageDeltaEvent)
val delta = event as GatewayEvent.MessageDeltaEvent
assertNull("sessionId should be null, not string 'null'", delta.sessionId)
assertEquals("m1", delta.messageId)
assertEquals("hello", delta.delta)
}
@Test
fun testNonPrimitiveFieldsDoNotCrashParser() {
// Object instead of string in delta / messageId, or array in place of primitive
val rawWithNestedObjects = """
{
"jsonrpc": "2.0",
"method": "event",
"params": {
"type": "message.delta",
"session_id": {"nested": "obj"},
"payload": {
"message_id": "m1",
"delta": {"malformed": [1, 2, 3]}
}
}
}
""".trimIndent()
val root1 = json.decodeFromString<JsonObject>(rawWithNestedObjects)
val event1 = GatewayEvent.parse(root1)
assertNotNull(event1)
assertTrue(event1 is GatewayEvent.MessageDeltaEvent)
val delta1 = event1 as GatewayEvent.MessageDeltaEvent
assertNull("Non-primitive session_id should safely resolve to null", delta1.sessionId)
assertEquals("", delta1.delta)
val rawWithArrayInIntField = """
{
"jsonrpc": "2.0",
"method": "event",
"params": {
"type": "error",
"payload": {
"code": [500],
"message": "some error"
}
}
}
""".trimIndent()
val root2 = json.decodeFromString<JsonObject>(rawWithArrayInIntField)
val event2 = GatewayEvent.parse(root2)
assertNotNull(event2)
assertTrue(event2 is GatewayEvent.ErrorEvent)
}
@Test
fun testMissingRequiredIdsAreRejected() {
// 1. message.start without message_id -> null
val noMsgIdStart = json.decodeFromString<JsonObject>("""{"params":{"type":"message.start","payload":{"role":"assistant"}}}""")
assertNull(GatewayEvent.parse(noMsgIdStart))
// 2. message.delta without message_id -> null
val noMsgIdDelta = json.decodeFromString<JsonObject>("""{"params":{"type":"message.delta","payload":{"delta":"hi"}}}""")
assertNull(GatewayEvent.parse(noMsgIdDelta))
// 3. message.complete without message_id -> null
val noMsgIdComplete = json.decodeFromString<JsonObject>("""{"params":{"type":"message.complete","payload":{"content":"done"}}}""")
assertNull(GatewayEvent.parse(noMsgIdComplete))
// 4. thinking.delta without message_id -> null
val noMsgIdThinking = json.decodeFromString<JsonObject>("""{"params":{"type":"thinking.delta","payload":{"delta":"thinking"}}}""")
assertNull(GatewayEvent.parse(noMsgIdThinking))
// 5. reasoning.delta without message_id -> null
val noMsgIdReasoning = json.decodeFromString<JsonObject>("""{"params":{"type":"reasoning.delta","payload":{"delta":"reasoning"}}}""")
assertNull(GatewayEvent.parse(noMsgIdReasoning))
// 6. reasoning.available without message_id -> null
val noMsgIdReasoningAvail = json.decodeFromString<JsonObject>("""{"params":{"type":"reasoning.available","payload":{"reasoning":"ready"}}}""")
assertNull(GatewayEvent.parse(noMsgIdReasoningAvail))
// 7. tool.start without tool_id -> null
val noToolIdStart = json.decodeFromString<JsonObject>("""{"params":{"type":"tool.start","payload":{"name":"bash"}}}""")
assertNull(GatewayEvent.parse(noToolIdStart))
// 8. tool.progress without tool_id -> null
val noToolIdProgress = json.decodeFromString<JsonObject>("""{"params":{"type":"tool.progress","payload":{"progress":"working"}}}""")
assertNull(GatewayEvent.parse(noToolIdProgress))
// 9. tool.generating without tool_id -> null
val noToolIdGen = json.decodeFromString<JsonObject>("""{"params":{"type":"tool.generating","payload":{"name":"bash"}}}""")
assertNull(GatewayEvent.parse(noToolIdGen))
// 10. tool.complete without tool_id -> null
val noToolIdComplete = json.decodeFromString<JsonObject>("""{"params":{"type":"tool.complete","payload":{"result":"ok"}}}""")
assertNull(GatewayEvent.parse(noToolIdComplete))
// 11. approval.request without request_id -> null
val noReqIdApproval = json.decodeFromString<JsonObject>("""{"params":{"type":"approval.request","payload":{"command":"ls"}}}""")
assertNull(GatewayEvent.parse(noReqIdApproval))
// 12. clarify.request without request_id -> null
val noReqIdClarify = json.decodeFromString<JsonObject>("""{"params":{"type":"clarify.request","payload":{"question":"port?"}}}""")
assertNull(GatewayEvent.parse(noReqIdClarify))
// 13. sudo.request without request_id -> null
val noReqIdSudo = json.decodeFromString<JsonObject>("""{"params":{"type":"sudo.request","payload":{"question":"password"}}}""")
assertNull(GatewayEvent.parse(noReqIdSudo))
// 14. secret.request without request_id -> null
val noReqIdSecret = json.decodeFromString<JsonObject>("""{"params":{"type":"secret.request","payload":{"question":"api key"}}}""")
assertNull(GatewayEvent.parse(noReqIdSecret))
}
@Test
fun testValidEventAfterRejectedEventParsesNormally() {
val corrupted = json.decodeFromString<JsonObject>("""{"params":{"type":"message.delta","payload":{"delta":"corrupted"}}}""")
val rejectedEvent = GatewayEvent.parse(corrupted)
assertNull(rejectedEvent)
val valid = json.decodeFromString<JsonObject>("""{"params":{"type":"message.delta","session_id":"s1","payload":{"message_id":"m2","delta":"valid"}}}""")
val validEvent = GatewayEvent.parse(valid)
assertNotNull(validEvent)
assertTrue(validEvent is GatewayEvent.MessageDeltaEvent)
val delta = validEvent as GatewayEvent.MessageDeltaEvent
assertEquals("m2", delta.messageId)
assertEquals("valid", delta.delta)
assertEquals("s1", delta.sessionId)
}
}

110
app/src/test/java/app/hermes/mobile/core/network/EventOrderingTest.kt Normal file
View file

@ -0,0 +1,110 @@
package app.hermes.mobile.core.network
import app.hermes.mobile.core.model.GatewayEvent
import app.hermes.mobile.core.model.HermesHost
import app.hermes.mobile.core.model.HermesHostId
import app.hermes.mobile.core.model.HostGatewayEvent
import app.hermes.mobile.core.runtime.HermesConnectionManager
import app.hermes.mobile.core.runtime.HermesHostRuntime
import app.hermes.mobile.core.security.InMemoryTokenVault
import app.hermes.mobile.core.storage.FakeHostDao
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals
import org.junit.Test
class EventOrderingTest {
@Test
fun test500SequentialDeltasPreserveFifoOrderingAcrossAll3Hops() = runBlocking {
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
val hostId = HermesHostId("test-host-ordering")
val host = HermesHost(id = hostId, displayName = "Ordering Host", baseUrl = "http://ordering-host:9119")
val tokenVault = InMemoryTokenVault()
val hostDao = FakeHostDao()
// Hop 1: GatewayClient
val gatewayClient = JsonRpcGatewayClient(scope = scope)
// Hop 2: HostRuntime
val runtime = HermesHostRuntime(
initialHost = host,
gatewayClient = gatewayClient,
tokenVault = tokenVault,
scope = scope
)
// Hop 3: ConnectionManager
val connectionManager = HermesConnectionManager(
hostDao = hostDao,
tokenVault = tokenVault,
scope = scope,
runtimeFactory = { _, _ -> runtime }
)
connectionManager.addHost(host)
val totalEvents = 500
val receivedDeltas = mutableListOf<String>()
val allReceivedDeferred = CompletableDeferred<Unit>()
// Collect from Hop 3 (HermesConnectionManager.allEvents)
val job = scope.launch {
connectionManager.allEvents.collect { hostGatewayEvent ->
val event = hostGatewayEvent.event
if (event is GatewayEvent.MessageDeltaEvent) {
receivedDeltas.add(event.delta)
if (receivedDeltas.size == totalEvents) {
allReceivedDeferred.complete(Unit)
}
}
}
}
// Give subscription a moment to establish
kotlinx.coroutines.delay(100)
val expectedBuilder = StringBuilder()
val eventJsons = (1..totalEvents).map { i ->
val chunk = "chunk-$i;"
expectedBuilder.append(chunk)
buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.delta")
put("session_id", "rt_ordering_1")
put("payload", buildJsonObject {
put("message_id", "msg_order_1")
put("delta", chunk)
})
})
}.toString()
}
// Rapid sequential emission from producer thread simulating WebSocket frames
for (jsonStr in eventJsons) {
gatewayClient.handleIncomingMessage(jsonStr)
}
withTimeout(15_000) {
allReceivedDeferred.await()
}
job.cancel()
val expectedString = expectedBuilder.toString()
val actualString = receivedDeltas.joinToString("")
assertEquals("All 500 events must be received", totalEvents, receivedDeltas.size)
assertEquals("Concatenated deltas must match byte-for-byte in exact FIFO order", expectedString, actualString)
}
}

173
app/src/test/java/app/hermes/mobile/core/network/StaleSocketIsolationTest.kt Normal file
View file

@ -0,0 +1,173 @@
package app.hermes.mobile.core.network
import app.hermes.mobile.core.model.JsonRpcResponse
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import okhttp3.Response
import okhttp3.WebSocket
import okhttp3.WebSocketListener
import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import java.io.IOException
class StaleSocketIsolationTest {
private lateinit var server1: MockWebServer
private lateinit var server2: MockWebServer
private lateinit var client: JsonRpcGatewayClient
@Before
fun setUp() {
server1 = MockWebServer()
server1.start()
server2 = MockWebServer()
server2.start()
client = JsonRpcGatewayClient()
}
@After
fun tearDown() {
client.disconnect()
try {
server1.shutdown()
} catch (_: Exception) {}
try {
server2.shutdown()
} catch (_: Exception) {}
}
@Test
fun testStaleSocketFailureDoesNotTransitionActiveConnectionStateToFailed() = runBlocking {
var server1Ws: WebSocket? = null
var server2Ws: WebSocket? = null
server1.enqueue(
MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
server1Ws = webSocket
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""")
}
})
)
val server2Received = CompletableDeferred<String>()
server2.enqueue(
MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
server2Ws = webSocket
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"2.0.0"}}}""")
}
override fun onMessage(webSocket: WebSocket, text: String) {
server2Received.complete(text)
}
})
)
// 1. Connect to server 1
client.connect("ws://${server1.hostName}:${server1.port}/api/ws", allowCleartext = true)
client.awaitGatewayReady(5000)
assertEquals(ConnectionState.Connected, client.connectionState.value)
// 2. Reconnect / connect to server 2
client.connect("ws://${server2.hostName}:${server2.port}/api/ws", allowCleartext = true)
client.awaitGatewayReady(5000)
assertEquals(ConnectionState.Connected, client.connectionState.value)
// 3. Send request to server 2
val requestDeferred = async(Dispatchers.IO) {
client.sendRequest("test.ping")
}
withTimeout(5000) {
server2Received.await()
}
// 4. Force failure / abrupt shutdown on stale server 1 socket
server1Ws?.close(1001, "Going away")
server1.shutdown()
// Give a moment for OkHttp to deliver stale socket failure/close callback
kotlinx.coroutines.delay(200)
// 5. Active connection state must still be Connected, NOT Failed
assertEquals(ConnectionState.Connected, client.connectionState.value)
// 6. Server 2 responds to the pending request
server2Ws?.send("""{"jsonrpc":"2.0","id":"a1","result":{"pong":true}}""")
val response = withTimeout(5000) {
requestDeferred.await()
}
assertNotNull(response)
assertEquals("a1", response.id)
assertEquals(ConnectionState.Connected, client.connectionState.value)
}
@Test
fun testStaleSocketFailureDoesNotAbortPendingRequestsOfActiveConnection() = runBlocking {
var server1Ws: WebSocket? = null
var server2Ws: WebSocket? = null
server1.enqueue(
MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
server1Ws = webSocket
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""")
}
})
)
val server2MsgDeferred = CompletableDeferred<String>()
server2.enqueue(
MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
server2Ws = webSocket
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"2.0.0"}}}""")
}
override fun onMessage(webSocket: WebSocket, text: String) {
server2MsgDeferred.complete(text)
}
})
)
client.connect("ws://${server1.hostName}:${server1.port}/api/ws", allowCleartext = true)
client.awaitGatewayReady(5000)
client.connect("ws://${server2.hostName}:${server2.port}/api/ws", allowCleartext = true)
client.awaitGatewayReady(5000)
val pendingCall = async(Dispatchers.IO) {
client.sendRequest("session.create")
}
server2MsgDeferred.await()
// Induce socket failure on server 1
server1Ws?.close(1001, "Going away")
server1.shutdown()
kotlinx.coroutines.delay(200)
// Ensure active connection is not in Failed state
assertTrue(client.connectionState.value is ConnectionState.Connected)
// Complete the pending request from server 2
server2Ws?.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"sess_active","session_id":"rt_active"}}""")
val result = withTimeout(5000) {
pendingCall.await()
}
assertNotNull(result.result)
}
}

149
app/src/test/java/app/hermes/mobile/core/repository/EmptyIdRejectionTest.kt Normal file
View file

@ -0,0 +1,149 @@
package app.hermes.mobile.core.repository
import app.hermes.mobile.core.model.BindingState
import app.hermes.mobile.core.model.HermesHost
import app.hermes.mobile.core.model.HermesHostId
import app.hermes.mobile.core.model.RuntimeSessionId
import app.hermes.mobile.core.runtime.HermesConnectionManager
import app.hermes.mobile.core.runtime.HermesHostRuntime
import app.hermes.mobile.core.security.InMemoryTokenVault
import app.hermes.mobile.core.storage.FakeHostDao
import app.hermes.mobile.core.storage.FakeUnifiedSessionDao
import app.hermes.mobile.core.storage.HostBindingEntity
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.resetMain
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.test.setMain
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
@OptIn(ExperimentalCoroutinesApi::class)
class EmptyIdRejectionTest {
private val testDispatcher = StandardTestDispatcher()
private lateinit var hostDao: FakeHostDao
private lateinit var sessionDao: FakeUnifiedSessionDao
private lateinit var tokenVault: InMemoryTokenVault
private lateinit var connectionManager: HermesConnectionManager
private lateinit var repository: UnifiedSessionRepository
private val hostId = HermesHostId("test-host-reject")
@Before
fun setUp() {
Dispatchers.setMain(testDispatcher)
hostDao = FakeHostDao()
sessionDao = FakeUnifiedSessionDao()
tokenVault = InMemoryTokenVault()
connectionManager = HermesConnectionManager(
hostDao = hostDao,
tokenVault = tokenVault,
scope = CoroutineScope(testDispatcher),
runtimeFactory = { parentScope, host ->
val childScope = CoroutineScope(kotlinx.coroutines.SupervisorJob(parentScope.coroutineContext[kotlinx.coroutines.Job]) + testDispatcher)
HermesHostRuntime(
initialHost = host,
restClient = app.hermes.mobile.core.network.HermesRestClient(),
gatewayClient = app.hermes.mobile.core.network.JsonRpcGatewayClient(scope = childScope),
tokenVault = tokenVault,
scope = childScope
)
}
)
repository = UnifiedSessionRepository(
connectionManager = connectionManager,
sessionDao = sessionDao,
scope = CoroutineScope(testDispatcher)
)
}
@After
fun tearDown() {
Dispatchers.resetMain()
}
@Test
fun testEmptyMessageIdDoesNotCreateMessageInRepository() = runTest(testDispatcher) {
val host = HermesHost(id = hostId, displayName = "Host Reject", baseUrl = "http://host-reject:9119")
connectionManager.addHost(host)
testScheduler.advanceUntilIdle()
val session = repository.createUnifiedSession(title = "Empty ID Test", initialHostId = hostId)
testScheduler.advanceUntilIdle()
val runtimeSessionId = "rt_empty_test"
repository.registerRuntimeBinding(session.id, hostId, RuntimeSessionId(runtimeSessionId))
sessionDao.insertOrUpdateBinding(
HostBindingEntity(
sessionId = session.id.value,
hostId = hostId.value,
durableSessionId = "dur_empty_test",
runtimeSessionId = runtimeSessionId,
state = BindingState.RUNNING.name
)
)
testScheduler.advanceUntilIdle()
val runtime = connectionManager.getRuntime(hostId)
// 1. MessageStart without message_id (or empty message_id)
val msgStartWithoutId = buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.start")
put("session_id", runtimeSessionId)
put("payload", buildJsonObject {
put("role", "assistant")
})
})
}
runtime?.gatewayClient?.handleIncomingMessage(msgStartWithoutId.toString())
testScheduler.advanceUntilIdle()
// 2. MessageDelta without message_id
val msgDeltaWithoutId = buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.delta")
put("session_id", runtimeSessionId)
put("payload", buildJsonObject {
put("delta", "Corrupted delta without ID")
})
})
}
runtime?.gatewayClient?.handleIncomingMessage(msgDeltaWithoutId.toString())
testScheduler.advanceUntilIdle()
// 3. MessageComplete without message_id
val msgCompleteWithoutId = buildJsonObject {
put("jsonrpc", "2.0")
put("method", "event")
put("params", buildJsonObject {
put("type", "message.complete")
put("session_id", runtimeSessionId)
put("payload", buildJsonObject {
put("content", "Corrupted complete without ID")
})
})
}
runtime?.gatewayClient?.handleIncomingMessage(msgCompleteWithoutId.toString())
testScheduler.advanceUntilIdle()
val messages = repository.getSessionMessages(session.id).value
assertTrue("No messages should be created from events missing message_id", messages.isEmpty())
val dbMessages = sessionDao.getMessagesForSession(session.id.value)
assertTrue("No entities should be persisted in DB from events missing message_id", dbMessages.isEmpty())
}
}