diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f60e30..a3c60cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ All notable Hermes Mobile source changes are recorded here. Entries are added du ## Unreleased ### Added +- 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. +- Added deterministic stream tests for token deltas, authoritative final-response reconciliation, cross-session rejection, missed-start recovery, and capability-gated structured tool events. - Added B3 foundation contracts for session search plus capability-negotiated session list/create/get/delete requests, without enabling unadvertised gateway routes. - Added durable Android selected-session, per-session draft, and pending-send state with fake repository/reducer coverage for duplicate-send protection and restoration. - Added one securely stored Hermes gateway profile with optional Local HTTPS and required Remote HTTPS routes, bearer/no-auth configuration, and visible active-route diagnostics. @@ -22,6 +26,9 @@ 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 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. - Changed Android gateway routing to prefer a compatible Local route and deterministically fall back to Remote without switching an in-flight lease. - Changed the Android shell to stop calling undocumented legacy companion `/api/*` routes; unavailable workspace/chat utilities now remain gated until direct upstream contracts are verified. - Changed Android networking to reject cleartext, credential-bearing, path-bearing, duplicate, or missing gateway URLs and disabled application cleartext traffic. @@ -35,11 +42,19 @@ 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 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 +- 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. +- `/root/.openclaw/workspace/scripts/hermes-mobile-preflight.sh` — blocked before execution because this sandbox denies read/execute access to the required script (`Permission denied`). +- `GRADLE_USER_HOME=/root/hermes-mobile/.gradle-user ./gradlew --no-daemon -Pkotlin.compiler.execution.strategy=in-process :app:compileDebugUnitTestKotlin :app:assembleDebug` — blocked before project compilation because the Gradle daemon cannot open its sandbox-prohibited TCP control socket (`java.net.SocketException: Operation not permitted`). +- Focused cached Kotlin compilation plus direct JUnit execution of `HermesStreamReducerTest` and `GatewayStreamingContractTest` — passed, 12 tests. +- `GRADLE_USER_HOME=/root/hermes-mobile/.gradle-user ./gradlew :app:testDebugUnitTest --tests 'cloud.molberg.hermesmobile.domain.HermesStreamReducerTest' --tests 'cloud.molberg.hermesmobile.streaming.GatewayStreamingContractTest'` — blocked before compilation because the sandbox cannot determine a usable wildcard IP. - Full Android Kotlin/Compose source and unit-test compilation through the cached Kotlin compiler plus direct JUnit execution — passed, 37 tests including B3 session capability mapping, durable chat JSON/state, fake chat reducer, and fake repositories. - `GRADLE_USER_HOME=/root/hermes-mobile/.gradle-user ./gradlew :app:testDebugUnitTest` — blocked before compilation because this sandbox prohibits Gradle daemon/control sockets (`java.net.SocketException: Operation not permitted`). - `npm run typecheck && npm run lint && npm run build` — passed for all workspaces. diff --git a/PROJECT_PLAN.md b/PROJECT_PLAN.md index 45e9c04..cfa701c 100644 --- a/PROJECT_PLAN.md +++ b/PROJECT_PLAN.md @@ -26,17 +26,17 @@ Complete the Android/client contract pivot before implementing backend behavior. ## Beta path — upstream-connected MVP -- [ ] **B3 — Direct gateway session contract and durable chat state** - - Fetch/list/search/select/create sessions through repository contracts after upstream session routes are verified. - - Restore selected session and protect against duplicate sends / lost drafts. - - Verification: fake-repository UI tests; direct gateway compatibility checks where available. - - Foundation verified 2026-07-24: repository search contract and fake implementation, capability-negotiated session list/create/get/delete request mapping, durable selected-session/draft/pending-send storage, and duplicate-send reducer/UI protection. - - Still blocked: official upstream docs do not document a session-search endpoint or query schema, and no live gateway compatibility exercise is available in this sandbox. Keep B3 unchecked until search and real gateway behavior are verified. +- [ ] **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. -- **Queued after B3 — B4: Direct gateway streaming, reconnects, and resilient long-running work** - - Stream event reducer: final response/token deltas first; tool lifecycle, session events, artifacts, cancel, and reconnect/backoff only where upstream capability is verified. - - Preserve partial output and make recovery obvious and safe. - - Verification: deterministic fake stream tests and real direct gateway exercise. +## Externally blocked + +- **B3 — Direct gateway session contract and durable chat state** + - Foundation verified 2026-07-24: repository search contract and fake implementation, capability-gated session list/create/get/delete mapping, durable selected-session/draft/pending-send storage, and duplicate-send reducer/UI protection. + - 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** - Compact tool cards with lifecycle/status, expandable output, copy/share/save affordances. @@ -71,6 +71,18 @@ 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] **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. + - Kept gateway `error` events separate from transport failure and excluded reconnect, cancellation, tool/artifact handling, and UI integration. + - Verification: focused cached Kotlin compilation and direct JUnit execution passed 15 reducer, stream-contract, and fake HTTP/SSE repository tests; Android Gradle build attempt and final checks are recorded in `CHANGELOG.md`. + +- [x] **B4a — Capability-gated direct streaming foundation** — 2026-07-24, commit `feat: add gateway streaming foundation`. + - Added capability-derived session SSE request construction using the advertised `POST` method and `{session_id}` path only. + - Bound each stream to an immutable route generation and session ID, decoded documented core lifecycle/delta/final events, and gated structured tool events on `tool_progress_events`. + - Made final assistant responses authoritative over accumulated deltas and ignored events from a different session lease. + - Verification: focused cached Kotlin compilation plus direct JUnit execution passed 12 reducer/stream contract tests; full checks recorded in `CHANGELOG.md`. + - [x] **B2 — Direct gateway connection, authentication, and local-first routing UX** — 2026-07-24, commit `e5cc27e` (`feat: add local-first gateway routing`). - Added one logical profile with optional Local HTTPS, required Remote HTTPS, Android-keystore-protected bearer/no-auth configuration, and active Local/Remote diagnostics. - Added bounded redirect-disabled `GET /health` probes followed by bearer-authenticated `GET /v1/models`, deterministic Local preference/Remote fallback, and immutable request leases. diff --git a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/domain/HermesStreamReducer.kt b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/domain/HermesStreamReducer.kt index 7dd4cd6..61bc938 100644 --- a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/domain/HermesStreamReducer.kt +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/domain/HermesStreamReducer.kt @@ -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.upsert(message: HermesMessage): List = if (any { it.id == message.id }) map { if (it.id == message.id) message else it } else this + message + private fun List.appendAssistantDelta(event: HermesStreamEvent.AssistantDelta): List { + 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)) diff --git a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/BoundGatewayStreamReducer.kt b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/BoundGatewayStreamReducer.kt new file mode 100644 index 0000000..d78364d --- /dev/null +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/BoundGatewayStreamReducer.kt @@ -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) + } +} diff --git a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionSseDecoder.kt b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionSseDecoder.kt new file mode 100644 index 0000000..8d236ba --- /dev/null +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionSseDecoder.kt @@ -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") +} 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 new file mode 100644 index 0000000..4e5f197 --- /dev/null +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepository.kt @@ -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() + + 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() + } +} diff --git a/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewayStreamingContract.kt b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewayStreamingContract.kt new file mode 100644 index 0000000..0038213 --- /dev/null +++ b/apps/mobile/android/app/src/main/java/cloud/molberg/hermesmobile/streaming/GatewayStreamingContract.kt @@ -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() + } +} diff --git a/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/domain/HermesStreamReducerTest.kt b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/domain/HermesStreamReducerTest.kt index 116c34c..f17711a 100644 --- a/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/domain/HermesStreamReducerTest.kt +++ b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/domain/HermesStreamReducerTest.kt @@ -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) diff --git a/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepositoryTest.kt b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepositoryTest.kt new file mode 100644 index 0000000..ea698a0 --- /dev/null +++ b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/GatewaySessionStreamRepositoryTest.kt @@ -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() + 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() + 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(), 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) + } +} diff --git a/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/GatewayStreamingContractTest.kt b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/GatewayStreamingContractTest.kt new file mode 100644 index 0000000..2ab1d44 --- /dev/null +++ b/apps/mobile/android/app/src/test/java/cloud/molberg/hermesmobile/streaming/GatewayStreamingContractTest.kt @@ -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 + ) + ) +} diff --git a/docs/BLOCKERS.md b/docs/BLOCKERS.md index 792f5ac..010c91d 100644 --- a/docs/BLOCKERS.md +++ b/docs/BLOCKERS.md @@ -11,8 +11,10 @@ Upstream compatibility needed for stronger binding: a documented stable gateway The official Hermes Agent API server documentation checked on 2026-07-24 documents authenticated `GET /v1/capabilities` and session endpoint capability entries for list, create, get, update, delete, messages, fork, chat, and chat streaming. It does not document a session-search endpoint, query parameter/body schema, response envelope, or a `session_search` capability entry. -B3 therefore keeps `searchSessions` in the Android repository contract and verifies it with the fake repository, but deliberately provides no direct-gateway HTTP mapping for search. List/create/get/delete request construction is safe only when the live gateway advertises the corresponding endpoint string through `GET /v1/capabilities`. +B3 therefore keeps `searchSessions` in the Android repository contract and verifies it with the fake repository, but deliberately provides no direct-gateway HTTP mapping for search. Current official upstream API-server behavior represents capability endpoints as structured `{method, path}` objects; the earlier Android B3 string-only endpoint parser has not been exercised against a live gateway and is not treated as verified compatibility. + +A local Hermes API-server configuration was present on 2026-07-24 with `API_SERVER_ENABLED=true`, host `0.0.0.0`, port `18793`, and a configured key. No process was listening on port `18793`, and `GET http://127.0.0.1:18793/health` was unreachable (`curl` HTTP code `000`), so no live session compatibility exercise was available. Upstream evidence needed to unblock B3: official documentation for session search including capability key, HTTP method, path, query/body schema, response envelope, and empty/error semantics, followed by a live compatibility exercise against an upstream Hermes gateway. -Source checked: `https://hermes-agent.nousresearch.com/docs/user-guide/features/api-server/` on 2026-07-24. +Sources checked on 2026-07-24: `https://hermes-agent.nousresearch.com/docs/user-guide/features/api-server/`; installed official upstream checkout `NousResearch/hermes-agent` at commit `36f2a966c`, including `gateway/platforms/api_server.py` capability and session-stream behavior. diff --git a/docs/DIRECT_GATEWAY_ARCHITECTURE.md b/docs/DIRECT_GATEWAY_ARCHITECTURE.md index c3db867..085b524 100644 --- a/docs/DIRECT_GATEWAY_ARCHITECTURE.md +++ b/docs/DIRECT_GATEWAY_ARCHITECTURE.md @@ -30,10 +30,24 @@ The current repository/context documents these upstream Hermes API-server endpoi | --- | --- | --- | | `GET /health` | Documented | Primary reachability and compatibility probe. Response shape beyond a successful JSON response is treated as server-specific. | | `GET /v1/models` | Documented | Authenticated compatibility probe for bearer-token connections and future model picker source. | +| `GET /v1/capabilities` | Documented | Capability negotiation source. Current upstream behavior advertises endpoints as `{method, path}` objects and feature support as booleans. | | `POST /v1/chat/completions` | Documented | Baseline OpenAI-compatible chat path. Mobile must assume this may not expose Hermes session IDs or structured tool events. | | `POST /v1/responses` | Documented | Preferred future request/streaming path if upstream exposes richer response events. Exact Hermes event mapping is not yet verified. | +| `/api/sessions/*` | Documented and capability-advertised | List/create/get/update/delete/messages/fork/chat/chat-stream routes exist, but mobile must use only the exact advertised method/path object and must not infer search or other routes. | +| `POST /api/sessions/{session_id}/chat/stream` | Documented and capability-advertised | SSE session turn. Core mobile support begins with `message.started`, `assistant.delta`, authoritative `assistant.completed`, `error`, and `done`; tool events remain gated by `tool_progress_events`. | -These are the only upstream API-server routes currently treated as documented by this repository. Any route not listed here must be labeled provisional until checked against official Hermes Agent docs or a running gateway. +Any route not listed here must be labeled provisional until checked against official Hermes Agent docs or a running gateway. Session search remains explicitly unverified and unmapped. + +## Direct stream lease contract + +- Parse `session_chat_streaming`, `tool_progress_events`, and `endpoints.session_chat_stream` from `GET /v1/capabilities`. +- Require the advertised session stream endpoint to be `POST`, use an absolute path containing `{session_id}`, and send the documented JSON `input` field. +- Bind the request URL, route generation, session ID, endpoint path, and structured-event gate into one immutable lease before the HTTP call starts. +- Execute that prebuilt request through OkHttp without consulting route selection again until the call ends. +- Never apply an SSE event whose `session_id` differs from the bound lease. A later Local/Remote re-evaluation applies only to a later request. +- Append `assistant.delta` text deterministically, then replace it with the authoritative `assistant.completed.content` response when present. +- Report HTTP, I/O, or malformed-SSE failure only against the route/session lease that owned the failed call; an upstream `error` event fails the session without invalidating the route. +- Decode tool lifecycle/progress only when `tool_progress_events` is advertised; B4 does not render or persist those structured payloads. ## Unverified Assumptions @@ -41,8 +55,8 @@ The following are useful design assumptions, not confirmed contracts: - Bearer authentication is accepted through an `Authorization: Bearer ` header on `/v1/*`. - `/health` returns a stable `status` field. -- `/v1/responses` can provide enough streaming metadata to preserve Hermes sessions, tool lifecycle, artifacts, cancellation, and reconnect state. -- Hermes session listing, creation, continuation, and deletion have stable direct HTTP endpoints. +- `/v1/responses` can provide enough streaming metadata to preserve Hermes sessions, artifacts, cancellation, and reconnect state. +- Hermes session search has a stable direct HTTP endpoint. - Tool result and artifact listing/download have stable direct HTTP endpoints. - Cancellation has a stable direct HTTP endpoint. - Error payloads consistently include `message` or `error`. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index c8b2df2..b3c932d 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -9,6 +9,7 @@ Done: - Android connection flow stores one logical gateway profile with optional Local HTTPS, required Remote HTTPS, Android-keystore-protected bearer/no-auth configuration, and active Local/Remote diagnostics. - Local-first route selection uses bounded `GET /health` probes, authenticated `GET /v1/models` compatibility checks, deterministic Remote fallback, and foreground/connectivity/request-failure re-evaluation without switching in-flight leases. - Android no longer calls undocumented legacy companion `/api/*` routes from the direct-gateway client path. +- Capability-gated session streaming binds each OkHttp call to an immutable route/session lease, parses SSE into authoritative final responses, ignores cross-session events, and reports transport failure only against the failed lease. - Chat UI supports conversation list/thread navigation, native composer, busy state, selectable replies, error bubbles, and new-chat reset. - Direct gateway compatibility contract is documented in `docs/DIRECT_GATEWAY_ARCHITECTURE.md`. - Legacy companion chat, terminal, and file utilities remain in the repo during migration but are not the target architecture. @@ -16,7 +17,7 @@ Done: - Companion and workspace TypeScript typecheck, build, and lint pass as of 2026-07-24. Remaining for beta: -- Complete B3 after upstream documents a session-search endpoint/query schema and a live gateway can verify advertised session CRUD behavior; the independently verifiable durable-state and capability-gated request foundation landed on 2026-07-24. +- Complete B3 after upstream documents a session-search endpoint/query schema and a live gateway can verify advertised session CRUD behavior; the independently verifiable durable-state foundation landed on 2026-07-24, but the current upstream `{method, path}` capability envelope has not been exercised live. - Exercise the connection/settings unhappy paths on a real device or emulator; this sandbox cannot start ADB because local control sockets are prohibited. - Exercise a real device or emulator against a running upstream Hermes gateway/API server. - Verify direct gateway behavior end to end beyond B2: chat request/response, session continuity expectations, streamed/final output handling, and failure recovery.