fix(ci): green pipeline fixes clippy androidtest deterministic tests and manifest cleanup (TASK-2026-08-25-11-green-pipeline)

This commit is contained in:
Ochenstarik 2026-08-25 09:45:13 +07:00
parent 2138c47536
commit 4c9d53e5a5
9 changed files with 220 additions and 486 deletions

View file

@ -22,7 +22,7 @@ import app.hermes.mobile.core.model.UnifiedSession
import app.hermes.mobile.core.model.UnifiedSessionId import app.hermes.mobile.core.model.UnifiedSessionId
import app.hermes.mobile.core.pairing.CanonicalEndpoint import app.hermes.mobile.core.pairing.CanonicalEndpoint
import app.hermes.mobile.core.pairing.HermesPairingPayload 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.encodeToString
import kotlinx.serialization.json.Json import kotlinx.serialization.json.Json
import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.buildJsonObject
@ -62,7 +62,8 @@ class ReleaseSerializationTest {
refreshToken = "refresh-secret", refreshToken = "refresh-secret",
tokenType = "Bearer", tokenType = "Bearer",
expiresIn = 3600L, expiresIn = 3600L,
scope = "read write" provider = "github",
userId = "user-123"
) )
val serializedTokens = json.encodeToString(tokens) val serializedTokens = json.encodeToString(tokens)
val deserializedTokens = json.decodeFromString<NativeAuthTokens>(serializedTokens) val deserializedTokens = json.decodeFromString<NativeAuthTokens>(serializedTokens)

View file

@ -4,7 +4,10 @@
<uses-permission android:name="android.permission.INTERNET" /> <uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" /> <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<uses-permission android:name="android.permission.CAMERA" /> <uses-permission android:name="android.permission.CAMERA" />
<uses-feature android:name="android.hardware.camera" android:required="false" /> <uses-permission android:name="android.permission.FOREGROUND_SERVICE" /> <uses-feature
android:name="android.hardware.camera"
android:required="false" />
<uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
<uses-permission android:name="android.permission.FOREGROUND_SERVICE_DATA_SYNC" /> <uses-permission android:name="android.permission.FOREGROUND_SERVICE_DATA_SYNC" />
<uses-permission android:name="android.permission.POST_NOTIFICATIONS" /> <uses-permission android:name="android.permission.POST_NOTIFICATIONS" />

View file

@ -72,8 +72,9 @@ class ReadyDeferredRaceTest {
// 1. Connect to Server 1 // 1. Connect to Server 1
client.connect(wsUrl1, allowCleartext = true) client.connect(wsUrl1, allowCleartext = true)
// 2. Start waiting for gateway.ready in background val awaiterStarted = CompletableDeferred<Unit>()
val awaiter = async { val awaiter = async {
awaiterStarted.complete(Unit)
try { try {
client.awaitGatewayReady(4000) client.awaitGatewayReady(4000)
true true
@ -82,8 +83,7 @@ class ReadyDeferredRaceTest {
} }
} }
// Give a brief moment for awaiter to capture deferred awaiterStarted.await()
delay(50)
// 3. Immediately re-invoke connect to Server 2 before Server 1 completes // 3. Immediately re-invoke connect to Server 2 before Server 1 completes
client.connect(wsUrl2, allowCleartext = true) client.connect(wsUrl2, allowCleartext = true)

View file

@ -9,12 +9,12 @@ import app.hermes.mobile.core.storage.FakeHostDao
import app.hermes.mobile.core.storage.FakeUnifiedSessionDao import app.hermes.mobile.core.storage.FakeUnifiedSessionDao
import app.hermes.mobile.core.storage.HostBindingEntity import app.hermes.mobile.core.storage.HostBindingEntity
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch 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.buildJsonObject
import kotlinx.serialization.json.put import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals import org.junit.Assert.assertEquals
@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap
* does not leak memory in repository caches (sessionMessagesState, hostExecutingState, * does not leak memory in repository caches (sessionMessagesState, hostExecutingState,
* sessionExecutingState, sessionHostMutexes, toolIdToMessageId). * sessionExecutingState, sessionHostMutexes, toolIdToMessageId).
*/ */
@OptIn(ExperimentalCoroutinesApi::class)
class CacheEvictionTest { class CacheEvictionTest {
@Suppress("UNCHECKED_CAST") @Suppress("UNCHECKED_CAST")
@ -42,21 +43,22 @@ class CacheEvictionTest {
} }
@Test @Test
fun test50SessionsCycleEvictsAndBoundsMemoryCaches() = runBlocking { fun test50SessionsCycleEvictsAndBoundsMemoryCaches() = runTest {
val testDispatcher = StandardTestDispatcher(testScheduler)
val hostId = HermesHostId("host-cache-1") val hostId = HermesHostId("host-cache-1")
val host = HermesHost(id = hostId, displayName = "Cache Host", baseUrl = "http://cache-host:9119") val host = HermesHost(id = hostId, displayName = "Cache Host", baseUrl = "http://cache-host:9119")
val hostDao = FakeHostDao() val hostDao = FakeHostDao()
val sessionDao = FakeUnifiedSessionDao() val sessionDao = FakeUnifiedSessionDao()
val tokenVault = InMemoryTokenVault() val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) val scope = CoroutineScope(SupervisorJob() + testDispatcher)
val connectionManager = HermesConnectionManager( val connectionManager = HermesConnectionManager(
hostDao = hostDao, hostDao = hostDao,
tokenVault = tokenVault, tokenVault = tokenVault,
scope = scope, scope = scope,
runtimeFactory = { parentScope, h -> runtimeFactory = { parentScope, h ->
val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default) val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + testDispatcher)
HermesHostRuntime( HermesHostRuntime(
initialHost = h, initialHost = h,
gatewayClient = JsonRpcGatewayClient(scope = childScope), gatewayClient = JsonRpcGatewayClient(scope = childScope),
@ -73,7 +75,7 @@ class CacheEvictionTest {
) )
connectionManager.addHost(host) connectionManager.addHost(host)
delay(50) testScheduler.advanceUntilIdle()
val runtime = connectionManager.getRuntime(hostId)!! val runtime = connectionManager.getRuntime(hostId)!!
val sessionCount = 50 val sessionCount = 50
@ -146,16 +148,17 @@ class CacheEvictionTest {
}) })
}.toString() }.toString()
runtime.gatewayClient.handleIncomingMessage(completeJson) runtime.gatewayClient.handleIncomingMessage(completeJson)
delay(20) testScheduler.advanceUntilIdle()
// Unsubscribe // Unsubscribe
job.cancel() job.cancel()
// Release session explicitly // Release session explicitly
repository.releaseSession(session.id) repository.releaseSession(session.id)
testScheduler.advanceUntilIdle()
} }
delay(100) testScheduler.advanceUntilIdle()
val messagesCacheSize = getInternalMapSize(repository, "sessionMessagesState") val messagesCacheSize = getInternalMapSize(repository, "sessionMessagesState")
val sessionExecSize = getInternalMapSize(repository, "sessionExecutingState") val sessionExecSize = getInternalMapSize(repository, "sessionExecutingState")

View file

@ -447,15 +447,19 @@ class MultiHostConcurrencyExecutionTest {
path.startsWith("/api/ws") || path.startsWith("/ws") -> { path.startsWith("/api/ws") || path.startsWith("/ws") -> {
MockResponse().withWebSocketUpgrade(object : WebSocketListener() { MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
try {
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") 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) { override fun onMessage(webSocket: WebSocket, text: String) {
try {
if (text.contains("session.resume")) { if (text.contains("session.resume")) {
webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"durable_persisted_99","session_id":"fresh_runtime_101"}}""") 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")) { } else if (text.contains("prompt.submit")) {
webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"turn_id":"t_101"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"turn_id":"t_101"}}""")
} }
} catch (_: Exception) {}
} }
}) })
} }
@ -510,6 +514,7 @@ class MultiHostConcurrencyExecutionTest {
runtime!!.connect() runtime!!.connect()
runtime.gatewayClient.awaitGatewayReady(5000) runtime.gatewayClient.awaitGatewayReady(5000)
try {
val turnId = freshRepo.sendPrompt(sessionId, "Hello after restart") val turnId = freshRepo.sendPrompt(sessionId, "Hello after restart")
assertEquals("t_101", turnId) assertEquals("t_101", turnId)
@ -519,9 +524,10 @@ class MultiHostConcurrencyExecutionTest {
assertEquals("durable_persisted_99", updatedBinding?.durableSessionId) assertEquals("durable_persisted_99", updatedBinding?.durableSessionId)
assertEquals("fresh_runtime_101", updatedBinding?.runtimeSessionId) assertEquals("fresh_runtime_101", updatedBinding?.runtimeSessionId)
assertEquals(BindingState.RUNNING.name, updatedBinding?.state) assertEquals(BindingState.RUNNING.name, updatedBinding?.state)
} finally {
runtime.disconnect() try { runtime.disconnect() } catch (_: Exception) {}
server.shutdown() try { server.shutdown() } catch (_: Exception) {}
}
} }
@Test @Test
@ -540,10 +546,13 @@ class MultiHostConcurrencyExecutionTest {
path.startsWith("/api/ws") || path.startsWith("/ws") -> { path.startsWith("/api/ws") || path.startsWith("/ws") -> {
MockResponse().withWebSocketUpgrade(object : WebSocketListener() { MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
try {
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") 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) { override fun onMessage(webSocket: WebSocket, text: String) {
try {
if (text.contains("session.resume")) { if (text.contains("session.resume")) {
sessionResumeCalled = true sessionResumeCalled = true
if (text.contains("valid_durable_888")) { if (text.contains("valid_durable_888")) {
@ -553,6 +562,7 @@ class MultiHostConcurrencyExecutionTest {
} else if (text.contains("prompt.submit")) { } 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":"a2","result":{"turn_id":"turn_ready_restart"}}""")
} }
} catch (_: Exception) {}
} }
}) })
} }
@ -607,6 +617,7 @@ class MultiHostConcurrencyExecutionTest {
runtime.connect() runtime.connect()
runtime.gatewayClient.awaitGatewayReady(5000) runtime.gatewayClient.awaitGatewayReady(5000)
try {
// Calling sendPrompt must trigger session.resume since in-memory attachment does not exist in new process // 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") val turnId = freshRepo.sendPrompt(sessionId, "Hello after clean restart")
assertEquals("turn_ready_restart", turnId) assertEquals("turn_ready_restart", turnId)
@ -618,9 +629,10 @@ class MultiHostConcurrencyExecutionTest {
assertNotNull(updatedBinding) assertNotNull(updatedBinding)
assertEquals("valid_durable_888", updatedBinding?.durableSessionId) assertEquals("valid_durable_888", updatedBinding?.durableSessionId)
assertEquals("fresh_runtime_777", updatedBinding?.runtimeSessionId) assertEquals("fresh_runtime_777", updatedBinding?.runtimeSessionId)
} finally {
runtime.disconnect() try { runtime.disconnect() } catch (_: Exception) {}
server.shutdown() try { server.shutdown() } catch (_: Exception) {}
}
} }
@Test @Test
@ -638,16 +650,20 @@ class MultiHostConcurrencyExecutionTest {
path.startsWith("/api/ws") || path.startsWith("/ws") -> { path.startsWith("/api/ws") || path.startsWith("/ws") -> {
MockResponse().withWebSocketUpgrade(object : WebSocketListener() { MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
try {
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") 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) { override fun onMessage(webSocket: WebSocket, text: String) {
try {
if (text.contains("session.create")) { if (text.contains("session.create")) {
sessionCreated = true sessionCreated = true
webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"dur_conn_1","session_id":"rt_initial"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"dur_conn_1","session_id":"rt_initial"}}""")
} else if (text.contains("session.resume")) { } 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"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_conn_1","session_id":"rt_after_reconnect"}}""")
} }
} catch (_: Exception) {}
} }
}) })
} }
@ -680,6 +696,7 @@ class MultiHostConcurrencyExecutionTest {
val session = testRepo.createUnifiedSession(title = "Reconnect Test", initialHostId = HermesHostId("host-windows")) val session = testRepo.createUnifiedSession(title = "Reconnect Test", initialHostId = HermesHostId("host-windows"))
val runtime = testConnectionManager.getRuntime(HermesHostId("host-windows"))!! val runtime = testConnectionManager.getRuntime(HermesHostId("host-windows"))!!
try {
// Initial connect // Initial connect
runtime.connect() runtime.connect()
runtime.gatewayClient.awaitGatewayReady(5000) runtime.gatewayClient.awaitGatewayReady(5000)
@ -698,9 +715,10 @@ class MultiHostConcurrencyExecutionTest {
// Reattach after reconnect // Reattach after reconnect
val binding2 = testRepo.ensureAttachedRuntimeSession(session.id, host1Id, runtime) val binding2 = testRepo.ensureAttachedRuntimeSession(session.id, host1Id, runtime)
assertEquals(RuntimeSessionId("rt_after_reconnect"), binding2.runtimeSessionId) assertEquals(RuntimeSessionId("rt_after_reconnect"), binding2.runtimeSessionId)
} finally {
runtime.disconnect() try { runtime.disconnect() } catch (_: Exception) {}
server.shutdown() try { server.shutdown() } catch (_: Exception) {}
}
} }
@Test @Test
@ -717,16 +735,20 @@ class MultiHostConcurrencyExecutionTest {
path.startsWith("/api/ws") || path.startsWith("/ws") -> { path.startsWith("/api/ws") || path.startsWith("/ws") -> {
MockResponse().withWebSocketUpgrade(object : WebSocketListener() { MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
try {
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") 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) { override fun onMessage(webSocket: WebSocket, text: String) {
try {
if (text.contains("session.resume")) { if (text.contains("session.resume")) {
webSocket.send("""{"jsonrpc":"2.0","id":"a1","result":{"stored_session_id":"dur_fail_test","session_id":"rt_fail_test"}}""") 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")) { } else if (text.contains("prompt.submit")) {
// Fail prompt submission with an RPC error // Fail prompt submission with an RPC error
webSocket.send("""{"jsonrpc":"2.0","id":"a2","error":{"code":-32000,"message":"Model overloaded"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a2","error":{"code":-32000,"message":"Model overloaded"}}""")
} }
} catch (_: Exception) {}
} }
}) })
} }
@ -775,6 +797,7 @@ class MultiHostConcurrencyExecutionTest {
runtime.connect() runtime.connect()
runtime.gatewayClient.awaitGatewayReady(5000) runtime.gatewayClient.awaitGatewayReady(5000)
try {
// Attempt sendPrompt which will fail at submitPrompt // Attempt sendPrompt which will fail at submitPrompt
var threw = false var threw = false
try { try {
@ -789,9 +812,10 @@ class MultiHostConcurrencyExecutionTest {
val binding = testSessionDao.getBindingsForSession(session.id.value).find { it.hostId == host1Id.value } val binding = testSessionDao.getBindingsForSession(session.id.value).find { it.hostId == host1Id.value }
assertEquals("msg_baseline", binding?.syncedThroughMessageId) assertEquals("msg_baseline", binding?.syncedThroughMessageId)
assertEquals(BindingState.ERROR.name, binding?.state) assertEquals(BindingState.ERROR.name, binding?.state)
} finally {
runtime.disconnect() try { runtime.disconnect() } catch (_: Exception) {}
server.shutdown() try { server.shutdown() } catch (_: Exception) {}
}
} }
@Test @Test
@ -809,10 +833,13 @@ class MultiHostConcurrencyExecutionTest {
path.startsWith("/api/ws") || path.startsWith("/ws") -> { path.startsWith("/api/ws") || path.startsWith("/ws") -> {
MockResponse().withWebSocketUpgrade(object : WebSocketListener() { MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
try {
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") 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) { override fun onMessage(webSocket: WebSocket, text: String) {
try {
if (text.contains("session.resume")) { if (text.contains("session.resume")) {
// Transient failure: 500 / -32000 Server overloaded // Transient failure: 500 / -32000 Server overloaded
webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":-32000,"message":"Transient server error"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":-32000,"message":"Transient server error"}}""")
@ -820,6 +847,7 @@ class MultiHostConcurrencyExecutionTest {
createCalled = true createCalled = true
webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_forbidden","session_id":"rt_forbidden"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_forbidden","session_id":"rt_forbidden"}}""")
} }
} catch (_: Exception) {}
} }
}) })
} }
@ -872,6 +900,7 @@ class MultiHostConcurrencyExecutionTest {
runtime.connect() runtime.connect()
runtime.gatewayClient.awaitGatewayReady(5000) runtime.gatewayClient.awaitGatewayReady(5000)
try {
var threw = false var threw = false
try { try {
testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime) testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime)
@ -885,9 +914,10 @@ class MultiHostConcurrencyExecutionTest {
// Verify binding was not destroyed // Verify binding was not destroyed
val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value }
assertEquals("dur_preserved_123", binding?.durableSessionId) assertEquals("dur_preserved_123", binding?.durableSessionId)
} finally {
runtime.disconnect() try { runtime.disconnect() } catch (_: Exception) {}
server.shutdown() try { server.shutdown() } catch (_: Exception) {}
}
} }
@Test @Test
@ -905,10 +935,13 @@ class MultiHostConcurrencyExecutionTest {
path.startsWith("/api/ws") || path.startsWith("/ws") -> { path.startsWith("/api/ws") || path.startsWith("/ws") -> {
MockResponse().withWebSocketUpgrade(object : WebSocketListener() { MockResponse().withWebSocketUpgrade(object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
try {
webSocket.send("""{"jsonrpc":"2.0","method":"event","params":{"type":"gateway.ready","payload":{"version":"1.0.0"}}}""") 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) { override fun onMessage(webSocket: WebSocket, text: String) {
try {
if (text.contains("session.resume")) { if (text.contains("session.resume")) {
// Definitively unrecoverable: 404 Session not found // Definitively unrecoverable: 404 Session not found
webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":404,"message":"Session not found"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a1","error":{"code":404,"message":"Session not found"}}""")
@ -916,6 +949,7 @@ class MultiHostConcurrencyExecutionTest {
createCalled = true createCalled = true
webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_new_fresh_999","session_id":"rt_new_fresh_999"}}""") webSocket.send("""{"jsonrpc":"2.0","id":"a2","result":{"stored_session_id":"dur_new_fresh_999","session_id":"rt_new_fresh_999"}}""")
} }
} catch (_: Exception) {}
} }
}) })
} }
@ -968,6 +1002,7 @@ class MultiHostConcurrencyExecutionTest {
runtime.connect() runtime.connect()
runtime.gatewayClient.awaitGatewayReady(5000) runtime.gatewayClient.awaitGatewayReady(5000)
try {
val attachedBinding = testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime) val attachedBinding = testRepo.ensureAttachedRuntimeSession(sessionId, host1Id, runtime)
assertTrue("session.create MUST be called on unrecoverable 404 resume error", createCalled) assertTrue("session.create MUST be called on unrecoverable 404 resume error", createCalled)
@ -978,9 +1013,10 @@ class MultiHostConcurrencyExecutionTest {
val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value } val binding = testSessionDao.getBindingsForSession(sessionId.value).find { it.hostId == host1Id.value }
assertEquals("dur_new_fresh_999", binding?.durableSessionId) assertEquals("dur_new_fresh_999", binding?.durableSessionId)
assertEquals("rt_new_fresh_999", binding?.runtimeSessionId) assertEquals("rt_new_fresh_999", binding?.runtimeSessionId)
} finally {
runtime.disconnect() try { runtime.disconnect() } catch (_: Exception) {}
server.shutdown() try { server.shutdown() } catch (_: Exception) {}
}
} }
@Test @Test

View file

@ -31,6 +31,8 @@ class SessionCreateRaceTest {
val repoScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) val repoScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
val createCallCount = AtomicInteger(0) val createCallCount = AtomicInteger(0)
val firstCreateStarted = CompletableDeferred<Unit>()
val unblockCreate = CompletableDeferred<Unit>()
val mockGatewayClient = mockk<JsonRpcGatewayClient>(relaxed = true) val mockGatewayClient = mockk<JsonRpcGatewayClient>(relaxed = true)
val runtimeScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) val runtimeScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
@ -40,7 +42,8 @@ class SessionCreateRaceTest {
coEvery { mockGatewayClient.awaitGatewayReady(any()) } returns Unit coEvery { mockGatewayClient.awaitGatewayReady(any()) } returns Unit
coEvery { mockGatewayClient.createSession(any(), any()) } coAnswers { coEvery { mockGatewayClient.createSession(any(), any()) } coAnswers {
createCallCount.incrementAndGet() createCallCount.incrementAndGet()
delay(100) // simulate network delay to expose race condition firstCreateStarted.complete(Unit)
unblockCreate.await()
CreateSessionResult( CreateSessionResult(
durableId = DurableSessionId("dur_race_1"), durableId = DurableSessionId("dur_race_1"),
runtimeId = RuntimeSessionId("rt_race_1") runtimeId = RuntimeSessionId("rt_race_1")
@ -71,10 +74,8 @@ class SessionCreateRaceTest {
) )
connectionManager.addHost(host) connectionManager.addHost(host)
delay(50)
val session = repository.createUnifiedSession(title = "Race Test Session", initialHostId = hostId) val session = repository.createUnifiedSession(title = "Race Test Session", initialHostId = hostId)
delay(50)
// Launch 2 concurrent sendPrompt calls for the same session & host // Launch 2 concurrent sendPrompt calls for the same session & host
val deferred1 = async(Dispatchers.Default) { val deferred1 = async(Dispatchers.Default) {
@ -84,6 +85,11 @@ class SessionCreateRaceTest {
repository.sendPrompt(session.id, "Prompt from coroutine 2") 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) awaitAll(deferred1, deferred2)
assertEquals("Exactly one session.create must be invoked for concurrent sendPrompt calls", 1, createCallCount.get()) assertEquals("Exactly one session.create must be invoked for concurrent sendPrompt calls", 1, createCallCount.get())

View file

@ -9,12 +9,11 @@ import app.hermes.mobile.core.storage.FakeHostDao
import app.hermes.mobile.core.storage.FakeUnifiedSessionDao import app.hermes.mobile.core.storage.FakeUnifiedSessionDao
import app.hermes.mobile.core.storage.HostBindingEntity import app.hermes.mobile.core.storage.HostBindingEntity
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.flow.first import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.runBlocking
import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals 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 * and thinking deltas, tools and thinking are strictly bound to their respective host and
* explicit messageId, avoiding false attribution. * explicit messageId, avoiding false attribution.
*/ */
@OptIn(ExperimentalCoroutinesApi::class)
class ToolAttributionTest { class ToolAttributionTest {
@Test @Test
fun testTwoConcurrentHostsAttributionAndThinkingIsolation() = runBlocking { fun testTwoConcurrentHostsAttributionAndThinkingIsolation() = runTest {
val testDispatcher = StandardTestDispatcher(testScheduler)
val hostAId = HermesHostId("host-a") val hostAId = HermesHostId("host-a")
val hostBId = HermesHostId("host-b") val hostBId = HermesHostId("host-b")
@ -40,14 +41,14 @@ class ToolAttributionTest {
val hostDao = FakeHostDao() val hostDao = FakeHostDao()
val sessionDao = FakeUnifiedSessionDao() val sessionDao = FakeUnifiedSessionDao()
val tokenVault = InMemoryTokenVault() val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) val scope = CoroutineScope(SupervisorJob() + testDispatcher)
val connectionManager = HermesConnectionManager( val connectionManager = HermesConnectionManager(
hostDao = hostDao, hostDao = hostDao,
tokenVault = tokenVault, tokenVault = tokenVault,
scope = scope, scope = scope,
runtimeFactory = { parentScope, h -> runtimeFactory = { parentScope, h ->
val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default) val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + testDispatcher)
HermesHostRuntime( HermesHostRuntime(
initialHost = h, initialHost = h,
gatewayClient = JsonRpcGatewayClient(scope = childScope), gatewayClient = JsonRpcGatewayClient(scope = childScope),
@ -65,7 +66,7 @@ class ToolAttributionTest {
connectionManager.addHost(hostA) connectionManager.addHost(hostA)
connectionManager.addHost(hostB) connectionManager.addHost(hostB)
delay(50) testScheduler.advanceUntilIdle()
val session = repository.createUnifiedSession(title = "Dual Host Attribution Test", initialHostId = hostAId) val session = repository.createUnifiedSession(title = "Dual Host Attribution Test", initialHostId = hostAId)
val rtSessionA = "rt_session_host_a" val rtSessionA = "rt_session_host_a"
@ -92,6 +93,7 @@ class ToolAttributionTest {
state = BindingState.RUNNING.name state = BindingState.RUNNING.name
) )
) )
testScheduler.advanceUntilIdle()
val runtimeA = connectionManager.getRuntime(hostAId)!! val runtimeA = connectionManager.getRuntime(hostAId)!!
val runtimeB = connectionManager.getRuntime(hostBId)!! val runtimeB = connectionManager.getRuntime(hostBId)!!
@ -127,7 +129,7 @@ class ToolAttributionTest {
}) })
}.toString()) }.toString())
delay(30) testScheduler.advanceUntilIdle()
// 3. Host A streams thinking delta for msgA // 3. Host A streams thinking delta for msgA
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject { runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
@ -187,7 +189,7 @@ class ToolAttributionTest {
}) })
}.toString()) }.toString())
delay(30) testScheduler.advanceUntilIdle()
// 7. Update tool progress and completion for host A // 7. Update tool progress and completion for host A
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject { runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
@ -246,7 +248,7 @@ class ToolAttributionTest {
}) })
}.toString()) }.toString())
delay(50) testScheduler.advanceUntilIdle()
val messages = repository.getSessionMessages(session.id).value val messages = repository.getSessionMessages(session.id).value
val msgA = messages.find { it.id == msgAId } val msgA = messages.find { it.id == msgAId }
@ -271,21 +273,22 @@ class ToolAttributionTest {
} }
@Test @Test
fun testThinkingDeltaWithExplicitMessageIdDoesNotFallBackToLastAssistant() = runBlocking { fun testThinkingDeltaWithExplicitMessageIdDoesNotFallBackToLastAssistant() = runTest {
val testDispatcher = StandardTestDispatcher(testScheduler)
val hostAId = HermesHostId("host-a") val hostAId = HermesHostId("host-a")
val hostA = HermesHost(id = hostAId, displayName = "Host A", baseUrl = "http://host-a:9119") val hostA = HermesHost(id = hostAId, displayName = "Host A", baseUrl = "http://host-a:9119")
val hostDao = FakeHostDao() val hostDao = FakeHostDao()
val sessionDao = FakeUnifiedSessionDao() val sessionDao = FakeUnifiedSessionDao()
val tokenVault = InMemoryTokenVault() val tokenVault = InMemoryTokenVault()
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) val scope = CoroutineScope(SupervisorJob() + testDispatcher)
val connectionManager = HermesConnectionManager( val connectionManager = HermesConnectionManager(
hostDao = hostDao, hostDao = hostDao,
tokenVault = tokenVault, tokenVault = tokenVault,
scope = scope, scope = scope,
runtimeFactory = { parentScope, h -> runtimeFactory = { parentScope, h ->
val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + Dispatchers.Default) val childScope = CoroutineScope(SupervisorJob(parentScope.coroutineContext[Job]) + testDispatcher)
HermesHostRuntime( HermesHostRuntime(
initialHost = h, initialHost = h,
gatewayClient = JsonRpcGatewayClient(scope = childScope), gatewayClient = JsonRpcGatewayClient(scope = childScope),
@ -302,7 +305,7 @@ class ToolAttributionTest {
) )
connectionManager.addHost(hostA) connectionManager.addHost(hostA)
delay(50) testScheduler.advanceUntilIdle()
val session = repository.createUnifiedSession(title = "Message Targeting Test", initialHostId = hostAId) val session = repository.createUnifiedSession(title = "Message Targeting Test", initialHostId = hostAId)
val rtSessionA = "rt_session_host_a" val rtSessionA = "rt_session_host_a"
@ -317,6 +320,7 @@ class ToolAttributionTest {
state = BindingState.RUNNING.name state = BindingState.RUNNING.name
) )
) )
testScheduler.advanceUntilIdle()
val runtimeA = connectionManager.getRuntime(hostAId)!! val runtimeA = connectionManager.getRuntime(hostAId)!!
@ -351,7 +355,7 @@ class ToolAttributionTest {
}) })
}.toString()) }.toString())
delay(30) testScheduler.advanceUntilIdle()
// Send thinking delta explicitly targeted at msg1Id // Send thinking delta explicitly targeted at msg1Id
runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject { runtimeA.gatewayClient.handleIncomingMessage(buildJsonObject {
@ -367,7 +371,7 @@ class ToolAttributionTest {
}) })
}.toString()) }.toString())
delay(30) testScheduler.advanceUntilIdle()
val messages = repository.getSessionMessages(session.id).value val messages = repository.getSessionMessages(session.id).value
val msg1 = messages.find { it.id == msg1Id } val msg1 = messages.find { it.id == msg1Id }

View file

@ -187,14 +187,14 @@ pub fn validate_payload(payload: &PairingPayloadV1, current_time: u64) -> Result
payload.host payload.host
))); )));
} }
if !trimmed_host.starts_with('[') || !trimmed_host.ends_with(']') { if (!trimmed_host.starts_with('[') || !trimmed_host.ends_with(']'))
if trimmed_host.contains(':') { && trimmed_host.contains(':')
{
return Err(PairingError::InvalidHost(format!( return Err(PairingError::InvalidHost(format!(
"Host '{}' contains forbidden colon delimiter outside IPv6 brackets", "Host '{}' contains forbidden colon delimiter outside IPv6 brackets",
payload.host payload.host
))); )));
} }
}
// 6. Port: 1..=65535 (u16 is <= 65535, port 0 is invalid) // 6. Port: 1..=65535 (u16 is <= 65535, port 0 is invalid)
if payload.port == 0 { if payload.port == 0 {

View file

@ -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')"
]
}
}