feat(ws): consume gateway change events (issue #784) (#837)

* feat(ws): consume gateway change events (cron/sessions/platforms/pairing) for targeted refresh (issue #784)

* test(chat): complete catch-all request stub — kill frozen-collector capture race (slash-focus flake, 4645a2b CI)

---------

Co-authored-by: M57 <hy4ri@users.noreply.github.com>
This commit is contained in:
M57andM57 authored and GitHub committed 2026-08-07 13:48:17 +03:00
1 parent 7d60057156
commit f34bfa7cf8
15 files changed
+320 -4

No files matched your search

@@ -0,0 +1,36 @@
package com.m57.hermescontrol.data.ws
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
/**
* Fan-out hub for gateway change events ([WsEvent.ChangeEvent]).
*
* [HermesWsClient] forwards the change events it parses into this hub; screens
* subscribe here instead of touching the [HermesWsClient] singleton directly.
*
* Why a separate hub: [HermesWsClient] is an Android-touching object whose real
* static init must never run inside unit tests (it poisons every later
* `mockkObject(HermesWsClient)` in the JVM). This hub is pure Kotlin — safe to
* subscribe to from a ViewModel's `init` in any environment. When the client is
* mocked in tests, nothing forwards into the hub and subscribers simply sit
* idle, which is also the correct behavior for backends without
* `change_events` (issue #784).
*/
object ChangeEventHub {
private val _events =
MutableSharedFlow<WsEvent.ChangeEvent>(
extraBufferCapacity = 16,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
/** Collect this from ViewModels to react to gateway change events. */
val events: SharedFlow<WsEvent.ChangeEvent> = _events.asSharedFlow()
/** Non-suspending emit — never blocks the WS event loop. */
fun emit(event: WsEvent.ChangeEvent) {
_events.tryEmit(event)
}
}
@@ -0,0 +1,20 @@
package com.m57.hermescontrol.data.ws
/**
* Change-event vocabulary for [WsEvent.ChangeEvent].
*
* The gateway advertises `change_events: true` in the `gateway.ready`
* handshake and broadcasts these when its on-disk signatures move
* (`tui_gateway/server.py` `_CHANGE_WATCHES`), so clients can demote
* blind polling to a slow backstop. Backends without the feature never
* broadcast — consumers simply stay quiet.
*
* `pet.changed` is deliberately absent: mobile has no pet feature, so it
* is neither parsed nor consumed.
*/
object ChangeEvents {
const val CRON = "cron.changed"
const val SESSIONS = "sessions.changed"
const val PLATFORMS = "platforms.changed"
const val PAIRING = "pairing.changed"
}
@@ -151,6 +151,13 @@ object EventParser {
WsEvent.BackgroundComplete(payload)
}
// Change events (issue #784): gateway watches on-disk signatures
// and broadcasts these so screens can refresh on change. pet.changed
// intentionally absent — mobile has no pet feature.
"cron.changed", "sessions.changed", "platforms.changed", "pairing.changed" -> {
WsEvent.ChangeEvent(eventType, payload)
}
"review.summary" -> {
val text = (payload?.get("text") as? String)?.trim() ?: ""
WsEvent.ReviewSummary(text, sessionId)
@@ -24,6 +24,7 @@ import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.filterIsInstance
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.shareIn
@@ -237,6 +238,14 @@ object HermesWsClient {
}
}
}
// Forward parsed change events (issue #784) to the hub so screens can
// refresh on change without touching this singleton — its real static
// init must never run inside unit tests (see ChangeEventHub).
wsScope.launch {
events
.filterIsInstance<WsEvent.ChangeEvent>()
.collect { ChangeEventHub.emit(it) }
}
}
// ── Connection helpers ────────────────────────────────────────────────
@@ -216,6 +216,22 @@ sealed class WsEvent {
val message: String?,
) : WsEvent()
// ── Gateway change events ───────────────────────────────────────────
/**
* Backend "something changed" broadcast. The gateway advertises
* `change_events: true` in the `gateway.ready` handshake and emits one of
* these when its on-disk signatures move (see [ChangeEvents] for the
* vocabulary), so screens can refresh on change instead of blind-polling.
* Backends without the feature simply never broadcast — consumers stay
* quiet. `pet.changed` exists on the backend but is intentionally NOT
* parsed: mobile has no pet feature.
*/
data class ChangeEvent(
val type: String,
val data: Map<String, Any?>? = null,
) : WsEvent()
// ── Background job completion ──────────────────────────────────────
/**
@@ -9,7 +9,9 @@ import com.m57.hermescontrol.data.model.TelegramOnboardingStartRequest
import com.m57.hermescontrol.data.model.TelegramOnboardingStartResponse
import com.m57.hermescontrol.data.remote.ApiClient
import com.m57.hermescontrol.data.remote.safeApiCall
import com.m57.hermescontrol.data.ws.ChangeEvents
import com.m57.hermescontrol.ui.common.ToastHost
import com.m57.hermescontrol.ui.common.refreshOnChange
import com.m57.hermescontrol.ui.common.safeLaunchLoad
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
@@ -60,6 +62,24 @@ class ChannelsViewModel :
private var launchJob: Job? = null
init {
// Issue #784: gateway broadcasts platforms.changed — refresh silently
// (no spinner) instead of waiting for a manual pull.
refreshOnChange(
eventType = ChangeEvents.PLATFORMS,
apiCall = { safeApiCall { ApiClient.hermesApi.getMessagingPlatforms() } },
onSuccess = { data ->
_uiState.update {
it.copy(
platforms = data.platforms.orEmpty(),
gatewayStartCommand = data.gatewayStartCommand,
envPath = data.envPath,
)
}
},
)
}
fun loadPlatforms() {
safeLaunchLoad(
apiCall = { safeApiCall { ApiClient.hermesApi.getMessagingPlatforms() } },
@@ -130,6 +130,9 @@ object ChatWsEventReducer {
// ReactionEvent is handled by the ViewModel — purely cosmetic animation
is WsEvent.ReactionEvent -> ReducerResult(state = state, streamingState = streamingState)
// Change events (issue #784) are consumed by their screens' ViewModels
is WsEvent.ChangeEvent -> ReducerResult(state = state, streamingState = streamingState)
}
// ── GatewayReady ──────────────────────────────────────────────────
@@ -3,7 +3,9 @@ package com.m57.hermescontrol.ui.common
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.ws.ChangeEventHub
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.launch
interface ToastHost {
@@ -31,3 +33,41 @@ inline fun <T> ViewModel.safeLaunchLoad(
}
}
}
/**
* Collect a gateway change event ([com.m57.hermescontrol.data.ws.ChangeEvents])
* from [ChangeEventHub] and run a **silent** refresh: no loading spinner, no
* error surface — stale data stays in place on failure. At most one request
* runs at a time; events arriving mid-flight are skipped, so a
* `sessions.changed` burst during a long streaming turn (the gateway floors
* those at 2s) cannot pile up requests.
*
* Call from `init`. Backends without `change_events` never broadcast, and in
* unit tests the hub is never fed (HermesWsClient is mocked or uninitialized),
* so this is a no-op there — exactly the "slow backstop" contract of issue
* #784.
*/
inline fun <T> ViewModel.refreshOnChange(
eventType: String,
crossinline apiCall: suspend () -> NetworkResult<T>,
crossinline onSuccess: (T) -> Unit,
): Job {
var refreshInFlight = false
return viewModelScope.launch {
ChangeEventHub.events
.filter { it.type == eventType }
.collect { _ ->
if (refreshInFlight) return@collect
refreshInFlight = true
try {
// No Dispatchers.IO hop — see safeLaunchLoad.
val result = apiCall()
if (result is NetworkResult.Success) {
onSuccess(result.data)
}
} finally {
refreshInFlight = false
}
}
}
}
@@ -11,8 +11,10 @@ import com.m57.hermescontrol.data.model.UpdateCronJobRequest
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.ChangeEvents
import com.m57.hermescontrol.data.ws.toJsonElement
import com.m57.hermescontrol.ui.common.ToastHost
import com.m57.hermescontrol.ui.common.refreshOnChange
import com.m57.hermescontrol.ui.common.safeLaunchLoad
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
@@ -83,6 +85,16 @@ class CronJobsViewModel :
private var loadJob: Job? = null
init {
// Issue #784: gateway broadcasts cron.changed — refresh the list
// silently on change instead of waiting for a manual pull.
refreshOnChange(
eventType = ChangeEvents.CRON,
apiCall = { safeApiCall { ApiClient.hermesApi.getCronJobs() } },
onSuccess = { data -> _uiState.update { it.copy(jobs = data.orEmpty()) } },
)
}
fun loadCronJobs() {
loadJob =
safeLaunchLoad(
@@ -8,7 +8,9 @@ import com.m57.hermescontrol.data.model.PairingRevokeRequest
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.ChangeEvents
import com.m57.hermescontrol.ui.common.ToastHost
import com.m57.hermescontrol.ui.common.refreshOnChange
import com.m57.hermescontrol.ui.common.safeLaunchLoad
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
@@ -36,6 +38,16 @@ class PairingViewModel :
private var launchJob: Job? = null
init {
// Issue #784: gateway broadcasts pairing.changed — refresh silently
// (no spinner) instead of waiting for a manual pull.
refreshOnChange(
eventType = ChangeEvents.PAIRING,
apiCall = { safeApiCall { ApiClient.hermesApi.getPairing() } },
onSuccess = { data -> _uiState.update { it.copy(pairing = data) } },
)
}
fun loadPairing() {
launchJob =
safeLaunchLoad(
@@ -10,7 +10,9 @@ import com.m57.hermescontrol.data.model.SessionSearchResult
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.ChangeEvents
import com.m57.hermescontrol.ui.common.ToastHost
import com.m57.hermescontrol.ui.common.refreshOnChange
import com.m57.hermescontrol.ui.common.safeLaunchLoad
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
@@ -62,6 +64,31 @@ class SessionsViewModel :
private var loadJob: Job? = null
private var statsJob: Job? = null
init {
// Issue #784: gateway broadcasts sessions.changed — refresh the list
// silently (no spinner, no selection reset) instead of blind polling.
refreshOnChange(
eventType = ChangeEvents.SESSIONS,
apiCall = {
safeApiCall {
ApiClient.hermesApi.getSessions(
limit = PAGE_SIZE,
offset = 0,
order = "recent",
)
}
},
onSuccess = { data ->
_uiState.update {
it.copy(
sessions = data.sessions.orEmpty(),
total = data.total,
)
}
},
)
}
/** Page size sent to the server — matches the default the gateway uses. */
private companion object {
const val PAGE_SIZE = 20
@@ -99,6 +99,36 @@ class EventParserTest {
assertEquals(mapOf("session_id" to "session-1"), gatewayReady.data)
}
@Test
fun testParseChangeEvents_returnsChangeEvent() {
// Issue #784: gateway broadcasts change events (pet.changed excluded —
// mobile has no pet feature). All four must map to ChangeEvent with the
// type preserved and the payload attached.
val cases =
mapOf(
"cron.changed" to mapOf("a" to "1"),
"sessions.changed" to mapOf("b" to "2"),
"platforms.changed" to mapOf("c" to "3"),
"pairing.changed" to mapOf("d" to "4"),
)
cases.forEach { (type, payload) ->
val response =
createJsonRpcResponse(
jsonrpc = "2.0",
id = null,
result = null,
error = null,
method = "event",
params = mapOf("type" to type, "payload" to payload),
)
val event = EventParser.parse(response)
assertTrue("expected ChangeEvent for $type, got $event", event is WsEvent.ChangeEvent)
val changeEvent = event as WsEvent.ChangeEvent
assertEquals(type, changeEvent.type)
assertEquals(payload, changeEvent.data)
}
}
@Test
fun testParseMessageToken_returnsMessageTokenEvent() {
val response =
@@ -101,11 +101,13 @@ class ChatViewModelTest {
// fires after unmockkAll and crashes an unrelated later test with
// MockK's "can't find stub" (UncaughtExceptionsBeforeTest). Mirror
// the real request()'s send() delegation (tests verify send calls)
// but return a never-completing deferred — same unanswered behavior,
// no leaked timer.
// but skip the timer. Must COMPLETE (Unit): a never-completing
// deferred freezes the VM's event collector at the SESSION_CREATE
// session.usage await and lets its late completion race per-test
// capture stubs (slash-focus flake, CI 31137581335 on 4645a2b).
every { HermesWsClient.request(any(), any(), any()) } answers {
HermesWsClient.send(arg(0), arg(1)) {}
CompletableDeferred<Any?>()
CompletableDeferred<Any?>(Unit)
}
mockkObject(ApiClient)
mockkObject(HermesDatabase)
@@ -107,10 +107,15 @@ class SlashCommandDispatchRpcTest {
// wsScope that fires after unmockkAll and crashes a later test with
// MockK's "can't find stub". Mirror the real send() delegation (so
// send-based verifications still see the call) but skip the timer.
// Must COMPLETE (Unit): a never-completing deferred freezes the VM's
// event collector at the SESSION_CREATE-triggered session.usage await,
// and that completion then races the per-test capture stubs — the slot
// can end up holding session.usage instead of command.dispatch (CI run
// 31137581335 on 4645a2b: slash-focus flake).
// Specific per-test request() stubs registered later take precedence.
every { HermesWsClient.request(any(), any(), any()) } answers {
HermesWsClient.send(arg(0), arg(1)) {}
CompletableDeferred<Any?>()
CompletableDeferred<Any?>(Unit)
}
}
@@ -0,0 +1,77 @@
package com.m57.hermescontrol.ui.common
import androidx.lifecycle.ViewModel
import com.m57.hermescontrol.data.remote.NetworkResult
import com.m57.hermescontrol.data.ws.ChangeEventHub
import com.m57.hermescontrol.data.ws.ChangeEvents
import com.m57.hermescontrol.data.ws.WsEvent
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.advanceUntilIdle
import kotlinx.coroutines.test.resetMain
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.test.setMain
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Before
import org.junit.Test
/**
* Regression guard for issue #784: `refreshOnChange` must react to hub events
* WITHOUT touching the HermesWsClient singleton — constructing a ViewModel
* that wires it must be safe in a plain unit-test JVM (no mockkObject of
* HermesWsClient, no static mocks). This used to poison the whole suite with
* ExceptionInInitializerError at MockKStub.kt:106.
*/
@OptIn(ExperimentalCoroutinesApi::class)
class RefreshOnChangeTest {
private val testDispatcher = StandardTestDispatcher()
private class TestViewModel : ViewModel() {
val state = MutableStateFlow("initial")
init {
refreshOnChange(
eventType = ChangeEvents.CRON,
apiCall = { NetworkResult.Success("refreshed") },
onSuccess = { state.value = it },
)
}
}
@Before
fun setUp() {
Dispatchers.setMain(testDispatcher)
}
@After
fun tearDown() {
Dispatchers.resetMain()
}
@Test
fun testRefreshOnChange_updatesOnMatchingEvent() =
runTest(testDispatcher) {
val viewModel = TestViewModel()
// Let the init coroutine run so the collector actually subscribes
// before the event fires (replay=0 — events before subscription
// are dropped, matching production where VMs subscribe at
// construction and events arrive afterwards).
advanceUntilIdle()
ChangeEventHub.emit(WsEvent.ChangeEvent(ChangeEvents.CRON))
advanceUntilIdle()
assertEquals("refreshed", viewModel.state.value)
}
@Test
fun testRefreshOnChange_ignoresOtherEventTypes() =
runTest(testDispatcher) {
val viewModel = TestViewModel()
advanceUntilIdle()
ChangeEventHub.emit(WsEvent.ChangeEvent(ChangeEvents.SESSIONS))
advanceUntilIdle()
assertEquals("initial", viewModel.state.value)
}
}