refactor(#1375): typed params for session.interrupt, steer, redirect (#1382)

This commit is contained in:
m57 authored and GitHub committed 2026-09-29 17:02:32 +03:00
1 parent a780df3261
commit aac42032fb
9 files changed
+480 -22

No files matched your search

@@ -13,6 +13,7 @@ import com.m57.hermescontrol.data.session.ActiveSessionHolder
import com.m57.hermescontrol.data.ws.contract.PromptSubmitParams
import com.m57.hermescontrol.data.ws.contract.RpcMethod
import com.m57.hermescontrol.data.ws.contract.RpcMethods
import com.m57.hermescontrol.data.ws.contract.SessionCorrectionParams
import com.m57.hermescontrol.data.ws.contract.SessionEventsSinceParams
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
@@ -1115,8 +1116,12 @@ object HermesWsClient {
onSent: ((String) -> Unit)? = null,
): String =
send(
method = WsMethods.SESSION_REDIRECT,
params = mapOf("session_id" to sessionId, "text" to text),
method = RpcMethods.SESSION_REDIRECT,
params =
SessionCorrectionParams(
sessionId = sessionId,
text = text,
),
onSent = onSent,
)
@@ -38,6 +38,27 @@ object RpcMethods {
PromptSubmitResult.serializer(),
)
val SESSION_INTERRUPT: RpcMethod<SessionInterruptParams, SessionInterruptResult> =
RpcMethod(
WsMethods.SESSION_INTERRUPT,
SessionInterruptParams.serializer(),
SessionInterruptResult.serializer(),
)
val SESSION_STEER: RpcMethod<SessionCorrectionParams, SessionCorrectionResult> =
RpcMethod(
WsMethods.SESSION_STEER,
SessionCorrectionParams.serializer(),
SessionCorrectionResult.serializer(),
)
val SESSION_REDIRECT: RpcMethod<SessionCorrectionParams, SessionCorrectionResult> =
RpcMethod(
WsMethods.SESSION_REDIRECT,
SessionCorrectionParams.serializer(),
SessionCorrectionResult.serializer(),
)
/** Every typed method. GatewayContractTest iterates this; future migrations append here. */
val all: List<RpcMethod<*, *>> =
listOf(
@@ -45,5 +66,8 @@ object RpcMethods {
SESSION_CREATE,
SESSION_RESUME,
PROMPT_SUBMIT,
SESSION_INTERRUPT,
SESSION_STEER,
SESSION_REDIRECT,
)
}
@@ -0,0 +1,26 @@
package com.m57.hermescontrol.data.ws.contract
import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable
@Serializable
data class SessionInterruptParams(
@SerialName("session_id") val sessionId: String,
)
@Serializable
data class SessionCorrectionParams(
@SerialName("session_id") val sessionId: String,
@SerialName("text") val text: String,
)
@Serializable
data class SessionInterruptResult(
val status: String? = null,
)
@Serializable
data class SessionCorrectionResult(
val status: String? = null,
val text: String? = null,
)
@@ -20,6 +20,7 @@ import com.m57.hermescontrol.data.ws.WsMethods
import com.m57.hermescontrol.data.ws.contract.PromptSubmitParams
import com.m57.hermescontrol.data.ws.contract.RpcMethods
import com.m57.hermescontrol.data.ws.contract.SessionCreateParams
import com.m57.hermescontrol.data.ws.contract.SessionInterruptParams
import com.m57.hermescontrol.data.ws.toJsonElement
import com.m57.hermescontrol.ui.chat.tool.ToolResultSummary
import kotlinx.coroutines.CompletableDeferred
@@ -909,8 +910,8 @@ class GroupChatViewModel(
for (sid in activeSessions.distinct()) {
try {
HermesWsClient.send(
WsMethods.SESSION_INTERRUPT,
mapOf("session_id" to sid),
RpcMethods.SESSION_INTERRUPT,
SessionInterruptParams(sid),
)
} catch (e: Exception) {
Log.w("GroupChatViewModel", "Failed to send interrupt for session $sid: ${e.message}")
@@ -37,6 +37,9 @@ import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.ws.JsonRpcError
import com.m57.hermescontrol.data.ws.WsEvent
import com.m57.hermescontrol.data.ws.WsMethods
import com.m57.hermescontrol.data.ws.contract.RpcMethods
import com.m57.hermescontrol.data.ws.contract.SessionCorrectionParams
import com.m57.hermescontrol.data.ws.contract.SessionInterruptParams
import com.m57.hermescontrol.data.ws.toAny
import com.m57.hermescontrol.data.ws.toJsonElement
import com.m57.hermescontrol.notification.ReplyNotificationTracker
@@ -2454,8 +2457,8 @@ class ChatViewModel(
deleteSnapshotOnRollback = true,
) {
wsClient.send(
WsMethods.SESSION_INTERRUPT,
mapOf("session_id" to owner.agentSessionId),
RpcMethods.SESSION_INTERRUPT,
SessionInterruptParams(owner.agentSessionId),
onSent = { id -> trackHardInterrupt(id, staged.id, owner, reservation.mainTurnEpoch) },
)
}
@@ -2821,8 +2824,8 @@ class ChatViewModel(
deleteSnapshotOnRollback = false,
) {
wsClient.send(
WsMethods.SESSION_INTERRUPT,
mapOf("session_id" to owner.agentSessionId),
RpcMethods.SESSION_INTERRUPT,
SessionInterruptParams(owner.agentSessionId),
onSent = { requestId -> trackHardInterrupt(requestId, id, owner, interruptedTurnEpoch) },
)
}
@@ -3043,8 +3046,8 @@ class ChatViewModel(
viewModelScope.launch {
commitReceiptAndEnqueue(owner, receipt, null, deleteSnapshotOnRollback = false) {
wsClient.send(
WsMethods.SESSION_INTERRUPT,
mapOf("session_id" to owner.agentSessionId),
RpcMethods.SESSION_INTERRUPT,
SessionInterruptParams(owner.agentSessionId),
onSent = { id -> trackHardInterrupt(id, userMessage.id, owner, interruptedTurnEpoch) },
)
}
@@ -3437,8 +3440,8 @@ class ChatViewModel(
if (mode == BusySendMode.GUIDE) {
enqueueOwned {
wsClient.send(
WsMethods.SESSION_STEER,
mapOf("session_id" to owner.agentSessionId, "text" to fullText),
RpcMethods.SESSION_STEER,
SessionCorrectionParams(owner.agentSessionId, fullText),
onSent = { id ->
trackOutgoingRequest(
id,
@@ -4328,8 +4331,8 @@ class ChatViewModel(
}
viewModelScope.launch(ioDispatcher) {
wsClient.send(
WsMethods.SESSION_INTERRUPT,
mapOf("session_id" to sessionId),
RpcMethods.SESSION_INTERRUPT,
SessionInterruptParams(sessionId),
onSent = { id -> trackRequest(id, WsMethods.SESSION_INTERRUPT) },
)
}
@@ -0,0 +1,281 @@
package com.m57.hermescontrol.data.ws
import android.util.Log
import com.m57.hermescontrol.data.local.AuthManager
import com.m57.hermescontrol.data.remote.CookieManager
import com.m57.hermescontrol.data.remote.DashboardSessionTokenRefresher
import com.m57.hermescontrol.data.remote.OkHttpProvider
import com.m57.hermescontrol.data.remote.buildFakePersistentCookieJar
import com.m57.hermescontrol.data.ws.contract.RpcMethods
import com.m57.hermescontrol.data.ws.contract.SessionCorrectionParams
import com.m57.hermescontrol.data.ws.contract.SessionInterruptParams
import io.mockk.coEvery
import io.mockk.every
import io.mockk.mockk
import io.mockk.mockkObject
import io.mockk.mockkStatic
import io.mockk.unmockkAll
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import okhttp3.Response
import okhttp3.WebSocket
import okhttp3.WebSocketListener
import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicReference
class HermesWsClientSessionControlTest {
private lateinit var mockWebServer: MockWebServer
@Before
fun setUp() {
mockkStatic(Log::class)
every { Log.d(any<String>(), any<String>()) } returns 0
every { Log.i(any<String>(), any<String>()) } returns 0
every { Log.w(any<String>(), any<String>()) } returns 0
every { Log.w(any<String>(), any<String>(), any<Throwable>()) } returns 0
every { Log.e(any<String>(), any<String>(), any()) } returns 0
mockWebServer = MockWebServer()
mockWebServer.start()
mockkObject(AuthManager)
every { AuthManager.wsUrl() } returns mockWebServer.url("/").toString().replace("http://", "ws://")
every { AuthManager.isAutoReconnect() } returns false
every { AuthManager.getSessionCookie() } returns null
every { AuthManager.serverStore } returns
mockk<com.m57.hermescontrol.data.config.ServerStore>().also {
every { it.getLatestState() } returns
com.m57.hermescontrol.data.config
.ServerStoreState()
}
CookieManager.setJarForTest(buildFakePersistentCookieJar())
mockkObject(DashboardSessionTokenRefresher)
coEvery { DashboardSessionTokenRefresher.refreshAsync() } returns null
every { DashboardSessionTokenRefresher.refresh() } returns null
HermesWsClient.disconnect(clearPendingMessages = true)
HermesWsClient.releaseExternalActivityConnectionLease()
HermesWsClient.releaseBackgroundConnectionLease()
HermesWsClient.setAppForeground(true)
AuthManager.resetAuthStateForTest()
}
@After
fun tearDown() {
AuthManager.resetAuthStateForTest()
HermesWsClient.releaseExternalActivityConnectionLease()
HermesWsClient.releaseBackgroundConnectionLease()
HermesWsClient.disconnect(clearPendingMessages = true)
runBlocking {
withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.DISCONNECTED } }
}
runCatching { mockWebServer.shutdown() }
unmockkAll()
}
private fun connectClient() {
HermesWsClient.connect()
runBlocking {
withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } }
}
}
@Test
fun sendRedirectPutsExpectedFrameAndInvokesOnSent() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "session.redirect") {
frameRef.set(json)
captured.countDown()
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
var onSentId: String? = null
val returnedId =
HermesWsClient.sendRedirect(
sessionId = "sess-red-1",
text = "change topic to testing",
onSent = { id -> onSentId = id },
)
assertTrue(returnedId.isNotBlank())
assertEquals(returnedId, onSentId)
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("session.redirect", (frame["method"] as? JsonPrimitive)?.content)
assertEquals(returnedId, (frame["id"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
assertNotNull(params)
val expectedParams =
buildJsonObject {
put("session_id", "sess-red-1")
put("text", "change topic to testing")
}
assertEquals(expectedParams, params)
assertEquals(setOf("session_id", "text"), params!!.keys)
}
@Test
fun typedSendSessionInterruptPutsExpectedFrameAndInvokesOnSent() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "session.interrupt") {
frameRef.set(json)
captured.countDown()
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
var onSentId: String? = null
val returnedId =
HermesWsClient.send(
method = RpcMethods.SESSION_INTERRUPT,
params = SessionInterruptParams(sessionId = "s1"),
onSent = { id -> onSentId = id },
)
assertTrue(returnedId.isNotBlank())
assertEquals(returnedId, onSentId)
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("session.interrupt", (frame["method"] as? JsonPrimitive)?.content)
assertEquals(returnedId, (frame["id"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
assertNotNull(params)
val expectedParams =
buildJsonObject {
put("session_id", "s1")
}
assertEquals(expectedParams, params)
assertEquals(setOf("session_id"), params!!.keys)
}
@Test
fun typedSendSessionSteerPutsExpectedFrameAndInvokesOnSent() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "session.steer") {
frameRef.set(json)
captured.countDown()
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
var onSentId: String? = null
val returnedId =
HermesWsClient.send(
method = RpcMethods.SESSION_STEER,
params = SessionCorrectionParams(sessionId = "s1", text = "steer towards plan B"),
onSent = { id -> onSentId = id },
)
assertTrue(returnedId.isNotBlank())
assertEquals(returnedId, onSentId)
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("session.steer", (frame["method"] as? JsonPrimitive)?.content)
assertEquals(returnedId, (frame["id"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
assertNotNull(params)
val expectedParams =
buildJsonObject {
put("session_id", "s1")
put("text", "steer towards plan B")
}
assertEquals(expectedParams, params)
assertEquals(setOf("session_id", "text"), params!!.keys)
}
}
@@ -334,7 +334,6 @@ class RawRpcParamsRatchetTest {
"data/ws/ConnectorRepository.kt::CONNECTORS_CONNECT",
"data/ws/ConnectorRepository.kt::CONNECTORS_LIST",
"data/ws/HermesWsClient.kt::CLIENT_CAPABILITIES",
"data/ws/HermesWsClient.kt::SESSION_REDIRECT",
"data/ws/ModelOptionsRepository.kt::MODEL_OPTIONS",
"data/ws/PluginManageRepository.kt::PLUGINS_MANAGE",
"data/ws/PluginManageRepository.kt::PLUGINS_MANAGE",
@@ -356,7 +355,6 @@ class RawRpcParamsRatchetTest {
"ui/bots/group/GroupChatViewModel.kt::PROFILES_CONFIGURE",
"ui/bots/group/GroupChatViewModel.kt::PROFILES_CONFIGURE",
"ui/bots/group/GroupChatViewModel.kt::PROFILES_LIST",
"ui/bots/group/GroupChatViewModel.kt::SESSION_INTERRUPT",
"ui/chat/ChatApprovalsDelegate.kt::APPROVAL_PENDING",
"ui/chat/ChatApprovalsDelegate.kt::APPROVAL_RECEIVED",
"ui/chat/ChatApprovalsDelegate.kt::APPROVAL_RESPOND",
@@ -380,13 +378,8 @@ class RawRpcParamsRatchetTest {
"ui/chat/ChatViewModel.kt::SESSION_COMPRESS",
"ui/chat/ChatViewModel.kt::SESSION_CONTEXT_BREAKDOWN",
"ui/chat/ChatViewModel.kt::SESSION_CREATE",
"ui/chat/ChatViewModel.kt::SESSION_INTERRUPT",
"ui/chat/ChatViewModel.kt::SESSION_INTERRUPT",
"ui/chat/ChatViewModel.kt::SESSION_INTERRUPT",
"ui/chat/ChatViewModel.kt::SESSION_INTERRUPT",
"ui/chat/ChatViewModel.kt::SESSION_LIST",
"ui/chat/ChatViewModel.kt::SESSION_RESUME",
"ui/chat/ChatViewModel.kt::SESSION_STEER",
"ui/chat/ChatViewModel.kt::SESSION_USAGE",
"ui/chat/ChatViewModel.kt::SLASH_EXEC",
"ui/process/ProcessesViewModel.kt::PROCESS_KILL",
@@ -0,0 +1,119 @@
package com.m57.hermescontrol.data.ws.contract
import com.m57.hermescontrol.data.remote.OkHttpProvider
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.decodeFromJsonElement
import kotlinx.serialization.json.encodeToJsonElement
import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Test
class SessionControlTypesTest {
@Test
fun sessionInterruptParamsEncodesSessionIdExactly() {
val params = SessionInterruptParams(sessionId = "sess-int-1")
val encoded = OkHttpProvider.json.encodeToJsonElement(SessionInterruptParams.serializer(), params)
val expected =
buildJsonObject {
put("session_id", "sess-int-1")
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("session_id"), (encoded as JsonObject).keys)
}
@Test
fun sessionCorrectionParamsEncodesSessionIdAndTextExactly() {
val params = SessionCorrectionParams(sessionId = "sess-corr-1", text = "stop that action")
val encoded = OkHttpProvider.json.encodeToJsonElement(SessionCorrectionParams.serializer(), params)
val expected =
buildJsonObject {
put("session_id", "sess-corr-1")
put("text", "stop that action")
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("session_id", "text"), (encoded as JsonObject).keys)
}
@Test
fun sessionInterruptResultDecodesFullPayloadAndIgnoresUnknownKeys() {
val payload =
buildJsonObject {
put("status", "interrupted")
put("unknown_key", "ignored_value")
put("extra_flag", true)
}
val decoded =
OkHttpProvider.json.decodeFromJsonElement(
SessionInterruptResult.serializer(),
payload,
)
assertEquals("interrupted", decoded.status)
}
@Test
fun sessionInterruptResultDecodesEmptyObjectToNulls() {
val payload = buildJsonObject {}
val decoded =
OkHttpProvider.json.decodeFromJsonElement(
SessionInterruptResult.serializer(),
payload,
)
assertNull(decoded.status)
}
@Test
fun sessionCorrectionResultDecodesFullPayloadAndIgnoresUnknownKeys() {
val payload =
buildJsonObject {
put("status", "applied")
put("text", "stop that action")
put("ignored_num", 42)
put("nested", buildJsonObject { put("ignored", "yes") })
}
val decoded =
OkHttpProvider.json.decodeFromJsonElement(
SessionCorrectionResult.serializer(),
payload,
)
assertEquals("applied", decoded.status)
assertEquals("stop that action", decoded.text)
}
@Test
fun sessionCorrectionResultDecodesEmptyObjectToNulls() {
val payload = buildJsonObject {}
val decoded =
OkHttpProvider.json.decodeFromJsonElement(
SessionCorrectionResult.serializer(),
payload,
)
assertNull(decoded.status)
assertNull(decoded.text)
}
@Test
fun rpcMethodsRegistrationMatchesControlContract() {
assertEquals("session.interrupt", RpcMethods.SESSION_INTERRUPT.name)
assertEquals("session.steer", RpcMethods.SESSION_STEER.name)
assertEquals("session.redirect", RpcMethods.SESSION_REDIRECT.name)
assertTrue(RpcMethods.all.contains(RpcMethods.SESSION_INTERRUPT))
assertTrue(RpcMethods.all.contains(RpcMethods.SESSION_STEER))
assertTrue(RpcMethods.all.contains(RpcMethods.SESSION_REDIRECT))
}
}
@@ -6343,7 +6343,13 @@ class ChatViewModelTest {
viewModel.interruptSession()
advanceUntilIdle()
verify { HermesWsClient.send(WsMethods.SESSION_INTERRUPT, mapOf("session_id" to sessionId), any()) }
verify {
HermesWsClient.send(
WsMethods.SESSION_INTERRUPT,
match { it["session_id"] == JsonPrimitive(sessionId) && it.size == 1 },
any(),
)
}
}
@Test