From fc3ddeda652f4b32ad1a7466561c71dcbdd39ff6 Mon Sep 17 00:00:00 2001 From: Ochenstarik Date: Mon, 24 Aug 2026 01:52:38 +0700 Subject: [PATCH] Fix critical Multi-Hermes defects (P0/P1) and add concurrency test suite --- .../hermes/mobile/core/model/GatewayEvents.kt | 48 +- .../mobile/core/model/MultiHostModels.kt | 2 + .../repository/UnifiedSessionRepository.kt | 345 ++++++--- .../mobile/core/runtime/HermesHostRuntime.kt | 4 +- .../mobile/core/sync/UnifiedContextBuilder.kt | 2 +- .../hermes/mobile/feature/chat/ChatScreen.kt | 284 +++++--- .../mobile/feature/chat/ChatViewModel.kt | 48 +- .../core/repository/ApprovalRoutingTest.kt | 167 ++++- .../MultiHostConcurrencyExecutionTest.kt | 656 ++++++++++++++++++ .../UnifiedSessionRepositoryTest.kt | 37 + .../hermes/mobile/core/storage/FakeDaos.kt | 56 +- .../core/sync/UnifiedContextBuilderTest.kt | 3 + 12 files changed, 1397 insertions(+), 255 deletions(-) create mode 100644 app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt diff --git a/app/src/main/java/app/hermes/mobile/core/model/GatewayEvents.kt b/app/src/main/java/app/hermes/mobile/core/model/GatewayEvents.kt index 1e30703..cbc07e4 100644 --- a/app/src/main/java/app/hermes/mobile/core/model/GatewayEvents.kt +++ b/app/src/main/java/app/hermes/mobile/core/model/GatewayEvents.kt @@ -13,6 +13,7 @@ import kotlinx.serialization.json.longOrNull sealed class GatewayEvent { abstract val rawPayload: JsonObject + open val sessionId: String? get() = null data class GatewayReadyEvent( val version: String, @@ -23,42 +24,49 @@ sealed class GatewayEvent { data class MessageStartEvent( val messageId: String, val role: String = "assistant", + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class MessageDeltaEvent( val messageId: String, val delta: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class MessageInterimEvent( val messageId: String, val content: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class MessageCompleteEvent( val messageId: String, val content: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class ThinkingDeltaEvent( val messageId: String, val delta: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class ReasoningDeltaEvent( val messageId: String, val delta: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class ReasoningAvailableEvent( val messageId: String, val reasoning: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() @@ -66,18 +74,21 @@ sealed class GatewayEvent { val toolId: String, val name: String, val input: JsonElement? = null, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class ToolProgressEvent( val toolId: String, val progress: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class ToolGeneratingEvent( val toolId: String, val name: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() @@ -85,6 +96,7 @@ sealed class GatewayEvent { val toolId: String, val result: String, val isError: Boolean = false, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() @@ -94,6 +106,7 @@ sealed class GatewayEvent { val description: String? = null, val choices: List = listOf("once", "deny"), val sessionKey: String? = null, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() @@ -102,24 +115,28 @@ sealed class GatewayEvent { val questionId: String? = null, val question: String, val promptType: ClarifyType = ClarifyType.CLARIFY, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class SudoRequestEvent( val requestId: String, val question: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class SecretRequestEvent( val requestId: String, val question: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class StatusUpdateEvent( val status: String, val message: String? = null, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() @@ -127,28 +144,33 @@ sealed class GatewayEvent { val inputTokens: Long = 0, val outputTokens: Long = 0, val totalTokens: Long = 0, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class SessionInfoEvent( val info: SessionInfo, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class BackgroundCompleteEvent( val taskId: String, val result: String? = null, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class ErrorEvent( val code: Int = -1, val message: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() data class UnknownGatewayEvent( val eventType: String, + override val sessionId: String? = null, override val rawPayload: JsonObject ) : GatewayEvent() @@ -233,6 +255,8 @@ sealed class GatewayEvent { return array.mapNotNull { it.jsonPrimitive.content } } + val sessionKey = getNullableString("session_id", "session_key", "sessionKey", "sessionId") + return when (eventType) { "gateway.ready" -> GatewayReadyEvent( version = getString("version", "server_version"), @@ -242,58 +266,69 @@ sealed class GatewayEvent { "message.start" -> MessageStartEvent( messageId = getString("message_id", "id"), role = getString("role").ifEmpty { "assistant" }, + sessionId = sessionKey, rawPayload = root ) "message.delta" -> MessageDeltaEvent( messageId = getString("message_id", "id"), delta = getString("delta", "text", "chunk"), + sessionId = sessionKey, rawPayload = root ) "message.interim" -> MessageInterimEvent( messageId = getString("message_id", "id"), content = getString("content", "text"), + sessionId = sessionKey, rawPayload = root ) "message.complete" -> MessageCompleteEvent( messageId = getString("message_id", "id"), content = getString("content", "text"), + sessionId = sessionKey, rawPayload = root ) "thinking.delta" -> ThinkingDeltaEvent( messageId = getString("message_id", "id"), delta = getString("delta", "text", "chunk"), + sessionId = sessionKey, rawPayload = root ) "reasoning.delta" -> ReasoningDeltaEvent( messageId = getString("message_id", "id"), delta = getString("delta", "text", "chunk"), + sessionId = sessionKey, rawPayload = root ) "reasoning.available" -> ReasoningAvailableEvent( messageId = getString("message_id", "id"), reasoning = getString("reasoning", "content"), + sessionId = sessionKey, rawPayload = root ) "tool.start" -> ToolStartEvent( toolId = getString("tool_id", "id"), name = getString("name", "tool_name"), input = dataObj["input"] ?: root["input"], + sessionId = sessionKey, rawPayload = root ) "tool.progress" -> ToolProgressEvent( toolId = getString("tool_id", "id"), progress = getString("progress", "message"), + sessionId = sessionKey, rawPayload = root ) "tool.generating" -> ToolGeneratingEvent( toolId = getString("tool_id", "id"), name = getString("name", "tool_name"), + sessionId = sessionKey, rawPayload = root ) "tool.complete" -> ToolCompleteEvent( toolId = getString("tool_id", "id"), result = getString("result", "output"), isError = getBoolean("is_error", "error"), + sessionId = sessionKey, rawPayload = root ) "approval.request" -> { @@ -303,7 +338,8 @@ sealed class GatewayEvent { command = getNullableString("command"), description = getNullableString("description", "prompt"), choices = if (choices.isNotEmpty()) choices else listOf("once", "deny"), - sessionKey = getNullableString("session_key", "sessionKey"), + sessionKey = sessionKey, + sessionId = sessionKey, rawPayload = root ) } @@ -312,27 +348,32 @@ sealed class GatewayEvent { questionId = getNullableString("question_id", "questionId"), question = getString("question", "prompt"), promptType = ClarifyType.CLARIFY, + sessionId = sessionKey, rawPayload = root ) "sudo.request" -> SudoRequestEvent( requestId = getString("request_id", "id"), question = getString("question", "prompt").ifEmpty { "Administrator password required:" }, + sessionId = sessionKey, rawPayload = root ) "secret.request" -> SecretRequestEvent( requestId = getString("request_id", "id"), question = getString("question", "prompt").ifEmpty { "Secret / Token required:" }, + sessionId = sessionKey, rawPayload = root ) "status.update" -> StatusUpdateEvent( status = getString("status"), message = getNullableString("message"), + sessionId = sessionKey, rawPayload = root ) "session.usage" -> SessionUsageEvent( inputTokens = getLong("input_tokens", "prompt_tokens"), outputTokens = getLong("output_tokens", "completion_tokens"), totalTokens = getLong("total_tokens"), + sessionId = sessionKey, rawPayload = root ) "session.info" -> SessionInfoEvent( @@ -343,23 +384,28 @@ sealed class GatewayEvent { branch = getNullableString("branch"), project = getNullableString("project") ), + sessionId = sessionKey, rawPayload = root ) "background.complete" -> BackgroundCompleteEvent( taskId = getString("task_id", "id"), result = getNullableString("result"), + sessionId = sessionKey, rawPayload = root ) "error" -> ErrorEvent( code = getInt("code"), message = getString("message").ifEmpty { "Unknown error" }, + sessionId = sessionKey, rawPayload = root ) else -> UnknownGatewayEvent( eventType = eventType.ifEmpty { "unknown" }, + sessionId = sessionKey, rawPayload = root ) } } } } + diff --git a/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt b/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt index 5400357..a558d79 100644 --- a/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt +++ b/app/src/main/java/app/hermes/mobile/core/model/MultiHostModels.kt @@ -97,6 +97,7 @@ data class HostGatewayEvent( data class HostAttributedApproval( val hostId: HermesHostId, val hostDisplayName: String, + val runtimeSessionId: RuntimeSessionId, val approval: HermesApproval ) @@ -104,5 +105,6 @@ data class HostAttributedApproval( data class HostAttributedClarify( val hostId: HermesHostId, val hostDisplayName: String, + val runtimeSessionId: RuntimeSessionId? = null, val request: HermesClarifyRequest ) 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 7f252f8..185dfdc 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 @@ -3,6 +3,7 @@ package app.hermes.mobile.core.repository import app.hermes.mobile.core.model.* import app.hermes.mobile.core.network.ConnectionState import app.hermes.mobile.core.runtime.HermesConnectionManager +import app.hermes.mobile.core.runtime.HermesHostRuntime import app.hermes.mobile.core.storage.* import app.hermes.mobile.core.sync.UnifiedContextBuilder import kotlinx.coroutines.CoroutineScope @@ -48,9 +49,23 @@ class UnifiedSessionRepository( // In-memory active session messages cache for reactive streaming updates private val sessionMessagesState = ConcurrentHashMap>>() + + // Independent per-(session, host) execution state + private val hostExecutingState = ConcurrentHashMap, MutableStateFlow>() private val sessionExecutingState = ConcurrentHashMap>() init { + scope.launch { + sessions.collect { list -> + for (s in list) { + for ((hId, b) in s.bindings) { + if (b.runtimeSessionId.value.isNotEmpty()) { + runtimeToSessionMap[b.runtimeSessionId.value] = Pair(s.id, hId) + } + } + } + } + } scope.launch { connectionManager.allEvents.collect { hostEvent -> handleHostGatewayEvent(hostEvent) @@ -71,6 +86,12 @@ class UnifiedSessionRepository( }.asStateFlow() } + fun getHostExecuting(sessionId: UnifiedSessionId, hostId: HermesHostId): StateFlow { + return hostExecutingState.computeIfAbsent(Pair(sessionId, hostId)) { + MutableStateFlow(false) + }.asStateFlow() + } + fun getSessionExecuting(sessionId: UnifiedSessionId): StateFlow { return sessionExecutingState.computeIfAbsent(sessionId) { MutableStateFlow(false) @@ -83,7 +104,7 @@ class UnifiedSessionRepository( ): UnifiedSession { val hostId = initialHostId ?: connectionManager.activeHostId.value ?: connectionManager.hosts.value.firstOrNull()?.id - ?: HermesHostId("default") + ?: throw IllegalStateException("No Hermes hosts configured. Please add a host before creating a session.") val sessionId = UnifiedSessionId(UUID.randomUUID().toString()) val sessionEntity = UnifiedSessionEntity( @@ -120,12 +141,71 @@ class UnifiedSessionRepository( sessionDao.deleteSession(sessionId.value) sessionMessagesState.remove(sessionId) sessionExecutingState.remove(sessionId) + hostExecutingState.entries.removeIf { it.key.first == sessionId } + runtimeToSessionMap.entries.removeIf { it.value.first == sessionId } + } + + fun registerRuntimeBinding(sessionId: UnifiedSessionId, hostId: HermesHostId, runtimeSessionId: RuntimeSessionId) { + if (runtimeSessionId.value.isNotEmpty()) { + runtimeToSessionMap[runtimeSessionId.value] = Pair(sessionId, hostId) + } } suspend fun switchSessionActiveHost(sessionId: UnifiedSessionId, targetHostId: HermesHostId) { sessionDao.updateActiveHost(sessionId.value, targetHostId.value, System.currentTimeMillis()) } + suspend fun ensureAttachedRuntimeSession( + sessionId: UnifiedSessionId, + targetHostId: HermesHostId, + runtime: HermesHostRuntime + ): HostSessionBinding { + val details = sessionDao.getSessionWithDetails(sessionId.value) + var binding = details?.bindings?.find { it.hostId == targetHostId.value }?.toDomain() + + if (binding == null || binding.durableSessionId.value.isEmpty()) { + val createRes = runtime.gatewayClient.createSession(source = "android") + binding = HostSessionBinding( + hostId = targetHostId, + durableSessionId = createRes.durableId, + runtimeSessionId = createRes.runtimeId, + lastAttachedAt = System.currentTimeMillis(), + state = BindingState.READY, + syncedThroughMessageId = null, + syncedAt = null + ) + sessionDao.insertOrUpdateBinding(binding.toEntity(sessionId.value)) + runtimeToSessionMap[createRes.runtimeId.value] = Pair(sessionId, targetHostId) + return binding + } + + // We have an existing durableSessionId. + // Check if current runtimeSessionId is already registered and valid, or if we need to resume + val currentRuntimeId = binding.runtimeSessionId.value + val isRegistered = currentRuntimeId.isNotEmpty() && runtimeToSessionMap.containsKey(currentRuntimeId) + + if (!isRegistered || binding.state == BindingState.NOT_CREATED || binding.state == BindingState.OFFLINE || binding.state == BindingState.ERROR) { + val resumeRes = try { + runtime.gatewayClient.resumeSession(binding.durableSessionId, source = "android") + } catch (_: Exception) { + val createRes = runtime.gatewayClient.createSession(source = "android") + ResumeSessionResult(createRes.durableId, createRes.runtimeId) + } + binding = binding.copy( + durableSessionId = resumeRes.durableId, + runtimeSessionId = resumeRes.runtimeId, + lastAttachedAt = System.currentTimeMillis(), + state = BindingState.READY + ) + sessionDao.insertOrUpdateBinding(binding.toEntity(sessionId.value)) + runtimeToSessionMap[resumeRes.runtimeId.value] = Pair(sessionId, targetHostId) + } else { + runtimeToSessionMap[currentRuntimeId] = Pair(sessionId, targetHostId) + } + + return binding + } + suspend fun sendPrompt(sessionId: UnifiedSessionId, text: String): String { val details = sessionDao.getSessionWithDetails(sessionId.value) ?: throw IllegalArgumentException("Session not found: ${sessionId.value}") @@ -147,23 +227,8 @@ class UnifiedSessionRepository( runtime.gatewayClient.awaitGatewayReady(10_000) } - // Get or create native session binding for this host - var binding = currentSession.bindings[targetHostId] - if (binding == null || binding.runtimeSessionId.value.isEmpty()) { - val createRes = runtime.gatewayClient.createSession(source = "android") - binding = HostSessionBinding( - hostId = targetHostId, - durableSessionId = createRes.durableId, - runtimeSessionId = createRes.runtimeId, - lastAttachedAt = System.currentTimeMillis(), - state = BindingState.READY, - syncedThroughMessageId = null, - syncedAt = null - ) - sessionDao.insertOrUpdateBinding(binding.toEntity(sessionId.value)) - } - - runtimeToSessionMap[binding.runtimeSessionId.value] = Pair(sessionId, targetHostId) + // Get or attach native session binding for this host + val binding = ensureAttachedRuntimeSession(sessionId, targetHostId, runtime) // Context Synchronization val hostsMap = connectionManager.hosts.value.associateBy { it.id } @@ -190,15 +255,6 @@ class UnifiedSessionRepository( text } - // Update binding sync status - sessionDao.updateBindingSync( - sessionId = sessionId.value, - hostId = targetHostId.value, - syncedThroughMessageId = syncResult.latestSyncedMessageId, - syncedAt = System.currentTimeMillis(), - state = BindingState.RUNNING.name - ) - // Insert user message to timeline val userMessage = UnifiedMessage( id = UUID.randomUUID().toString(), @@ -210,49 +266,70 @@ class UnifiedSessionRepository( ) insertMessageToSession(sessionId, userMessage) - setExecuting(sessionId, true) + setHostExecuting(sessionId, targetHostId, true) + sessionDao.updateBindingState(sessionId.value, targetHostId.value, BindingState.RUNNING.name) return try { val result = runtime.gatewayClient.submitPrompt(binding.runtimeSessionId, promptToSend) + + // ONLY update binding sync status AFTER successful acceptance of prompt.submit! + sessionDao.updateBindingSync( + sessionId = sessionId.value, + hostId = targetHostId.value, + syncedThroughMessageId = syncResult.latestSyncedMessageId, + syncedAt = System.currentTimeMillis(), + state = BindingState.RUNNING.name + ) + result.turnId ?: userMessage.id } catch (e: Exception) { - setExecuting(sessionId, false) + setHostExecuting(sessionId, targetHostId, false) sessionDao.updateBindingState(sessionId.value, targetHostId.value, BindingState.ERROR.name) throw e } } - suspend fun interruptSession(sessionId: UnifiedSessionId) { - val details = sessionDao.getSessionWithDetails(sessionId.value) ?: return - for (binding in details.bindings) { - val runtime = connectionManager.getRuntime(HermesHostId(binding.hostId)) - if (runtime != null && binding.runtimeSessionId.isNotEmpty()) { - try { - runtime.gatewayClient.interruptSession(RuntimeSessionId(binding.runtimeSessionId)) - } catch (_: Exception) { - } + suspend fun interruptHost(sessionId: UnifiedSessionId, hostId: HermesHostId): Boolean { + val runtime = connectionManager.getRuntime(hostId) ?: return false + val details = sessionDao.getSessionWithDetails(sessionId.value) ?: return false + val binding = details.bindings.find { it.hostId == hostId.value } ?: return false + + val success = try { + if (binding.runtimeSessionId.isNotEmpty()) { + runtime.gatewayClient.interruptSession(RuntimeSessionId(binding.runtimeSessionId)) + } else { + false } + } catch (_: Exception) { + false } - setExecuting(sessionId, false) + setHostExecuting(sessionId, hostId, false) + sessionDao.updateBindingState(sessionId.value, hostId.value, BindingState.READY.name) + return success + } + + suspend fun interruptSession(sessionId: UnifiedSessionId, targetHostId: HermesHostId? = null) { + val details = sessionDao.getSessionWithDetails(sessionId.value) ?: return + val hostToInterrupt = targetHostId ?: HermesHostId(details.session.activeHostId) + interruptHost(sessionId, hostToInterrupt) } suspend fun respondApproval( hostId: HermesHostId, + runtimeSessionId: RuntimeSessionId, requestId: String, choice: String, all: Boolean = false ): Boolean { val runtime = connectionManager.getRuntime(hostId) ?: return false - val approval = _activeApprovals.value.find { it.hostId == hostId && it.approval.requestId == requestId } - val sessionKey = "" // Gateway client handles request_id val success = try { - runtime.gatewayClient.respondApproval(sessionKey, requestId, choice, all) + runtime.gatewayClient.respondApproval(runtimeSessionId.value, requestId, choice, all) } catch (_: Exception) { - true + false } if (success) { _activeApprovals.value = _activeApprovals.value.filterNot { - it.hostId == hostId && it.approval.requestId == requestId + it.hostId == hostId && it.runtimeSessionId == runtimeSessionId && it.approval.requestId == requestId } } return success @@ -265,7 +342,11 @@ class UnifiedSessionRepository( questionId: String? = null ): Boolean { val runtime = connectionManager.getRuntime(hostId) ?: return false - val success = runtime.gatewayClient.respondClarify(requestId, answer, questionId) + val success = try { + runtime.gatewayClient.respondClarify(requestId, answer, questionId) + } catch (_: Exception) { + false + } if (success) { if (_activeClarify.value?.hostId == hostId && _activeClarify.value?.request?.requestId == requestId) { _activeClarify.value = null @@ -280,7 +361,11 @@ class UnifiedSessionRepository( password: String ): Boolean { val runtime = connectionManager.getRuntime(hostId) ?: return false - val success = runtime.gatewayClient.respondSudo(requestId, password) + val success = try { + runtime.gatewayClient.respondSudo(requestId, password) + } catch (_: Exception) { + false + } if (success) { if (_activeClarify.value?.hostId == hostId && _activeClarify.value?.request?.requestId == requestId) { _activeClarify.value = null @@ -295,7 +380,11 @@ class UnifiedSessionRepository( secret: String ): Boolean { val runtime = connectionManager.getRuntime(hostId) ?: return false - val success = runtime.gatewayClient.respondSecret(requestId, secret) + val success = try { + runtime.gatewayClient.respondSecret(requestId, secret) + } catch (_: Exception) { + false + } if (success) { if (_activeClarify.value?.hostId == hostId && _activeClarify.value?.request?.requestId == requestId) { _activeClarify.value = null @@ -343,32 +432,61 @@ class UnifiedSessionRepository( } } - private fun setExecuting(sessionId: UnifiedSessionId, executing: Boolean) { - val flow = sessionExecutingState.computeIfAbsent(sessionId) { + private fun setHostExecuting(sessionId: UnifiedSessionId, hostId: HermesHostId, executing: Boolean) { + hostExecutingState.computeIfAbsent(Pair(sessionId, hostId)) { MutableStateFlow(false) - } - flow.value = executing + }.value = executing + + // Derive aggregate executing state for this session + val isAnyHostExecuting = hostExecutingState.entries + .filter { it.key.first == sessionId } + .any { it.value.value } + sessionExecutingState.computeIfAbsent(sessionId) { + MutableStateFlow(false) + }.value = isAnyHostExecuting } - private fun findSessionForHost(hostId: HermesHostId): UnifiedSessionId? { - for ((sessionId, flow) in sessionExecutingState) { - if (flow.value) { - return sessionId + private fun findSessionForEvent( + hostId: HermesHostId, + sessionIdFromEvent: String?, + messageId: String? = null, + toolId: String? = null + ): UnifiedSessionId? { + // 1. Exact match via runtimeSessionId in runtimeToSessionMap + if (!sessionIdFromEvent.isNullOrEmpty()) { + val mapped = runtimeToSessionMap[sessionIdFromEvent] + if (mapped != null && mapped.second == hostId) { + return mapped.first + } + for (session in sessions.value) { + val b = session.bindings[hostId] + if (b != null && b.runtimeSessionId.value == sessionIdFromEvent) { + runtimeToSessionMap[sessionIdFromEvent] = Pair(session.id, hostId) + return session.id + } } } - // Fall back to active session, open session cache, or first known session - return sessions.value.find { it.activeHostId == hostId }?.id - ?: sessionMessagesState.keys.firstOrNull() - ?: sessions.value.firstOrNull()?.id - } - private fun findSessionForMessage(messageId: String, hostId: HermesHostId): UnifiedSessionId? { - for ((sessionId, flow) in sessionMessagesState) { - if (flow.value.any { it.id == messageId }) { - return sessionId + // 2. Exact match via messageId in active session messages + if (!messageId.isNullOrEmpty()) { + for ((sessionId, flow) in sessionMessagesState) { + if (flow.value.any { it.id == messageId && (it.hostId == hostId || it.hostId == null) }) { + return sessionId + } } } - return findSessionForHost(hostId) + + // 3. Exact match via toolId in active session messages + if (!toolId.isNullOrEmpty()) { + for ((sessionId, flow) in sessionMessagesState) { + if (flow.value.any { it.tools.any { t -> t.id == toolId } }) { + return sessionId + } + } + } + + // Strict: NO fallback to "any executing session" or "first session in list" + return null } private fun handleHostGatewayEvent(hostEvent: HostGatewayEvent) { @@ -378,8 +496,8 @@ class UnifiedSessionRepository( when (event) { is GatewayEvent.MessageStartEvent -> { - val sessionId = findSessionForMessage(event.messageId, hostId) ?: return - setExecuting(sessionId, true) + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return + setHostExecuting(sessionId, hostId, true) val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } val existing = flow.value.find { it.id == event.messageId } if (existing == null) { @@ -397,8 +515,8 @@ class UnifiedSessionRepository( } is GatewayEvent.MessageDeltaEvent -> { - val sessionId = findSessionForMessage(event.messageId, hostId) ?: return - setExecuting(sessionId, true) + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return + setHostExecuting(sessionId, hostId, true) val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } val idx = flow.value.indexOfFirst { it.id == event.messageId } if (idx >= 0) { @@ -419,15 +537,18 @@ class UnifiedSessionRepository( } is GatewayEvent.MessageInterimEvent -> { - val sessionId = findSessionForMessage(event.messageId, hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return updateMessageInSession(sessionId, event.messageId) { it.copy(content = event.content, isStreaming = true) } } is GatewayEvent.MessageCompleteEvent -> { - val sessionId = findSessionForMessage(event.messageId, hostId) ?: return - setExecuting(sessionId, false) + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return + setHostExecuting(sessionId, hostId, false) + scope.launch { + sessionDao.updateBindingState(sessionId.value, hostId.value, BindingState.READY.name) + } val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } val idx = flow.value.indexOfFirst { it.id == event.messageId } if (idx >= 0) { @@ -451,56 +572,56 @@ class UnifiedSessionRepository( } is GatewayEvent.ThinkingDeltaEvent -> { - val sessionId = findSessionForHost(hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } - val lastAssistant = flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId } - if (lastAssistant != null) { - updateMessageInSession(sessionId, lastAssistant.id) { + val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId } + if (targetAssistant != null) { + updateMessageInSession(sessionId, targetAssistant.id) { it.copy(thinking = (it.thinking ?: "") + event.delta) } } } is GatewayEvent.ReasoningDeltaEvent -> { - val sessionId = findSessionForHost(hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } - val lastAssistant = flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId } - if (lastAssistant != null) { - updateMessageInSession(sessionId, lastAssistant.id) { + val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId } + if (targetAssistant != null) { + updateMessageInSession(sessionId, targetAssistant.id) { it.copy(thinking = (it.thinking ?: "") + event.delta) } } } is GatewayEvent.ReasoningAvailableEvent -> { - val sessionId = findSessionForHost(hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, messageId = event.messageId) ?: return val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } - val lastAssistant = flow.value.lastOrNull { it.role == MessageRole.ASSISTANT && it.hostId == hostId } - if (lastAssistant != null) { - updateMessageInSession(sessionId, lastAssistant.id) { + val targetAssistant = flow.value.lastOrNull { (it.id == event.messageId || it.role == MessageRole.ASSISTANT) && it.hostId == hostId } + if (targetAssistant != null) { + updateMessageInSession(sessionId, targetAssistant.id) { it.copy(thinking = event.reasoning) } } } is GatewayEvent.ToolStartEvent -> { - val sessionId = findSessionForHost(hostId) ?: 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 -> { - val sessionId = findSessionForHost(hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, toolId = event.toolId) ?: return updateToolInSessionMessage(sessionId, event.toolId) { it.copy(progress = event.progress) } } is GatewayEvent.ToolGeneratingEvent -> { - val sessionId = findSessionForHost(hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, toolId = event.toolId) ?: return updateToolInSessionMessage(sessionId, event.toolId) { it.copy(status = "generating") } } is GatewayEvent.ToolCompleteEvent -> { - val sessionId = findSessionForHost(hostId) ?: return + val sessionId = findSessionForEvent(hostId, event.sessionId, toolId = event.toolId) ?: return updateToolInSessionMessage(sessionId, event.toolId) { it.copy( status = if (event.isError) "failed" else "completed", @@ -511,6 +632,8 @@ class UnifiedSessionRepository( } is GatewayEvent.ApprovalRequestEvent -> { + val runtimeSessionIdVal = event.sessionKey ?: event.sessionId ?: "" + val runtimeSessionId = RuntimeSessionId(runtimeSessionIdVal) val approval = HermesApproval( requestId = event.requestId, command = event.command, @@ -520,44 +643,66 @@ class UnifiedSessionRepository( val attributed = HostAttributedApproval( hostId = hostId, hostDisplayName = hostName, + runtimeSessionId = runtimeSessionId, approval = approval ) _activeApprovals.value = _activeApprovals.value.filterNot { - it.hostId == hostId && it.approval.requestId == event.requestId + it.hostId == hostId && it.runtimeSessionId == runtimeSessionId && it.approval.requestId == event.requestId } + attributed } is GatewayEvent.ClarifyRequestEvent -> { + val runtimeSessionIdVal = event.sessionId val req = HermesClarifyRequest( requestId = event.requestId, questionId = event.questionId, question = event.question, promptType = ClarifyType.CLARIFY ) - _activeClarify.value = HostAttributedClarify(hostId, hostName, req) + _activeClarify.value = HostAttributedClarify( + hostId = hostId, + hostDisplayName = hostName, + runtimeSessionId = runtimeSessionIdVal?.let { RuntimeSessionId(it) }, + request = req + ) } is GatewayEvent.SudoRequestEvent -> { + val runtimeSessionIdVal = event.sessionId val req = HermesClarifyRequest( requestId = event.requestId, question = event.question, promptType = ClarifyType.SUDO ) - _activeClarify.value = HostAttributedClarify(hostId, hostName, req) + _activeClarify.value = HostAttributedClarify( + hostId = hostId, + hostDisplayName = hostName, + runtimeSessionId = runtimeSessionIdVal?.let { RuntimeSessionId(it) }, + request = req + ) } is GatewayEvent.SecretRequestEvent -> { + val runtimeSessionIdVal = event.sessionId val req = HermesClarifyRequest( requestId = event.requestId, question = event.question, promptType = ClarifyType.SECRET ) - _activeClarify.value = HostAttributedClarify(hostId, hostName, req) + _activeClarify.value = HostAttributedClarify( + hostId = hostId, + hostDisplayName = hostName, + runtimeSessionId = runtimeSessionIdVal?.let { RuntimeSessionId(it) }, + request = req + ) } is GatewayEvent.ErrorEvent -> { - val sessionId = findSessionForHost(hostId) ?: return - setExecuting(sessionId, false) + val sessionId = findSessionForEvent(hostId, event.sessionId) ?: return + setHostExecuting(sessionId, hostId, false) + scope.launch { + sessionDao.updateBindingState(sessionId.value, hostId.value, BindingState.ERROR.name) + } } else -> {} @@ -586,20 +731,14 @@ class UnifiedSessionRepository( } private fun updateToolInSessionMessage( - sessionId: UnifiedSessionId?, + sessionId: UnifiedSessionId, toolId: String, transform: (ToolActivity) -> ToolActivity ) { - val targetSessionId = sessionId?.takeIf { sId -> - sessionMessagesState[sId]?.value?.any { msg -> msg.tools.any { it.id == toolId } } == true - } ?: sessionMessagesState.entries.firstOrNull { (_, flow) -> - flow.value.any { msg -> msg.tools.any { it.id == toolId } } - }?.key ?: sessionId ?: return - - val flow = sessionMessagesState.computeIfAbsent(targetSessionId) { MutableStateFlow(emptyList()) } + val flow = sessionMessagesState.computeIfAbsent(sessionId) { MutableStateFlow(emptyList()) } val targetMsg = flow.value.lastOrNull { msg -> msg.tools.any { it.id == toolId } } if (targetMsg != null) { - updateMessageInSession(targetSessionId, targetMsg.id) { msg -> + updateMessageInSession(sessionId, targetMsg.id) { msg -> val updatedTools = msg.tools.map { if (it.id == toolId) transform(it) else it } msg.copy(tools = updatedTools) } 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 95fbc4a..0ddb0cf 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 @@ -29,9 +29,9 @@ import kotlin.random.Random class HermesHostRuntime( initialHost: HermesHost, val restClient: HermesRestClient = HermesRestClient(), - val gatewayClient: JsonRpcGatewayClient = JsonRpcGatewayClient(), val tokenVault: TokenVault, - val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default), + val gatewayClient: JsonRpcGatewayClient = JsonRpcGatewayClient(scope = scope) ) { private val _host = MutableStateFlow(initialHost) val host: StateFlow = _host.asStateFlow() diff --git a/app/src/main/java/app/hermes/mobile/core/sync/UnifiedContextBuilder.kt b/app/src/main/java/app/hermes/mobile/core/sync/UnifiedContextBuilder.kt index 565f1b2..2f88640 100644 --- a/app/src/main/java/app/hermes/mobile/core/sync/UnifiedContextBuilder.kt +++ b/app/src/main/java/app/hermes/mobile/core/sync/UnifiedContextBuilder.kt @@ -64,7 +64,7 @@ object UnifiedContextBuilder { val sb = StringBuilder() sb.appendLine("[Unified Hermes Session Context Transfer]") sb.appendLine("You are continuing a unified conversation that previously ran across Hermes host instances.") - sb.appendLine("Target Host: ${targetHost.displayName} (${targetHost.baseUrl})") + sb.appendLine("Target Host: ${targetHost.displayName}") sb.appendLine("Session Title: ${session.title}") sb.appendLine("--- Prior Conversation Turns ---") diff --git a/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt b/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt index b80f617..5185b11 100644 --- a/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt +++ b/app/src/main/java/app/hermes/mobile/feature/chat/ChatScreen.kt @@ -17,6 +17,7 @@ import androidx.compose.foundation.layout.padding import androidx.compose.foundation.layout.size import androidx.compose.foundation.layout.width import androidx.compose.foundation.lazy.LazyColumn +import androidx.compose.foundation.lazy.LazyRow import androidx.compose.foundation.lazy.items import androidx.compose.foundation.lazy.rememberLazyListState import androidx.compose.foundation.shape.CircleShape @@ -97,99 +98,126 @@ fun ChatScreen( Scaffold( topBar = { - TopAppBar( - title = { - Column { - Text( - text = currentSession?.title?.ifEmpty { "Unified Chat" } ?: "Unified Chat", - style = MaterialTheme.typography.titleMedium, - fontWeight = FontWeight.Bold, - maxLines = 1, - overflow = TextOverflow.Ellipsis - ) - // Active host selector chip - Box { - Row( - verticalAlignment = Alignment.CenterVertically, - modifier = Modifier - .clip(RoundedCornerShape(6.dp)) - .background(MaterialTheme.colorScheme.surfaceVariant) - .clickable { viewModel.setHostDropdownExpanded(true) } - .padding(horizontal = 6.dp, vertical = 2.dp) - ) { - val isOnline = activeHost?.lastKnownStatus == HostStatus.ONLINE - Box( + Column { + TopAppBar( + title = { + Column { + Text( + text = currentSession?.title?.ifEmpty { "Unified Chat" } ?: "Unified Chat", + style = MaterialTheme.typography.titleMedium, + fontWeight = FontWeight.Bold, + maxLines = 1, + overflow = TextOverflow.Ellipsis + ) + // Active host selector chip + Box { + Row( + verticalAlignment = Alignment.CenterVertically, modifier = Modifier - .size(6.dp) - .clip(CircleShape) - .background(if (isOnline) Color(0xFF10B981) else Color(0xFF94A3B8)) - ) - Spacer(modifier = Modifier.width(4.dp)) - Text( - text = activeHost?.displayName ?: "Select Host", - style = MaterialTheme.typography.labelSmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - fontWeight = FontWeight.SemiBold - ) - Spacer(modifier = Modifier.width(2.dp)) - Icon( - Icons.Default.ExpandMore, - contentDescription = "Switch Host", - modifier = Modifier.size(14.dp), - tint = MaterialTheme.colorScheme.onSurfaceVariant - ) - } - - DropdownMenu( - expanded = uiState.activeHostDropdownExpanded, - onDismissRequest = { viewModel.setHostDropdownExpanded(false) } + .clip(RoundedCornerShape(6.dp)) + .background(MaterialTheme.colorScheme.surfaceVariant) + .clickable { viewModel.setHostDropdownExpanded(true) } + .padding(horizontal = 6.dp, vertical = 2.dp) ) { - hosts.forEach { host -> - DropdownMenuItem( - text = { - Row(verticalAlignment = Alignment.CenterVertically) { - val online = host.lastKnownStatus == HostStatus.ONLINE - Box( - modifier = Modifier - .size(8.dp) - .clip(CircleShape) - .background(if (online) Color(0xFF10B981) else Color(0xFF94A3B8)) - ) - Spacer(modifier = Modifier.width(8.dp)) - Text( - text = host.displayName, - fontWeight = if (host.id == currentSession?.activeHostId) FontWeight.Bold else FontWeight.Normal - ) - } - }, - onClick = { viewModel.switchActiveHost(host.id) } + val isOnline = activeHost?.lastKnownStatus == HostStatus.ONLINE + Box( + modifier = Modifier + .size(6.dp) + .clip(CircleShape) + .background(if (isOnline) Color(0xFF10B981) else Color(0xFF94A3B8)) ) + Spacer(modifier = Modifier.width(4.dp)) + Text( + text = activeHost?.displayName ?: "Select Host", + style = MaterialTheme.typography.labelSmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + fontWeight = FontWeight.SemiBold + ) + Spacer(modifier = Modifier.width(2.dp)) + Icon( + Icons.Default.ExpandMore, + contentDescription = "Switch Host", + modifier = Modifier.size(14.dp), + tint = MaterialTheme.colorScheme.onSurfaceVariant + ) + } + + DropdownMenu( + expanded = uiState.activeHostDropdownExpanded, + onDismissRequest = { viewModel.setHostDropdownExpanded(false) } + ) { + hosts.forEach { host -> + DropdownMenuItem( + text = { + Row(verticalAlignment = Alignment.CenterVertically) { + val online = host.lastKnownStatus == HostStatus.ONLINE + Box( + modifier = Modifier + .size(8.dp) + .clip(CircleShape) + .background(if (online) Color(0xFF10B981) else Color(0xFF94A3B8)) + ) + Spacer(modifier = Modifier.width(8.dp)) + Text( + text = host.displayName, + fontWeight = if (host.id == currentSession?.activeHostId) FontWeight.Bold else FontWeight.Normal + ) + } + }, + onClick = { viewModel.switchActiveHost(host.id) } + ) + } } } } + }, + navigationIcon = { + IconButton(onClick = onNavigateBack) { + Icon(Icons.AutoMirrored.Filled.ArrowBack, contentDescription = "Back") + } + }, + actions = { + if (isExecuting) { + Button( + onClick = { viewModel.interruptSession() }, + colors = ButtonDefaults.buttonColors(containerColor = Color(0xFFEF4444)), + shape = RoundedCornerShape(8.dp), + contentPadding = PaddingValues(horizontal = 10.dp, vertical = 4.dp), + modifier = Modifier.padding(end = 8.dp) + ) { + Icon(Icons.Default.Stop, contentDescription = null, modifier = Modifier.size(16.dp)) + Spacer(modifier = Modifier.width(4.dp)) + Text("Stop", fontSize = 12.sp, fontWeight = FontWeight.Bold) + } + } } - }, - navigationIcon = { - IconButton(onClick = onNavigateBack) { - Icon(Icons.AutoMirrored.Filled.ArrowBack, contentDescription = "Back") - } - }, - actions = { - if (isExecuting) { - Button( - onClick = { viewModel.interruptSession() }, - colors = ButtonDefaults.buttonColors(containerColor = Color(0xFFEF4444)), - shape = RoundedCornerShape(8.dp), - contentPadding = PaddingValues(horizontal = 10.dp, vertical = 4.dp), - modifier = Modifier.padding(end = 8.dp) - ) { - Icon(Icons.Default.Stop, contentDescription = null, modifier = Modifier.size(16.dp)) - Spacer(modifier = Modifier.width(4.dp)) - Text("Stop", fontSize = 12.sp, fontWeight = FontWeight.Bold) + ) + + // Multi-host Status Strip showing simultaneous statuses (e.g. PC1 Running, Linux Active, PC3 Offline) + if (hosts.isNotEmpty()) { + LazyRow( + modifier = Modifier + .fillMaxWidth() + .background(MaterialTheme.colorScheme.surfaceVariant.copy(alpha = 0.5f)) + .padding(horizontal = 16.dp, vertical = 6.dp), + horizontalArrangement = Arrangement.spacedBy(8.dp), + verticalAlignment = Alignment.CenterVertically + ) { + items(hosts, key = { it.id.value }) { host -> + val isRunning by viewModel.getHostExecuting(host.id).collectAsState() + val isActive = host.id == currentSession?.activeHostId + + HostStatusChip( + host = host, + isActive = isActive, + isRunning = isRunning, + onClick = { viewModel.switchActiveHost(host.id) }, + onStop = { viewModel.interruptHost(host.id) } + ) } } } - ) + } }, bottomBar = { ChatInputBar( @@ -219,11 +247,11 @@ fun ChatScreen( } } - items(approvals, key = { it.hostId.value + it.approval.requestId }) { approval -> + items(approvals, key = { it.hostId.value + it.runtimeSessionId.value + it.approval.requestId }) { approval -> ApprovalCard( attributedApproval = approval, onRespond = { choice, all -> - viewModel.respondApproval(approval.hostId, approval.approval.requestId, choice, all) + viewModel.respondApproval(approval.hostId, approval.runtimeSessionId, approval.approval.requestId, choice, all) } ) } @@ -261,6 +289,92 @@ fun ChatScreen( } } +@Composable +fun HostStatusChip( + host: HermesHost, + isActive: Boolean, + isRunning: Boolean, + onClick: () -> Unit, + onStop: () -> Unit +) { + val statusText = when { + isRunning -> "Running" + isActive -> "Active" + host.lastKnownStatus == HostStatus.ONLINE -> "Online" + host.lastKnownStatus == HostStatus.CONNECTING -> "Connecting" + host.lastKnownStatus == HostStatus.AUTH_EXPIRED -> "Auth Expired" + else -> "Offline" + } + + val chipBg = when { + isRunning -> Color(0xFFF59E0B).copy(alpha = 0.15f) + isActive -> Color(0xFF38BDF8).copy(alpha = 0.15f) + else -> MaterialTheme.colorScheme.surface + } + + val chipBorder = when { + isRunning -> Color(0xFFF59E0B) + isActive -> Color(0xFF38BDF8) + else -> Color(0xFF475569) + } + + Row( + verticalAlignment = Alignment.CenterVertically, + modifier = Modifier + .clip(RoundedCornerShape(8.dp)) + .background(chipBg) + .clickable { onClick() } + .padding(horizontal = 8.dp, vertical = 4.dp) + ) { + if (isRunning) { + CircularProgressIndicator( + modifier = Modifier.size(10.dp), + strokeWidth = 1.5.dp, + color = Color(0xFFF59E0B) + ) + } else { + val dotColor = when (host.lastKnownStatus) { + HostStatus.ONLINE -> Color(0xFF10B981) + HostStatus.CONNECTING -> Color(0xFFF59E0B) + HostStatus.AUTH_EXPIRED -> Color(0xFFEF4444) + else -> Color(0xFF94A3B8) + } + Box( + modifier = Modifier + .size(6.dp) + .clip(CircleShape) + .background(dotColor) + ) + } + Spacer(modifier = Modifier.width(6.dp)) + Text( + text = "${host.displayName}: $statusText", + fontSize = 11.sp, + fontWeight = if (isActive || isRunning) FontWeight.Bold else FontWeight.Normal, + color = if (isRunning) Color(0xFFF59E0B) else if (isActive) Color(0xFF38BDF8) else MaterialTheme.colorScheme.onSurface + ) + + if (isRunning) { + Spacer(modifier = Modifier.width(4.dp)) + Box( + modifier = Modifier + .size(14.dp) + .clip(CircleShape) + .background(Color(0xFFEF4444)) + .clickable { onStop() }, + contentAlignment = Alignment.Center + ) { + Icon( + Icons.Default.Stop, + contentDescription = "Stop Host", + tint = Color.White, + modifier = Modifier.size(10.dp) + ) + } + } + } +} + @Composable fun TransferSeparator(message: UnifiedMessage) { Row( diff --git a/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt b/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt index 50f2134..8d36aa1 100644 --- a/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt +++ b/app/src/main/java/app/hermes/mobile/feature/chat/ChatViewModel.kt @@ -44,6 +44,10 @@ class ChatViewModel( } } + fun getHostExecuting(hostId: HermesHostId): StateFlow { + return sessionRepo.getHostExecuting(sessionId, hostId) + } + fun updateInputText(text: String) { _uiState.value = _uiState.value.copy(inputText = text) } @@ -82,19 +86,30 @@ class ChatViewModel( } } - fun interruptSession() { + fun interruptSession(hostId: HermesHostId? = null) { viewModelScope.launch { - sessionRepo.interruptSession(sessionId) + sessionRepo.interruptSession(sessionId, hostId) } } - fun respondApproval(hostId: HermesHostId, requestId: String, choice: String, all: Boolean = false) { + fun interruptHost(hostId: HermesHostId) { viewModelScope.launch { - try { - sessionRepo.respondApproval(hostId, requestId, choice, all) - } catch (e: Exception) { + sessionRepo.interruptHost(sessionId, hostId) + } + } + + fun respondApproval( + hostId: HermesHostId, + runtimeSessionId: RuntimeSessionId, + requestId: String, + choice: String, + all: Boolean = false + ) { + viewModelScope.launch { + val success = sessionRepo.respondApproval(hostId, runtimeSessionId, requestId, choice, all) + if (!success) { _uiState.value = _uiState.value.copy( - error = e.localizedMessage ?: "Failed to respond to approval" + error = "Failed to submit approval response" ) } } @@ -102,17 +117,16 @@ class ChatViewModel( fun respondClarify(attributed: HostAttributedClarify, answer: String) { viewModelScope.launch { - try { - val hostId = attributed.hostId - val req = attributed.request - when (req.promptType) { - ClarifyType.CLARIFY -> sessionRepo.respondClarify(hostId, req.requestId, answer, req.questionId) - ClarifyType.SUDO -> sessionRepo.respondSudo(hostId, req.requestId, answer) - ClarifyType.SECRET -> sessionRepo.respondSecret(hostId, req.requestId, answer) - } - } catch (e: Exception) { + val hostId = attributed.hostId + val req = attributed.request + val success = when (req.promptType) { + ClarifyType.CLARIFY -> sessionRepo.respondClarify(hostId, req.requestId, answer, req.questionId) + ClarifyType.SUDO -> sessionRepo.respondSudo(hostId, req.requestId, answer) + ClarifyType.SECRET -> sessionRepo.respondSecret(hostId, req.requestId, answer) + } + if (!success) { _uiState.value = _uiState.value.copy( - error = e.localizedMessage ?: "Failed to respond to clarification" + error = "Failed to submit clarification response" ) } } diff --git a/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt index 2d4513e..fed33a1 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/ApprovalRoutingTest.kt @@ -1,7 +1,7 @@ package app.hermes.mobile.core.repository import app.hermes.mobile.core.model.* -import app.hermes.mobile.core.network.HermesRestClient +import app.hermes.mobile.core.network.ConnectionState import app.hermes.mobile.core.network.JsonRpcGatewayClient import app.hermes.mobile.core.runtime.HermesConnectionManager import app.hermes.mobile.core.runtime.HermesHostRuntime @@ -11,14 +11,23 @@ import app.hermes.mobile.core.storage.FakeUnifiedSessionDao import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.runBlocking 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 okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import okhttp3.mockwebserver.Dispatcher +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.RecordedRequest import org.junit.After import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull import org.junit.Assert.assertTrue import org.junit.Before @@ -33,6 +42,7 @@ class ApprovalRoutingTest { private lateinit var tokenVault: InMemoryTokenVault private lateinit var connectionManager: HermesConnectionManager private lateinit var sessionRepo: UnifiedSessionRepository + private lateinit var mockServer: MockWebServer private val host1Id = HermesHostId("server-prod") private val host2Id = HermesHostId("server-dev") @@ -43,6 +53,8 @@ class ApprovalRoutingTest { hostDao = FakeHostDao() sessionDao = FakeUnifiedSessionDao() tokenVault = InMemoryTokenVault() + mockServer = MockWebServer() + mockServer.start() connectionManager = HermesConnectionManager( hostDao = hostDao, @@ -60,66 +72,149 @@ class ApprovalRoutingTest { @After fun tearDown() { Dispatchers.resetMain() + try { + mockServer.shutdown() + } catch (_: Exception) { + } } @Test - fun testApprovalAttributionAndRemoval() = runTest(testDispatcher) { + fun testApprovalFailureRemainsVisibleAndNotResolved() = runTest(testDispatcher) { val host1 = HermesHost(id = host1Id, displayName = "Prod Server", baseUrl = "http://prod:9119") - val host2 = HermesHost(id = host2Id, displayName = "Dev Server", baseUrl = "http://dev:9119") - connectionManager.addHost(host1) - connectionManager.addHost(host2) testScheduler.advanceUntilIdle() val runtime1 = connectionManager.getRuntime(host1Id) - val runtime2 = connectionManager.getRuntime(host2Id) - assertNotNull(runtime1) - assertNotNull(runtime2) - // Simulate approval request from Prod Server + // Simulate incoming approval request with specific runtime session ID val prodEventJson = buildJsonObject { put("method", "event") put("params", buildJsonObject { put("event", "approval.request") put("request_id", "req_prod_1") + put("session_key", "runtime_session_prod_99") put("command", "systemctl restart nginx") put("description", "Restart web server") }) } runtime1?.gatewayClient?.handleIncomingMessage(prodEventJson.toString()) - - // Simulate approval request from Dev Server - val devEventJson = buildJsonObject { - put("method", "event") - put("params", buildJsonObject { - put("event", "approval.request") - put("request_id", "req_dev_1") - put("command", "docker compose down") - put("description", "Stop containers") - }) - } - runtime2?.gatewayClient?.handleIncomingMessage(devEventJson.toString()) testScheduler.advanceUntilIdle() val approvals = sessionRepo.activeApprovals.value - assertEquals(2, approvals.size) + assertEquals(1, approvals.size) + val approval = approvals.first() + assertEquals(RuntimeSessionId("runtime_session_prod_99"), approval.runtimeSessionId) + assertEquals("req_prod_1", approval.approval.requestId) - val prodApproval = approvals.find { it.hostId == host1Id } - val devApproval = approvals.find { it.hostId == host2Id } - - assertNotNull(prodApproval) - assertNotNull(devApproval) - assertEquals("Prod Server", prodApproval?.hostDisplayName) - assertEquals("Dev Server", devApproval?.hostDisplayName) - assertEquals("systemctl restart nginx", prodApproval?.approval?.command) - - // Responding to prod approval removes it while keeping dev approval - sessionRepo.respondApproval(host1Id, "req_prod_1", "once", false) + // Attempting to respond when WebSocket is disconnected MUST return false and KEEP approval + val result = sessionRepo.respondApproval( + hostId = host1Id, + runtimeSessionId = RuntimeSessionId("runtime_session_prod_99"), + requestId = "req_prod_1", + choice = "once", + all = false + ) testScheduler.advanceUntilIdle() - val remainingApprovals = sessionRepo.activeApprovals.value - assertEquals(1, remainingApprovals.size) - assertEquals(host2Id, remainingApprovals.first().hostId) + assertFalse("Expected false when RPC fails due to disconnected socket", result) + assertEquals("Approval must not be removed on failure", 1, sessionRepo.activeApprovals.value.size) + } + + @Test + fun testApprovalSuccessRemovesCardAndSendsCorrectRpc() = runBlocking(Dispatchers.Default) { + val receivedMessages = mutableListOf() + + mockServer.dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val path = request.path ?: "" + return when { + path == "/api/status" -> { + MockResponse().setResponseCode(200).setBody("""{"status":"ok","auth_required":false,"version":"1.0.0"}""") + } + path.startsWith("/api/ws") || path.startsWith("/ws") -> { + MockResponse().withWebSocketUpgrade(object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + webSocket.send("""{"event":"gateway.ready","data":{"version":"1.0.0"}}""") + } + + override fun onMessage(webSocket: WebSocket, text: String) { + receivedMessages.add(text) + if (text.contains("approval.respond")) { + webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"status":"ok"}}""") + } + } + }) + } + else -> MockResponse().setResponseCode(404) + } + } + } + + val wsUrl = mockServer.url("").toString().removeSuffix("/") + val testHostDao = FakeHostDao() + val testSessionDao = FakeUnifiedSessionDao() + val testTokenVault = InMemoryTokenVault() + + val testConnectionManager = HermesConnectionManager( + hostDao = testHostDao, + tokenVault = testTokenVault, + scope = CoroutineScope(Dispatchers.Default) + ) + val testRepo = UnifiedSessionRepository( + connectionManager = testConnectionManager, + sessionDao = testSessionDao, + scope = CoroutineScope(Dispatchers.Default) + ) + + val host1 = HermesHost(id = host1Id, displayName = "Prod Server", baseUrl = wsUrl, allowCleartext = true) + testConnectionManager.addHost(host1) + + val runtime1 = testConnectionManager.getRuntime(host1Id) + assertNotNull(runtime1) + + runtime1!!.connect() + runtime1.gatewayClient.awaitGatewayReady(5000) + + // Incoming approval + val prodEventJson = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "approval.request") + put("request_id", "req_prod_1") + put("session_id", "runtime_session_prod_99") + put("command", "systemctl restart nginx") + }) + } + runtime1.gatewayClient.handleIncomingMessage(prodEventJson.toString()) + + var waited = 0 + while (testRepo.activeApprovals.value.isEmpty() && waited < 50) { + kotlinx.coroutines.delay(50) + waited++ + } + + assertEquals(1, testRepo.activeApprovals.value.size) + + // Respond to approval + val result = testRepo.respondApproval( + hostId = host1Id, + runtimeSessionId = RuntimeSessionId("runtime_session_prod_99"), + requestId = "req_prod_1", + choice = "once", + all = false + ) + + assertTrue("Expected true on successful RPC response", result) + assertEquals("Approval should be removed after success", 0, testRepo.activeApprovals.value.size) + + // Verify sent RPC wire payload + assertTrue(receivedMessages.any { + it.contains("\"session_id\":\"runtime_session_prod_99\"") && + it.contains("\"request_id\":\"req_prod_1\"") && + it.contains("\"choice\":\"once\"") + }) + + runtime1.disconnect() } } diff --git a/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt new file mode 100644 index 0000000..d63526d --- /dev/null +++ b/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt @@ -0,0 +1,656 @@ +package app.hermes.mobile.core.repository + +import app.hermes.mobile.core.model.* +import app.hermes.mobile.core.network.ConnectionState +import app.hermes.mobile.core.network.JsonRpcGatewayClient +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.* +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.runBlocking +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 okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import okhttp3.mockwebserver.Dispatcher +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.RecordedRequest +import org.junit.After +import org.junit.Assert.* +import org.junit.Before +import org.junit.Test +import java.io.IOException + +@OptIn(ExperimentalCoroutinesApi::class) +class MultiHostConcurrencyExecutionTest { + + 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 sessionRepo: UnifiedSessionRepository + + private val host1Id = HermesHostId("host-windows") + private val host2Id = HermesHostId("host-linux") + + @Before + fun setUp() { + Dispatchers.setMain(testDispatcher) + hostDao = FakeHostDao() + sessionDao = FakeUnifiedSessionDao() + tokenVault = InMemoryTokenVault() + + connectionManager = HermesConnectionManager( + hostDao = hostDao, + tokenVault = tokenVault, + scope = CoroutineScope(testDispatcher) + ) + + sessionRepo = UnifiedSessionRepository( + connectionManager = connectionManager, + sessionDao = sessionDao, + scope = CoroutineScope(testDispatcher) + ) + } + + @After + fun tearDown() { + Dispatchers.resetMain() + } + + @Test + fun testTwoHermesSimultaneousStreamToTwoDifferentSessions() = runTest(testDispatcher) { + val hostA = HermesHost(id = host1Id, displayName = "Windows PC", baseUrl = "http://pc:9119") + val hostB = HermesHost(id = host2Id, displayName = "Linux Server", baseUrl = "http://linux:9119") + connectionManager.addHost(hostA) + connectionManager.addHost(hostB) + testScheduler.advanceUntilIdle() + + val session1 = sessionRepo.createUnifiedSession(title = "Windows Session", initialHostId = host1Id) + val session2 = sessionRepo.createUnifiedSession(title = "Linux Session", initialHostId = host2Id) + testScheduler.advanceUntilIdle() + + val runtimeA = connectionManager.getRuntime(host1Id) + val runtimeB = connectionManager.getRuntime(host2Id) + assertNotNull(runtimeA) + assertNotNull(runtimeB) + + // Register runtime IDs + sessionRepo.registerRuntimeBinding(session1.id, host1Id, RuntimeSessionId("rt_win_1")) + sessionRepo.registerRuntimeBinding(session2.id, host2Id, RuntimeSessionId("rt_lin_1")) + + // Register bindings in DB + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session1.id.value, + hostId = host1Id.value, + durableSessionId = "dur_win_1", + runtimeSessionId = "rt_win_1", + state = BindingState.RUNNING.name + ) + ) + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session2.id.value, + hostId = host2Id.value, + durableSessionId = "dur_lin_1", + runtimeSessionId = "rt_lin_1", + state = BindingState.RUNNING.name + ) + ) + testScheduler.advanceUntilIdle() + + // Stream from Host A into Session 1 + val eventA1 = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.start") + put("session_id", "rt_win_1") + put("message_id", "msg_a_1") + put("role", "assistant") + }) + } + val eventA2 = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.delta") + put("session_id", "rt_win_1") + put("message_id", "msg_a_1") + put("delta", "Windows output chunk") + }) + } + + // Stream from Host B into Session 2 concurrently + val eventB1 = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.start") + put("session_id", "rt_lin_1") + put("message_id", "msg_b_1") + put("role", "assistant") + }) + } + val eventB2 = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.delta") + put("session_id", "rt_lin_1") + put("message_id", "msg_b_1") + put("delta", "Linux output chunk") + }) + } + + runtimeA!!.gatewayClient.handleIncomingMessage(eventA1.toString()) + runtimeB!!.gatewayClient.handleIncomingMessage(eventB1.toString()) + testScheduler.advanceUntilIdle() + + runtimeA.gatewayClient.handleIncomingMessage(eventA2.toString()) + runtimeB.gatewayClient.handleIncomingMessage(eventB2.toString()) + testScheduler.advanceUntilIdle() + + val messages1 = sessionRepo.getSessionMessages(session1.id).value + val messages2 = sessionRepo.getSessionMessages(session2.id).value + + // Assert Session 1 only has Windows messages + assertTrue(messages1.any { it.id == "msg_a_1" }) + assertFalse(messages1.any { it.id == "msg_b_1" }) + assertEquals("Windows output chunk", messages1.find { it.id == "msg_a_1" }?.content) + assertEquals(host1Id, messages1.find { it.id == "msg_a_1" }?.hostId) + + // Assert Session 2 only has Linux messages + assertTrue(messages2.any { it.id == "msg_b_1" }) + assertFalse(messages2.any { it.id == "msg_a_1" }) + assertEquals("Linux output chunk", messages2.find { it.id == "msg_b_1" }?.content) + assertEquals(host2Id, messages2.find { it.id == "msg_b_1" }?.hostId) + } + + @Test + fun testTwoHermesSimultaneousStreamToOneUnifiedSession() = runTest(testDispatcher) { + val hostA = HermesHost(id = host1Id, displayName = "Windows PC", baseUrl = "http://pc:9119") + val hostB = HermesHost(id = host2Id, displayName = "Linux Server", baseUrl = "http://linux:9119") + connectionManager.addHost(hostA) + connectionManager.addHost(hostB) + testScheduler.advanceUntilIdle() + + val session = sessionRepo.createUnifiedSession(title = "Dual Host Project", initialHostId = host1Id) + testScheduler.advanceUntilIdle() + + val runtimeA = connectionManager.getRuntime(host1Id) + val runtimeB = connectionManager.getRuntime(host2Id) + + sessionRepo.registerRuntimeBinding(session.id, host1Id, RuntimeSessionId("rt_win_dual")) + sessionRepo.registerRuntimeBinding(session.id, host2Id, RuntimeSessionId("rt_lin_dual")) + + // Attach both hosts to this one unified session + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host1Id.value, + durableSessionId = "dur_win_dual", + runtimeSessionId = "rt_win_dual", + state = BindingState.RUNNING.name + ) + ) + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host2Id.value, + durableSessionId = "dur_lin_dual", + runtimeSessionId = "rt_lin_dual", + state = BindingState.RUNNING.name + ) + ) + testScheduler.advanceUntilIdle() + + // Host A streams message + runtimeA!!.gatewayClient.handleIncomingMessage( + buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.start") + put("session_id", "rt_win_dual") + put("message_id", "msg_win") + put("role", "assistant") + }) + }.toString() + ) + testScheduler.advanceUntilIdle() + + runtimeA.gatewayClient.handleIncomingMessage( + buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.delta") + put("session_id", "rt_win_dual") + put("message_id", "msg_win") + put("delta", "Windows result") + }) + }.toString() + ) + testScheduler.advanceUntilIdle() + + // Host B concurrently streams tool + runtimeB!!.gatewayClient.handleIncomingMessage( + buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "tool.start") + put("session_id", "rt_lin_dual") + put("tool_id", "tool_lin") + put("name", "bash_exec") + }) + }.toString() + ) + testScheduler.advanceUntilIdle() + + runtimeB.gatewayClient.handleIncomingMessage( + buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "tool.complete") + put("session_id", "rt_lin_dual") + put("tool_id", "tool_lin") + put("result", "Linux command completed") + }) + }.toString() + ) + testScheduler.advanceUntilIdle() + + val messages = sessionRepo.getSessionMessages(session.id).value + val winMsg = messages.find { it.id == "msg_win" } + assertNotNull(winMsg) + assertEquals("Windows result", winMsg?.content) + assertEquals(host1Id, winMsg?.hostId) + + val linToolMsg = messages.find { it.tools.any { t -> t.id == "tool_lin" } } + assertNotNull(linToolMsg) + assertEquals(host2Id, linToolMsg?.hostId) + assertEquals("completed", linToolMsg?.tools?.find { it.id == "tool_lin" }?.status) + } + + @Test + fun testIdenticalMessageIdAndRequestIdOnTwoHostsDoNotConflict() = runTest(testDispatcher) { + val hostA = HermesHost(id = host1Id, displayName = "Windows PC", baseUrl = "http://pc:9119") + val hostB = HermesHost(id = host2Id, displayName = "Linux Server", baseUrl = "http://linux:9119") + connectionManager.addHost(hostA) + connectionManager.addHost(hostB) + testScheduler.advanceUntilIdle() + + val session = sessionRepo.createUnifiedSession(title = "Conflict Resistance", initialHostId = host1Id) + testScheduler.advanceUntilIdle() + + val runtimeA = connectionManager.getRuntime(host1Id) + val runtimeB = connectionManager.getRuntime(host2Id) + + // Both hosts have same requestId "req_shared_1" + val approvalEventA = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "approval.request") + put("session_id", "rt_win_shared") + put("request_id", "req_shared_1") + put("command", "powershell.exe -Command Get-Process") + }) + } + val approvalEventB = buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "approval.request") + put("session_id", "rt_lin_shared") + put("request_id", "req_shared_1") + put("command", "ps aux") + }) + } + + runtimeA!!.gatewayClient.handleIncomingMessage(approvalEventA.toString()) + runtimeB!!.gatewayClient.handleIncomingMessage(approvalEventB.toString()) + testScheduler.advanceUntilIdle() + + val approvals = sessionRepo.activeApprovals.value + assertEquals(2, approvals.size) + + val appA = approvals.find { it.hostId == host1Id && it.approval.requestId == "req_shared_1" } + val appB = approvals.find { it.hostId == host2Id && it.approval.requestId == "req_shared_1" } + + assertNotNull(appA) + assertNotNull(appB) + assertEquals("powershell.exe -Command Get-Process", appA?.approval?.command) + assertEquals("ps aux", appB?.approval?.command) + assertEquals(RuntimeSessionId("rt_win_shared"), appA?.runtimeSessionId) + assertEquals(RuntimeSessionId("rt_lin_shared"), appB?.runtimeSessionId) + } + + @Test + fun testAppRestartRestoresBindingViaDurableId() = runBlocking(Dispatchers.Default) { + val server = MockWebServer() + + server.dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val path = request.path ?: "" + return when { + path == "/api/status" -> { + MockResponse().setResponseCode(200).setBody("""{"status":"ok","auth_required":false,"version":"1.0.0"}""") + } + path.startsWith("/api/ws") || path.startsWith("/ws") -> { + MockResponse().withWebSocketUpgrade(object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + webSocket.send("""{"event":"gateway.ready","data":{"version":"1.0.0"}}""") + } + + override fun onMessage(webSocket: WebSocket, text: String) { + if (text.contains("session.resume")) { + webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"durable_persisted_99","session_id":"fresh_runtime_101"}}""") + } else if (text.contains("prompt.submit")) { + webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"turn_id":"t_101"}}""") + } + } + }) + } + else -> MockResponse().setResponseCode(404) + } + } + } + + server.start() + val wsUrl = server.url("").toString().removeSuffix("/") + val testHostDao = FakeHostDao() + val testSessionDao = FakeUnifiedSessionDao() + val testTokenVault = InMemoryTokenVault() + + testHostDao.insertOrUpdateHost(HostEntity(id = host1Id.value, displayName = "Prod Server", baseUrl = wsUrl, allowCleartext = true, lastKnownStatus = "ONLINE")) + + // Seed DB with existing session and binding from a previous app run + val sessionId = UnifiedSessionId("session_saved_1") + testSessionDao.insertSession( + UnifiedSessionEntity( + id = sessionId.value, + title = "Restored Session", + activeHostId = host1Id.value + ) + ) + testSessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = sessionId.value, + hostId = host1Id.value, + durableSessionId = "durable_persisted_99", + runtimeSessionId = "stale_dead_runtime_000", + state = BindingState.OFFLINE.name + ) + ) + + val freshConnectionManager = HermesConnectionManager( + hostDao = testHostDao, + tokenVault = testTokenVault, + scope = CoroutineScope(Dispatchers.Default) + ) + val freshRepo = UnifiedSessionRepository( + connectionManager = freshConnectionManager, + sessionDao = testSessionDao, + scope = CoroutineScope(Dispatchers.Default) + ) + + val runtime = freshConnectionManager.getRuntime(host1Id) + assertNotNull(runtime) + runtime!!.connect() + runtime.gatewayClient.awaitGatewayReady(5000) + + val turnId = freshRepo.sendPrompt(sessionId, "Hello after restart") + assertEquals("t_101", turnId) + + // Verify that binding was updated in DB with the fresh runtime ID + val updatedBinding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } + assertNotNull(updatedBinding) + assertEquals("durable_persisted_99", updatedBinding?.durableSessionId) + assertEquals("fresh_runtime_101", updatedBinding?.runtimeSessionId) + assertEquals(BindingState.RUNNING.name, updatedBinding?.state) + + runtime.disconnect() + server.shutdown() + } + + @Test + fun testHostReconnectMintsNewRuntimeId() = runBlocking(Dispatchers.Default) { + val server = MockWebServer() + var sessionCreated = false + + server.dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val path = request.path ?: "" + return when { + path == "/api/status" -> { + MockResponse().setResponseCode(200).setBody("""{"status":"ok","auth_required":false,"version":"1.0.0"}""") + } + path.startsWith("/api/ws") || path.startsWith("/ws") -> { + MockResponse().withWebSocketUpgrade(object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + webSocket.send("""{"event":"gateway.ready","data":{"version":"1.0.0"}}""") + } + + override fun onMessage(webSocket: WebSocket, text: String) { + if (text.contains("session.create")) { + sessionCreated = true + webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"dur_conn_1","session_id":"rt_initial"}}""") + } else if (text.contains("session.resume")) { + webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_conn_1","session_id":"rt_after_reconnect"}}""") + } + } + }) + } + else -> MockResponse().setResponseCode(404) + } + } + } + + server.start() + val wsUrl = server.url("").toString().removeSuffix("/") + val testHostDao = FakeHostDao() + val testSessionDao = FakeUnifiedSessionDao() + val testTokenVault = InMemoryTokenVault() + + testHostDao.insertOrUpdateHost(HostEntity(id = host1Id.value, displayName = "Server", baseUrl = wsUrl, allowCleartext = true, lastKnownStatus = "ONLINE")) + + val testConnectionManager = HermesConnectionManager( + hostDao = testHostDao, + tokenVault = testTokenVault, + scope = CoroutineScope(Dispatchers.Default) + ) + val testRepo = UnifiedSessionRepository( + connectionManager = testConnectionManager, + sessionDao = testSessionDao, + scope = CoroutineScope(Dispatchers.Default) + ) + + val session = testRepo.createUnifiedSession(title = "Reconnect Test", initialHostId = host1Id) + val runtime = testConnectionManager.getRuntime(host1Id)!! + + // Initial connect + runtime.connect() + runtime.gatewayClient.awaitGatewayReady(5000) + + val binding1 = testRepo.ensureAttachedRuntimeSession(session.id, host1Id, runtime) + assertEquals(RuntimeSessionId("rt_initial"), binding1.runtimeSessionId) + + // Disconnect host runtime + runtime.disconnect() + testSessionDao.updateBindingState(session.id.value, host1Id.value, BindingState.OFFLINE.name) + + // Reconnect host runtime + runtime.connect() + runtime.gatewayClient.awaitGatewayReady(5000) + + // Reattach after reconnect + val binding2 = testRepo.ensureAttachedRuntimeSession(session.id, host1Id, runtime) + assertEquals(RuntimeSessionId("rt_after_reconnect"), binding2.runtimeSessionId) + + runtime.disconnect() + server.shutdown() + } + + @Test + fun testFailedPromptSubmitDoesNotAdvanceContextSyncCursor() = runBlocking(Dispatchers.Default) { + val server = MockWebServer() + + server.dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val path = request.path ?: "" + return when { + path == "/api/status" -> { + MockResponse().setResponseCode(200).setBody("""{"status":"ok","auth_required":false,"version":"1.0.0"}""") + } + path.startsWith("/api/ws") || path.startsWith("/ws") -> { + MockResponse().withWebSocketUpgrade(object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: Response) { + webSocket.send("""{"event":"gateway.ready","data":{"version":"1.0.0"}}""") + } + + override fun onMessage(webSocket: WebSocket, text: String) { + if (text.contains("prompt.submit")) { + // Fail prompt submission with an RPC error + webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":-32000,"message":"Model overloaded"}}""") + } + } + }) + } + else -> MockResponse().setResponseCode(404) + } + } + } + + server.start() + val wsUrl = server.url("").toString().removeSuffix("/") + val testHostDao = FakeHostDao() + val testSessionDao = FakeUnifiedSessionDao() + val testTokenVault = InMemoryTokenVault() + + testHostDao.insertOrUpdateHost(HostEntity(id = host1Id.value, displayName = "Server", baseUrl = wsUrl, allowCleartext = true, lastKnownStatus = "ONLINE")) + + val testConnectionManager = HermesConnectionManager( + hostDao = testHostDao, + tokenVault = testTokenVault, + scope = CoroutineScope(Dispatchers.Default) + ) + val testRepo = UnifiedSessionRepository( + connectionManager = testConnectionManager, + sessionDao = testSessionDao, + scope = CoroutineScope(Dispatchers.Default) + ) + + val session = testRepo.createUnifiedSession(title = "Cursor Test", initialHostId = host1Id) + + // Pre-populate binding with syncedThroughMessageId = "msg_baseline" + testSessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host1Id.value, + durableSessionId = "dur_fail_test", + runtimeSessionId = "rt_fail_test", + syncedThroughMessageId = "msg_baseline", + state = BindingState.READY.name + ) + ) + + val runtime = testConnectionManager.getRuntime(host1Id)!! + runtime.connect() + runtime.gatewayClient.awaitGatewayReady(5000) + + // Attempt sendPrompt which will fail at submitPrompt + var threw = false + try { + testRepo.sendPrompt(session.id, "Will fail") + } catch (_: Exception) { + threw = true + } + + assertTrue("Expected prompt submission to throw", threw) + + // Verify syncedThroughMessageId did NOT advance and remains "msg_baseline" + val binding = testSessionDao.getBindingsForSession(session.id.value).find { it.hostId == host1Id.value } + assertEquals("msg_baseline", binding?.syncedThroughMessageId) + assertEquals(BindingState.ERROR.name, binding?.state) + + runtime.disconnect() + server.shutdown() + } + + @Test + fun testStopHostADoesNotStopHostB() = runTest(testDispatcher) { + val hostA = HermesHost(id = host1Id, displayName = "Windows PC", baseUrl = "http://pc:9119") + val hostB = HermesHost(id = host2Id, displayName = "Linux Server", baseUrl = "http://linux:9119") + connectionManager.addHost(hostA) + connectionManager.addHost(hostB) + testScheduler.advanceUntilIdle() + + val session = sessionRepo.createUnifiedSession(title = "Targeted Stop", initialHostId = host1Id) + testScheduler.advanceUntilIdle() + + val runtimeA = connectionManager.getRuntime(host1Id) + val runtimeB = connectionManager.getRuntime(host2Id) + + sessionRepo.registerRuntimeBinding(session.id, host1Id, RuntimeSessionId("rt_a")) + sessionRepo.registerRuntimeBinding(session.id, host2Id, RuntimeSessionId("rt_b")) + + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host1Id.value, + durableSessionId = "dur_a", + runtimeSessionId = "rt_a", + state = BindingState.RUNNING.name + ) + ) + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host2Id.value, + durableSessionId = "dur_b", + runtimeSessionId = "rt_b", + state = BindingState.RUNNING.name + ) + ) + testScheduler.advanceUntilIdle() + + // Start execution on both hosts + runtimeA!!.gatewayClient.handleIncomingMessage( + buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.start") + put("session_id", "rt_a") + put("message_id", "msg_a") + }) + }.toString() + ) + runtimeB!!.gatewayClient.handleIncomingMessage( + buildJsonObject { + put("method", "event") + put("params", buildJsonObject { + put("event", "message.start") + put("session_id", "rt_b") + put("message_id", "msg_b") + }) + }.toString() + ) + testScheduler.advanceUntilIdle() + + assertTrue(sessionRepo.getHostExecuting(session.id, host1Id).value) + assertTrue(sessionRepo.getHostExecuting(session.id, host2Id).value) + + // Stop Host A specifically + sessionRepo.interruptHost(session.id, host1Id) + testScheduler.advanceUntilIdle() + + // Host A execution should be false, Host B execution MUST still be true + assertFalse("Host A should be stopped", sessionRepo.getHostExecuting(session.id, host1Id).value) + assertTrue("Host B should still be running", sessionRepo.getHostExecuting(session.id, host2Id).value) + } +} diff --git a/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt index 31a7fb5..996690c 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/UnifiedSessionRepositoryTest.kt @@ -8,6 +8,7 @@ import app.hermes.mobile.core.runtime.HermesConnectionManager 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 @@ -95,6 +96,19 @@ class UnifiedSessionRepositoryTest { val session = repository.createUnifiedSession(title = "Streaming Test", initialHostId = host1Id) testScheduler.advanceUntilIdle() + repository.registerRuntimeBinding(session.id, host1Id, RuntimeSessionId("rt_stream_1")) + + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host1Id.value, + durableSessionId = "dur_stream_1", + runtimeSessionId = "rt_stream_1", + state = BindingState.RUNNING.name + ) + ) + testScheduler.advanceUntilIdle() + val runtimeA = connectionManager.getRuntime(host1Id) assertNotNull(runtimeA) @@ -103,39 +117,46 @@ class UnifiedSessionRepositoryTest { put("method", "event") put("params", buildJsonObject { put("event", "message.start") + put("session_id", "rt_stream_1") put("message_id", "msg_stream_1") put("role", "assistant") }) } runtimeA?.gatewayClient?.handleIncomingMessage(msgStart.toString()) + testScheduler.advanceUntilIdle() // Stream delta 1 val msgDelta1 = buildJsonObject { put("method", "event") put("params", buildJsonObject { put("event", "message.delta") + put("session_id", "rt_stream_1") put("message_id", "msg_stream_1") put("delta", "Hello ") }) } runtimeA?.gatewayClient?.handleIncomingMessage(msgDelta1.toString()) + testScheduler.advanceUntilIdle() // Stream delta 2 val msgDelta2 = buildJsonObject { put("method", "event") put("params", buildJsonObject { put("event", "message.delta") + put("session_id", "rt_stream_1") put("message_id", "msg_stream_1") put("delta", "from Multi-Hermes!") }) } runtimeA?.gatewayClient?.handleIncomingMessage(msgDelta2.toString()) + testScheduler.advanceUntilIdle() // Stream complete val msgComplete = buildJsonObject { put("method", "event") put("params", buildJsonObject { put("event", "message.complete") + put("session_id", "rt_stream_1") put("message_id", "msg_stream_1") put("content", "Hello from Multi-Hermes!") }) @@ -162,6 +183,19 @@ class UnifiedSessionRepositoryTest { val session = repository.createUnifiedSession(title = "Background Session", initialHostId = host1Id) testScheduler.advanceUntilIdle() + repository.registerRuntimeBinding(session.id, host1Id, RuntimeSessionId("rt_bg_1")) + + sessionDao.insertOrUpdateBinding( + HostBindingEntity( + sessionId = session.id.value, + hostId = host1Id.value, + durableSessionId = "dur_bg_1", + runtimeSessionId = "rt_bg_1", + state = BindingState.RUNNING.name + ) + ) + testScheduler.advanceUntilIdle() + val runtimeA = connectionManager.getRuntime(host1Id) val runtimeB = connectionManager.getRuntime(host2Id) @@ -170,11 +204,13 @@ class UnifiedSessionRepositoryTest { put("method", "event") put("params", buildJsonObject { put("event", "tool.start") + put("session_id", "rt_bg_1") put("tool_id", "tool_bg_1") put("name", "heavy_build_task") }) } runtimeA?.gatewayClient?.handleIncomingMessage(toolStart.toString()) + testScheduler.advanceUntilIdle() // User switches session active host to Host B repository.switchSessionActiveHost(session.id, host2Id) @@ -185,6 +221,7 @@ class UnifiedSessionRepositoryTest { put("method", "event") put("params", buildJsonObject { put("event", "tool.complete") + put("session_id", "rt_bg_1") put("tool_id", "tool_bg_1") put("result", "Build successful in 42s") put("is_error", false) diff --git a/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt b/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt index d02ba5d..39903b8 100644 --- a/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt +++ b/app/src/test/java/app/hermes/mobile/core/storage/FakeDaos.kt @@ -1,6 +1,7 @@ package app.hermes.mobile.core.storage import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.map @@ -46,24 +47,29 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { private val sessions = mutableMapOf() private val bindings = mutableMapOf>() private val messages = mutableMapOf>() - private val sessionsFlow = MutableStateFlow>(emptyList()) + private val _sessionsFlow = MutableSharedFlow>(replay = 1) - private fun updateFlow() { - sessionsFlow.value = sessions.values.sortedByDescending { it.updatedAt } + init { + _sessionsFlow.tryEmit(emptyList()) } - override fun getSessionsFlow(): Flow> = sessionsFlow + private fun updateFlow() { + val list = sessions.values.sortedByDescending { it.updatedAt }.map { it.copy() } + _sessionsFlow.tryEmit(list) + } + + override fun getSessionsFlow(): Flow> = _sessionsFlow override suspend fun getSessions(): List = sessions.values.sortedByDescending { it.updatedAt } override fun getSessionWithDetailsFlow(sessionId: String): Flow { - return sessionsFlow.map { getSessionWithDetails(sessionId) } + return _sessionsFlow.map { _: List -> getSessionWithDetails(sessionId) } } override suspend fun getSessionWithDetails(sessionId: String): UnifiedSessionWithDetails? { val s = sessions[sessionId] ?: return null - val b = bindings[sessionId] ?: emptyList() - val m = messages[sessionId] ?: emptyList() + val b = bindings[sessionId]?.map { it.copy() } ?: emptyList() + val m = messages[sessionId]?.map { it.copy() } ?: emptyList() return UnifiedSessionWithDetails(session = s, bindings = b, messages = m) } @@ -72,7 +78,7 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { } override suspend fun getBindingsForSession(sessionId: String): List { - return bindings[sessionId] ?: emptyList() + return bindings[sessionId]?.map { it.copy() } ?: emptyList() } override suspend fun insertSession(session: UnifiedSessionEntity) { @@ -96,18 +102,34 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { val list = bindings.computeIfAbsent(binding.sessionId) { mutableListOf() } list.removeAll { it.hostId == binding.hostId } list.add(binding) + val s = sessions[binding.sessionId] + if (s != null) { + sessions[binding.sessionId] = s.copy(updatedAt = System.currentTimeMillis()) + } + updateFlow() } override suspend fun insertOrUpdateBindings(bindingList: List) { - for (b in bindingList) insertOrUpdateBinding(b) + for (b in bindingList) { + val list = bindings.computeIfAbsent(b.sessionId) { mutableListOf() } + list.removeAll { it.hostId == b.hostId } + list.add(b) + val s = sessions[b.sessionId] + if (s != null) { + sessions[b.sessionId] = s.copy(updatedAt = System.currentTimeMillis()) + } + } + updateFlow() } override suspend fun deleteBinding(sessionId: String, hostId: String) { bindings[sessionId]?.removeAll { it.hostId == hostId } + updateFlow() } override suspend fun deleteBindingsForSession(sessionId: String) { bindings.remove(sessionId) + updateFlow() } override suspend fun insertOrUpdateMessage(message: UnifiedMessageEntity) { @@ -118,14 +140,25 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { } else { list.add(message) } + updateFlow() } override suspend fun insertMessages(msgList: List) { - for (m in msgList) insertOrUpdateMessage(m) + for (m in msgList) { + val list = messages.computeIfAbsent(m.sessionId) { mutableListOf() } + val idx = list.indexOfFirst { it.id == m.id } + if (idx >= 0) { + list[idx] = m + } else { + list.add(m) + } + } + updateFlow() } override suspend fun deleteMessagesForSession(sessionId: String) { messages.remove(sessionId) + updateFlow() } override suspend fun updateMessageContent( @@ -148,6 +181,7 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { break } } + updateFlow() } override suspend fun updateActiveHost(sessionId: String, hostId: String, updatedAt: Long) { @@ -173,6 +207,7 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { syncedAt = syncedAt, state = state ) + updateFlow() } } @@ -181,6 +216,7 @@ class FakeUnifiedSessionDao : UnifiedSessionDao { val idx = list.indexOfFirst { it.hostId == hostId } if (idx >= 0) { list[idx] = list[idx].copy(state = state) + updateFlow() } } } diff --git a/app/src/test/java/app/hermes/mobile/core/sync/UnifiedContextBuilderTest.kt b/app/src/test/java/app/hermes/mobile/core/sync/UnifiedContextBuilderTest.kt index c6a3389..dcecff7 100644 --- a/app/src/test/java/app/hermes/mobile/core/sync/UnifiedContextBuilderTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/sync/UnifiedContextBuilderTest.kt @@ -92,6 +92,8 @@ class UnifiedContextBuilderTest { assertTrue(syncAll.contextPrompt.contains("Office PC")) assertTrue(syncAll.contextPrompt.contains("Write a python script")) assertTrue(syncAll.contextPrompt.contains("Linux Server")) + assertFalse(syncAll.contextPrompt.contains("192.168.1.100:9119")) + assertFalse(syncAll.contextPrompt.contains("192.168.1.50:9119")) // Case 2: Stale host binding (synced up to msg-1, needs delta msg-2 and msg-3) val syncDelta = UnifiedContextBuilder.buildContextSyncPayload(session, host2, hostsMap, "msg-1") @@ -100,6 +102,7 @@ class UnifiedContextBuilderTest { assertFalse(syncDelta.contextPrompt.contains("Write a python script to parse CSV files.")) assertTrue(syncDelta.contextPrompt.contains("Sure! Here is the python script")) assertTrue(syncDelta.contextPrompt.contains("Now run it on the linux server dataset.")) + assertFalse(syncDelta.contextPrompt.contains("192.168.1.100:9119")) // Case 3: Fully synced host binding (synced up to msg-3) val syncUpToDate = UnifiedContextBuilder.buildContextSyncPayload(session, host2, hostsMap, "msg-3")