feat: wire durable gateway SSE sends

2026-07-24 12:58:38 +00:00
parent 2a03caa7a9
commit c75dddae2b
5 changed files with 364 additions and 9 deletions
@@ -0,0 +1,101 @@
+package cloud.molberg.hermesmobile.streaming
+
+import cloud.molberg.hermesmobile.connection.GatewayRouteLease
+import cloud.molberg.hermesmobile.domain.ChatStateRepository
+import cloud.molberg.hermesmobile.domain.HermesConversationState
+import cloud.molberg.hermesmobile.domain.SessionStatus
+import java.util.concurrent.Executor
+
+sealed interface GatewaySendStart {
+    data class Started(val sessionId: String, val requestId: String) : GatewaySendStart
+    data object NoSelectedSession : GatewaySendStart
+    data object BlankDraft : GatewaySendStart
+    data object ConversationMismatch : GatewaySendStart
+    data object DuplicateSend : GatewaySendStart
+}
+
+data class GatewaySendState(
+    val profileId: String,
+    val sessionId: String,
+    val requestId: String,
+    val conversation: HermesConversationState
+)
+
+class DurableGatewaySendCoordinator(
+    private val chatStateRepository: ChatStateRepository,
+    private val streamingContract: GatewayStreamingContract,
+    private val streamRepository: GatewaySessionStreamRepository,
+    private val executor: Executor,
+    private val publishState: (GatewaySendState) -> Unit
+) {
+    fun sendSelected(
+        profileId: String,
+        routeLease: GatewayRouteLease,
+        bearerToken: String,
+        requestId: String,
+        initialState: HermesConversationState
+    ): GatewaySendStart {
+        val durableState = chatStateRepository.load(profileId)
+        val sessionId = durableState.selectedSessionId ?: return GatewaySendStart.NoSelectedSession
+        val prompt = durableState.drafts[sessionId]?.trim().orEmpty()
+        if (prompt.isBlank()) return GatewaySendStart.BlankDraft
+        if (initialState.session.id != sessionId) return GatewaySendStart.ConversationMismatch
+        if (!chatStateRepository.reserveSend(profileId, sessionId, requestId, prompt)) {
+            return GatewaySendStart.DuplicateSend
+        }
+
+        val streamRequest = try {
+            streamingContract.sessionChatRequest(routeLease, bearerToken, sessionId, prompt)
+        } catch (failure: RuntimeException) {
+            chatStateRepository.finishSend(profileId, sessionId, requestId)
+            throw failure
+        }
+
+        try {
+            executor.execute {
+                try {
+                    val finalState = streamRepository.execute(streamRequest, initialState) { state ->
+                        if (!state.session.status.isTerminal()) {
+                            publishIfCurrent(profileId, sessionId, requestId, state)
+                        }
+                    }
+                    if (finalState.session.status.isTerminal()) {
+                        finishAndPublishIfCurrent(profileId, sessionId, requestId, finalState)
+                    }
+                } catch (failure: RuntimeException) {
+                    chatStateRepository.finishSend(profileId, sessionId, requestId)
+                    throw failure
+                }
+            }
+        } catch (failure: RuntimeException) {
+            chatStateRepository.finishSend(profileId, sessionId, requestId)
+            throw failure
+        }
+        return GatewaySendStart.Started(sessionId, requestId)
+    }
+
+    private fun publishIfCurrent(
+        profileId: String,
+        sessionId: String,
+        requestId: String,
+        state: HermesConversationState
+    ) {
+        if (chatStateRepository.load(profileId).pendingSends[sessionId]?.requestId != requestId) return
+        publishState(GatewaySendState(profileId, sessionId, requestId, state))
+    }
+
+    private fun finishAndPublishIfCurrent(
+        profileId: String,
+        sessionId: String,
+        requestId: String,
+        state: HermesConversationState
+    ) {
+        if (chatStateRepository.load(profileId).pendingSends[sessionId]?.requestId != requestId) return
+        chatStateRepository.finishSend(profileId, sessionId, requestId)
+        if (chatStateRepository.load(profileId).pendingSends.containsKey(sessionId)) return
+        publishState(GatewaySendState(profileId, sessionId, requestId, state))
+    }
+
+    private fun SessionStatus.isTerminal(): Boolean =
+        this == SessionStatus.Completed || this == SessionStatus.Failed || this == SessionStatus.Cancelled
+}
@@ -14,7 +14,8 @@ class GatewaySessionStreamRepository(
 ) {
    fun execute(
        streamRequest: GatewayStreamRequest,
-        initialState: HermesConversationState
+        initialState: HermesConversationState,
+        publishState: (HermesConversationState) -> Unit = {}
    ): HermesConversationState {
        val lease = streamRequest.lease
        require(initialState.session.id == lease.sessionId) {
@@ -31,6 +32,7 @@ class GatewaySessionStreamRepository(
                    when (val decoded = GatewaySessionSseDecoder.decode(lease, eventName, data)) {
                        is GatewayDecodedStreamEvent.Domain -> {
                            state = BoundGatewayStreamReducer.reduce(lease, state, decoded.event)
+                            publishState(state)
                        }

                        is GatewayDecodedStreamEvent.Structured, null -> Unit
@@ -39,9 +41,9 @@ class GatewaySessionStreamRepository(
            }
            state
        } catch (failure: IOException) {
-            transportFailure(lease, state)
+            transportFailure(lease, state).also(publishState)
        } catch (failure: JSONException) {
-            transportFailure(lease, state)
+            transportFailure(lease, state).also(publishState)
        }
    }

@@ -0,0 +1,245 @@
+package cloud.molberg.hermesmobile.streaming
+
+import cloud.molberg.hermesmobile.connection.ConnectionRoute
+import cloud.molberg.hermesmobile.connection.GatewayRouteLease
+import cloud.molberg.hermesmobile.domain.HermesConversationState
+import cloud.molberg.hermesmobile.domain.HermesSession
+import cloud.molberg.hermesmobile.domain.MessagePart
+import cloud.molberg.hermesmobile.domain.SessionStatus
+import cloud.molberg.hermesmobile.fake.FakeChatStateRepository
+import java.io.IOException
+import java.util.concurrent.Executor
+import okhttp3.Interceptor
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.Protocol
+import okhttp3.Response
+import okhttp3.ResponseBody.Companion.toResponseBody
+import org.json.JSONObject
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class DurableGatewaySendCoordinatorTest {
+    @Test
+    fun successReservesOnceExecutesAsynchronouslyAndRestoresDurableSelectionAndDraft() {
+        val chatState = preparedChatState()
+        val executor = QueuedExecutor()
+        val published = mutableListOf<GatewaySendState>()
+        var requestedUrl: String? = null
+        var requestedInput: String? = null
+        val coordinator = coordinator(
+            chatState = chatState,
+            executor = executor,
+            published = published,
+            client = client { chain ->
+                requestedUrl = chain.request().url.toString()
+                requestedInput = JSONObject(chain.request().body!!.bodyUtf8()).getString("input")
+                response(
+                    chain,
+                    """
+                        event: assistant.delta
+                        data: {"session_id":"session-1","message_id":"message-1","delta":"Draft"}
+
+                        event: assistant.completed
+                        data: {"session_id":"session-1","message_id":"message-1","content":"Final answer"}
+
+                        event: done
+                        data: {"session_id":"session-1"}
+
+                    """.trimIndent()
+                )
+            }
+        )
+
+        assertEquals(
+            GatewaySendStart.Started("session-1", "request-1"),
+            coordinator.sendSelected(PROFILE_ID, LOCAL_LEASE, "token", "request-1", conversation())
+        )
+        assertEquals(
+            GatewaySendStart.DuplicateSend,
+            coordinator.sendSelected(PROFILE_ID, REMOTE_LEASE, "other", "request-2", conversation())
+        )
+        assertNull(requestedUrl)
+
+        executor.runNext()
+
+        assertEquals("https://local.example.test/api/sessions/session-1/chat/stream", requestedUrl)
+        assertEquals("durable prompt", requestedInput)
+        assertEquals(SessionStatus.Completed, published.last().conversation.session.status)
+        assertEquals(
+            "Final answer",
+            (published.last().conversation.messages.single().parts.single() as MessagePart.Text).text
+        )
+        assertDurableRestored(chatState)
+    }
+
+    @Test
+    fun gatewayFailurePublishesFailureAndClearsOnlyItsReservation() {
+        val chatState = preparedChatState()
+        val executor = QueuedExecutor()
+        val published = mutableListOf<GatewaySendState>()
+        val reportedLeases = mutableListOf<GatewayRouteLease>()
+        val coordinator = coordinator(
+            chatState = chatState,
+            executor = executor,
+            published = published,
+            reportedLeases = reportedLeases,
+            client = client { chain ->
+                response(
+                    chain,
+                    """
+                        event: error
+                        data: {"session_id":"session-1","message":"Hermes rejected the turn."}
+
+                    """.trimIndent()
+                )
+            }
+        )
+
+        coordinator.sendSelected(PROFILE_ID, LOCAL_LEASE, "", "request-1", conversation())
+        executor.runNext()
+
+        assertEquals(SessionStatus.Failed, published.single().conversation.session.status)
+        assertEquals("Hermes rejected the turn.", published.single().conversation.error)
+        assertTrue(reportedLeases.isEmpty())
+        assertDurableRestored(chatState)
+    }
+
+    @Test
+    fun transportFailurePublishesFailureAndReportsTheCapturedRouteLease() {
+        val chatState = preparedChatState()
+        val executor = QueuedExecutor()
+        val published = mutableListOf<GatewaySendState>()
+        val reportedLeases = mutableListOf<GatewayRouteLease>()
+        val coordinator = coordinator(
+            chatState = chatState,
+            executor = executor,
+            published = published,
+            reportedLeases = reportedLeases,
+            client = client { throw IOException("synthetic disconnect") }
+        )
+
+        coordinator.sendSelected(PROFILE_ID, LOCAL_LEASE, "", "request-1", conversation())
+        executor.runNext()
+
+        assertEquals(listOf(LOCAL_LEASE), reportedLeases)
+        assertEquals(SessionStatus.Failed, published.single().conversation.session.status)
+        assertEquals("Gateway stream transport failed.", published.single().conversation.error)
+        assertDurableRestored(chatState)
+    }
+
+    @Test
+    fun staleRequestIdCannotPublishOrClearANewerReservation() {
+        val chatState = preparedChatState()
+        val executor = QueuedExecutor()
+        val published = mutableListOf<GatewaySendState>()
+        val coordinator = coordinator(
+            chatState = chatState,
+            executor = executor,
+            published = published,
+            client = client { chain ->
+                response(
+                    chain,
+                    """
+                        event: done
+                        data: {"session_id":"session-1"}
+
+                    """.trimIndent()
+                )
+            }
+        )
+
+        coordinator.sendSelected(PROFILE_ID, LOCAL_LEASE, "", "request-1", conversation())
+        chatState.finishSend(PROFILE_ID, "session-1", "request-1")
+        assertTrue(chatState.reserveSend(PROFILE_ID, "session-1", "request-2", "newer prompt"))
+
+        executor.runNext()
+
+        assertTrue(published.isEmpty())
+        assertEquals("request-2", chatState.load(PROFILE_ID).pendingSends["session-1"]?.requestId)
+        assertEquals("session-1", chatState.load(PROFILE_ID).selectedSessionId)
+        assertEquals(" durable prompt ", chatState.load(PROFILE_ID).drafts["session-1"])
+    }
+
+    private fun coordinator(
+        chatState: FakeChatStateRepository,
+        executor: Executor,
+        published: MutableList<GatewaySendState>,
+        client: OkHttpClient,
+        reportedLeases: MutableList<GatewayRouteLease> = mutableListOf()
+    ): DurableGatewaySendCoordinator = DurableGatewaySendCoordinator(
+        chatStateRepository = chatState,
+        streamingContract = GatewayStreamingContract(
+            GatewayStreamingCapabilities(
+                sessionChatStreaming = true,
+                sessionChatStream = GatewayCapabilityEndpoint("POST", "/api/sessions/{session_id}/chat/stream")
+            )
+        ),
+        streamRepository = GatewaySessionStreamRepository(client, reportedLeases::add),
+        executor = executor,
+        publishState = published::add
+    )
+
+    private fun preparedChatState(): FakeChatStateRepository = FakeChatStateRepository().also { repository ->
+        repository.selectSession(PROFILE_ID, "session-1")
+        repository.saveDraft(PROFILE_ID, "session-1", " durable prompt ")
+    }
+
+    private fun assertDurableRestored(repository: FakeChatStateRepository) {
+        val durable = repository.load(PROFILE_ID)
+        assertEquals("session-1", durable.selectedSessionId)
+        assertEquals(" durable prompt ", durable.drafts["session-1"])
+        assertNull(durable.pendingSends["session-1"])
+    }
+
+    private fun conversation(): HermesConversationState = HermesConversationState(
+        session = HermesSession(
+            id = "session-1",
+            title = "Session",
+            subtitle = "Ready",
+            status = SessionStatus.Idle,
+            model = null,
+            workspace = null,
+            messageCount = 0,
+            createdAt = null,
+            updatedAt = null
+        )
+    )
+
+    private fun client(interceptor: (Interceptor.Chain) -> Response): OkHttpClient =
+        OkHttpClient.Builder().addInterceptor { chain -> interceptor(chain) }.build()
+
+    private fun response(chain: Interceptor.Chain, body: String): Response = Response.Builder()
+        .request(chain.request())
+        .protocol(Protocol.HTTP_1_1)
+        .code(200)
+        .message("OK")
+        .body(body.toResponseBody("text/event-stream".toMediaType()))
+        .build()
+
+    private fun okhttp3.RequestBody.bodyUtf8(): String {
+        val buffer = okio.Buffer()
+        writeTo(buffer)
+        return buffer.readUtf8()
+    }
+
+    private class QueuedExecutor : Executor {
+        private val tasks = ArrayDeque<Runnable>()
+
+        override fun execute(command: Runnable) {
+            tasks.addLast(command)
+        }
+
+        fun runNext() {
+            tasks.removeFirst().run()
+        }
+    }
+
+    private companion object {
+        const val PROFILE_ID = "profile-1"
+        val LOCAL_LEASE = GatewayRouteLease(ConnectionRoute.Local, "https://local.example.test", 11L)
+        val REMOTE_LEASE = GatewayRouteLease(ConnectionRoute.Remote, "https://remote.example.test", 12L)
+    }
+}