feat: execute gateway session SSE

2026-07-24 12:39:51 +00:00
parent 8e7b5c6b44
commit 2a03caa7a9
13 changed files with 713 additions and 19 deletions
@@ -15,9 +15,8 @@ object HermesStreamReducer {
            )

            is HermesStreamEvent.AssistantDelta -> state.copy(
-                messages = state.messages.map { message ->
-                    if (message.id == event.messageId) message.appendText(event.text).copy(status = MessageStatus.Streaming) else message
-                },
+                messages = state.messages.appendAssistantDelta(event),
+                session = state.session.copy(status = SessionStatus.Running, subtitle = "Hermes is streaming..."),
                error = null
            )

@@ -51,6 +50,20 @@ object HermesStreamReducer {
    private fun List<HermesMessage>.upsert(message: HermesMessage): List<HermesMessage> =
        if (any { it.id == message.id }) map { if (it.id == message.id) message else it } else this + message

+    private fun List<HermesMessage>.appendAssistantDelta(event: HermesStreamEvent.AssistantDelta): List<HermesMessage> {
+        val existing = firstOrNull { it.id == event.messageId }
+        val streamingMessage = existing?.appendText(event.text)?.copy(status = MessageStatus.Streaming)
+            ?: HermesMessage(
+                id = event.messageId,
+                sessionId = event.sessionId,
+                role = MessageRole.Assistant,
+                parts = listOf(MessagePart.Text(event.text)),
+                status = MessageStatus.Streaming,
+                createdAt = event.timestamp
+            )
+        return upsert(streamingMessage)
+    }
+
    private fun HermesMessage.appendText(delta: String): HermesMessage {
        val lastTextIndex = parts.indexOfLast { it is MessagePart.Text }
        if (lastTextIndex < 0) return copy(parts = parts + MessagePart.Text(delta))
@@ -0,0 +1,16 @@
+package cloud.molberg.hermesmobile.streaming
+
+import cloud.molberg.hermesmobile.domain.HermesConversationState
+import cloud.molberg.hermesmobile.domain.HermesStreamEvent
+import cloud.molberg.hermesmobile.domain.HermesStreamReducer
+
+object BoundGatewayStreamReducer {
+    fun reduce(
+        lease: GatewayStreamLease,
+        state: HermesConversationState,
+        event: HermesStreamEvent
+    ): HermesConversationState {
+        if (state.session.id != lease.sessionId || event.sessionId != lease.sessionId) return state
+        return HermesStreamReducer.reduce(state, event)
+    }
+}
@@ -0,0 +1,94 @@
+package cloud.molberg.hermesmobile.streaming
+
+import cloud.molberg.hermesmobile.domain.HermesMessage
+import cloud.molberg.hermesmobile.domain.HermesStreamEvent
+import cloud.molberg.hermesmobile.domain.MessagePart
+import cloud.molberg.hermesmobile.domain.MessageRole
+import cloud.molberg.hermesmobile.domain.MessageStatus
+import org.json.JSONObject
+
+sealed interface GatewayDecodedStreamEvent {
+    data class Domain(val event: HermesStreamEvent) : GatewayDecodedStreamEvent
+    data class Structured(val name: String, val payload: JSONObject) : GatewayDecodedStreamEvent
+}
+
+object GatewaySessionSseDecoder {
+    fun decode(lease: GatewayStreamLease, eventName: String, data: String): GatewayDecodedStreamEvent? {
+        if (!lease.accepts(eventName)) return null
+        val payload = JSONObject(data)
+        val sessionId = payload.optString("session_id").ifBlank { lease.sessionId }
+        val timestamp = payload.optDouble("ts").takeUnless { it.isNaN() }?.toString()
+        val eventId = listOf(payload.optString("run_id"), payload.optString("seq"))
+            .filter { it.isNotBlank() }
+            .joinToString(":")
+            .ifBlank { eventName }
+        return when (eventName) {
+            "message.started" -> payload.optJSONObject("message")?.let { message ->
+                val messageId = message.optString("id")
+                if (messageId.isBlank()) null else GatewayDecodedStreamEvent.Domain(
+                    HermesStreamEvent.MessageStarted(
+                        eventId,
+                        sessionId,
+                        timestamp,
+                        HermesMessage(
+                            id = messageId,
+                            sessionId = sessionId,
+                            role = MessageRole.Assistant,
+                            parts = emptyList(),
+                            status = MessageStatus.Streaming,
+                            createdAt = timestamp
+                        )
+                    )
+                )
+            }
+
+            "assistant.delta" -> payload.requiredString("message_id")?.let { messageId ->
+                GatewayDecodedStreamEvent.Domain(
+                    HermesStreamEvent.AssistantDelta(
+                        eventId,
+                        sessionId,
+                        timestamp,
+                        messageId,
+                        payload.optString("delta")
+                    )
+                )
+            }
+
+            "assistant.completed" -> payload.requiredString("message_id")?.let { messageId ->
+                GatewayDecodedStreamEvent.Domain(
+                    HermesStreamEvent.MessageCompleted(
+                        eventId,
+                        sessionId,
+                        timestamp,
+                        HermesMessage(
+                            id = messageId,
+                            sessionId = sessionId,
+                            role = MessageRole.Assistant,
+                            parts = listOf(MessagePart.Text(payload.optString("content"))),
+                            status = MessageStatus.Complete,
+                            createdAt = timestamp
+                        )
+                    )
+                )
+            }
+
+            "error" -> GatewayDecodedStreamEvent.Domain(
+                HermesStreamEvent.Failed(
+                    eventId,
+                    sessionId,
+                    timestamp,
+                    payload.optString("message").ifBlank { "Gateway stream failed." },
+                    recoverable = true
+                )
+            )
+
+            "done" -> GatewayDecodedStreamEvent.Domain(HermesStreamEvent.Done(eventId, sessionId, timestamp))
+            in STRUCTURED_EVENTS -> GatewayDecodedStreamEvent.Structured(eventName, payload)
+            else -> null
+        }
+    }
+
+    private fun JSONObject.requiredString(name: String): String? = optString(name).takeIf { it.isNotBlank() }
+
+    private val STRUCTURED_EVENTS = setOf("tool.progress", "tool.started", "tool.completed", "tool.failed")
+}
@@ -0,0 +1,95 @@
+package cloud.molberg.hermesmobile.streaming
+
+import cloud.molberg.hermesmobile.connection.GatewayRouteLease
+import cloud.molberg.hermesmobile.domain.HermesConversationState
+import cloud.molberg.hermesmobile.domain.HermesStreamEvent
+import java.io.IOException
+import okhttp3.OkHttpClient
+import okio.BufferedSource
+import org.json.JSONException
+
+class GatewaySessionStreamRepository(
+    private val client: OkHttpClient,
+    private val reportTransportFailure: (GatewayRouteLease) -> Unit
+) {
+    fun execute(
+        streamRequest: GatewayStreamRequest,
+        initialState: HermesConversationState
+    ): HermesConversationState {
+        val lease = streamRequest.lease
+        require(initialState.session.id == lease.sessionId) {
+            "Gateway stream state must match the leased session."
+        }
+        var state = initialState
+        return try {
+            client.newCall(streamRequest.request).execute().use { response ->
+                if (!response.isSuccessful) {
+                    throw IOException("Gateway stream returned HTTP ${response.code}.")
+                }
+                val body = response.body ?: throw IOException("Gateway stream response body is missing.")
+                readEvents(body.source()) { eventName, data ->
+                    when (val decoded = GatewaySessionSseDecoder.decode(lease, eventName, data)) {
+                        is GatewayDecodedStreamEvent.Domain -> {
+                            state = BoundGatewayStreamReducer.reduce(lease, state, decoded.event)
+                        }
+
+                        is GatewayDecodedStreamEvent.Structured, null -> Unit
+                    }
+                }
+            }
+            state
+        } catch (failure: IOException) {
+            transportFailure(lease, state)
+        } catch (failure: JSONException) {
+            transportFailure(lease, state)
+        }
+    }
+
+    private fun transportFailure(
+        lease: GatewayStreamLease,
+        state: HermesConversationState
+    ): HermesConversationState {
+        reportTransportFailure(lease.routeLease)
+        return BoundGatewayStreamReducer.reduce(
+            lease,
+            state,
+            HermesStreamEvent.Failed(
+                id = "transport:${lease.routeLease.generation}",
+                sessionId = lease.sessionId,
+                timestamp = null,
+                message = "Gateway stream transport failed.",
+                recoverable = true
+            )
+        )
+    }
+
+    private fun readEvents(source: BufferedSource, consume: (String, String) -> Unit) {
+        var eventName: String? = null
+        val dataLines = mutableListOf<String>()
+
+        fun dispatch() {
+            val name = eventName
+            if (name != null && dataLines.isNotEmpty()) consume(name, dataLines.joinToString("\n"))
+            eventName = null
+            dataLines.clear()
+        }
+
+        while (true) {
+            val line = source.readUtf8Line() ?: break
+            if (line.isEmpty()) {
+                dispatch()
+                continue
+            }
+            if (line.startsWith(':')) continue
+            val separator = line.indexOf(':')
+            val field = if (separator < 0) line else line.substring(0, separator)
+            val rawValue = if (separator < 0) "" else line.substring(separator + 1)
+            val value = rawValue.removePrefix(" ")
+            when (field) {
+                "event" -> eventName = value
+                "data" -> dataLines += value
+            }
+        }
+        dispatch()
+    }
+}
@@ -0,0 +1,103 @@
+package cloud.molberg.hermesmobile.streaming
+
+import cloud.molberg.hermesmobile.connection.GatewayRouteLease
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.Request
+import okhttp3.RequestBody.Companion.toRequestBody
+import org.json.JSONObject
+
+data class GatewayCapabilityEndpoint(
+    val method: String,
+    val path: String
+)
+
+data class GatewayStreamingCapabilities(
+    val sessionChatStreaming: Boolean = false,
+    val toolProgressEvents: Boolean = false,
+    val sessionChatStream: GatewayCapabilityEndpoint? = null
+) {
+    companion object {
+        fun fromCapabilities(json: JSONObject): GatewayStreamingCapabilities {
+            val features = json.optJSONObject("features")
+            val endpoint = json.optJSONObject("endpoints")
+                ?.optJSONObject("session_chat_stream")
+                ?.let { GatewayCapabilityEndpoint(it.optString("method"), it.optString("path")) }
+                ?.takeIf { it.method.isNotBlank() && it.path.isNotBlank() }
+            return GatewayStreamingCapabilities(
+                sessionChatStreaming = features?.optBoolean("session_chat_streaming") == true,
+                toolProgressEvents = features?.optBoolean("tool_progress_events") == true,
+                sessionChatStream = endpoint
+            )
+        }
+    }
+}
+
+data class GatewayStreamLease(
+    val routeLease: GatewayRouteLease,
+    val sessionId: String,
+    val endpointPath: String,
+    val structuredEventsEnabled: Boolean
+) {
+    fun accepts(eventName: String): Boolean =
+        eventName in CORE_EVENTS || structuredEventsEnabled && eventName in STRUCTURED_EVENTS
+
+    private companion object {
+        val CORE_EVENTS = setOf(
+            "run.started",
+            "message.started",
+            "assistant.delta",
+            "assistant.completed",
+            "run.completed",
+            "error",
+            "done"
+        )
+        val STRUCTURED_EVENTS = setOf("tool.progress", "tool.started", "tool.completed", "tool.failed")
+    }
+}
+
+data class GatewayStreamRequest(
+    val lease: GatewayStreamLease,
+    val request: Request
+)
+
+class GatewayStreamingContract(private val capabilities: GatewayStreamingCapabilities) {
+    fun sessionChatRequest(
+        routeLease: GatewayRouteLease,
+        bearerToken: String,
+        sessionId: String,
+        input: String
+    ): GatewayStreamRequest {
+        require(capabilities.sessionChatStreaming) { "Gateway does not advertise session_chat_streaming." }
+        val endpoint = requireNotNull(capabilities.sessionChatStream) {
+            "Gateway does not advertise the session_chat_stream endpoint."
+        }
+        require(endpoint.method == "POST") { "Gateway session_chat_stream method must be POST." }
+        require(endpoint.path.startsWith('/')) { "Gateway session_chat_stream path must be absolute." }
+        require(endpoint.path.contains("{session_id}")) {
+            "Gateway session_chat_stream path must document a {session_id} placeholder."
+        }
+        val endpointPath = endpoint.path.replace("{session_id}", encode(sessionId))
+        val lease = GatewayStreamLease(
+            routeLease = routeLease,
+            sessionId = sessionId,
+            endpointPath = endpointPath,
+            structuredEventsEnabled = capabilities.toolProgressEvents
+        )
+        val body = JSONObject().put("input", input).toString().toRequestBody(JSON_MEDIA_TYPE)
+        val builder = Request.Builder()
+            .url(routeLease.baseUrl.trimEnd('/') + endpointPath)
+            .header("Accept", "text/event-stream")
+            .post(body)
+        if (bearerToken.isNotBlank()) builder.header("Authorization", "Bearer $bearerToken")
+        return GatewayStreamRequest(lease, builder.build())
+    }
+
+    private fun encode(value: String): String =
+        URLEncoder.encode(value, StandardCharsets.UTF_8.name()).replace("+", "%20")
+
+    private companion object {
+        val JSON_MEDIA_TYPE = "application/json; charset=utf-8".toMediaType()
+    }
+}
@@ -32,6 +32,42 @@ class HermesStreamReducerTest {
        assertNull(result.error)
    }

+    @Test
+    fun assistantDeltaCreatesStreamingMessageWhenStartEventWasMissed() {
+        val result = HermesStreamReducer.reduce(
+            baseState(),
+            HermesStreamEvent.AssistantDelta("e1", "s1", null, "a1", "Recovered delta")
+        )
+
+        assertEquals("Recovered delta", (result.messages.single().parts.single() as MessagePart.Text).text)
+        assertEquals(MessageStatus.Streaming, result.messages.single().status)
+        assertEquals(SessionStatus.Running, result.session.status)
+    }
+
+    @Test
+    fun finalResponseReplacesAccumulatedDeltaText() {
+        val streamed = HermesStreamReducer.reduce(
+            baseState(),
+            HermesStreamEvent.AssistantDelta("e1", "s1", null, "a1", "Partial answer")
+        )
+        val finalMessage = HermesMessage(
+            "a1",
+            "s1",
+            MessageRole.Assistant,
+            listOf(MessagePart.Text("Authoritative final answer")),
+            MessageStatus.Complete,
+            null
+        )
+
+        val result = HermesStreamReducer.reduce(
+            streamed,
+            HermesStreamEvent.MessageCompleted("e2", "s1", null, finalMessage)
+        )
+
+        assertEquals("Authoritative final answer", (result.messages.single().parts.single() as MessagePart.Text).text)
+        assertEquals(SessionStatus.Completed, result.session.status)
+    }
+
    @Test
    fun toolAndArtifactEventsAreStoredById() {
        val tool = ToolResult("t1", "call-1", "workspace.inspect", ToolStatus.Running, "Running", null, "", null, null, null)
@@ -0,0 +1,143 @@
+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 java.io.IOException
+import okhttp3.Interceptor
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.Protocol
+import okhttp3.Response
+import okhttp3.ResponseBody.Companion.toResponseBody
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNull
+import org.junit.Test
+
+class GatewaySessionStreamRepositoryTest {
+    @Test
+    fun fakeHttpSseCallReducesFramesThroughTheBoundLease() {
+        val streamRequest = request(routeLease = LOCAL_LEASE)
+        var requestedUrl: String? = null
+        val repository = GatewaySessionStreamRepository(
+            client = client { chain ->
+                requestedUrl = chain.request().url.toString()
+                response(
+                    chain,
+                    """
+                        : keepalive
+
+                        event: message.started
+                        data: {"session_id":"session-1","run_id":"run-1","seq":1,"message":{"id":"message-1"}}
+
+                        event: assistant.delta
+                        data: {"session_id":"session-1","run_id":"run-1","seq":2,"message_id":"message-1","delta":"Draft"}
+
+                        event: assistant.completed
+                        data: {"session_id":"session-1","run_id":"run-1","seq":3,"message_id":"message-1","content":"Authoritative final"}
+
+                        event: done
+                        data: {"session_id":"session-1","run_id":"run-1","seq":4}
+
+                    """.trimIndent()
+                )
+            },
+            reportTransportFailure = { error("Successful stream must not report route failure.") }
+        )
+
+        val result = repository.execute(streamRequest, state("session-1"))
+
+        assertEquals("https://local.example.test/api/sessions/session-1/chat/stream", requestedUrl)
+        assertEquals("Authoritative final", (result.messages.single().parts.single() as MessagePart.Text).text)
+        assertEquals(SessionStatus.Completed, result.session.status)
+        assertNull(result.error)
+    }
+
+    @Test
+    fun transportFailureIsReportedAndReducedOnlyAgainstTheOriginalLease() {
+        val streamRequest = request(routeLease = LOCAL_LEASE)
+        val reportedLeases = mutableListOf<GatewayRouteLease>()
+        val repository = GatewaySessionStreamRepository(
+            client = client { throw IOException("synthetic disconnect") },
+            reportTransportFailure = reportedLeases::add
+        )
+
+        val result = repository.execute(streamRequest, state("session-1"))
+
+        assertEquals(listOf(LOCAL_LEASE), reportedLeases)
+        assertEquals("session-1", result.session.id)
+        assertEquals(SessionStatus.Failed, result.session.status)
+        assertEquals("Gateway stream transport failed.", result.error)
+    }
+
+    @Test
+    fun gatewayErrorFrameDoesNotReportTransportFailure() {
+        val reportedLeases = mutableListOf<GatewayRouteLease>()
+        val repository = GatewaySessionStreamRepository(
+            client = client { chain ->
+                response(
+                    chain,
+                    """
+                        event: error
+                        data: {"session_id":"session-1","message":"Hermes rejected the turn."}
+
+                    """.trimIndent()
+                )
+            },
+            reportTransportFailure = reportedLeases::add
+        )
+
+        val result = repository.execute(request(LOCAL_LEASE), state("session-1"))
+
+        assertEquals(emptyList<GatewayRouteLease>(), reportedLeases)
+        assertEquals(SessionStatus.Failed, result.session.status)
+        assertEquals("Hermes rejected the turn.", result.error)
+    }
+
+    private fun request(routeLease: GatewayRouteLease): GatewayStreamRequest =
+        GatewayStreamingContract(
+            GatewayStreamingCapabilities(
+                sessionChatStreaming = true,
+                sessionChatStream = GatewayCapabilityEndpoint(
+                    method = "POST",
+                    path = "/api/sessions/{session_id}/chat/stream"
+                )
+            )
+        ).sessionChatRequest(routeLease, "", "session-1", "hello")
+
+    private fun state(sessionId: String): HermesConversationState =
+        HermesConversationState(
+            HermesSession(
+                id = sessionId,
+                title = sessionId,
+                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 companion object {
+        val LOCAL_LEASE = GatewayRouteLease(ConnectionRoute.Local, "https://local.example.test", 11L)
+    }
+}
@@ -0,0 +1,150 @@
+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 okio.Buffer
+import org.json.JSONObject
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertThrows
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class GatewayStreamingContractTest {
+    @Test
+    fun capabilitiesProvideStructuredSessionStreamEndpointAndToolGate() {
+        val capabilities = capabilities(toolProgressEvents = true)
+
+        assertTrue(capabilities.sessionChatStreaming)
+        assertEquals("POST", capabilities.sessionChatStream?.method)
+        assertEquals("/api/sessions/{session_id}/chat/stream", capabilities.sessionChatStream?.path)
+        assertTrue(capabilities.toolProgressEvents)
+    }
+
+    @Test
+    fun requestUsesAdvertisedEndpointAndBindsImmutableRouteSessionLease() {
+        val routeLease = GatewayRouteLease(ConnectionRoute.Local, "https://local.example.test", 7L)
+        val stream = GatewayStreamingContract(capabilities()).sessionChatRequest(
+            routeLease,
+            "secret-token",
+            "session / one",
+            "Continue the build"
+        )
+        val body = Buffer().also { stream.request.body!!.writeTo(it) }.readUtf8()
+        val laterRoute = GatewayRouteLease(ConnectionRoute.Remote, "https://remote.example.test", 8L)
+
+        assertEquals("https://local.example.test/api/sessions/session%20%2F%20one/chat/stream", stream.request.url.toString())
+        assertEquals("Bearer secret-token", stream.request.header("Authorization"))
+        assertEquals("Continue the build", JSONObject(body).getString("input"))
+        assertEquals(routeLease, stream.lease.routeLease)
+        assertFalse(stream.lease.routeLease == laterRoute)
+    }
+
+    @Test
+    fun absentOrInvalidStreamCapabilityStaysGated() {
+        assertThrows(IllegalArgumentException::class.java) {
+            GatewayStreamingContract(GatewayStreamingCapabilities()).sessionChatRequest(
+                GatewayRouteLease(ConnectionRoute.Remote, "https://remote.example.test", 1L),
+                "",
+                "session-1",
+                "hello"
+            )
+        }
+    }
+
+    @Test
+    fun deterministicFakeStreamReducesDeltasThenAuthoritativeFinalResponse() {
+        val lease = GatewayStreamingContract(capabilities()).sessionChatRequest(
+            GatewayRouteLease(ConnectionRoute.Remote, "https://remote.example.test", 3L),
+            "",
+            "session-1",
+            "hello"
+        ).lease
+        val events = listOf(
+            "message.started" to """{"session_id":"session-1","run_id":"run-1","seq":1,"message":{"id":"message-1","role":"assistant"}}""",
+            "assistant.delta" to """{"session_id":"session-1","run_id":"run-1","seq":2,"message_id":"message-1","delta":"Hello"}""",
+            "assistant.delta" to """{"session_id":"session-1","run_id":"run-1","seq":3,"message_id":"message-1","delta":" draft"}""",
+            "assistant.completed" to """{"session_id":"session-1","run_id":"run-1","seq":4,"message_id":"message-1","content":"Hello final"}""",
+            "done" to """{"session_id":"session-1","run_id":"run-1","seq":5}"""
+        )
+        val result = events.fold(baseState()) { state, (name, data) ->
+            val decoded = GatewaySessionSseDecoder.decode(lease, name, data)
+            assertNotNull(decoded)
+            BoundGatewayStreamReducer.reduce(lease, state, (decoded as GatewayDecodedStreamEvent.Domain).event)
+        }
+
+        assertEquals("Hello final", (result.messages.single().parts.single() as MessagePart.Text).text)
+        assertEquals(SessionStatus.Completed, result.session.status)
+    }
+
+    @Test
+    fun foreignSessionEventsAndUnadvertisedStructuredEventsAreIgnored() {
+        val lease = GatewayStreamingContract(capabilities(toolProgressEvents = false)).sessionChatRequest(
+            GatewayRouteLease(ConnectionRoute.Remote, "https://remote.example.test", 3L),
+            "",
+            "session-1",
+            "hello"
+        ).lease
+        val foreign = GatewaySessionSseDecoder.decode(
+            lease,
+            "assistant.delta",
+            """{"session_id":"session-2","message_id":"message-2","delta":"wrong session"}"""
+        ) as GatewayDecodedStreamEvent.Domain
+
+        assertEquals(baseState(), BoundGatewayStreamReducer.reduce(lease, baseState(), foreign.event))
+        assertNull(GatewaySessionSseDecoder.decode(lease, "tool.started", """{"tool_name":"terminal"}"""))
+    }
+
+    @Test
+    fun structuredEventsDecodeOnlyWhenCapabilityIsAdvertised() {
+        val lease = GatewayStreamingContract(capabilities(toolProgressEvents = true)).sessionChatRequest(
+            GatewayRouteLease(ConnectionRoute.Remote, "https://remote.example.test", 3L),
+            "",
+            "session-1",
+            "hello"
+        ).lease
+
+        val decoded = GatewaySessionSseDecoder.decode(lease, "tool.started", """{"tool_name":"terminal"}""")
+
+        assertTrue(decoded is GatewayDecodedStreamEvent.Structured)
+    }
+
+    private fun capabilities(toolProgressEvents: Boolean = false): GatewayStreamingCapabilities =
+        GatewayStreamingCapabilities.fromCapabilities(
+            JSONObject(
+                """{
+                    "features": {
+                        "session_chat_streaming": true,
+                        "tool_progress_events": $toolProgressEvents
+                    },
+                    "endpoints": {
+                        "session_chat_stream": {
+                            "method": "POST",
+                            "path": "/api/sessions/{session_id}/chat/stream"
+                        }
+                    }
+                }""".trimIndent()
+            )
+        )
+
+    private fun baseState(): HermesConversationState =
+        HermesConversationState(
+            session = HermesSession(
+                id = "session-1",
+                title = "Test",
+                subtitle = "Ready",
+                status = SessionStatus.Idle,
+                model = null,
+                workspace = null,
+                messageCount = 0,
+                createdAt = null,
+                updatedAt = null
+            )
+        )
+}