fix(#1119): handle gateway server requests (#1125)

This commit is contained in:
M57 authored and GitHub committed 2026-09-15 13:09:11 +03:00
1 parent fb0f93c000
commit e4be8b146e
16 files changed
+1169 -119

No files matched your search

@@ -19,9 +19,17 @@ object EventParser {
response: JsonRpcResponse,
rawJson: String = "",
): WsEvent {
// ── RPC response (has id) ────────────────────────────────────────
// JSON-RPC is peer-to-peer: an inbound frame with both `id` and `method`
// is a request from the gateway, not a response to one of our calls.
val id = response.id
if (id != null) {
if (response.method != null && id != null) {
@Suppress("UNCHECKED_CAST")
val requestParams = response.params?.toAny() as? Map<String, Any?> ?: emptyMap()
return WsEvent.ServerRequest(id, response.method, requestParams)
}
// A response has no method and carries a result or error member.
if (response.method == null && id != null && (response.result != null || response.error != null)) {
return if (response.error != null) {
WsEvent.RpcError(id, response.error)
} else {
@@ -220,6 +228,13 @@ object EventParser {
WsEvent.ClarifyExpire(clarifyId, sessionId)
}
"request.cancel" -> {
val requestId = payload?.get("id") as? String ?: ""
val requestMethod = payload?.get("method") as? String ?: ""
val reason = payload?.get("reason") as? String ?: ""
WsEvent.ServerRequestCancelled(requestId, requestMethod, reason, sessionId)
}
"status.update" -> {
val status = payload?.get("status") as? String
WsEvent.StatusUpdate(status, payload)
@@ -33,6 +33,8 @@ import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import okhttp3.Request
import okhttp3.RequestBody.Companion.toRequestBody
import okhttp3.Response
@@ -808,6 +810,55 @@ object HermesWsClient {
}
}
/**
* Answer a gateway-originated JSON-RPC request. This deliberately bypasses
* [send]: server requests already have an id allocated by the gateway and
* responses must not contain a method or receive a new client id.
*/
fun respondToServerRequest(
id: String,
result: JsonElement,
): Boolean {
if (id.isBlank()) return false
val json =
buildJsonObject {
put("jsonrpc", "2.0")
put("id", id)
put("result", result)
}.toString()
synchronized(outboundLock) {
val ws = webSocket
if (ws == null || !connected.get()) return false
return ws.send(json)
}
}
/** Answer a gateway-originated request with a standard JSON-RPC error. */
fun respondToServerRequestError(
id: String,
code: Int,
message: String,
): Boolean {
if (id.isBlank()) return false
val json =
buildJsonObject {
put("jsonrpc", "2.0")
put("id", id)
put(
"error",
buildJsonObject {
put("code", code)
put("message", message)
},
)
}.toString()
synchronized(outboundLock) {
val ws = webSocket
if (ws == null || !connected.get()) return false
return ws.send(json)
}
}
// ── Send helpers ─────────────────────────────────────────────────────
/**
@@ -1249,6 +1300,21 @@ object HermesWsClient {
resolvePending(rpcId, rpc.result, null)
}
// Resume/replay responses carry still-open server requests
// separately from the event ring. Re-deliver them through the
// same generic path before exposing the RPC result.
@Suppress("UNCHECKED_CAST")
val openRequests =
(rpc.result?.toAny() as? Map<*, *>)?.get("open_requests") as? List<Map<*, *>>
openRequests?.forEach { openRequest ->
val openId = openRequest["id"] as? String ?: return@forEach
val method = openRequest["method"] as? String ?: return@forEach
@Suppress("UNCHECKED_CAST")
val params = openRequest["params"] as? Map<String, Any?> ?: emptyMap()
parsedEvents.tryEmit(WsEvent.ServerRequest(openId, method, params, replayed = true))
}
if (rpc.id == null) {
@Suppress("UNCHECKED_CAST")
val params = rpc.params?.toAny() as? Map<String, Any?>
@@ -23,7 +23,9 @@ data class JsonRpcRequest(
)
/**
* Incoming JSON-RPC 2.0 response **or** server-pushed notification.
* Incoming JSON-RPC 2.0 response, server-to-client request, or notification.
* Frames carrying both [id] and [method] are classified as server requests by
* [EventParser], never as responses to a client RPC.
*/
@Serializable
data class JsonRpcResponse(
@@ -152,6 +152,8 @@ sealed class WsEvent {
val questionId: String? = null,
val multiSelect: Boolean = false,
val questions: List<ClarifyQuestion> = emptyList(),
val serverRequestId: String? = null,
val lockedAnswers: Map<String, String> = emptyMap(),
) : WsEvent()
data class ClarifyExpire(
@@ -213,6 +215,25 @@ sealed class WsEvent {
val sessionId: String? = null,
) : WsEvent()
/**
* Generic JSON-RPC server-to-client request. The gateway keeps the request
* open until the client sends a response frame with this exact [id].
*/
data class ServerRequest(
val id: String,
val method: String,
val params: Map<String, Any?> = emptyMap(),
val replayed: Boolean = false,
) : WsEvent()
/** The gateway withdrew a still-open server request (timeout/interruption). */
data class ServerRequestCancelled(
val id: String,
val method: String,
val reason: String,
val sessionId: String? = null,
) : WsEvent()
// ── RPC responses ────────────────────────────────────────────────────
data class RpcResult(
@@ -256,18 +277,21 @@ sealed class WsEvent {
* restricts to once/deny only (desktop `smartDenied` parity).
*/
val smartDenied: Boolean? = null,
/** JSON-RPC server-request id (`srq-*`) used for the response frame. */
val serverRequestId: String? = null,
) : WsEvent()
// ── Sudo / secret requests ─────────────────────────────────────────
/**
* Backend needs the user's sudo password to continue a turn
* (desktop: `sudo.request` → `sudo.respond {request_id, password}`).
* Mobile previously dropped this and the agent hung forever.
* New gateways use a same-ID JSON-RPC result frame; legacy gateways fall
* back to `sudo.respond`.
*/
data class SudoRequest(
val requestId: String?,
val sessionId: String?,
val serverRequestId: String? = null,
) : WsEvent()
/**
@@ -278,12 +302,13 @@ sealed class WsEvent {
data class SudoExpire(
val requestId: String?,
val sessionId: String?,
val serverRequestId: String? = null,
) : WsEvent()
/**
* Backend needs a secret value (password / token) to continue a turn
* (desktop: `secret.request` → `secret.respond {request_id, value}`).
* Mobile previously dropped this and the agent hung forever.
* New gateways use a same-ID JSON-RPC result frame; legacy gateways fall
* back to `secret.respond`.
* `envVar`/`prompt` mirror desktop: title = envVar ?: secretTitle,
* body = prompt ?: secretDesc.
*/
@@ -292,6 +317,7 @@ sealed class WsEvent {
val sessionId: String?,
val envVar: String? = null,
val prompt: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
/**
@@ -301,6 +327,7 @@ sealed class WsEvent {
data class SecretExpire(
val requestId: String?,
val sessionId: String?,
val serverRequestId: String? = null,
) : WsEvent()
// ── Credential Vault prompts (issue #1090) ──────────────────────────
@@ -310,11 +337,13 @@ sealed class WsEvent {
val sessionId: String? = null,
val backend: String? = null,
val displayName: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
data class VaultUnlockExpire(
val requestId: String?,
val sessionId: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
data class VaultSaveLoginRequest(
@@ -322,11 +351,13 @@ sealed class WsEvent {
val sessionId: String? = null,
val origin: String? = null,
val site: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
data class VaultSaveLoginExpire(
val requestId: String?,
val sessionId: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
data class VaultCodeRequest(
@@ -334,11 +365,13 @@ sealed class WsEvent {
val sessionId: String? = null,
val site: String? = null,
val hint: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
data class VaultCodeExpire(
val requestId: String?,
val sessionId: String? = null,
val serverRequestId: String? = null,
) : WsEvent()
// ── Gateway-level errors ───────────────────────────────────────────