feat(connectors): surface active session integrations (#1091) (#1100)

app/src/main/java/com/m57/hermescontrol/data/local/AuthManager.kt

@@ -2,6 +2,7 @@ package com.m57.hermescontrol.data.local
 
 import android.content.Context
 import android.content.SharedPreferences
+import androidx.annotation.VisibleForTesting
 import androidx.security.crypto.EncryptedSharedPreferences
 import androidx.security.crypto.MasterKeys
 import com.m57.hermescontrol.data.config.ConnectionProfile
@@ -500,6 +501,11 @@ object AuthManager {
         _baseUrlFlow.value = ""
     }
 
+    @VisibleForTesting
+    fun setBaseUrlForTest(baseUrl: String) {
+        _baseUrlFlow.value = baseUrl
+    }
+
     fun getToken(): String? {
         if (tokenInitialized) return cachedToken
         synchronized(this) {

app/src/main/java/com/m57/hermescontrol/data/model/ConnectorModels.kt

@@ -0,0 +1,233 @@
+package com.m57.hermescontrol.data.model
+
+import com.m57.hermescontrol.util.ConnectorUrlValidator
+
+/**
+ * High-level connection status for a connector item.
+ * Deliberate precedence: if [connected] is true, status resolves to [CONNECTED]
+ * regardless of expired/revoked/stale statuses reported by backend.
+ */
+enum class ConnectorStatus {
+    CONNECTED,
+    INITIATED,
+    EXPIRED,
+    REVOKED,
+    FAILED,
+    DISCONNECTED,
+    UNKNOWN,
+    ;
+
+    val isConnected: Boolean get() = this == CONNECTED
+
+    companion object {
+        fun from(
+            connected: Boolean,
+            rawStatus: String?,
+        ): ConnectorStatus = from(connected = connected as Boolean?, rawStatus = rawStatus)
+
+        fun from(
+            connected: Boolean?,
+            rawStatus: String?,
+        ): ConnectorStatus {
+            if (connected == true) return CONNECTED
+            if (connected == false) {
+                return when (rawStatus?.lowercase()?.trim()) {
+                    "expired" -> EXPIRED
+                    "revoked" -> REVOKED
+                    "initiated" -> INITIATED
+                    "failed", "error" -> FAILED
+                    "disconnected" -> DISCONNECTED
+                    null, "", "active", "connected", "authorized" -> DISCONNECTED
+                    else -> UNKNOWN
+                }
+            }
+            // Missing boolean: "authorized", "active", or "connected" yields effective CONNECTED
+            return when (rawStatus?.lowercase()?.trim()) {
+                "authorized", "active", "connected" -> CONNECTED
+                "expired" -> EXPIRED
+                "revoked" -> REVOKED
+                "initiated" -> INITIATED
+                "failed", "error" -> FAILED
+                "disconnected" -> DISCONNECTED
+                null, "" -> DISCONNECTED
+                else -> UNKNOWN
+            }
+        }
+    }
+}
+
+/**
+ * Status resulting from a connect/reconnect initiation.
+ */
+enum class ConnectorConnectStatus {
+    ACTIVE,
+    INITIATED,
+    FAILED,
+    UNKNOWN,
+    ;
+
+    val isActive: Boolean get() = this == ACTIVE
+
+    companion object {
+        fun from(rawStatus: String?): ConnectorConnectStatus =
+            when (rawStatus?.lowercase()?.trim()) {
+                "active", "connected" -> ACTIVE
+                "initiated" -> INITIATED
+                "failed", "error" -> FAILED
+                null, "" -> UNKNOWN
+                else -> UNKNOWN
+            }
+    }
+}
+
+/**
+ * Individual connector item returned by `connectors.list`.
+ */
+data class ConnectorItem(
+    val connector: String,
+    val name: String = connector,
+    val connected: Boolean = false,
+    val enabled: Boolean = true,
+    val connectionStatus: String? = null,
+    val status: ConnectorStatus = ConnectorStatus.from(connected, connectionStatus),
+    val connectUrl: String? = null,
+) {
+    val isConnected: Boolean get() = connected || status.isConnected
+
+    /** Safe validated HTTPS URL, or null if invalid or absent. */
+    val safeConnectUrl: String?
+        get() = if (ConnectorUrlValidator.isValidHttpsUrl(connectUrl)) connectUrl else null
+
+    override fun toString(): String {
+        val redactedUrl = if (connectUrl != null) "[REDACTED]" else "null"
+        return "ConnectorItem(" +
+            "connector=$connector, " +
+            "name=$name, " +
+            "connected=$connected, " +
+            "enabled=$enabled, " +
+            "status=$status, " +
+            "connectUrl=$redactedUrl" +
+            ")"
+    }
+}
+
+/**
+ * Result of a connect/reconnect action for an individual connector.
+ * Instructions from arbitrary backends are intentionally not exposed for UI display.
+ */
+data class ConnectorConnectItem(
+    val connector: String,
+    val status: ConnectorConnectStatus,
+    val connectUrl: String? = null,
+) {
+    val isConnectUrlValid: Boolean
+        get() = ConnectorUrlValidator.isValidHttpsUrl(connectUrl)
+
+    val safeConnectUrl: String?
+        get() = if (isConnectUrlValid) connectUrl else null
+
+    override fun toString(): String {
+        val redactedUrl = if (connectUrl != null) "[REDACTED]" else "null"
+        return "ConnectorConnectItem(" +
+            "connector=$connector, " +
+            "status=$status, " +
+            "connectUrl=$redactedUrl, " +
+            "isConnectUrlValid=$isConnectUrlValid" +
+            ")"
+    }
+}
+
+/**
+ * Summary counts returned by `connectors.connect`.
+ */
+data class ConnectorConnectSummary(
+    val total: Int = 0,
+    val active: Int = 0,
+    val initiated: Int = 0,
+    val failed: Int = 0,
+)
+
+/**
+ * Typed safe connector errors mapped from RPC codes and reasons.
+ * Raw exceptions or auth tokens are never exposed.
+ */
+sealed interface ConnectorError {
+    val message: String
+
+    data class InvalidParams(
+        override val message: String = "Invalid connector parameters.",
+    ) : ConnectorError
+
+    data class NotOwner(
+        override val message: String = "Session is not owned by this transport.",
+    ) : ConnectorError
+
+    data class Unavailable(
+        override val message: String = "Connectors are not available in this session.",
+    ) : ConnectorError
+
+    data class UnsupportedRuntime(
+        override val message: String = "Connectors must be managed on the session's compute host.",
+    ) : ConnectorError
+
+    data class RequestFailed(
+        override val message: String = "Connector request failed. Please try again.",
+    ) : ConnectorError
+
+    data class InvalidResponse(
+        override val message: String = "Connector service returned an invalid response.",
+    ) : ConnectorError
+
+    data class UnsupportedBackend(
+        override val message: String = "Connector service is not supported by this backend.",
+    ) : ConnectorError
+
+    data class MalformedEnvelope(
+        override val message: String = "Malformed response envelope received from gateway.",
+    ) : ConnectorError
+
+    data class NetworkError(
+        override val message: String = "Failed to communicate with gateway.",
+    ) : ConnectorError
+
+    data class Other(
+        val code: Int = 0,
+        override val message: String = "Connector operation failed.",
+    ) : ConnectorError
+}
+
+/**
+ * Result envelope for `connectors.list`.
+ */
+sealed interface ConnectorListResult {
+    data class Success(
+        val available: Boolean,
+        val connectors: List<ConnectorItem>,
+    ) : ConnectorListResult
+
+    data class Error(
+        val error: ConnectorError,
+    ) : ConnectorListResult
+
+    fun getOrNull(): List<ConnectorItem>? = (this as? Success)?.connectors
+
+    fun errorOrNull(): ConnectorError? = (this as? Error)?.error
+}
+
+/**
+ * Result envelope for `connectors.connect`.
+ */
+sealed interface ConnectorConnectResult {
+    data class Success(
+        val results: List<ConnectorConnectItem>,
+        val summary: ConnectorConnectSummary,
+    ) : ConnectorConnectResult
+
+    data class Error(
+        val error: ConnectorError,
+    ) : ConnectorConnectResult
+
+    fun getOrNull(): List<ConnectorConnectItem>? = (this as? Success)?.results
+
+    fun errorOrNull(): ConnectorError? = (this as? Error)?.error
+}

app/src/main/java/com/m57/hermescontrol/data/ws/ConnectorParser.kt

@@ -0,0 +1,389 @@
+package com.m57.hermescontrol.data.ws
+
+import com.m57.hermescontrol.data.model.ConnectorConnectItem
+import com.m57.hermescontrol.data.model.ConnectorConnectResult
+import com.m57.hermescontrol.data.model.ConnectorConnectStatus
+import com.m57.hermescontrol.data.model.ConnectorConnectSummary
+import com.m57.hermescontrol.data.model.ConnectorError
+import com.m57.hermescontrol.data.model.ConnectorItem
+import com.m57.hermescontrol.data.model.ConnectorListResult
+import com.m57.hermescontrol.data.model.ConnectorStatus
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonArray
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonNull
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.booleanOrNull
+import kotlinx.serialization.json.intOrNull
+
+/**
+ * Dedicated parser for connector payloads with strict envelope validation,
+ * resilient type coercion, legacy alias normalization, and safe error mapping.
+ */
+object ConnectorParser {
+    private val json =
+        Json {
+            ignoreUnknownKeys = true
+            isLenient = true
+        }
+
+    /**
+     * Parse raw result from `connectors.list` RPC.
+     * Fails with [ConnectorError.MalformedEnvelope] if the required envelope structure is missing.
+     */
+    fun parseListResult(raw: Any?): ConnectorListResult {
+        val root =
+            toJsonElement(raw) as? JsonObject
+                ?: return ConnectorListResult.Error(
+                    ConnectorError.MalformedEnvelope("Response envelope must be a JSON object."),
+                )
+
+        // Required envelope check: "connectors" must exist and be an array.
+        val connectorsArray =
+            root["connectors"] as? JsonArray
+                ?: return ConnectorListResult.Error(
+                    ConnectorError.MalformedEnvelope("Envelope missing required 'connectors' array."),
+                )
+
+        // Optional 'available' flag (defaults to true if omitted, unless explicitly false)
+        val available = parseBoolean(root["available"], default = true)
+
+        val items = mutableListOf<ConnectorItem>()
+        val seenSlugs = mutableSetOf<String>()
+        for (element in connectorsArray) {
+            val itemObj = element as? JsonObject ?: continue
+            val item = parseConnectorItem(itemObj) ?: continue
+            if (seenSlugs.add(item.connector)) {
+                items.add(item)
+            }
+        }
+
+        // Entirely invalid nonempty connectors array must fail rather than looking silently valid empty.
+        if (connectorsArray.isNotEmpty() && items.isEmpty()) {
+            return ConnectorListResult.Error(
+                ConnectorError.MalformedEnvelope("Connectors list contains only invalid items."),
+            )
+        }
+
+        return ConnectorListResult.Success(
+            available = available,
+            connectors = items,
+        )
+    }
+
+    /**
+     * Parse raw result from `connectors.connect` RPC.
+     * Fails with [ConnectorError.MalformedEnvelope] or [ConnectorError.InvalidResponse] if invalid.
+     */
+    fun parseConnectResult(raw: Any?): ConnectorConnectResult {
+        val rawElement = toJsonElement(raw)
+
+        // Tolerate purposeful legacy bare array form: [ { "connector": "linear", ... } ]
+        if (rawElement is JsonArray) {
+            val results = mutableListOf<ConnectorConnectItem>()
+            val seenSlugs = mutableSetOf<String>()
+            for (element in rawElement) {
+                val itemObj = element as? JsonObject ?: continue
+                val item = parseConnectItem(itemObj) ?: continue
+                if (seenSlugs.add(item.connector)) {
+                    results.add(item)
+                }
+            }
+            if (results.isNotEmpty()) {
+                val summary =
+                    ConnectorConnectSummary(
+                        total = results.size,
+                        active = results.count { it.status == ConnectorConnectStatus.ACTIVE },
+                        initiated = results.count { it.status == ConnectorConnectStatus.INITIATED },
+                        failed = results.count { it.status == ConnectorConnectStatus.FAILED },
+                    )
+                return ConnectorConnectResult.Success(
+                    results = results,
+                    summary = summary,
+                )
+            }
+            return ConnectorConnectResult.Error(
+                ConnectorError.InvalidResponse("Connect response contains no valid results."),
+            )
+        }
+
+        val root =
+            rawElement as? JsonObject
+                ?: return ConnectorConnectResult.Error(
+                    ConnectorError.MalformedEnvelope("Connect response must be a JSON object."),
+                )
+
+        // Required envelope check: "results" must exist and be an array.
+        val resultsArray =
+            root["results"] as? JsonArray
+                ?: return ConnectorConnectResult.Error(
+                    ConnectorError.MalformedEnvelope("Connect envelope missing required 'results' array."),
+                )
+
+        val results = mutableListOf<ConnectorConnectItem>()
+        val seenSlugs = mutableSetOf<String>()
+        for (element in resultsArray) {
+            val itemObj = element as? JsonObject ?: continue
+            val item = parseConnectItem(itemObj) ?: continue
+            if (seenSlugs.add(item.connector)) {
+                results.add(item)
+            }
+        }
+
+        // Connect must return at least nonempty valid results
+        if (results.isEmpty()) {
+            return ConnectorConnectResult.Error(
+                ConnectorError.InvalidResponse("Connect response contains no valid results."),
+            )
+        }
+
+        // Current summary object is required in standard envelope
+        val summaryObj =
+            root["summary"] as? JsonObject
+                ?: return ConnectorConnectResult.Error(
+                    ConnectorError.MalformedEnvelope("Connect envelope missing required 'summary' object."),
+                )
+        val summary = parseConnectSummary(summaryObj)
+
+        return ConnectorConnectResult.Success(
+            results = results,
+            summary = summary,
+        )
+    }
+
+    private fun parseConnectorItem(obj: JsonObject): ConnectorItem? {
+        // Strict string identifier with backend slug validation; legacy aliases supported
+        val slug =
+            parseIdentifier(obj["connector"])
+                ?: parseIdentifier(obj["slug"])
+                ?: parseIdentifier(obj["id"])
+                ?: return null // Skip items without valid connector identifier
+
+        // Display name: primary "name", legacy aliases "displayName", "title", fallback to slug
+        val name =
+            parseString(obj["name"])
+                ?: parseString(obj["displayName"])
+                ?: parseString(obj["title"])
+                ?: slug
+
+        val rawConnected = parseBooleanOrNull(obj["connected"])
+        val enabled = parseBoolean(obj["enabled"], default = true)
+
+        // Connection status string: primary "connectionStatus", legacy aliases "status", "connection_status"
+        val rawStatus =
+            parseString(obj["connectionStatus"])
+                ?: parseString(obj["status"])
+                ?: parseString(obj["connection_status"])
+
+        // Compute typed status respecting connected precedence:
+        // When connected boolean is missing (legacy), status "authorized"/"connected"/"active" yields CONNECTED.
+        // Explicit connected=false is authoritative and not overridden by status active.
+        val status = ConnectorStatus.from(connected = rawConnected, rawStatus = rawStatus)
+        val effectiveConnected = rawConnected ?: status.isConnected
+
+        // Connect URL (transient): primary "connect_url", legacy aliases "connectUrl", "auth_url", "authUrl"
+        val connectUrl =
+            parseString(obj["connect_url"])
+                ?: parseString(obj["connectUrl"])
+                ?: parseString(obj["auth_url"])
+                ?: parseString(obj["authUrl"])
+
+        return ConnectorItem(
+            connector = slug,
+            name = name,
+            connected = effectiveConnected,
+            enabled = enabled,
+            connectionStatus = rawStatus,
+            status = status,
+            connectUrl = connectUrl,
+        )
+    }
+
+    private fun parseConnectItem(obj: JsonObject): ConnectorConnectItem? {
+        val slug =
+            parseIdentifier(obj["connector"])
+                ?: parseIdentifier(obj["slug"])
+                ?: parseIdentifier(obj["id"])
+                ?: return null
+
+        val rawStatus =
+            parseString(obj["status"])
+                ?: parseString(obj["connectionStatus"])
+                ?: parseString(obj["connection_status"])
+
+        val status = ConnectorConnectStatus.from(rawStatus)
+
+        val connectUrl =
+            parseString(obj["connect_url"])
+                ?: parseString(obj["connectUrl"])
+                ?: parseString(obj["auth_url"])
+                ?: parseString(obj["authUrl"])
+
+        return ConnectorConnectItem(
+            connector = slug,
+            status = status,
+            connectUrl = connectUrl,
+        )
+    }
+
+    private fun parseConnectSummary(obj: JsonObject?): ConnectorConnectSummary {
+        if (obj == null) return ConnectorConnectSummary()
+        return ConnectorConnectSummary(
+            total = parseInt(obj["total"], default = 0),
+            active = parseInt(obj["active"], default = 0),
+            initiated = parseInt(obj["initiated"], default = 0),
+            failed = parseInt(obj["failed"], default = 0),
+        )
+    }
+
+    /**
+     * Map RPC error codes and reasons into typed [ConnectorError]s.
+     * Raw server strings and sensitive parameters are never forwarded or exposed in error messages.
+     */
+    fun mapRpcError(
+        code: Int,
+        message: String? = null,
+        data: JsonElement? = null,
+    ): ConnectorError {
+        val reason =
+            when (data) {
+                is JsonObject -> (data["reason"] as? JsonPrimitive)?.content
+                is JsonPrimitive -> data.content
+                else -> null
+            }?.uppercase()?.trim()
+
+        return when {
+            code == 4000 || reason == "INVALID_PARAMS" -> {
+                ConnectorError.InvalidParams()
+            }
+
+            code == 4001 || reason == "NOT_OWNER" -> {
+                ConnectorError.NotOwner()
+            }
+
+            code == 4031 || reason == "CONNECTORS_UNAVAILABLE" -> {
+                ConnectorError.Unavailable()
+            }
+
+            code == 5033 || reason == "UNSUPPORTED_RUNTIME" -> {
+                ConnectorError.UnsupportedRuntime()
+            }
+
+            code == 5034 && reason == "INVALID_CONNECTOR_RESPONSE" -> {
+                ConnectorError.InvalidResponse()
+            }
+
+            code == 5034 -> {
+                ConnectorError.RequestFailed()
+            }
+
+            code == -32601 -> {
+                ConnectorError.UnsupportedBackend()
+            }
+
+            else -> {
+                ConnectorError.Other(code = code)
+            }
+        }
+    }
+
+    internal fun parseStrictString(element: JsonElement?): String? {
+        if (element == null || element is JsonNull) return null
+        if (element is JsonPrimitive && element.isString) {
+            return element.content
+        }
+        return null
+    }
+
+    internal fun parseIdentifier(element: JsonElement?): String? {
+        val str = parseStrictString(element)?.trim() ?: return null
+        return if (ConnectorRepository.isValidSlug(str)) str else null
+    }
+
+    internal fun parseString(element: JsonElement?): String? {
+        if (element == null || element is JsonNull) return null
+        if (element is JsonPrimitive) {
+            if (!element.isString && element.content.isBlank()) return null
+            return element.content
+        }
+        return null
+    }
+
+    internal fun parseBooleanOrNull(element: JsonElement?): Boolean? {
+        if (element == null || element is JsonNull) return null
+        if (element is JsonPrimitive) {
+            element.booleanOrNull?.let { return it }
+            val content = element.content.trim()
+            if (content.equals("true", ignoreCase = true) || content == "1") return true
+            if (content.equals("false", ignoreCase = true) || content == "0") return false
+        }
+        return null
+    }
+
+    internal fun parseBoolean(
+        element: JsonElement?,
+        default: Boolean,
+    ): Boolean = parseBooleanOrNull(element) ?: default
+
+    internal fun parseInt(
+        element: JsonElement?,
+        default: Int,
+    ): Int {
+        if (element == null || element is JsonNull) return default
+        if (element is JsonPrimitive) {
+            element.intOrNull?.let { return it }
+            element.content.toIntOrNull()?.let { return it }
+        }
+        return default
+    }
+
+    fun toJsonElement(value: Any?): JsonElement =
+        when (value) {
+            null -> {
+                JsonNull
+            }
+
+            is JsonElement -> {
+                value
+            }
+
+            is Map<*, *> -> {
+                JsonObject(
+                    @Suppress("UNCHECKED_CAST")
+                    (value as Map<String, Any?>).mapValues { (_, v) -> toJsonElement(v) },
+                )
+            }
+
+            is List<*> -> {
+                JsonArray(value.map { toJsonElement(it) })
+            }
+
+            is String -> {
+                val trimmed = value.trim()
+                if ((trimmed.startsWith("{") && trimmed.endsWith("}")) ||
+                    (trimmed.startsWith("[") && trimmed.endsWith("]"))
+                ) {
+                    try {
+                        json.parseToJsonElement(value)
+                    } catch (_: Exception) {
+                        JsonPrimitive(value)
+                    }
+                } else {
+                    JsonPrimitive(value)
+                }
+            }
+
+            is Boolean -> {
+                JsonPrimitive(value)
+            }
+
+            is Number -> {
+                JsonPrimitive(value)
+            }
+
+            else -> {
+                JsonPrimitive(value.toString())
+            }
+        }
+}

app/src/main/java/com/m57/hermescontrol/data/ws/ConnectorRepository.kt

@@ -0,0 +1,126 @@
+package com.m57.hermescontrol.data.ws
+
+import com.m57.hermescontrol.data.model.ConnectorConnectResult
+import com.m57.hermescontrol.data.model.ConnectorError
+import com.m57.hermescontrol.data.model.ConnectorListResult
+import kotlinx.coroutines.CancellationException
+
+/**
+ * Repository interface for connector operations: `connectors.list` and `connectors.connect` (issue #1091).
+ * Public injectable interface enabling isolated testing in ViewModels and controllers.
+ */
+interface ConnectorRepository {
+    /**
+     * Request connector catalog and authorization status for [sessionId].
+     */
+    suspend fun listConnectors(sessionId: String): ConnectorListResult
+
+    /**
+     * Initiate authorization flow for [connectors] slugs in [sessionId].
+     * If [reconnect] is true, restarts authorization flow even if already connected.
+     */
+    suspend fun connect(
+        sessionId: String,
+        connectors: List<String>,
+        reconnect: Boolean = false,
+    ): ConnectorConnectResult
+
+    companion object : ConnectorRepository by HermesConnectorRepository() {
+        private val SLUG_REGEX = Regex("^[a-z0-9][a-z0-9_-]*$")
+
+        fun isValidSlug(slug: String?): Boolean {
+            if (slug.isNullOrBlank()) return false
+            return SLUG_REGEX.matches(slug)
+        }
+    }
+}
+
+/**
+ * Default implementation of [ConnectorRepository] backed by [HermesWsClient].
+ */
+class HermesConnectorRepository(
+    private val rpcRequest: suspend (method: String, params: Map<String, Any>) -> Any? = { method, params ->
+        val deferred = HermesWsClient.request(method, params)
+        try {
+            deferred.await()
+        } catch (e: CancellationException) {
+            deferred.cancel(e)
+            throw e
+        } finally {
+            if (!deferred.isCompleted) {
+                deferred.cancel()
+            }
+        }
+    },
+) : ConnectorRepository {
+    override suspend fun listConnectors(sessionId: String): ConnectorListResult {
+        if (sessionId.isBlank()) {
+            return ConnectorListResult.Error(
+                ConnectorError.InvalidParams("session_id required"),
+            )
+        }
+
+        return try {
+            val params = mapOf("session_id" to sessionId)
+            val result = rpcRequest(WsMethods.CONNECTORS_LIST, params)
+            ConnectorParser.parseListResult(result)
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: HermesWsClient.HermesRpcException) {
+            ConnectorListResult.Error(
+                ConnectorParser.mapRpcError(code = e.code, message = e.message ?: "", data = e.data),
+            )
+        } catch (_: Exception) {
+            ConnectorListResult.Error(
+                ConnectorError.NetworkError("Failed to communicate with gateway."),
+            )
+        }
+    }
+
+    override suspend fun connect(
+        sessionId: String,
+        connectors: List<String>,
+        reconnect: Boolean,
+    ): ConnectorConnectResult {
+        if (sessionId.isBlank()) {
+            return ConnectorConnectResult.Error(
+                ConnectorError.InvalidParams("session_id required"),
+            )
+        }
+
+        if (connectors.isEmpty()) {
+            return ConnectorConnectResult.Error(
+                ConnectorError.InvalidParams("connectors must be nonempty slugs; reconnect must be boolean"),
+            )
+        }
+
+        for (slug in connectors) {
+            if (!ConnectorRepository.isValidSlug(slug)) {
+                return ConnectorConnectResult.Error(
+                    ConnectorError.InvalidParams("connectors must be nonempty slugs; reconnect must be boolean"),
+                )
+            }
+        }
+
+        return try {
+            val params =
+                mapOf(
+                    "session_id" to sessionId,
+                    "connectors" to connectors,
+                    "reconnect" to reconnect,
+                )
+            val result = rpcRequest(WsMethods.CONNECTORS_CONNECT, params)
+            ConnectorParser.parseConnectResult(result)
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: HermesWsClient.HermesRpcException) {
+            ConnectorConnectResult.Error(
+                ConnectorParser.mapRpcError(code = e.code, message = e.message ?: "", data = e.data),
+            )
+        } catch (_: Exception) {
+            ConnectorConnectResult.Error(
+                ConnectorError.NetworkError("Failed to communicate with gateway."),
+            )
+        }
+    }
+}

app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt

@@ -9,6 +9,7 @@ import com.m57.hermescontrol.data.remote.DashboardSessionTokenRefresher
 import com.m57.hermescontrol.data.remote.NetworkMonitor
 import com.m57.hermescontrol.data.remote.OkHttpProvider
 import com.m57.hermescontrol.data.session.ActiveSessionHolder
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -30,6 +31,8 @@ import kotlinx.coroutines.withContext
 import kotlinx.serialization.decodeFromString
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
 import okhttp3.Request
 import okhttp3.RequestBody.Companion.toRequestBody
 import okhttp3.Response
@@ -702,6 +705,8 @@ object HermesWsClient {
     /** Thrown when an awaited [request] is neither answered nor rejected within [REQUEST_TIMEOUT_MS], or is rejected by a disconnect. */
     class HermesRpcException(
         message: String,
+        val code: Int = 0,
+        val data: JsonElement? = null,
     ) : Exception(message)
 
     /**
@@ -741,6 +746,15 @@ object HermesWsClient {
             send(method, params) { reqId ->
                 pendingCalls[reqId] = PendingCall(method, deferred)
             }
+        deferred.invokeOnCompletion { cause ->
+            if (cause is CancellationException) {
+                pendingCalls.remove(id)?.let { call ->
+                    call.timeoutJob?.cancel()
+                    removeQueuedMessage(id)
+                    disconnectIfIdleInBackground()
+                }
+            }
+        }
         // Arm the per-request timeout (fires if the server never answers).
         pendingCalls[id]?.timeoutJob =
             wsScope.launch {
@@ -760,7 +774,13 @@ object HermesWsClient {
         removeQueuedMessage(id)
         call.timeoutJob?.cancel()
         if (error != null) {
-            call.deferred.completeExceptionally(HermesRpcException(error.message))
+            call.deferred.completeExceptionally(
+                HermesRpcException(
+                    message = error.message,
+                    code = error.code,
+                    data = error.data,
+                ),
+            )
         } else {
             call.deferred.complete(result)
         }
@@ -808,7 +828,9 @@ object HermesWsClient {
                 params = decoratedParams.mapValues { it.value.toJsonElement() },
             )
         val json = OkHttpProvider.json.encodeToString(request)
-        if (BuildConfig.DEBUG) Log.d(TAG, "→ $json")
+        if (BuildConfig.DEBUG) {
+            Log.d(TAG, formatSafeOutgoingFrameLog(id, method, request.params.keys, json.length, queued = false))
+        }
         var reconnect = false
         synchronized(outboundLock) {
             if (method == WsMethods.PROMPT_SUBMIT) {
@@ -1191,7 +1213,7 @@ object HermesWsClient {
                         markQueuedMessageSent(msg)
                         continue
                     }
-                    if (BuildConfig.DEBUG) Log.d(TAG, "→ (queued) $msg")
+                    if (BuildConfig.DEBUG) Log.d(TAG, formatSafeQueuedFrameLog(msg))
                     if (!webSocket.send(msg)) {
                         recoverRejectedSocket(webSocket)
                         break
@@ -1209,7 +1231,7 @@ object HermesWsClient {
             text: String,
         ) {
             if (!isCurrent() || HermesWsClient.webSocket !== webSocket) return
-            if (BuildConfig.DEBUG) Log.d(TAG, "← $text")
+            if (BuildConfig.DEBUG) Log.d(TAG, formatSafeIncomingFrameLog(text))
             lastPongTimestamp = monotonicTimeMs()
             // Resolve any in-flight `request()` awaiting this RPC result/error
             // (issue #526) before fanning the parsed event out to collectors.
@@ -1269,7 +1291,7 @@ object HermesWsClient {
                         parsed
                     }
                 } catch (e: Exception) {
-                    Log.e(TAG, "Failed to parse message", e)
+                    Log.e(TAG, "Failed to parse message: ${e.javaClass.simpleName}")
                     WsEvent.Unknown(text)
                 }
             when (event) {
@@ -1381,4 +1403,60 @@ object HermesWsClient {
             }
         }
     }
+
+    @VisibleForTesting
+    internal fun setConnectionStatusForTest(status: ConnectionStatus) {
+        _connectionStatus.value = status
+    }
+
+    @VisibleForTesting
+    internal fun formatSafeOutgoingFrameLog(
+        id: String,
+        method: String,
+        paramsKeys: Collection<String>?,
+        byteLength: Int,
+        queued: Boolean = false,
+    ): String {
+        val prefix = if (queued) "→ (queued)" else "→"
+        val keys = paramsKeys?.joinToString(",", prefix = "[", postfix = "]") ?: "[]"
+        return "$prefix id=$id method=$method paramsKeys=$keys bytes=$byteLength"
+    }
+
+    @VisibleForTesting
+    internal fun formatSafeQueuedFrameLog(msg: String): String =
+        try {
+            val req = OkHttpProvider.json.decodeFromString<JsonRpcRequest>(msg)
+            formatSafeOutgoingFrameLog(req.id, req.method, req.params.keys, msg.length, queued = true)
+        } catch (_: Throwable) {
+            "→ (queued) frame bytes=${msg.length}"
+        }
+
+    @VisibleForTesting
+    internal fun formatSafeIncomingFrameLog(text: String): String =
+        try {
+            val element = OkHttpProvider.json.parseToJsonElement(text)
+            if (element is JsonObject) {
+                val id = (element["id"] as? JsonPrimitive)?.content
+                val method = (element["method"] as? JsonPrimitive)?.content
+                val hasResult = element.containsKey("result")
+                val resultKeys =
+                    (element["result"] as? JsonObject)?.keys?.joinToString(
+                        ",",
+                        prefix = "[",
+                        postfix = "]",
+                    )
+                val errorObj = element["error"] as? JsonObject
+                val errorCode = (errorObj?.get("code") as? JsonPrimitive)?.content
+                val sb = StringBuilder("← ")
+                if (id != null) sb.append("id=").append(id).append(" ")
+                if (method != null) sb.append("method=").append(method).append(" ")
+                if (hasResult) sb.append("result=").append(resultKeys ?: "present").append(" ")
+                if (errorCode != null) sb.append("errorCode=").append(errorCode).append(" ")
+                sb.append("bytes=").append(text.length).toString()
+            } else {
+                "← non-object frame bytes=${text.length}"
+            }
+        } catch (_: Throwable) {
+            "← frame bytes=${text.length}"
+        }
 }

app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt

@@ -91,6 +91,14 @@ object WsMethods {
     /** Rolling tail snapshot of a subagent's live execution transcript. */
     const val SUBAGENT_TAIL = "subagent.tail"
 
+    // ── Connectors (issue #1091) ───────────────────────────────────────────
+
+    /** List available third-party connectors and authorization state for a session. */
+    const val CONNECTORS_LIST = "connectors.list"
+
+    /** Initiate or restart authorization flow for one or more connectors. */
+    const val CONNECTORS_CONNECT = "connectors.connect"
+
     // ── Billing / subscription (issue #628) ─────────────────────────────
     // Adopted from the backend release audit (hermes-agent 0bf44d557..614dc194e).
     // `credits.view` was REMOVED upstream; these replace it. `usage.bars`