diff --git a/app/src/main/AndroidManifest.xml b/app/src/main/AndroidManifest.xml index 2760767..b94b5e4 100644 --- a/app/src/main/AndroidManifest.xml +++ b/app/src/main/AndroidManifest.xml @@ -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"> 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 { 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"), - role = getString("role").ifEmpty { "assistant" }, - sessionId = sessionId, - rawPayload = root - ) - "message.delta" -> MessageDeltaEvent( - messageId = getString("message_id", "id"), - delta = getString("delta", "text", "chunk"), - sessionId = sessionId, - rawPayload = root - ) - "message.interim" -> MessageInterimEvent( - messageId = getString("message_id", "id"), - content = getString("content", "text"), - sessionId = sessionId, - rawPayload = root - ) - "message.complete" -> MessageCompleteEvent( - messageId = getString("message_id", "id"), - content = getString("content", "text"), - sessionId = sessionId, - rawPayload = root - ) - "thinking.delta" -> ThinkingDeltaEvent( - messageId = getString("message_id", "id"), - delta = getString("delta", "text", "chunk"), - sessionId = sessionId, - rawPayload = root - ) - "reasoning.delta" -> ReasoningDeltaEvent( - messageId = getString("message_id", "id"), - delta = getString("delta", "text", "chunk"), - sessionId = sessionId, - rawPayload = root - ) - "reasoning.available" -> ReasoningAvailableEvent( - messageId = getString("message_id", "id"), - reasoning = getString("reasoning", "content"), - sessionId = sessionId, - rawPayload = root - ) - "tool.start" -> ToolStartEvent( - toolId = getString("tool_id", "id"), - name = getString("name", "tool_name"), - input = payloadObj["input"], - sessionId = sessionId, - rawPayload = root - ) - "tool.progress" -> ToolProgressEvent( - toolId = getString("tool_id", "id"), - progress = getString("progress", "message"), - sessionId = sessionId, - rawPayload = root - ) - "tool.generating" -> ToolGeneratingEvent( - toolId = getString("tool_id", "id"), - name = getString("name", "tool_name"), - sessionId = sessionId, - rawPayload = root - ) - "tool.complete" -> ToolCompleteEvent( - toolId = getString("tool_id", "id"), - result = getString("result", "output"), - isError = getBoolean("is_error", "error"), - sessionId = sessionId, - rawPayload = root - ) + "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" -> { + 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" -> { + val messageId = getString("message_id", "id") + if (messageId.isBlank()) return null + MessageInterimEvent( + messageId = messageId, + content = getString("content", "text"), + sessionId = sessionId, + rawPayload = root + ) + } + "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" -> { + 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" -> { + 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" -> { + val messageId = getString("message_id", "id") + if (messageId.isBlank()) return null + ReasoningAvailableEvent( + messageId = messageId, + reasoning = getString("reasoning", "content"), + sessionId = sessionId, + rawPayload = root + ) + } + "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" -> { + val toolId = getString("tool_id", "id") + if (toolId.isBlank()) return null + ToolProgressEvent( + toolId = toolId, + progress = getString("progress", "message"), + sessionId = sessionId, + rawPayload = root + ) + } + "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" -> { + 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"), - questionId = getNullableString("question_id", "questionId"), - question = getString("question", "prompt"), - promptType = ClarifyType.CLARIFY, - sessionId = sessionId, - rawPayload = root - ) - "sudo.request" -> SudoRequestEvent( - requestId = getString("request_id", "id"), - question = getString("question", "prompt").ifEmpty { "Administrator password required:" }, - sessionId = sessionId, - rawPayload = root - ) - "secret.request" -> SecretRequestEvent( - requestId = getString("request_id", "id"), - question = getString("question", "prompt").ifEmpty { "Secret / Token required:" }, - sessionId = sessionId, - rawPayload = root - ) + "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" -> { + 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" -> { + 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"), - result = getNullableString("result"), - sessionId = sessionId, - rawPayload = root - ) + "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 +} + + diff --git a/app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt b/app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt index c8c9ec4..aee03e1 100644 --- a/app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt +++ b/app/src/main/java/app/hermes/mobile/core/network/JsonRpcGatewayClient.kt @@ -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>() private var gatewayReadyDeferred = CompletableDeferred() + @Volatile private var activeWebSocket: WebSocket? = null + @Volatile + private var currentListener: WebSocketListener? = null private val _connectionState = MutableStateFlow(ConnectionState.Disconnected) val connectionState: StateFlow = _connectionState.asStateFlow() @@ -80,6 +91,32 @@ class JsonRpcGatewayClient( private val _events = MutableSharedFlow(extraBufferCapacity = 64) val events: SharedFlow = _events.asSharedFlow() + private val eventQueue = ConcurrentLinkedQueue() + 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}") } } diff --git a/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt b/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt index b7765ec..e9a8821 100644 --- a/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt +++ b/app/src/main/java/app/hermes/mobile/core/repository/UnifiedSessionRepository.kt @@ -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, diff --git a/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt b/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt index b3192c9..2cd22b0 100644 --- a/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt +++ b/app/src/main/java/app/hermes/mobile/core/runtime/HermesConnectionManager.kt @@ -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(extraBufferCapacity = 128) val allEvents: SharedFlow = _allEvents.asSharedFlow() + private val eventQueue = ConcurrentLinkedQueue() + 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 diff --git a/app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt b/app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt index 0ddb0cf..fd52278 100644 --- a/app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt +++ b/app/src/main/java/app/hermes/mobile/core/runtime/HermesHostRuntime.kt @@ -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(extraBufferCapacity = 64) val events: SharedFlow = _events.asSharedFlow() + private val eventQueue = ConcurrentLinkedQueue() + 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)) } } diff --git a/app/src/main/res/xml/network_security_config.xml b/app/src/main/res/xml/network_security_config.xml new file mode 100644 index 0000000..5198d41 --- /dev/null +++ b/app/src/main/res/xml/network_security_config.xml @@ -0,0 +1,13 @@ + + + + + + + + + + diff --git a/app/src/test/java/app/hermes/mobile/core/model/GatewayEventValidationTest.kt b/app/src/test/java/app/hermes/mobile/core/model/GatewayEventValidationTest.kt new file mode 100644 index 0000000..4f79bbc --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/model/GatewayEventValidationTest.kt @@ -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(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(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(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("""{"params":{"type":"message.start","payload":{"role":"assistant"}}}""") + assertNull(GatewayEvent.parse(noMsgIdStart)) + + // 2. message.delta without message_id -> null + val noMsgIdDelta = json.decodeFromString("""{"params":{"type":"message.delta","payload":{"delta":"hi"}}}""") + assertNull(GatewayEvent.parse(noMsgIdDelta)) + + // 3. message.complete without message_id -> null + val noMsgIdComplete = json.decodeFromString("""{"params":{"type":"message.complete","payload":{"content":"done"}}}""") + assertNull(GatewayEvent.parse(noMsgIdComplete)) + + // 4. thinking.delta without message_id -> null + val noMsgIdThinking = json.decodeFromString("""{"params":{"type":"thinking.delta","payload":{"delta":"thinking"}}}""") + assertNull(GatewayEvent.parse(noMsgIdThinking)) + + // 5. reasoning.delta without message_id -> null + val noMsgIdReasoning = json.decodeFromString("""{"params":{"type":"reasoning.delta","payload":{"delta":"reasoning"}}}""") + assertNull(GatewayEvent.parse(noMsgIdReasoning)) + + // 6. reasoning.available without message_id -> null + val noMsgIdReasoningAvail = json.decodeFromString("""{"params":{"type":"reasoning.available","payload":{"reasoning":"ready"}}}""") + assertNull(GatewayEvent.parse(noMsgIdReasoningAvail)) + + // 7. tool.start without tool_id -> null + val noToolIdStart = json.decodeFromString("""{"params":{"type":"tool.start","payload":{"name":"bash"}}}""") + assertNull(GatewayEvent.parse(noToolIdStart)) + + // 8. tool.progress without tool_id -> null + val noToolIdProgress = json.decodeFromString("""{"params":{"type":"tool.progress","payload":{"progress":"working"}}}""") + assertNull(GatewayEvent.parse(noToolIdProgress)) + + // 9. tool.generating without tool_id -> null + val noToolIdGen = json.decodeFromString("""{"params":{"type":"tool.generating","payload":{"name":"bash"}}}""") + assertNull(GatewayEvent.parse(noToolIdGen)) + + // 10. tool.complete without tool_id -> null + val noToolIdComplete = json.decodeFromString("""{"params":{"type":"tool.complete","payload":{"result":"ok"}}}""") + assertNull(GatewayEvent.parse(noToolIdComplete)) + + // 11. approval.request without request_id -> null + val noReqIdApproval = json.decodeFromString("""{"params":{"type":"approval.request","payload":{"command":"ls"}}}""") + assertNull(GatewayEvent.parse(noReqIdApproval)) + + // 12. clarify.request without request_id -> null + val noReqIdClarify = json.decodeFromString("""{"params":{"type":"clarify.request","payload":{"question":"port?"}}}""") + assertNull(GatewayEvent.parse(noReqIdClarify)) + + // 13. sudo.request without request_id -> null + val noReqIdSudo = json.decodeFromString("""{"params":{"type":"sudo.request","payload":{"question":"password"}}}""") + assertNull(GatewayEvent.parse(noReqIdSudo)) + + // 14. secret.request without request_id -> null + val noReqIdSecret = json.decodeFromString("""{"params":{"type":"secret.request","payload":{"question":"api key"}}}""") + assertNull(GatewayEvent.parse(noReqIdSecret)) + } + + @Test + fun testValidEventAfterRejectedEventParsesNormally() { + val corrupted = json.decodeFromString("""{"params":{"type":"message.delta","payload":{"delta":"corrupted"}}}""") + val rejectedEvent = GatewayEvent.parse(corrupted) + assertNull(rejectedEvent) + + val valid = json.decodeFromString("""{"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) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/network/EventOrderingTest.kt b/app/src/test/java/app/hermes/mobile/core/network/EventOrderingTest.kt new file mode 100644 index 0000000..b19950d --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/network/EventOrderingTest.kt @@ -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() + val allReceivedDeferred = CompletableDeferred() + + // 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) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/network/StaleSocketIsolationTest.kt b/app/src/test/java/app/hermes/mobile/core/network/StaleSocketIsolationTest.kt new file mode 100644 index 0000000..ac4491d --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/network/StaleSocketIsolationTest.kt @@ -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() + 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() + 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) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/repository/EmptyIdRejectionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/EmptyIdRejectionTest.kt new file mode 100644 index 0000000..2c05f4e --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/repository/EmptyIdRejectionTest.kt @@ -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()) + } +}