diff --git a/app/src/androidTest/java/app/hermes/mobile/ReleaseSerializationTest.kt b/app/src/androidTest/java/app/hermes/mobile/ReleaseSerializationTest.kt index a040f46..c1bef6c 100644 --- a/app/src/androidTest/java/app/hermes/mobile/ReleaseSerializationTest.kt +++ b/app/src/androidTest/java/app/hermes/mobile/ReleaseSerializationTest.kt @@ -22,7 +22,7 @@ import app.hermes.mobile.core.model.UnifiedSession import app.hermes.mobile.core.model.UnifiedSessionId import app.hermes.mobile.core.pairing.CanonicalEndpoint import app.hermes.mobile.core.pairing.HermesPairingPayload -import app.hermes.mobile.core.security.NativeAuthTokens +import app.hermes.mobile.core.model.NativeAuthTokens import kotlinx.serialization.encodeToString import kotlinx.serialization.json.Json import kotlinx.serialization.json.buildJsonObject @@ -62,7 +62,8 @@ class ReleaseSerializationTest { refreshToken = "refresh-secret", tokenType = "Bearer", expiresIn = 3600L, - scope = "read write" + provider = "github", + userId = "user-123" ) val serializedTokens = json.encodeToString(tokens) val deserializedTokens = json.decodeFromString(serializedTokens) diff --git a/app/src/main/AndroidManifest.xml b/app/src/main/AndroidManifest.xml index 85df2e4..419d5a7 100644 --- a/app/src/main/AndroidManifest.xml +++ b/app/src/main/AndroidManifest.xml @@ -4,7 +4,10 @@ - + + diff --git a/app/src/test/java/app/hermes/mobile/core/network/ReadyDeferredRaceTest.kt b/app/src/test/java/app/hermes/mobile/core/network/ReadyDeferredRaceTest.kt index 031a296..24d0352 100644 --- a/app/src/test/java/app/hermes/mobile/core/network/ReadyDeferredRaceTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/network/ReadyDeferredRaceTest.kt @@ -72,8 +72,9 @@ class ReadyDeferredRaceTest { // 1. Connect to Server 1 client.connect(wsUrl1, allowCleartext = true) - // 2. Start waiting for gateway.ready in background + val awaiterStarted = CompletableDeferred() val awaiter = async { + awaiterStarted.complete(Unit) try { client.awaitGatewayReady(4000) true @@ -82,8 +83,7 @@ class ReadyDeferredRaceTest { } } - // Give a brief moment for awaiter to capture deferred - delay(50) + awaiterStarted.await() // 3. Immediately re-invoke connect to Server 2 before Server 1 completes client.connect(wsUrl2, allowCleartext = true) diff --git a/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt index c4c9fe3..ef6a56e 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/CacheEvictionTest.kt @@ -9,12 +9,12 @@ 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.Job import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.delay import kotlinx.coroutines.launch -import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.put import org.junit.Assert.assertEquals @@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap * does not leak memory in repository caches (sessionMessagesState, hostExecutingState, * sessionExecutingState, sessionHostMutexes, toolIdToMessageId). */ +@OptIn(ExperimentalCoroutinesApi::class) class CacheEvictionTest { @Suppress("UNCHECKED_CAST") @@ -42,21 +43,22 @@ class CacheEvictionTest { } @Test - fun test50SessionsCycleEvictsAndBoundsMemoryCaches() = runBlocking { + fun test50SessionsCycleEvictsAndBoundsMemoryCaches() = runTest { + val testDispatcher = StandardTestDispatcher(testScheduler) val hostId = HermesHostId("host-cache-1") val host = HermesHost(id = hostId, displayName = "Cache Host", baseUrl = "http://cache-host:9119") val hostDao = FakeHostDao() val sessionDao = FakeUnifiedSessionDao() val tokenVault = InMemoryTokenVault() - val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + val scope = CoroutineScope(SupervisorJob() + testDispatcher) val connectionManager = HermesConnectionManager( hostDao = hostDao, tokenVault = tokenVault, scope = scope, runtimeFactory = { parentScope, h -> - val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default) + val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + testDispatcher) HermesHostRuntime( initialHost = h, gatewayClient = JsonRpcGatewayClient(scope = childScope), @@ -73,7 +75,7 @@ class CacheEvictionTest { ) connectionManager.addHost(host) - delay(50) + testScheduler.advanceUntilIdle() val runtime = connectionManager.getRuntime(hostId)!! val sessionCount = 50 @@ -146,16 +148,17 @@ class CacheEvictionTest { }) }.toString() runtime.gatewayClient.handleIncomingMessage(completeJson) - delay(20) + testScheduler.advanceUntilIdle() // Unsubscribe job.cancel() // Release session explicitly repository.releaseSession(session.id) + testScheduler.advanceUntilIdle() } - delay(100) + testScheduler.advanceUntilIdle() val messagesCacheSize = getInternalMapSize(repository, "sessionMessagesState") val sessionExecSize = getInternalMapSize(repository, "sessionExecutingState") 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 index 6fbfda9..97304a4 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/MultiHostConcurrencyExecutionTest.kt @@ -447,15 +447,19 @@ class MultiHostConcurrencyExecutionTest { path.startsWith("/api/ws") || path.startsWith("/ws") -> { MockResponse().withWebSocketUpgrade(object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: Response) { - webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + try { + webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + } catch (_: Exception) {} } 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"}}""") - } + try { + 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"}}""") + } + } catch (_: Exception) {} } }) } @@ -510,18 +514,20 @@ class MultiHostConcurrencyExecutionTest { runtime!!.connect() runtime.gatewayClient.awaitGatewayReady(5000) - val turnId = freshRepo.sendPrompt(sessionId, "Hello after restart") - assertEquals("t_101", turnId) + try { + 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() + // 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) + } finally { + try { runtime.disconnect() } catch (_: Exception) {} + try { server.shutdown() } catch (_: Exception) {} + } } @Test @@ -540,19 +546,23 @@ class MultiHostConcurrencyExecutionTest { path.startsWith("/api/ws") || path.startsWith("/ws") -> { MockResponse().withWebSocketUpgrade(object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: Response) { - webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + try { + webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + } catch (_: Exception) {} } override fun onMessage(webSocket: WebSocket, text: String) { - if (text.contains("session.resume")) { - sessionResumeCalled = true - if (text.contains("valid_durable_888")) { - resumedDurableId = "valid_durable_888" + try { + if (text.contains("session.resume")) { + sessionResumeCalled = true + if (text.contains("valid_durable_888")) { + resumedDurableId = "valid_durable_888" + } + webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"valid_durable_888","session_id":"fresh_runtime_777"}}""") + } else if (text.contains("prompt.submit")) { + webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"turn_id":"turn_ready_restart"}}""") } - webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"valid_durable_888","session_id":"fresh_runtime_777"}}""") - } else if (text.contains("prompt.submit")) { - webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"turn_id":"turn_ready_restart"}}""") - } + } catch (_: Exception) {} } }) } @@ -607,20 +617,22 @@ class MultiHostConcurrencyExecutionTest { runtime.connect() runtime.gatewayClient.awaitGatewayReady(5000) - // Calling sendPrompt must trigger session.resume since in-memory attachment does not exist in new process - val turnId = freshRepo.sendPrompt(sessionId, "Hello after clean restart") - assertEquals("turn_ready_restart", turnId) + try { + // Calling sendPrompt must trigger session.resume since in-memory attachment does not exist in new process + val turnId = freshRepo.sendPrompt(sessionId, "Hello after clean restart") + assertEquals("turn_ready_restart", turnId) - assertTrue("session.resume MUST be called even if persisted state was READY", sessionResumeCalled) - assertEquals("valid_durable_888", resumedDurableId) + assertTrue("session.resume MUST be called even if persisted state was READY", sessionResumeCalled) + assertEquals("valid_durable_888", resumedDurableId) - val updatedBinding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } - assertNotNull(updatedBinding) - assertEquals("valid_durable_888", updatedBinding?.durableSessionId) - assertEquals("fresh_runtime_777", updatedBinding?.runtimeSessionId) - - runtime.disconnect() - server.shutdown() + val updatedBinding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } + assertNotNull(updatedBinding) + assertEquals("valid_durable_888", updatedBinding?.durableSessionId) + assertEquals("fresh_runtime_777", updatedBinding?.runtimeSessionId) + } finally { + try { runtime.disconnect() } catch (_: Exception) {} + try { server.shutdown() } catch (_: Exception) {} + } } @Test @@ -638,16 +650,20 @@ class MultiHostConcurrencyExecutionTest { path.startsWith("/api/ws") || path.startsWith("/ws") -> { MockResponse().withWebSocketUpgrade(object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: Response) { - webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + try { + webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + } catch (_: Exception) {} } 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"}}""") - } + try { + 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"}}""") + } + } catch (_: Exception) {} } }) } @@ -680,27 +696,29 @@ class MultiHostConcurrencyExecutionTest { val session = testRepo.createUnifiedSession(title = "Reconnect Test", initialHostId = HermesHostId("host-windows")) val runtime = testConnectionManager.getRuntime(HermesHostId("host-windows"))!! - // Initial connect - runtime.connect() - runtime.gatewayClient.awaitGatewayReady(5000) + try { + // Initial connect + runtime.connect() + runtime.gatewayClient.awaitGatewayReady(5000) - val binding1 = testRepo.ensureAttachedRuntimeSession(session.id, host1Id, runtime) - assertEquals(RuntimeSessionId("rt_initial"), binding1.runtimeSessionId) + 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) + // Disconnect host runtime + runtime.disconnect() + testSessionDao.updateBindingState(session.id.value, host1Id.value, BindingState.OFFLINE.name) - // Reconnect host runtime - runtime.connect() - runtime.gatewayClient.awaitGatewayReady(5000) + // 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() + // Reattach after reconnect + val binding2 = testRepo.ensureAttachedRuntimeSession(session.id, host1Id, runtime) + assertEquals(RuntimeSessionId("rt_after_reconnect"), binding2.runtimeSessionId) + } finally { + try { runtime.disconnect() } catch (_: Exception) {} + try { server.shutdown() } catch (_: Exception) {} + } } @Test @@ -717,16 +735,20 @@ class MultiHostConcurrencyExecutionTest { path.startsWith("/api/ws") || path.startsWith("/ws") -> { MockResponse().withWebSocketUpgrade(object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: Response) { - webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + try { + webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + } catch (_: Exception) {} } override fun onMessage(webSocket: WebSocket, text: String) { - if (text.contains("session.resume")) { - webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"dur_fail_test","session_id":"rt_fail_test"}}""") - } else if (text.contains("prompt.submit")) { - // Fail prompt submission with an RPC error - webSocket.send("""{"jsonrpc":"2.0","id":"a2","error":{"code":-32000,"message":"Model overloaded"}}""") - } + try { + if (text.contains("session.resume")) { + webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"dur_fail_test","session_id":"rt_fail_test"}}""") + } else if (text.contains("prompt.submit")) { + // Fail prompt submission with an RPC error + webSocket.send("""{"jsonrpc":"2.0","id":"a2","error":{"code":-32000,"message":"Model overloaded"}}""") + } + } catch (_: Exception) {} } }) } @@ -775,23 +797,25 @@ class MultiHostConcurrencyExecutionTest { 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 + // 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) + } finally { + try { runtime.disconnect() } catch (_: Exception) {} + try { server.shutdown() } catch (_: Exception) {} } - - 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 @@ -809,17 +833,21 @@ class MultiHostConcurrencyExecutionTest { path.startsWith("/api/ws") || path.startsWith("/ws") -> { MockResponse().withWebSocketUpgrade(object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: Response) { - webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + try { + webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + } catch (_: Exception) {} } override fun onMessage(webSocket: WebSocket, text: String) { - if (text.contains("session.resume")) { - // Transient failure: 500 / -32000 Server overloaded - webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":-32000,"message":"Transient server error"}}""") - } else if (text.contains("session.create")) { - createCalled = true - webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_forbidden","session_id":"rt_forbidden"}}""") - } + try { + if (text.contains("session.resume")) { + // Transient failure: 500 / -32000 Server overloaded + webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":-32000,"message":"Transient server error"}}""") + } else if (text.contains("session.create")) { + createCalled = true + webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_forbidden","session_id":"rt_forbidden"}}""") + } + } catch (_: Exception) {} } }) } @@ -872,22 +900,24 @@ class MultiHostConcurrencyExecutionTest { runtime.connect() runtime.gatewayClient.awaitGatewayReady(5000) - var threw = false try { - testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime) - } catch (_: Exception) { - threw = true + var threw = false + try { + testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime) + } catch (_: Exception) { + threw = true + } + + assertTrue("Expected transient resume error to throw", threw) + assertFalse("session.create MUST NOT be called on transient resume error", createCalled) + + // Verify binding was not destroyed + val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } + assertEquals("dur_preserved_123", binding?.durableSessionId) + } finally { + try { runtime.disconnect() } catch (_: Exception) {} + try { server.shutdown() } catch (_: Exception) {} } - - assertTrue("Expected transient resume error to throw", threw) - assertFalse("session.create MUST NOT be called on transient resume error", createCalled) - - // Verify binding was not destroyed - val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } - assertEquals("dur_preserved_123", binding?.durableSessionId) - - runtime.disconnect() - server.shutdown() } @Test @@ -905,17 +935,21 @@ class MultiHostConcurrencyExecutionTest { path.startsWith("/api/ws") || path.startsWith("/ws") -> { MockResponse().withWebSocketUpgrade(object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: Response) { - webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + try { + webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") + } catch (_: Exception) {} } override fun onMessage(webSocket: WebSocket, text: String) { - if (text.contains("session.resume")) { - // Definitively unrecoverable: 404 Session not found - webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":404,"message":"Session not found"}}""") - } else if (text.contains("session.create")) { - createCalled = true - webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_new_fresh_999","session_id":"rt_new_fresh_999"}}""") - } + try { + if (text.contains("session.resume")) { + // Definitively unrecoverable: 404 Session not found + webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":404,"message":"Session not found"}}""") + } else if (text.contains("session.create")) { + createCalled = true + webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_new_fresh_999","session_id":"rt_new_fresh_999"}}""") + } + } catch (_: Exception) {} } }) } @@ -968,19 +1002,21 @@ class MultiHostConcurrencyExecutionTest { runtime.connect() runtime.gatewayClient.awaitGatewayReady(5000) - val attachedBinding = testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime) + try { + val attachedBinding = testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime) - assertTrue("session.create MUST be called on unrecoverable 404 resume error", createCalled) - assertEquals(DurableSessionId("dur_new_fresh_999"), attachedBinding.durableSessionId) - assertEquals(RuntimeSessionId("rt_new_fresh_999"), attachedBinding.runtimeSessionId) + assertTrue("session.create MUST be called on unrecoverable 404 resume error", createCalled) + assertEquals(DurableSessionId("dur_new_fresh_999"), attachedBinding.durableSessionId) + assertEquals(RuntimeSessionId("rt_new_fresh_999"), attachedBinding.runtimeSessionId) - // Verify Room DB binding was updated - val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } - assertEquals("dur_new_fresh_999", binding?.durableSessionId) - assertEquals("rt_new_fresh_999", binding?.runtimeSessionId) - - runtime.disconnect() - server.shutdown() + // Verify Room DB binding was updated + val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } + assertEquals("dur_new_fresh_999", binding?.durableSessionId) + assertEquals("rt_new_fresh_999", binding?.runtimeSessionId) + } finally { + try { runtime.disconnect() } catch (_: Exception) {} + try { server.shutdown() } catch (_: Exception) {} + } } @Test diff --git a/app/src/test/java/app/hermes/mobile/core/repository/SessionCreateRaceTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/SessionCreateRaceTest.kt index a2650c8..d57b803 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/SessionCreateRaceTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/SessionCreateRaceTest.kt @@ -31,6 +31,8 @@ class SessionCreateRaceTest { val repoScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) val createCallCount = AtomicInteger(0) + val firstCreateStarted = CompletableDeferred() + val unblockCreate = CompletableDeferred() val mockGatewayClient = mockk(relaxed = true) val runtimeScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) @@ -40,7 +42,8 @@ class SessionCreateRaceTest { coEvery { mockGatewayClient.awaitGatewayReady(any()) } returns Unit coEvery { mockGatewayClient.createSession(any(), any()) } coAnswers { createCallCount.incrementAndGet() - delay(100) // simulate network delay to expose race condition + firstCreateStarted.complete(Unit) + unblockCreate.await() CreateSessionResult( durableId = DurableSessionId("dur_race_1"), runtimeId = RuntimeSessionId("rt_race_1") @@ -71,10 +74,8 @@ class SessionCreateRaceTest { ) connectionManager.addHost(host) - delay(50) val session = repository.createUnifiedSession(title = "Race Test Session", initialHostId = hostId) - delay(50) // Launch 2 concurrent sendPrompt calls for the same session & host val deferred1 = async(Dispatchers.Default) { @@ -84,6 +85,11 @@ class SessionCreateRaceTest { repository.sendPrompt(session.id, "Prompt from coroutine 2") } + // Wait until the first createSession call enters and acquires the lock + firstCreateStarted.await() + // Unblock creation + unblockCreate.complete(Unit) + awaitAll(deferred1, deferred2) assertEquals("Exactly one session.create must be invoked for concurrent sendPrompt calls", 1, createCallCount.get()) diff --git a/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt b/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt index 6e2621f..b8d0fb0 100644 --- a/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt +++ b/app/src/test/java/app/hermes/mobile/core/repository/ToolAttributionTest.kt @@ -9,12 +9,11 @@ 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.Job import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.delay -import kotlinx.coroutines.flow.first -import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.put import org.junit.Assert.assertEquals @@ -27,10 +26,12 @@ import org.junit.Test * and thinking deltas, tools and thinking are strictly bound to their respective host and * explicit messageId, avoiding false attribution. */ +@OptIn(ExperimentalCoroutinesApi::class) class ToolAttributionTest { @Test - fun testTwoConcurrentHostsAttributionAndThinkingIsolation() = runBlocking { + fun testTwoConcurrentHostsAttributionAndThinkingIsolation() = runTest { + val testDispatcher = StandardTestDispatcher(testScheduler) val hostAId = HermesHostId("host-a") val hostBId = HermesHostId("host-b") @@ -40,14 +41,14 @@ class ToolAttributionTest { val hostDao = FakeHostDao() val sessionDao = FakeUnifiedSessionDao() val tokenVault = InMemoryTokenVault() - val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + val scope = CoroutineScope(SupervisorJob() + testDispatcher) val connectionManager = HermesConnectionManager( hostDao = hostDao, tokenVault = tokenVault, scope = scope, runtimeFactory = { parentScope, h -> - val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default) + val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + testDispatcher) HermesHostRuntime( initialHost = h, gatewayClient = JsonRpcGatewayClient(scope = childScope), @@ -65,7 +66,7 @@ class ToolAttributionTest { connectionManager.addHost(hostA) connectionManager.addHost(hostB) - delay(50) + testScheduler.advanceUntilIdle() val session = repository.createUnifiedSession(title = "Dual Host Attribution Test", initialHostId = hostAId) val rtSessionA = "rt_session_host_a" @@ -92,6 +93,7 @@ class ToolAttributionTest { state = BindingState.RUNNING.name ) ) + testScheduler.advanceUntilIdle() val runtimeA = connectionManager.getRuntime(hostAId)!! val runtimeB = connectionManager.getRuntime(hostBId)!! @@ -127,7 +129,7 @@ class ToolAttributionTest { }) }.toString()) - delay(30) + testScheduler.advanceUntilIdle() // 3. Host A streams thinking delta for msgA runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject { @@ -187,7 +189,7 @@ class ToolAttributionTest { }) }.toString()) - delay(30) + testScheduler.advanceUntilIdle() // 7. Update tool progress and completion for host A runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject { @@ -246,7 +248,7 @@ class ToolAttributionTest { }) }.toString()) - delay(50) + testScheduler.advanceUntilIdle() val messages = repository.getSessionMessages(session.id).value val msgA = messages.find { it.id == msgAId } @@ -271,21 +273,22 @@ class ToolAttributionTest { } @Test - fun testThinkingDeltaWithExplicitMessageIdDoesNotFallBackToLastAssistant() = runBlocking { + fun testThinkingDeltaWithExplicitMessageIdDoesNotFallBackToLastAssistant() = runTest { + val testDispatcher = StandardTestDispatcher(testScheduler) val hostAId = HermesHostId("host-a") val hostA = HermesHost(id = hostAId, displayName = "Host A", baseUrl = "http://host-a:9119") val hostDao = FakeHostDao() val sessionDao = FakeUnifiedSessionDao() val tokenVault = InMemoryTokenVault() - val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + val scope = CoroutineScope(SupervisorJob() + testDispatcher) val connectionManager = HermesConnectionManager( hostDao = hostDao, tokenVault = tokenVault, scope = scope, runtimeFactory = { parentScope, h -> - val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default) + val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + testDispatcher) HermesHostRuntime( initialHost = h, gatewayClient = JsonRpcGatewayClient(scope = childScope), @@ -302,7 +305,7 @@ class ToolAttributionTest { ) connectionManager.addHost(hostA) - delay(50) + testScheduler.advanceUntilIdle() val session = repository.createUnifiedSession(title = "Message Targeting Test", initialHostId = hostAId) val rtSessionA = "rt_session_host_a" @@ -317,6 +320,7 @@ class ToolAttributionTest { state = BindingState.RUNNING.name ) ) + testScheduler.advanceUntilIdle() val runtimeA = connectionManager.getRuntime(hostAId)!! @@ -351,7 +355,7 @@ class ToolAttributionTest { }) }.toString()) - delay(30) + testScheduler.advanceUntilIdle() // Send thinking delta explicitly targeted at msg1Id runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject { @@ -367,7 +371,7 @@ class ToolAttributionTest { }) }.toString()) - delay(30) + testScheduler.advanceUntilIdle() val messages = repository.getSessionMessages(session.id).value val msg1 = messages.find { it.id == msg1Id } diff --git a/hermes-pair/src/pairing.rs b/hermes-pair/src/pairing.rs index 4b4f36f..abf1c28 100644 --- a/hermes-pair/src/pairing.rs +++ b/hermes-pair/src/pairing.rs @@ -187,13 +187,13 @@ pub fn validate_payload(payload: &PairingPayloadV1, current_time: u64) -> Result payload.host ))); } - if !trimmed_host.starts_with('[') || !trimmed_host.ends_with(']') { - if trimmed_host.contains(':') { - return Err(PairingError::InvalidHost(format!( - "Host '{}' contains forbidden colon delimiter outside IPv6 brackets", - payload.host - ))); - } + if (!trimmed_host.starts_with('[') || !trimmed_host.ends_with(']')) + && trimmed_host.contains(':') + { + return Err(PairingError::InvalidHost(format!( + "Host '{}' contains forbidden colon delimiter outside IPv6 brackets", + payload.host + ))); } // 6. Port: 1..=65535 (u16 is <= 65535, port 0 is invalid) diff --git a/schemas/1.json b/schemas/1.json deleted file mode 100644 index d7f00a1..0000000 --- a/schemas/1.json +++ /dev/null @@ -1,319 +0,0 @@ -{ - "formatVersion": 1, - "database": { - "version": 1, - "identityHash": "6dad88d50705023f76f91b35349d8c60", - "entities": [ - { - "tableName": "hosts", - "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `displayName` TEXT NOT NULL, `baseUrl` TEXT NOT NULL, `allowCleartext` INTEGER NOT NULL, `enabled` INTEGER NOT NULL, `lastSeenAt` INTEGER NOT NULL, `lastKnownStatus` TEXT NOT NULL, PRIMARY KEY(`id`))", - "fields": [ - { - "fieldPath": "id", - "columnName": "id", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "displayName", - "columnName": "displayName", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "baseUrl", - "columnName": "baseUrl", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "allowCleartext", - "columnName": "allowCleartext", - "affinity": "INTEGER", - "notNull": true - }, - { - "fieldPath": "enabled", - "columnName": "enabled", - "affinity": "INTEGER", - "notNull": true - }, - { - "fieldPath": "lastSeenAt", - "columnName": "lastSeenAt", - "affinity": "INTEGER", - "notNull": true - }, - { - "fieldPath": "lastKnownStatus", - "columnName": "lastKnownStatus", - "affinity": "TEXT", - "notNull": true - } - ], - "primaryKey": { - "autoGenerate": false, - "columnNames": [ - "id" - ] - }, - "indices": [], - "foreignKeys": [] - }, - { - "tableName": "unified_sessions", - "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `title` TEXT NOT NULL, `activeHostId` TEXT NOT NULL, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, PRIMARY KEY(`id`))", - "fields": [ - { - "fieldPath": "id", - "columnName": "id", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "title", - "columnName": "title", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "activeHostId", - "columnName": "activeHostId", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "createdAt", - "columnName": "createdAt", - "affinity": "INTEGER", - "notNull": true - }, - { - "fieldPath": "updatedAt", - "columnName": "updatedAt", - "affinity": "INTEGER", - "notNull": true - } - ], - "primaryKey": { - "autoGenerate": false, - "columnNames": [ - "id" - ] - }, - "indices": [], - "foreignKeys": [] - }, - { - "tableName": "host_bindings", - "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`sessionId` TEXT NOT NULL, `hostId` TEXT NOT NULL, `durableSessionId` TEXT NOT NULL, `runtimeSessionId` TEXT NOT NULL, `lastAttachedAt` INTEGER NOT NULL, `state` TEXT NOT NULL, `syncedThroughMessageId` TEXT, `syncedAt` INTEGER, PRIMARY KEY(`sessionId`, `hostId`), FOREIGN KEY(`sessionId`) REFERENCES `unified_sessions`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )", - "fields": [ - { - "fieldPath": "sessionId", - "columnName": "sessionId", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "hostId", - "columnName": "hostId", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "durableSessionId", - "columnName": "durableSessionId", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "runtimeSessionId", - "columnName": "runtimeSessionId", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "lastAttachedAt", - "columnName": "lastAttachedAt", - "affinity": "INTEGER", - "notNull": true - }, - { - "fieldPath": "state", - "columnName": "state", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "syncedThroughMessageId", - "columnName": "syncedThroughMessageId", - "affinity": "TEXT", - "notNull": false - }, - { - "fieldPath": "syncedAt", - "columnName": "syncedAt", - "affinity": "INTEGER", - "notNull": false - } - ], - "primaryKey": { - "autoGenerate": false, - "columnNames": [ - "sessionId", - "hostId" - ] - }, - "indices": [ - { - "name": "index_host_bindings_sessionId", - "unique": false, - "columnNames": [ - "sessionId" - ], - "orders": [], - "createSql": "CREATE INDEX IF NOT EXISTS `index_host_bindings_sessionId` ON `${TABLE_NAME}` (`sessionId`)" - }, - { - "name": "index_host_bindings_hostId", - "unique": false, - "columnNames": [ - "hostId" - ], - "orders": [], - "createSql": "CREATE INDEX IF NOT EXISTS `index_host_bindings_hostId` ON `${TABLE_NAME}` (`hostId`)" - } - ], - "foreignKeys": [ - { - "table": "unified_sessions", - "onDelete": "CASCADE", - "onUpdate": "NO ACTION", - "columns": [ - "sessionId" - ], - "referencedColumns": [ - "id" - ] - } - ] - }, - { - "tableName": "unified_messages", - "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `sessionId` TEXT NOT NULL, `role` TEXT NOT NULL, `content` TEXT NOT NULL, `hostId` TEXT, `source` TEXT NOT NULL, `createdAt` INTEGER NOT NULL, `nativeMessageId` TEXT, `thinking` TEXT, `toolsJson` TEXT, `isStreaming` INTEGER NOT NULL, PRIMARY KEY(`id`), FOREIGN KEY(`sessionId`) REFERENCES `unified_sessions`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )", - "fields": [ - { - "fieldPath": "id", - "columnName": "id", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "sessionId", - "columnName": "sessionId", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "role", - "columnName": "role", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "content", - "columnName": "content", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "hostId", - "columnName": "hostId", - "affinity": "TEXT", - "notNull": false - }, - { - "fieldPath": "source", - "columnName": "source", - "affinity": "TEXT", - "notNull": true - }, - { - "fieldPath": "createdAt", - "columnName": "createdAt", - "affinity": "INTEGER", - "notNull": true - }, - { - "fieldPath": "nativeMessageId", - "columnName": "nativeMessageId", - "affinity": "TEXT", - "notNull": false - }, - { - "fieldPath": "thinking", - "columnName": "thinking", - "affinity": "TEXT", - "notNull": false - }, - { - "fieldPath": "toolsJson", - "columnName": "toolsJson", - "affinity": "TEXT", - "notNull": false - }, - { - "fieldPath": "isStreaming", - "columnName": "isStreaming", - "affinity": "INTEGER", - "notNull": true - } - ], - "primaryKey": { - "autoGenerate": false, - "columnNames": [ - "id" - ] - }, - "indices": [ - { - "name": "index_unified_messages_sessionId", - "unique": false, - "columnNames": [ - "sessionId" - ], - "orders": [], - "createSql": "CREATE INDEX IF NOT EXISTS `index_unified_messages_sessionId` ON `${TABLE_NAME}` (`sessionId`)" - }, - { - "name": "index_unified_messages_createdAt", - "unique": false, - "columnNames": [ - "createdAt" - ], - "orders": [], - "createSql": "CREATE INDEX IF NOT EXISTS `index_unified_messages_createdAt` ON `${TABLE_NAME}` (`createdAt`)" - } - ], - "foreignKeys": [ - { - "table": "unified_sessions", - "onDelete": "CASCADE", - "onUpdate": "NO ACTION", - "columns": [ - "sessionId" - ], - "referencedColumns": [ - "id" - ] - } - ] - } - ], - "views": [], - "setupQueries": [ - "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)", - "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '6dad88d50705023f76f91b35349d8c60')" - ] - } -}