feat(#1218): support backend-owned connector setup operations (#1229)

app/src/test/java/com/m57/hermescontrol/data/ws/ConnectionOperationEventParserTest.kt

@@ -0,0 +1,83 @@
+package com.m57.hermescontrol.data.ws
+
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class ConnectionOperationEventParserTest {
+    @Test
+    fun connectionRequest_usesEnvelopeSessionAndPayloadSequence() {
+        val event =
+            EventParser.parseParams(
+                mapOf(
+                    "type" to "connection.request",
+                    "session_id" to "runtime-session",
+                    "seq" to 999,
+                    "payload" to operationPayload(seq = 4L),
+                ),
+            )
+
+        assertTrue(event is WsEvent.ConnectionRequest)
+        val snapshot = (event as WsEvent.ConnectionRequest).snapshot
+        assertEquals("runtime-session", snapshot.sessionId)
+        assertEquals(4L, snapshot.seq)
+    }
+
+    @Test
+    fun connectionUpdate_parsesFullSettledSnapshot() {
+        val event =
+            EventParser.parseParams(
+                mapOf(
+                    "type" to "connection.update",
+                    "session_id" to "runtime-session",
+                    "payload" to
+                        operationPayload(seq = 5L) +
+                        mapOf(
+                            "settled" to true,
+                            "settled_by" to "all_resolved",
+                            "future_field" to "ignored",
+                        ),
+                ),
+            )
+
+        assertTrue(event is WsEvent.ConnectionUpdate)
+        val snapshot = (event as WsEvent.ConnectionUpdate).snapshot
+        assertTrue(snapshot.settled)
+        assertEquals("all_resolved", snapshot.settledBy)
+    }
+
+    @Test
+    fun malformedOperation_degradesWithoutRetainingRawFrame() {
+        val event =
+            EventParser.parseParams(
+                params =
+                    mapOf(
+                        "type" to "connection.request",
+                        "session_id" to "runtime-session",
+                        "payload" to operationPayload(seq = 1L) - "op_id",
+                    ),
+                rawJson = "issue1218-secret-canary",
+            )
+
+        assertTrue(event is WsEvent.Unknown)
+        assertEquals("", (event as WsEvent.Unknown).raw)
+    }
+
+    private fun operationPayload(seq: Long): Map<String, Any?> =
+        mapOf(
+            "op_id" to "op-1",
+            "seq" to seq,
+            "deadline_at" to 2_000_000_000.0,
+            "timeout_seconds" to 120.0,
+            "tool_call_id" to "tool-1",
+            "targets" to
+                listOf(
+                    mapOf(
+                        "name" to "github",
+                        "kind" to "connector",
+                        "action" to "authorize",
+                        "state" to "pending",
+                    ),
+                ),
+        )
+}

app/src/test/java/com/m57/hermescontrol/data/ws/ConnectionOperationParserTest.kt

@@ -0,0 +1,169 @@
+package com.m57.hermescontrol.data.ws
+
+import com.m57.hermescontrol.data.model.ConnectionTargetAction
+import com.m57.hermescontrol.data.model.ConnectionTargetKind
+import com.m57.hermescontrol.data.model.ConnectionTargetState
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class ConnectionOperationParserTest {
+    @Test
+    fun parseCompleteSnapshot_preservesBackendOrderAndMetadata() {
+        val snapshot = ConnectionOperationParser.parse(completePayload(), "runtime-session")
+
+        assertNotNull(snapshot)
+        snapshot!!
+        assertEquals("runtime-session", snapshot.sessionId)
+        assertEquals("op-1218", snapshot.opId)
+        assertEquals(7L, snapshot.seq)
+        assertEquals(2, snapshot.targets.size)
+        val target = snapshot.targets.first()
+        assertEquals("reports", target.name)
+        assertEquals(ConnectionTargetKind.MCP, target.kind)
+        assertEquals(ConnectionTargetAction.INSTALL, target.action)
+        assertEquals(ConnectionTargetState.PENDING, target.state)
+        assertEquals("install-2", target.attempt)
+        assertEquals("Use a workspace token", target.hint)
+        assertEquals(listOf("REGION", "API_TOKEN"), target.requiredEnv.map { it.name })
+        assertEquals("us-east", target.requiredEnv[0].defaultValue)
+        assertNull(target.requiredEnv[1].defaultValue)
+        assertTrue(target.requiredEnv[1].secret)
+        assertEquals(listOf("search", "export"), target.tools)
+    }
+
+    @Test
+    fun unknownFieldsAndEnumValues_degradeWithoutDroppingUsableTarget() {
+        val payload =
+            completePayload().toMutableMap().apply {
+                put("future_operation_field", mapOf("nested" to true))
+                put(
+                    "targets",
+                    listOf(
+                        targetPayload().toMutableMap().apply {
+                            put("kind", "future-kind")
+                            put("action", "future-action")
+                            put("state", "future-state")
+                            put("future_target_field", "ignored")
+                        },
+                    ),
+                )
+            }
+
+        val target = ConnectionOperationParser.parse(payload, "runtime-session")!!.targets.single()
+
+        assertEquals(ConnectionTargetKind.UNKNOWN, target.kind)
+        assertEquals(ConnectionTargetAction.UNKNOWN, target.action)
+        assertEquals(ConnectionTargetState.UNKNOWN, target.state)
+    }
+
+    @Test
+    fun malformedIdentityOrDeadline_isRejected() {
+        assertNull(ConnectionOperationParser.parse(completePayload() - "op_id", "runtime-session"))
+        assertNull(ConnectionOperationParser.parse(completePayload() - "seq", "runtime-session"))
+        assertNull(ConnectionOperationParser.parse(completePayload() - "deadline_at", "runtime-session"))
+        assertNull(ConnectionOperationParser.parse(completePayload() + ("seq" to 1.5), "runtime-session"))
+        assertNull(
+            ConnectionOperationParser.parse(completePayload() + ("deadline_at" to Double.NaN), "runtime-session"),
+        )
+        assertNull(
+            ConnectionOperationParser.parse(completePayload() + ("targets" to emptyList<Any>()), "runtime-session"),
+        )
+    }
+
+    @Test
+    fun secretDefaultsAndOAuthUrls_areNeverExposedByModels() {
+        val canary = "issue1218-secret-canary"
+        val payload =
+            completePayload() +
+                (
+                    "targets" to
+                        listOf(
+                            targetPayload() +
+                                (
+                                    "connect_url" to
+                                        "https://example.com/oauth?token=$canary"
+                                ) +
+                                (
+                                    "required_env" to
+                                        listOf(
+                                            mapOf(
+                                                "name" to "API_TOKEN",
+                                                "required" to true,
+                                                "secret" to true,
+                                                "default" to canary,
+                                            ),
+                                        )
+                                ),
+                        )
+                )
+
+        val snapshot = ConnectionOperationParser.parse(payload, "runtime-session")!!
+
+        assertNull(
+            snapshot.targets
+                .single()
+                .requiredEnv
+                .single()
+                .defaultValue,
+        )
+        assertFalse(snapshot.toString().contains(canary))
+        assertTrue(snapshot.toString().contains("[REDACTED]"))
+    }
+
+    private fun completePayload(): Map<String, Any?> =
+        mapOf(
+            "op_id" to "op-1218",
+            "seq" to 7L,
+            "deadline_at" to 2_000_000_000.0,
+            "timeout_seconds" to 300.0,
+            "tool_call_id" to "tool-1",
+            "settled" to false,
+            "targets" to
+                listOf(
+                    targetPayload(),
+                    mapOf(
+                        "name" to "github",
+                        "kind" to "connector",
+                        "action" to "authorize",
+                        "state" to "initiated",
+                    ),
+                ),
+        )
+
+    private fun targetPayload(): Map<String, Any?> =
+        mapOf(
+            "name" to "reports",
+            "kind" to "mcp",
+            "action" to "install",
+            "state" to "pending",
+            "detail" to "Credentials required",
+            "instructions" to "Enter the service credentials",
+            "discovery_error" to null,
+            "connect_url" to null,
+            "connection_id" to "vendor-account-1",
+            "attempt" to "install-2",
+            "hint" to "Use a workspace token",
+            "required_env" to
+                listOf(
+                    mapOf(
+                        "name" to "REGION",
+                        "required" to false,
+                        "secret" to false,
+                        "default" to "us-east",
+                        "prompt" to "Region",
+                    ),
+                    mapOf(
+                        "name" to "API_TOKEN",
+                        "required" to true,
+                        "secret" to true,
+                        "default" to "must-not-be-retained",
+                        "prompt" to "API token",
+                    ),
+                ),
+            "tools" to listOf("search", "export"),
+        )
+}

app/src/test/java/com/m57/hermescontrol/ui/chat/ChatConnectionOperationDelegateTest.kt

@@ -0,0 +1,314 @@
+package com.m57.hermescontrol.ui.chat
+
+import com.m57.hermescontrol.data.model.ConnectionEnvField
+import com.m57.hermescontrol.data.model.ConnectionOperationSnapshot
+import com.m57.hermescontrol.data.model.ConnectionOperationTarget
+import com.m57.hermescontrol.data.model.ConnectionTargetAction
+import com.m57.hermescontrol.data.model.ConnectionTargetKind
+import com.m57.hermescontrol.data.model.ConnectionTargetState
+import com.m57.hermescontrol.data.ws.HermesWsClient
+import com.m57.hermescontrol.data.ws.WsMethods
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class ChatConnectionOperationDelegateTest {
+    @Test
+    fun requestAndUpdate_areStrictlyOrderedAndCorrelated() {
+        val delegate = delegate()
+        delegate.reset("session-a")
+
+        assertTrue(delegate.acceptRequest(snapshot(opId = "op-a", seq = 1L)))
+        assertFalse(delegate.acceptUpdate(snapshot(opId = "op-a", seq = 1L)))
+        assertFalse(delegate.acceptUpdate(snapshot(opId = "op-old", seq = 99L)))
+        assertEquals(
+            "op-a",
+            delegate.state.value.operation
+                ?.opId,
+        )
+        assertTrue(delegate.acceptUpdate(snapshot(opId = "op-a", seq = 2L)))
+        assertEquals(
+            2L,
+            delegate.state.value.operation
+                ?.seq,
+        )
+
+        assertTrue(delegate.acceptRequest(snapshot(opId = "op-b", seq = 1L)))
+        assertFalse(delegate.acceptUpdate(snapshot(opId = "op-a", seq = 3L)))
+        assertEquals(
+            "op-b",
+            delegate.state.value.operation
+                ?.opId,
+        )
+    }
+
+    @Test
+    fun wrongSessionAndSettledOperation_areNeverReopened() {
+        val delegate = delegate()
+        delegate.reset("session-a")
+
+        assertFalse(delegate.acceptRequest(snapshot(sessionId = "session-b")))
+        assertTrue(delegate.acceptRequest(snapshot(opId = "op-a", seq = 1L)))
+        assertTrue(delegate.acceptUpdate(snapshot(opId = "op-a", seq = 2L, settled = true)))
+        assertNull(delegate.state.value.operation)
+        assertFalse(delegate.acceptRequest(snapshot(opId = "op-a", seq = 3L)))
+    }
+
+    @Test
+    fun resumeReconciliation_cannotRegressOrClearNewerLiveRequest() {
+        val delegate = delegate()
+        delegate.reset("session-a")
+        delegate.acceptRequest(snapshot(opId = "op-a", seq = 2L))
+        val checkpoint = delegate.resumeCheckpoint()
+
+        delegate.acceptRequest(snapshot(opId = "op-b", seq = 1L))
+
+        assertFalse(delegate.reconcileResume(null, checkpoint))
+        assertFalse(delegate.reconcileResume(snapshot(opId = "op-a", seq = 2L), checkpoint))
+        assertEquals(
+            "op-b",
+            delegate.state.value.operation
+                ?.opId,
+        )
+    }
+
+    @Test
+    fun absentResumeSnapshot_clearsOnlyUnchangedOperation() {
+        val delegate = delegate()
+        delegate.reset("session-a")
+        delegate.acceptRequest(snapshot(opId = "op-a", seq = 2L))
+        val checkpoint = delegate.resumeCheckpoint()
+
+        assertTrue(delegate.reconcileResume(null, checkpoint))
+        assertNull(delegate.state.value.operation)
+    }
+
+    @Test
+    fun approvePayload_isExactAndSecretsNeverEnterUiState() =
+        runTest {
+            val requester = RecordingRequester()
+            val delegate = delegate(requester)
+            delegate.reset("session-a")
+            delegate.acceptRequest(snapshot(requiredEnv = listOf(envField("TOKEN", secret = true))))
+            val canary = "issue1218-secret-canary"
+
+            delegate.respond(
+                target = "github",
+                env = mapOf("TOKEN" to canary, "UNDECLARED" to "drop-me"),
+                approved = true,
+            )
+
+            assertEquals(1, requester.calls.size)
+            val call = requester.calls.single()
+            assertEquals(WsMethods.CONNECTION_RESPOND, call.first)
+            assertEquals("session-a", call.second["session_id"])
+            assertEquals("op-a", call.second["op_id"])
+            @Suppress("UNCHECKED_CAST")
+            val result = call.second["result"] as Map<String, Any>
+
+            @Suppress("UNCHECKED_CAST")
+            val answer = (result["targets"] as List<Map<String, Any>>).single()
+            assertEquals("github", answer["name"])
+            assertEquals("approved", answer["status"])
+            assertEquals(mapOf("TOKEN" to canary), answer["env"])
+            assertFalse(
+                delegate.state.value
+                    .toString()
+                    .contains(canary),
+            )
+            assertTrue(delegate.state.value.pendingAction is ConnectionPendingAction.Respond)
+        }
+
+    @Test
+    fun rpcSuccessKeepsExactlyOnceLock_untilNewerSnapshotArrives() =
+        runTest {
+            val requester = RecordingRequester()
+            val delegate = delegate(requester)
+            delegate.reset("session-a")
+            delegate.acceptRequest(snapshot(seq = 1L))
+
+            delegate.respond("github", emptyMap(), approved = false)
+            delegate.respond("github", emptyMap(), approved = false)
+
+            assertEquals(1, requester.calls.size)
+            assertEquals(
+                ConnectionTargetState.PENDING,
+                delegate.state.value.operation
+                    ?.targets
+                    ?.single()
+                    ?.state,
+            )
+            assertTrue(delegate.state.value.pendingAction is ConnectionPendingAction.Respond)
+
+            delegate.acceptUpdate(snapshot(seq = 2L, state = ConnectionTargetState.PENDING))
+            assertNull(delegate.state.value.pendingAction)
+            delegate.respond("github", emptyMap(), approved = false)
+            assertEquals(2, requester.calls.size)
+        }
+
+    @Test
+    fun skipAndContinuePayloads_matchBackendContract() =
+        runTest {
+            val skipRequester = RecordingRequester()
+            val skipDelegate = delegate(skipRequester)
+            skipDelegate.reset("session-a")
+            skipDelegate.acceptRequest(snapshot())
+            skipDelegate.respond("github", mapOf("TOKEN" to "must-drop"), approved = false)
+
+            @Suppress("UNCHECKED_CAST")
+            val skipResult = skipRequester.calls.single().second["result"] as Map<String, Any>
+
+            @Suppress("UNCHECKED_CAST")
+            val skipAnswer = (skipResult["targets"] as List<Map<String, Any>>).single()
+            assertEquals(mapOf("name" to "github", "status" to "skipped"), skipAnswer)
+
+            val continueRequester = RecordingRequester()
+            val continueDelegate = delegate(continueRequester)
+            continueDelegate.reset("session-a")
+            continueDelegate.acceptRequest(snapshot())
+            continueDelegate.continueOperation()
+
+            assertEquals(
+                mapOf("settled_by" to "continue"),
+                continueRequester.calls.single().second["result"],
+            )
+        }
+
+    @Test
+    fun rpcFailureUnlocksWithSanitizedError() =
+        runTest {
+            val requester = RecordingRequester(failure = IllegalStateException("secret backend detail"))
+            val delegate = delegate(requester)
+            delegate.reset("session-a")
+            delegate.acceptRequest(snapshot())
+
+            delegate.respond("github", emptyMap(), approved = false)
+
+            assertNull(delegate.state.value.pendingAction)
+            assertEquals(
+                "request_failed",
+                delegate.state.value.error
+                    ?.message,
+            )
+            assertFalse(
+                delegate.state.value
+                    .toString()
+                    .contains("secret backend detail"),
+            )
+        }
+
+    @Test
+    fun lateOldGenerationFailure_cannotMutateResetState() =
+        runTest {
+            val gate = CompletableDeferred<Unit>()
+            val requester =
+                ConnectionOperationRequester { _, _ ->
+                    gate.await()
+                    throw IllegalStateException("old failure")
+                }
+            val delegate = ChatConnectionOperationDelegate(requester)
+            delegate.reset("session-a")
+            delegate.acceptRequest(snapshot())
+
+            val action = launch { delegate.respond("github", emptyMap(), approved = false) }
+            runCurrent()
+            delegate.reset("session-b")
+            gate.complete(Unit)
+            action.join()
+
+            assertNull(delegate.state.value.operation)
+            assertNull(delegate.state.value.error)
+        }
+
+    @Test
+    fun wakeUnknownOperation_isHarmlessSettledRace() =
+        runTest {
+            val requester =
+                RecordingRequester(
+                    failure = HermesWsClient.HermesRpcException("gone", code = 4004),
+                )
+            val delegate = delegate(requester)
+            delegate.reset("session-a")
+            delegate.acceptRequest(snapshot())
+
+            delegate.wake("op-a")
+
+            assertNull(delegate.state.value.operation)
+            assertNull(delegate.state.value.error)
+        }
+
+    private fun delegate(requester: RecordingRequester = RecordingRequester()): ChatConnectionOperationDelegate =
+        ChatConnectionOperationDelegate(requester)
+
+    private fun snapshot(
+        sessionId: String = "session-a",
+        opId: String = "op-a",
+        seq: Long = 1L,
+        state: ConnectionTargetState = ConnectionTargetState.PENDING,
+        settled: Boolean = false,
+        requiredEnv: List<ConnectionEnvField> = emptyList(),
+    ): ConnectionOperationSnapshot =
+        ConnectionOperationSnapshot(
+            sessionId = sessionId,
+            opId = opId,
+            seq = seq,
+            deadlineAt = 2_000_000_000.0,
+            timeoutSeconds = 300.0,
+            toolCallId = "tool-a",
+            settled = settled,
+            settledBy = if (settled) "all_resolved" else null,
+            targets =
+                listOf(
+                    ConnectionOperationTarget(
+                        name = "github",
+                        kind = ConnectionTargetKind.CONNECTOR,
+                        action = ConnectionTargetAction.AUTHORIZE,
+                        state = state,
+                        detail = null,
+                        instructions = null,
+                        discoveryError = null,
+                        connectUrl = null,
+                        connectionId = null,
+                        attempt = null,
+                        requiredEnv = requiredEnv,
+                        tools = emptyList(),
+                        hint = null,
+                    ),
+                ),
+        )
+
+    private fun envField(
+        name: String,
+        secret: Boolean,
+    ): ConnectionEnvField =
+        ConnectionEnvField(
+            name = name,
+            required = true,
+            secret = secret,
+            defaultValue = null,
+            prompt = null,
+        )
+
+    private class RecordingRequester(
+        private val failure: Throwable? = null,
+    ) : ConnectionOperationRequester {
+        val calls = mutableListOf<Pair<String, Map<String, Any>>>()
+
+        override suspend fun request(
+            method: String,
+            params: Map<String, Any>,
+        ): Any? {
+            calls += method to params
+            failure?.let { throw it }
+            return Unit
+        }
+    }
+}

app/src/test/java/com/m57/hermescontrol/ui/chat/ChatViewModelTest.kt

@@ -24,6 +24,7 @@ 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.ConnectionOperationParser
 import com.m57.hermescontrol.data.ws.ConnectionStatus
 import com.m57.hermescontrol.data.ws.HermesWsClient
 import com.m57.hermescontrol.data.ws.JsonRpcError
@@ -411,6 +412,77 @@ class ChatViewModelTest {
         return Pair(viewModel, "session-123")
     }
 
+    @Test
+    fun connectionOperation_liveEventRoutesAndSessionResetClearsIt() =
+        runTest {
+            val (viewModel, sessionId) = createViewModelWithSession()
+            val snapshot = connectionOperationSnapshot(sessionId = sessionId)
+
+            mockEventsFlow.emit(WsEvent.ConnectionRequest(snapshot))
+            advanceUntilIdle()
+            assertEquals(
+                "op-1218",
+                viewModel.connectionOperationState.value.operation
+                    ?.opId,
+            )
+
+            viewModel.createNewSession()
+            advanceUntilIdle()
+            assertNull(viewModel.connectionOperationState.value.operation)
+
+            mockEventsFlow.emit(WsEvent.ConnectionUpdate(snapshot.copy(seq = 2L)))
+            advanceUntilIdle()
+            assertNull(viewModel.connectionOperationState.value.operation)
+        }
+
+    @Test
+    fun connectionOperation_sessionResumeRestoresPendingSnapshot() =
+        runTest {
+            val viewModel = createViewModel()
+            advanceUntilIdle()
+            viewModel.switchSession("stored-session")
+            advanceUntilIdle()
+            val resumeRequestId = sentRequestMethods.last { it.first == WsMethods.SESSION_RESUME }.second
+
+            mockEventsFlow.emit(
+                WsEvent.RpcResult(
+                    resumeRequestId,
+                    mapOf(
+                        "session_id" to "runtime-session",
+                        "pending_connection" to connectionOperationPayload(),
+                    ),
+                ),
+            )
+            advanceUntilIdle()
+
+            val operation = viewModel.connectionOperationState.value.operation
+            assertEquals("runtime-session", operation?.sessionId)
+            assertEquals("op-1218", operation?.opId)
+        }
+
+    private fun connectionOperationSnapshot(
+        sessionId: String,
+        seq: Long = 1L,
+    ) = checkNotNull(ConnectionOperationParser.parse(connectionOperationPayload(seq), sessionId))
+
+    private fun connectionOperationPayload(seq: Long = 1L): Map<String, Any?> =
+        mapOf(
+            "op_id" to "op-1218",
+            "seq" to seq,
+            "deadline_at" to 2_000_000_000.0,
+            "timeout_seconds" to 300.0,
+            "tool_call_id" to "tool-1218",
+            "targets" to
+                listOf(
+                    mapOf(
+                        "name" to "github",
+                        "kind" to "connector",
+                        "action" to "authorize",
+                        "state" to "pending",
+                    ),
+                ),
+        )
+
     // ── Slash command tests ──────────────────────────────────────────────────
 
     @Test

app/src/test/java/com/m57/hermescontrol/ui/chat/SlashCommandDispatchRpcTest.kt

@@ -53,11 +53,13 @@ class SlashCommandDispatchRpcTest {
     private lateinit var app: Application
     private lateinit var fakeRepo: FakeChatPersistenceRepository
     private var reqCount = 0
+    private var sessionCreateRequestId: String? = null
 
     @Before
     fun setUp() {
         Dispatchers.setMain(testDispatcher)
         reqCount = 0
+        sessionCreateRequestId = null
 
         mockkStatic(android.util.Log::class)
         every { android.util.Log.d(any(), any()) } returns 0
@@ -114,6 +116,9 @@ class SlashCommandDispatchRpcTest {
         every { HermesWsClient.send(any(), any(), any()) } answers {
             reqCount++
             val id = "req-id-$reqCount"
+            if (arg<String>(0) == WsMethods.SESSION_CREATE) {
+                sessionCreateRequestId = id
+            }
             arg<((String) -> Unit)?>(2)?.invoke(id)
             id
         }
@@ -152,8 +157,8 @@ class SlashCommandDispatchRpcTest {
         mockConnectionStatus.value = ConnectionStatus.CONNECTED
         mockEventsFlow.emit(WsEvent.GatewayReady(null))
         advanceUntilIdle()
-        // req-id-3 = session.create (after loadSessions + fetchCommandCatalog)
-        mockEventsFlow.emit(WsEvent.RpcResult("req-id-3", mapOf("session_id" to "session-xyz")))
+        val createRequestId = requireNotNull(sessionCreateRequestId) { "session.create was not sent" }
+        mockEventsFlow.emit(WsEvent.RpcResult(createRequestId, mapOf("session_id" to "session-xyz")))
         advanceUntilIdle()
         return Pair(vm, "session-xyz")
     }