feat(profiles): phase 2 — atomic switch coordinator (#824)

* feat(profiles): ProfileSwitchCoordinator — atomic re-home on profile switch (phase 2)

* fix(ws): deterministic reconnect backoff in tests (testAutoReconnect flake)

* test(ws): harden testAutoReconnect against CI latency (stub token refresh, generous latches)

* test(profiles): remove static Dispatchers mock from coordinator test — it bled into HermesWsClientTest in full-suite runs

---------

Co-authored-by: M57 <hy4ri@users.noreply.github.com>
This commit is contained in:
M57andM57 authored and GitHub committed 2026-08-06 15:07:09 +03:00
1 parent 398268d105
commit 531219cdb5
8 files changed
+241 -84

No files matched your search

@@ -0,0 +1,50 @@
package com.m57.hermescontrol.data.session
import com.m57.hermescontrol.data.local.AuthManager
import com.m57.hermescontrol.data.model.SetActiveProfileRequest
import com.m57.hermescontrol.data.remote.ApiClient
import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.remote.safeApiCall
import com.m57.hermescontrol.data.ws.HermesWsClient
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.withContext
/**
* The single flow that performs a profile switch — the mobile equivalent of
* desktop's re-home (``requestFreshSession`` + socket swap). Every surface
* that switches profiles goes through here, so the switch is atomic instead
* of a pile of scattered patches.
*
* Order matters:
* 1. Flip the server's sticky active profile (REST).
* 2. Persist the LOCAL selection — the REST interceptor (``?profile=``) and
* the WS params injector (``params.profile``) now scope everything to the
* new profile. The per-server token fallback (phase 1) keeps auth intact:
* no re-login, restart-safe.
* 3. Emit [switched] BEFORE the socket re-dial, so chat wipes its stale
* conversation first — when the reconnected socket delivers
* ``gateway.ready``, ``handleGatewayReady`` sees no open session and
* auto-creates a FRESH session in the new profile (desktop parity).
* 4. Re-dial the WebSocket so the gateway re-homes chat to the new profile.
*/
object ProfileSwitchCoordinator {
private val _switched = MutableSharedFlow<String>(extraBufferCapacity = 1)
val switched: SharedFlow<String> = _switched.asSharedFlow()
suspend fun switchProfile(name: String): NetworkResult<Unit> {
val result =
withContext(Dispatchers.IO) {
safeApiCall { ApiClient.hermesApi.setActiveProfile(SetActiveProfileRequest(name)) }
}
if (result !is NetworkResult.Success) return result
AuthManager.setSelectedProfileId(name)
_switched.emit(name)
HermesWsClient.disconnect()
HermesWsClient.connect()
return result
}
}
@@ -65,9 +65,21 @@ object HermesWsClient {
// ── Backoff settings ─────────────────────────────────────────────────
private const val INITIAL_BACKOFF_MS = 1_000L
private var initialBackoffMs = 1_000L
private const val MAX_BACKOFF_MS = 30_000L
private const val BACKOFF_MULTIPLIER = 2.0
/**
* Test hook: the initial reconnect backoff, overridable so reconnect
* tests are deterministic instead of racing real time (CI runners
* routinely exceed fixed latch windows). Production keeps 1s.
*/
@VisibleForTesting
internal fun setReconnectBackoffForTest(initialMillis: Long) {
initialBackoffMs = initialMillis
currentBackoff = initialMillis
}
private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024
private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L
@@ -89,7 +101,7 @@ object HermesWsClient {
private var closingSocket: WebSocket? = null
@Volatile
private var currentBackoff = INITIAL_BACKOFF_MS
private var currentBackoff = initialBackoffMs
private val wsScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@@ -260,7 +272,7 @@ object HermesWsClient {
// therefore allowed to leave the terminal AUTH_EXPIRED state.
intentionalClose.set(false)
_connectionStatus.value = ConnectionStatus.CONNECTING
currentBackoff = INITIAL_BACKOFF_MS
currentBackoff = initialBackoffMs
true
}
if (socketToCancel != null) {
@@ -282,7 +294,7 @@ object HermesWsClient {
return
}
Log.d(TAG, "Default network changed — reconnecting WebSocket")
currentBackoff = INITIAL_BACKOFF_MS
currentBackoff = initialBackoffMs
reconnectJob?.cancel()
reconnectJob = null
outboundDrainJob?.cancel()
@@ -796,7 +808,7 @@ object HermesWsClient {
Log.i(TAG, "WebSocket opened")
connected.set(true)
_connectionStatus.value = ConnectionStatus.CONNECTED
currentBackoff = INITIAL_BACKOFF_MS
currentBackoff = initialBackoffMs
startHealthTracking()
while (true) {
@@ -24,6 +24,7 @@ import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.remote.OkHttpProvider
import com.m57.hermescontrol.data.remote.safeApiCall
import com.m57.hermescontrol.data.session.ActiveSessionHolder
import com.m57.hermescontrol.data.session.ProfileSwitchCoordinator
import com.m57.hermescontrol.data.ws.CommandBlocklist
import com.m57.hermescontrol.data.ws.CommandCatalog
import com.m57.hermescontrol.data.ws.ConnectionStatus
@@ -39,7 +40,6 @@ import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.drop
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
@@ -505,16 +505,15 @@ class ChatViewModel(
}
}
}
// Desktop parity (requestFreshSession): when the selected profile
// changes, wipe the open conversation and reload the session list so
// the previous profile's context never leaks into the new profile's
// chat (the gateway re-homes on profile switch).
// Desktop parity (requestFreshSession): when the profile switch
// coordinator fires, wipe the open conversation. The re-dialed socket
// then delivers gateway.ready → handleGatewayReady loads the new
// profile's session list and auto-creates a FRESH session, so the
// previous profile's context never leaks into the new profile's chat.
viewModelScope.launch {
AuthManager.selectedProfileFlow
.drop(1)
ProfileSwitchCoordinator.switched
.collect { _ ->
resetSessionState(sessionId = null, title = "Hermes", isLoading = true)
loadSessions()
}
}
if (wsClient.connectionStatus.value == ConnectionStatus.CONNECTED) {
@@ -11,7 +11,6 @@ import com.m57.hermescontrol.data.model.PinnedModel
import com.m57.hermescontrol.data.model.ProfileDescribeAutoRequest
import com.m57.hermescontrol.data.model.ProfileInfo
import com.m57.hermescontrol.data.model.RenameProfileRequest
import com.m57.hermescontrol.data.model.SetActiveProfileRequest
import com.m57.hermescontrol.data.model.Skill
import com.m57.hermescontrol.data.model.UpdateProfileDescriptionRequest
import com.m57.hermescontrol.data.model.UpdateProfileModelRequest
@@ -19,7 +18,7 @@ import com.m57.hermescontrol.data.model.UpdateProfileSoulRequest
import com.m57.hermescontrol.data.remote.ApiClient
import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.remote.safeApiCall
import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.session.ProfileSwitchCoordinator
import com.m57.hermescontrol.ui.common.ToastHost
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
@@ -106,30 +105,11 @@ class ProfilesViewModel :
_uiState.update { it.copy(activeProfileName = name) }
viewModelScope.launch {
val result =
withContext(Dispatchers.IO) {
safeApiCall { ApiClient.hermesApi.setActiveProfile(SetActiveProfileRequest(name)) }
}
when (result) {
// The coordinator performs the whole re-home atomically: server
// flip → local selection persist → switch broadcast → socket
// re-dial (desktop's requestFreshSession + socket swap).
when (val result = ProfileSwitchCoordinator.switchProfile(name)) {
is NetworkResult.Success -> {
// Persist the LOCAL selection so ProfileScopeInterceptor scopes
// every management REST call (?profile=) to the new profile,
// and re-home the WebSocket so the live gateway follows too
// (mirrors desktop's re-home-on-switch).
//
// Token inheritance: the app keys tokens per profile
// (token_<profileId>) because connections can point at
// different dashboards, but profiles listed on THIS screen
// are all served by the SAME connected dashboard — so a
// profile created/selected in-app inherits the current
// connection's token instead of forcing a re-login.
val currentToken = AuthManager.getToken()
if (currentToken != null && AuthManager.getProfileToken(name) == null) {
AuthManager.setProfileToken(name, currentToken)
}
AuthManager.setSelectedProfileId(name)
HermesWsClient.disconnect()
HermesWsClient.connect()
_uiState.update { it.copy(toastMessage = "Switched to profile $name") }
loadProfiles()
}
@@ -0,0 +1,116 @@
package com.m57.hermescontrol.data.session
import com.m57.hermescontrol.data.local.AuthManager
import com.m57.hermescontrol.data.model.SetActiveProfileRequest
import com.m57.hermescontrol.data.remote.ApiClient
import com.m57.hermescontrol.data.remote.HermesApiService
import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.ws.HermesWsClient
import io.mockk.Ordering
import io.mockk.coEvery
import io.mockk.coVerify
import io.mockk.every
import io.mockk.mockk
import io.mockk.mockkObject
import io.mockk.unmockkAll
import io.mockk.verify
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import okhttp3.ResponseBody.Companion.toResponseBody
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import retrofit2.Response
/**
* The switch coordinator is the single atomic profile-switch flow — every
* surface (Profiles screen, future quick-switch) routes through it. These
* tests pin the ORDER of operations, because chat's fresh-session behavior
* depends on the switch broadcast landing BEFORE the socket re-dial.
*/
@OptIn(ExperimentalCoroutinesApi::class)
class ProfileSwitchCoordinatorTest {
private lateinit var mockApi: HermesApiService
@Before
fun setUp() {
// NOTE: no mockkStatic(Dispatchers) here — a static Dispatchers mock
// bleeds into later test classes in the same JVM (it hijacks
// Dispatchers.IO for HermesWsClient's reconnect coroutines), which
// deterministically broke HermesWsClientTest.testAutoReconnect in
// full-suite runs. Real Dispatchers.IO is fine: the network layer is
// mocked, so withContext(Dispatchers.IO) hops are instant.
mockkObject(ApiClient)
mockApi = mockk(relaxed = true)
every { ApiClient.hermesApi } returns mockApi
mockkObject(AuthManager)
every { AuthManager.setSelectedProfileId(any()) } returns Unit
mockkObject(HermesWsClient)
every { HermesWsClient.disconnect() } returns Unit
every { HermesWsClient.connect() } returns Unit
}
@After
fun tearDown() {
unmockkAll()
}
@Test
fun `success flips server then persists selection then re-dials socket`() =
runTest {
coEvery { mockApi.setActiveProfile(any()) } returns Response.success(Unit)
val result = ProfileSwitchCoordinator.switchProfile("meow")
assertTrue(result is NetworkResult.Success)
coVerify { mockApi.setActiveProfile(SetActiveProfileRequest("meow")) }
verify(ordering = Ordering.SEQUENCE) {
AuthManager.setSelectedProfileId("meow")
HermesWsClient.disconnect()
HermesWsClient.connect()
}
}
@Test
fun `success broadcasts the switch so chat wipes before the re-dial`() =
runTest {
coEvery { mockApi.setActiveProfile(any()) } returns Response.success(Unit)
// Subscribe BEFORE the switch — exactly how ChatViewModel does it.
val received = Channel<String>(Channel.UNLIMITED)
backgroundScope.launch {
ProfileSwitchCoordinator.switched.collect { received.send(it) }
}
runCurrent()
ProfileSwitchCoordinator.switchProfile("meow")
runCurrent()
// The broadcast is buffered with capacity 1, so chat observes the
// wipe BEFORE gateway.ready arrives on the re-dialed socket and
// auto-creates the fresh session (desktop requestFreshSession).
assertEquals("meow", received.tryReceive().getOrNull())
}
@Test
fun `failure touches nothing`() =
runTest {
coEvery { mockApi.setActiveProfile(any()) } returns errorResponse(500)
val result = ProfileSwitchCoordinator.switchProfile("meow")
assertTrue(result is NetworkResult.Failure)
verify(exactly = 0) { AuthManager.setSelectedProfileId(any()) }
verify(exactly = 0) { HermesWsClient.disconnect() }
verify(exactly = 0) { HermesWsClient.connect() }
}
private fun <T> errorResponse(code: Int): Response<T> = Response.error(code, "{}".toResponseBody(null))
}
@@ -907,6 +907,15 @@ class HermesWsClientTest {
@Test
fun testAutoReconnect() {
every { AuthManager.isAutoReconnect() } returns true
// Deterministic reconnect: zero backoff removes the real-time race
// that flaked this test on loaded CI runners (1s backoff + fixed
// 5-6s latch windows routinely overshoot under parallel load).
HermesWsClient.setReconnectBackoffForTest(0L)
// The reconnect path refreshes the WS token over the network before
// opening the socket. Stub it so no real HTTP call sits inside the
// test's timing window (CI network latency was the dominant flake).
mockkObject(DashboardSessionTokenRefresher)
every { DashboardSessionTokenRefresher.fetch(any(), any()) } returns null
var serverSocket1: WebSocket? = null
var serverSocket2: WebSocket? = null
@@ -944,8 +953,10 @@ class HermesWsClientTest {
HermesWsClient.connect()
assertTrue("Failed initial connection", connect1Latch.await(5, TimeUnit.SECONDS))
runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } }
assertTrue("Failed initial connection", connect1Latch.await(15, TimeUnit.SECONDS))
runBlocking {
withTimeout(15_000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } }
}
assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value)
// Force server to close socket 1 to trigger reconnect
@@ -954,13 +965,17 @@ class HermesWsClientTest {
// Wait for status to become RECONNECTING
runBlocking {
withTimeout(
5000,
15_000,
) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.RECONNECTING } }
}
// The client should now attempt to reconnect after initial backoff (1000ms)
// Wait for the second connection to hit the server
assertTrue("Failed to reconnect", connect2Latch.await(6, TimeUnit.SECONDS))
// Wait for the second connection to hit the server. Generous ceiling:
// loaded CI runners routinely stretch short wall-clock windows.
assertTrue("Failed to reconnect", connect2Latch.await(30, TimeUnit.SECONDS))
// Restore production backoff so later tests see the default.
HermesWsClient.setReconnectBackoffForTest(1_000L)
}
// ── TEST-10: WS reconnect state recovery ────────────────────────────
@@ -968,6 +983,9 @@ class HermesWsClientTest {
@Test
fun testBackoffResetsOnSuccessfulConnect() {
every { AuthManager.isAutoReconnect() } returns true
// Pin the initial backoff so this test's reset assertion is
// deterministic regardless of what earlier tests configured.
HermesWsClient.setReconnectBackoffForTest(1_000L)
val serverLatch = CountDownLatch(1)
mockWebServer.enqueue(
@@ -15,6 +15,7 @@ import com.m57.hermescontrol.data.remote.GatewayFile
import com.m57.hermescontrol.data.remote.GatewayFileClient
import com.m57.hermescontrol.data.remote.GatewayFileResult
import com.m57.hermescontrol.data.session.ActiveSessionHolder
import com.m57.hermescontrol.data.session.ProfileSwitchCoordinator
import com.m57.hermescontrol.data.ws.ConnectionStatus
import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.ws.JsonRpcError
@@ -63,7 +64,7 @@ class ChatViewModelTest {
private val testDispatcher = StandardTestDispatcher()
private val mockEventsFlow = MutableSharedFlow<WsEvent>(extraBufferCapacity = 64)
private val mockConnectionStatus = MutableStateFlow(ConnectionStatus.DISCONNECTED)
private val mockSelectedProfileFlow = MutableStateFlow<String?>(null)
private val mockSwitchFlow = MutableSharedFlow<String>(extraBufferCapacity = 8)
private lateinit var app: Application
private lateinit var fakeRepo: FakeChatPersistenceRepository
@@ -111,7 +112,8 @@ class ChatViewModelTest {
every { AuthManager.getToken() } returns "test-token"
every { AuthManager.getBaseUrl() } returns "http://test.local/"
every { AuthManager.getSelectedProfileId() } returns null
every { AuthManager.selectedProfileFlow } returns mockSelectedProfileFlow
mockkObject(ProfileSwitchCoordinator)
every { ProfileSwitchCoordinator.switched } returns mockSwitchFlow
every { AuthManager.isTypingEffectEnabled() } returns true
every { AuthManager.getTypingEffectDelayMs() } returns 30
every { AuthManager.isAutoReconnect() } returns false
@@ -244,21 +246,20 @@ class ChatViewModelTest {
}
@Test
fun profileSwitch_resetsOpenSessionAndReloadsSessionList() =
fun profileSwitch_wipesOpenSessionForFreshStart() =
runTest {
val (viewModel, sessionId) = createViewModelWithSession()
assertEquals(sessionId, viewModel.uiState.value.currentSessionId)
// Selected profile changes (switch in ProfilesScreen) → chat must
// wipe the stale conversation and reload the session list
// (desktop's requestFreshSession parity — the gateway re-homes
// on profile switch and the old session id is no longer valid).
mockSelectedProfileFlow.value = "meow"
// Profile switch fires → chat wipes the stale conversation. The
// re-dialed socket's gateway.ready then auto-creates a FRESH
// session in the new profile (handleGatewayReady, desktop
// requestFreshSession parity).
mockSwitchFlow.emit("meow")
advanceUntilIdle()
assertNull(viewModel.uiState.value.currentSessionId)
assertEquals("Hermes", viewModel.uiState.value.chatTitle)
verify(atLeast = 1) { HermesWsClient.send(WsMethods.SESSION_LIST, any(), any()) }
}
@Test
@@ -13,6 +13,9 @@ import com.m57.hermescontrol.data.model.ProfilesResponse
import com.m57.hermescontrol.data.model.RenameProfileRequest
import com.m57.hermescontrol.data.remote.ApiClient
import com.m57.hermescontrol.data.remote.HermesApiService
import com.m57.hermescontrol.data.remote.NetworkError
import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.session.ProfileSwitchCoordinator
import com.m57.hermescontrol.data.ws.HermesWsClient
import io.mockk.coEvery
import io.mockk.coVerify
@@ -21,7 +24,6 @@ import io.mockk.mockk
import io.mockk.mockkObject
import io.mockk.mockkStatic
import io.mockk.unmockkAll
import io.mockk.verify
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.StandardTestDispatcher
@@ -92,6 +94,9 @@ class ProfilesViewModelTest {
every { HermesWsClient.disconnect() } returns Unit
every { HermesWsClient.connect() } returns Unit
mockkObject(ProfileSwitchCoordinator)
coEvery { ProfileSwitchCoordinator.switchProfile(any()) } returns NetworkResult.Success(Unit)
mockkObject(ApiClient)
mockApi = mockk(relaxed = true)
every { ApiClient.hermesApi } returns mockApi
@@ -283,52 +288,28 @@ class ProfilesViewModelTest {
}
@Test
fun `selectActiveProfile success persists selection and rehomes websocket`() {
coEvery { mockApi.setActiveProfile(any()) } returns Response.success(Unit)
fun `selectActiveProfile success delegates to coordinator and reloads`() {
coEvery { ProfileSwitchCoordinator.switchProfile("work") } returns NetworkResult.Success(Unit)
val vm = createViewModel()
vm.selectActiveProfile("work")
testDispatcher.scheduler.advanceUntilIdle()
// Local selection persisted → ProfileScopeInterceptor now scopes all
// management REST calls (?profile=work) to the new profile.
assertEquals("work", storedSelectedProfile)
// Token inherited from the current connection so switching to a
// same-dashboard profile doesn't force a re-login (token_<profile>
// is keyed per profile and a freshly-created profile has none).
assertEquals("tok-abc", storedProfileToken)
// WebSocket re-homed so the live gateway follows the new profile.
verify { HermesWsClient.disconnect() }
verify { HermesWsClient.connect() }
coVerify { ProfileSwitchCoordinator.switchProfile("work") }
assertTrue(vm.uiState.value.toastMessage!!.contains("Switched to profile work"))
coVerify { mockApi.getProfiles() }
}
@Test
fun `selectActiveProfile keeps existing token for the target profile`() {
storedProfileToken = "tok-meow"
coEvery { mockApi.setActiveProfile(any()) } returns Response.success(Unit)
val vm = createViewModel()
vm.selectActiveProfile("meow")
testDispatcher.scheduler.advanceUntilIdle()
// A profile that already has its own token (e.g. connected elsewhere)
// must NOT be clobbered by inheritance.
assertEquals("tok-meow", storedProfileToken)
assertEquals("meow", storedSelectedProfile)
}
@Test
fun `selectActiveProfile failure rolls back and keeps local selection`() {
coEvery { mockApi.setActiveProfile(any()) } returns errorResponse(500)
fun `selectActiveProfile failure rolls back optimistic state`() {
coEvery {
ProfileSwitchCoordinator.switchProfile("work")
} returns NetworkResult.Failure(NetworkError.Http(code = 500, message = "boom"))
val vm = createViewModel()
vm.selectActiveProfile("work")
testDispatcher.scheduler.advanceUntilIdle()
assertNull(storedSelectedProfile)
assertNull(storedProfileToken)
assertNull(vm.uiState.value.activeProfileName)
assertTrue(vm.uiState.value.toastMessage!!.contains("Failed to switch profile"))
}