This commit is contained in:
7 files changed
+165
-18
No files matched your search
@@ -24,7 +24,7 @@ object ConnectionOperationParser {
|
||||
return null
|
||||
}
|
||||
return ConnectionOperationSnapshot(
|
||||
sessionId = (sessionId ?: payload["session_id"] as? String)?.takeIf { it.isNotBlank() },
|
||||
sessionId = resolveSessionId(payload, sessionId),
|
||||
opId = opId,
|
||||
seq = seq,
|
||||
deadlineAt = deadlineAt,
|
||||
@@ -39,6 +39,21 @@ object ConnectionOperationParser {
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Session that owns the operation. Since hermes-agent v0.21.5 updates carry `owner` (#1281):
|
||||
* session-owned ones name their session; account-owned ones never bind to a chat.
|
||||
*/
|
||||
private fun resolveSessionId(
|
||||
payload: Map<String, Any?>,
|
||||
envelopeSessionId: String?,
|
||||
): String? {
|
||||
val owner = payload["owner"] as? Map<*, *>
|
||||
if (owner?.get("type") == "account") return null
|
||||
val ownerSessionId = (owner?.get("session_id") as? String).takeIf { owner?.get("type") == "session" }
|
||||
return listOf(envelopeSessionId, ownerSessionId, payload["session_id"] as? String)
|
||||
.firstOrNull { !it.isNullOrBlank() }
|
||||
}
|
||||
|
||||
private fun parseTarget(raw: Map<*, *>?): ConnectionOperationTarget? {
|
||||
val map = raw ?: return null
|
||||
val name = (map["name"] as? String)?.trim() ?: return null
|
||||
|
||||
@@ -32,6 +32,9 @@ interface ConnectorRepository {
|
||||
if (slug.isNullOrBlank()) return false
|
||||
return SLUG_REGEX.matches(slug)
|
||||
}
|
||||
|
||||
/** `owner` discriminated union required by connector RPCs since hermes-agent v0.21.5 (#1281). */
|
||||
fun sessionOwner(sessionId: String): Map<String, String> = mapOf("type" to "session", "session_id" to sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -61,7 +64,7 @@ class HermesConnectorRepository(
|
||||
}
|
||||
|
||||
return try {
|
||||
val params = mapOf("session_id" to sessionId)
|
||||
val params = mapOf("owner" to ConnectorRepository.sessionOwner(sessionId))
|
||||
val result = rpcRequest(WsMethods.CONNECTORS_LIST, params)
|
||||
ConnectorParser.parseListResult(result)
|
||||
} catch (e: CancellationException) {
|
||||
@@ -105,7 +108,7 @@ class HermesConnectorRepository(
|
||||
return try {
|
||||
val params =
|
||||
mapOf(
|
||||
"session_id" to sessionId,
|
||||
"owner" to ConnectorRepository.sessionOwner(sessionId),
|
||||
"connectors" to connectors,
|
||||
"reconnect" to reconnect,
|
||||
)
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package com.m57.hermescontrol.ui.chat
|
||||
|
||||
import com.m57.hermescontrol.data.model.ConnectionOperationSnapshot
|
||||
import com.m57.hermescontrol.data.model.ConnectorError
|
||||
import com.m57.hermescontrol.data.ws.ConnectorParser
|
||||
import com.m57.hermescontrol.data.ws.ConnectorRepository
|
||||
import com.m57.hermescontrol.data.ws.HermesWsClient
|
||||
import com.m57.hermescontrol.data.ws.WsMethods
|
||||
import kotlinx.coroutines.CancellationException
|
||||
@@ -167,7 +170,7 @@ class ChatConnectionOperationDelegate(
|
||||
method = WsMethods.CONNECTION_RESPOND,
|
||||
params =
|
||||
mapOf(
|
||||
"session_id" to requireSessionId(),
|
||||
"owner" to ConnectorRepository.sessionOwner(requireSessionId()),
|
||||
"op_id" to snapshot.opId,
|
||||
"result" to mapOf("targets" to listOf(answer)),
|
||||
),
|
||||
@@ -180,7 +183,7 @@ class ChatConnectionOperationDelegate(
|
||||
method = WsMethods.CONNECTION_RESPOND,
|
||||
params =
|
||||
mapOf(
|
||||
"session_id" to requireSessionId(),
|
||||
"owner" to ConnectorRepository.sessionOwner(requireSessionId()),
|
||||
"op_id" to snapshot.opId,
|
||||
"result" to mapOf("settled_by" to "continue"),
|
||||
),
|
||||
@@ -193,7 +196,7 @@ class ChatConnectionOperationDelegate(
|
||||
val snapshot = begin(ConnectionPendingAction.Wake(current.opId, current.seq)) ?: return
|
||||
dispatch(
|
||||
method = WsMethods.CONNECTORS_OPERATION_WAKE,
|
||||
params = mapOf("session_id" to requireSessionId(), "op_id" to snapshot.opId),
|
||||
params = mapOf("owner" to ConnectorRepository.sessionOwner(requireSessionId()), "op_id" to snapshot.opId),
|
||||
clearOnSuccess = true,
|
||||
unknownOperationSettles = true,
|
||||
)
|
||||
@@ -225,7 +228,7 @@ class ChatConnectionOperationDelegate(
|
||||
_state.value =
|
||||
_state.value.copy(
|
||||
pendingAction = null,
|
||||
error = ConnectionOperationError("request_failed"),
|
||||
error = operationError(error),
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -245,6 +248,16 @@ class ChatConnectionOperationDelegate(
|
||||
return current
|
||||
}
|
||||
|
||||
/** Typed, sanitized error: ownership/runtime failures cannot succeed on retry (#1281). */
|
||||
private fun operationError(error: Exception): ConnectionOperationError {
|
||||
val rpc = error as? HermesWsClient.HermesRpcException ?: return ConnectionOperationError(REQUEST_FAILED)
|
||||
return when (ConnectorParser.mapRpcError(code = rpc.code, data = rpc.data)) {
|
||||
is ConnectorError.NotOwner -> ConnectionOperationError(NOT_OWNER, retryable = false)
|
||||
is ConnectorError.UnsupportedRuntime -> ConnectionOperationError(UNSUPPORTED_RUNTIME, retryable = false)
|
||||
else -> ConnectionOperationError(REQUEST_FAILED)
|
||||
}
|
||||
}
|
||||
|
||||
private fun matchesSession(snapshot: ConnectionOperationSnapshot): Boolean =
|
||||
!sessionId.isNullOrBlank() && snapshot.sessionId == sessionId
|
||||
|
||||
@@ -263,7 +276,10 @@ class ChatConnectionOperationDelegate(
|
||||
_state.value = ConnectionOperationUiState()
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val MAX_SETTLED_IDS = 32
|
||||
companion object {
|
||||
private const val MAX_SETTLED_IDS = 32
|
||||
const val REQUEST_FAILED = "request_failed"
|
||||
const val NOT_OWNER = "not_owner"
|
||||
const val UNSUPPORTED_RUNTIME = "unsupported_runtime"
|
||||
}
|
||||
}
|
||||
@@ -29,6 +29,7 @@ import androidx.compose.ui.unit.dp
|
||||
import com.m57.hermescontrol.R
|
||||
import com.m57.hermescontrol.data.model.ConnectionOperationTarget
|
||||
import com.m57.hermescontrol.data.model.ConnectionTargetState
|
||||
import com.m57.hermescontrol.ui.chat.ChatConnectionOperationDelegate
|
||||
import com.m57.hermescontrol.ui.chat.ConnectionOperationUiState
|
||||
|
||||
@OptIn(ExperimentalMaterial3Api::class)
|
||||
@@ -224,9 +225,24 @@ internal fun ConnectionSetupContent(
|
||||
}
|
||||
}
|
||||
|
||||
state.error?.let {
|
||||
state.error?.let { error ->
|
||||
Text(
|
||||
text = stringResource(R.string.connection_setup_request_failed),
|
||||
text =
|
||||
stringResource(
|
||||
when (error.message) {
|
||||
ChatConnectionOperationDelegate.NOT_OWNER -> {
|
||||
R.string.session_integrations_err_not_owner
|
||||
}
|
||||
|
||||
ChatConnectionOperationDelegate.UNSUPPORTED_RUNTIME -> {
|
||||
R.string.session_integrations_err_unsupported_runtime
|
||||
}
|
||||
|
||||
else -> {
|
||||
R.string.connection_setup_request_failed
|
||||
}
|
||||
},
|
||||
),
|
||||
color = MaterialTheme.colorScheme.error,
|
||||
modifier = Modifier.testTag("connection_setup_error"),
|
||||
)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package com.m57.hermescontrol.data.ws
|
||||
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
||||
@@ -63,6 +64,39 @@ class ConnectionOperationEventParserTest {
|
||||
assertEquals("", (event as WsEvent.Unknown).raw)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun connectionUpdate_bindsToSessionOwner() {
|
||||
val event =
|
||||
EventParser.parseParams(
|
||||
mapOf(
|
||||
"type" to "connection.update",
|
||||
"payload" to
|
||||
operationPayload(seq = 2L) +
|
||||
mapOf("owner" to mapOf("type" to "session", "session_id" to "owner-session")),
|
||||
),
|
||||
)
|
||||
|
||||
assertTrue(event is WsEvent.ConnectionUpdate)
|
||||
assertEquals("owner-session", (event as WsEvent.ConnectionUpdate).snapshot.sessionId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun connectionUpdate_accountOwnerNeverBindsToSession() {
|
||||
val event =
|
||||
EventParser.parseParams(
|
||||
mapOf(
|
||||
"type" to "connection.update",
|
||||
"session_id" to "",
|
||||
"payload" to
|
||||
operationPayload(seq = 2L) +
|
||||
mapOf("owner" to mapOf("type" to "account"), "session_id" to "stale-session"),
|
||||
),
|
||||
)
|
||||
|
||||
assertTrue(event is WsEvent.ConnectionUpdate)
|
||||
assertNull((event as WsEvent.ConnectionUpdate).snapshot.sessionId)
|
||||
}
|
||||
|
||||
private fun operationPayload(seq: Long): Map<String, Any?> =
|
||||
mapOf(
|
||||
"op_id" to "op-1",
|
||||
|
||||
@@ -44,7 +44,10 @@ class ConnectorRepositoryTest {
|
||||
val result = repo.listConnectors("sess-abc-123")
|
||||
assertTrue(result is ConnectorListResult.Success)
|
||||
assertEquals(WsMethods.CONNECTORS_LIST, recordedMethod)
|
||||
assertEquals(mapOf("session_id" to "sess-abc-123"), recordedParams)
|
||||
assertEquals(
|
||||
mapOf("owner" to mapOf("type" to "session", "session_id" to "sess-abc-123")),
|
||||
recordedParams,
|
||||
)
|
||||
// Verify no profile or singular slug mutations
|
||||
assertFalse(recordedParams?.containsKey("profile") == true)
|
||||
assertFalse(recordedParams?.containsKey("connector") == true)
|
||||
@@ -80,9 +83,14 @@ class ConnectorRepositoryTest {
|
||||
|
||||
assertTrue(result is ConnectorConnectResult.Success)
|
||||
assertEquals(WsMethods.CONNECTORS_CONNECT, recordedMethod)
|
||||
assertEquals("sess-456", recordedParams?.get("session_id"))
|
||||
assertEquals(listOf("linear", "github_app-1"), recordedParams?.get("connectors"))
|
||||
assertEquals(true, recordedParams?.get("reconnect"))
|
||||
assertEquals(
|
||||
mapOf(
|
||||
"owner" to mapOf("type" to "session", "session_id" to "sess-456"),
|
||||
"connectors" to listOf("linear", "github_app-1"),
|
||||
"reconnect" to true,
|
||||
),
|
||||
recordedParams,
|
||||
)
|
||||
|
||||
// Verify no singular slug key mutation
|
||||
assertFalse(recordedParams?.containsKey("connector") == true)
|
||||
|
||||
+58
-3
@@ -13,6 +13,8 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.test.runCurrent
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.JsonPrimitive
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNull
|
||||
@@ -109,7 +111,8 @@ class ChatConnectionOperationDelegateTest {
|
||||
assertEquals(1, requester.calls.size)
|
||||
val call = requester.calls.single()
|
||||
assertEquals(WsMethods.CONNECTION_RESPOND, call.first)
|
||||
assertEquals("session-a", call.second["session_id"])
|
||||
assertEquals(setOf("owner", "op_id", "result"), call.second.keys)
|
||||
assertEquals(mapOf("type" to "session", "session_id" to "session-a"), call.second["owner"])
|
||||
assertEquals("op-a", call.second["op_id"])
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val result = call.second["result"] as Map<String, Any>
|
||||
@@ -177,11 +180,63 @@ class ChatConnectionOperationDelegateTest {
|
||||
continueDelegate.continueOperation()
|
||||
|
||||
assertEquals(
|
||||
mapOf("settled_by" to "continue"),
|
||||
continueRequester.calls.single().second["result"],
|
||||
mapOf(
|
||||
"owner" to mapOf("type" to "session", "session_id" to "session-a"),
|
||||
"op_id" to "op-a",
|
||||
"result" to mapOf("settled_by" to "continue"),
|
||||
),
|
||||
continueRequester.calls.single().second,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun wakePayload_sendsSessionOwnerNotTopLevelSessionId() =
|
||||
runTest {
|
||||
val requester = RecordingRequester()
|
||||
val delegate = delegate(requester)
|
||||
delegate.reset("session-a")
|
||||
delegate.acceptRequest(snapshot())
|
||||
|
||||
delegate.wake("op-a")
|
||||
|
||||
val call = requester.calls.single()
|
||||
assertEquals(WsMethods.CONNECTORS_OPERATION_WAKE, call.first)
|
||||
assertEquals(
|
||||
mapOf(
|
||||
"owner" to mapOf("type" to "session", "session_id" to "session-a"),
|
||||
"op_id" to "op-a",
|
||||
),
|
||||
call.second,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun ownershipAndRuntimeFailures_mapToTypedNonRetryableErrors() =
|
||||
runTest {
|
||||
val cases =
|
||||
listOf(
|
||||
HermesWsClient.HermesRpcException("x", code = 4001) to
|
||||
ChatConnectionOperationDelegate.NOT_OWNER,
|
||||
HermesWsClient.HermesRpcException(
|
||||
"x",
|
||||
code = 5033,
|
||||
data = JsonObject(mapOf("reason" to JsonPrimitive("UNSUPPORTED_RUNTIME"))),
|
||||
) to ChatConnectionOperationDelegate.UNSUPPORTED_RUNTIME,
|
||||
)
|
||||
for ((failure, expected) in cases) {
|
||||
val delegate = delegate(RecordingRequester(failure = failure))
|
||||
delegate.reset("session-a")
|
||||
delegate.acceptRequest(snapshot())
|
||||
|
||||
delegate.continueOperation()
|
||||
|
||||
val error = delegate.state.value.error
|
||||
assertEquals(expected, error?.message)
|
||||
assertFalse(error?.retryable ?: true)
|
||||
assertNull(delegate.state.value.pendingAction)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun rpcFailureUnlocksWithSanitizedError() =
|
||||
runTest {
|
||||
|
||||
Reference in new issue
Block a user