From c75dddae2bb691e0860ea3d08b3cca7c607bb1fe Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 24 Jul 2026 12:58:38 +0000 Subject: [PATCH] feat: wire durable gateway SSE sends --- CHANGELOG.md | 7 + PROJECT_PLAN.md | 12 +- .../DurableGatewaySendCoordinator.kt | 101 ++++++++ .../GatewaySessionStreamRepository.kt | 8 +- .../DurableGatewaySendCoordinatorTest.kt | 245 ++++++++++++++++++ 5 files changed, 364 insertions(+), 9 deletions(-) create mode 100644 apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinator.kt create mode 100644 apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinatorTest.kt diff --git a/CHANGELOG.md b/CHANGELOG.md index a3c60cb..f096c6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ All notable Hermes Mobile source changes are recorded here. Entries are added du ## Unreleased ### Added +- Added a durable direct-gateway SSE send coordinator that captures the selected session, draft, request ID, and immutable route lease before asynchronous execution. +- Added deterministic coordinator tests for successful SSE completion, upstream gateway failure, transport failure, and stale request IDs. - Added an OkHttp-backed Android session stream repository that executes the capability-advertised SSE request and reduces decoded frames through the existing lease-bound foundation. - Added deterministic fake HTTP/SSE repository tests for authoritative completion, original-lease transport failure, and gateway error events that do not invalidate the route. - Added a capability-gated direct session SSE foundation with structured `{method, path}` endpoint parsing, immutable route/session leases, documented `input` request encoding, and core event decoding. @@ -26,6 +28,7 @@ All notable Hermes Mobile source changes are recorded here. Entries are added du - Added `PROJECT_PLAN.md` as the project-controlled milestone plan. ### Changed +- Changed direct session streaming to publish reducer updates to the owning durable send coordinator while retaining the original immutable route/session lease for the entire OkHttp call. - Changed direct session streaming so one immutable route/session lease owns the request and every decoded event for the full OkHttp call. - Changed streamed assistant reduction so deltas can recover a missing start event and the final upstream response replaces accumulated partial text. - Restructured the delivery plan so externally blocked B3 remains incomplete while durable chat-send integration with the completed SSE repository is the sole current atomic task. @@ -42,12 +45,16 @@ All notable Hermes Mobile source changes are recorded here. Entries are added du - Ignored `.gradle-user/` and `.dev/` alongside existing generated build artifacts. ### Fixed +- Fixed durable stream completion so a stale request ID cannot publish a terminal state or clear a newer reservation. - Fixed stream failure handling so HTTP, I/O, and malformed-SSE failures are reported and reduced only against the original lease, while upstream `error` events remain session failures rather than route failures. - Fixed bearer handling so `/health` is checked without credentials and the token is sent only to the documented authenticated `/v1/models` compatibility route. - Fixed legacy stored HTTP URLs so they cannot become active routes, and sanitized connection/request diagnostics so tokens and internal URLs are not emitted. - Fixed companion/mobile lint issues from missing Node globals and an unused React settings value. ### Verification +- `/root/.openclaw/workspace/scripts/hermes-mobile-preflight.sh` — passed; Android `:app:assembleDebug` BUILD SUCCESSFUL (35 tasks, 4 executed). +- Full Android Kotlin/Compose main-source and unit-test compilation through the cached Kotlin compiler plus direct JUnit execution — passed, 52 tests including B4c coordinator success, gateway-failure, transport-failure, and stale-request-ID coverage. +- `git diff --check` — passed for B4c. - Full Android Kotlin/Compose main-source and unit-test compilation through the cached Kotlin compiler plus direct JUnit execution — passed, 48 tests including the B4b reducer, stream-contract, and fake HTTP/SSE repository coverage. - `GRADLE_USER_HOME=/root/hermes-mobile/.gradle-user ./gradlew --no-daemon -Pkotlin.compiler.execution.strategy=in-process :app:testDebugUnitTest :app:assembleDebug` — blocked before project compilation because this sandbox prohibits Gradle's TCP control socket (`java.net.SocketException: Operation not permitted`). - `git diff --check` — passed for B4b. diff --git a/PROJECT_PLAN.md b/PROJECT_PLAN.md index cfa701c..abafd95 100644 --- a/PROJECT_PLAN.md +++ b/PROJECT_PLAN.md @@ -26,11 +26,6 @@ Complete the Android/client contract pivot before implementing backend behavior. ## Beta path — upstream-connected MVP -- [ ] **B4c — Wire gateway SSE into durable chat sends** - - Add one send coordinator that reserves the selected-session send, executes the capability-built stream request asynchronously, and publishes reducer state without reselecting the route mid-call. - - Clear only the matching durable pending-send reservation on terminal completion or failure; preserve duplicate-send protection and current draft/session restoration behavior. - - Verification: deterministic coordinator tests for success, gateway failure, transport failure, and stale request IDs; full cached Android Kotlin/JUnit verification. - ## Externally blocked - **B3 — Direct gateway session contract and durable chat state** @@ -38,7 +33,7 @@ Complete the Android/client contract pivot before implementing backend behavior. - Not complete: official upstream documentation still provides no session-search HTTP contract, and the configured local API server was not listening for a live compatibility exercise on 2026-07-24. - Unblock evidence is maintained in `docs/BLOCKERS.md`; do not mark B3 complete until documented search and real gateway behavior are verified. -- **Queued — B5: Tool, artifact, and output experience** +- [ ] **B5 — Tool, artifact, and output experience** - Compact tool cards with lifecycle/status, expandable output, copy/share/save affordances. - Safe rendering for large logs and code; no unbounded nested scroll failures. - Artifact/file handoff only where upstream capability is actually supported. @@ -71,6 +66,11 @@ Complete the Android/client contract pivot before implementing backend behavior. _Move finished items here with date, commit, and verification. Keep this section factual; do not claim unverified work._ +- [x] **B4c — Wire gateway SSE into durable chat sends** — 2026-07-24, commit `feat: wire durable gateway SSE sends`. + - Added a direct-gateway send coordinator that captures the selected session, draft, route lease, and capability-built SSE request before asynchronous execution; it never reselects a route or switches an active stream. + - Publishes only the currently reserved request's intermediate and terminal reducer states; terminal completion, gateway failure, and transport failure clear only the matching durable reservation while retaining the selected session and draft. + - Verification: `/root/.openclaw/workspace/scripts/hermes-mobile-preflight.sh` passed (Android `:app:assembleDebug`, 35 tasks, 4 executed); direct cached Kotlin/JUnit verification passed 52 tests including deterministic B4c success, gateway failure, transport failure, and stale request-ID tests; `git diff --check` passed. + - [x] **B4b — Execute session SSE through the Android repository** — 2026-07-24, commit `feat: execute gateway session SSE`. - Added a thin OkHttp repository that executes the capability-built session stream request and parses SSE frames into the existing decoder and lease-bound reducer. - Kept the route generation, request URL, session ID, endpoint, and event gate immutable for the full call; transport/protocol failure invalidates and fails only that original lease. diff --git a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinator.kt b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinator.kt new file mode 100644 index 0000000..f3f3c6f --- /dev/null +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinator.kt @@ -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 +} diff --git a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepository.kt b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepository.kt index 4e5f197..ef6297f 100644 --- a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepository.kt +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepository.kt @@ -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) } } diff --git a/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinatorTest.kt b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinatorTest.kt new file mode 100644 index 0000000..c79a7a6 --- /dev/null +++ b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/DurableGatewaySendCoordinatorTest.kt @@ -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() + 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() + val reportedLeases = mutableListOf() + 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() + val reportedLeases = mutableListOf() + 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() + 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, + client: OkHttpClient, + reportedLeases: MutableList = 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() + + 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) + } +}