mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-10 00:11:19 +00:00
fix: handle terminal chat send acknowledgements (#91049)
* test: cover terminal chat send acknowledgements * test: cover Swift terminal chat send acknowledgement * fix: handle terminal chat send acknowledgements * fix: align terminal ack web lifecycle options * test: fix Android terminal ack style * fix: tidy Android terminal ack helpers * fix: clear mic pending run after terminal ack * fix: handle terminal talk mode chat send acks * fix: handle terminal tui chat send acks * fix: handle terminal acp chat send acks * test: add Swift chat message text helper * test: cover steer terminal chat send acknowledgements * fix: handle terminal steer chat send acks * test: cover terminal realtime consult send acks * fix: reject terminal realtime consult send acks * test: cover Swift terminal ok chat send ack * fix: clear Swift pending run on terminal ok ack * test: cover terminal ack helper callers * fix: preserve terminal ack helper semantics * fix: narrow terminal ack type guard * test: cover mic terminal ack statuses * fix: preserve mic terminal ack status * fix: keep mic ack contract internal * test: fix mic ack import order * test: cover acp terminal ok ack * test: narrow acp ok ack assertion * test: cover redirect terminal acknowledgements * fix: handle redirect terminal acknowledgements * fix: settle terminal ack reconnect prompts * fix: surface Android terminal ack timeouts * fix(tui): handle detached terminal chat acknowledgements * fix(tui): report terminal timeout send failures * fix: satisfy iOS talk-mode SwiftFormat * fix: keep iOS talk logs compile-safe
This commit is contained in:
David
GitHub
parent
daa382611f
commit
3ff0c29f9d
31 changed files with 1911 additions and 217 deletions
|
|
@ -14,6 +14,7 @@ import ai.openclaw.app.gateway.GatewayTlsProbeFailure
|
|||
import ai.openclaw.app.gateway.GatewayTlsProbeResult
|
||||
import ai.openclaw.app.gateway.GatewayUpdateAvailableSummary
|
||||
import ai.openclaw.app.gateway.normalizeGatewayTlsFingerprint
|
||||
import ai.openclaw.app.gateway.parseChatSendAck
|
||||
import ai.openclaw.app.gateway.probeGatewayTlsFingerprint
|
||||
import ai.openclaw.app.node.A2UIHandler
|
||||
import ai.openclaw.app.node.CalendarHandler
|
||||
|
|
@ -632,7 +633,11 @@ class NodeRuntime(
|
|||
put("idempotencyKey", JsonPrimitive(idempotencyKey))
|
||||
}
|
||||
val response = operatorSession.request("chat.send", params.toString())
|
||||
parseChatSendRunId(response) ?: idempotencyKey
|
||||
val ack = parseChatSendAck(json, response)
|
||||
ack.copy(runId = ack.runId ?: idempotencyKey)
|
||||
},
|
||||
refreshAfterTerminalSuccess = {
|
||||
chat.refresh()
|
||||
},
|
||||
speakAssistantReply = { text ->
|
||||
// Voice-tab replies should speak through the dedicated reply speaker.
|
||||
|
|
@ -1843,15 +1848,6 @@ class NodeRuntime(
|
|||
}
|
||||
}
|
||||
|
||||
private fun parseChatSendRunId(response: String): String? {
|
||||
return try {
|
||||
val root = json.parseToJsonElement(response).asObjectOrNull() ?: return null
|
||||
root["runId"].asStringOrNull()
|
||||
} catch (_: Throwable) {
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
private fun parseTalkSessionId(response: String): String {
|
||||
val root = json.parseToJsonElement(response).asObjectOrNull()
|
||||
val sessionId =
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package ai.openclaw.app.chat
|
||||
|
||||
import ai.openclaw.app.gateway.GatewaySession
|
||||
import ai.openclaw.app.gateway.parseChatSendAck
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.delay
|
||||
|
|
@ -19,11 +20,21 @@ import java.util.UUID
|
|||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.atomic.AtomicLong
|
||||
|
||||
class ChatController(
|
||||
class ChatController internal constructor(
|
||||
private val scope: CoroutineScope,
|
||||
private val session: GatewaySession,
|
||||
private val json: Json,
|
||||
private val requestGateway: suspend (method: String, paramsJson: String?) -> String,
|
||||
) {
|
||||
constructor(
|
||||
scope: CoroutineScope,
|
||||
session: GatewaySession,
|
||||
json: Json,
|
||||
) : this(
|
||||
scope = scope,
|
||||
json = json,
|
||||
requestGateway = { method, paramsJson -> session.request(method, paramsJson) },
|
||||
)
|
||||
|
||||
private var appliedMainSessionKey = "main"
|
||||
private val _sessionKey = MutableStateFlow("main")
|
||||
val sessionKey: StateFlow<String> = _sessionKey.asStateFlow()
|
||||
|
|
@ -267,8 +278,9 @@ class ChatController(
|
|||
)
|
||||
}
|
||||
}
|
||||
val res = session.request("chat.send", params.toString())
|
||||
val actualRunId = parseRunId(res) ?: runId
|
||||
val res = requestGateway("chat.send", params.toString())
|
||||
val ack = parseChatSendAck(json, res)
|
||||
val actualRunId = ack.runId ?: runId
|
||||
if (actualRunId != runId) {
|
||||
// Gateway may return a canonical run id; move all pending bookkeeping to that id.
|
||||
optimisticMessagesByRunId[actualRunId] = optimisticMessagesByRunId.remove(runId) ?: optimisticMessage
|
||||
|
|
@ -279,7 +291,24 @@ class ChatController(
|
|||
_pendingRunCount.value = pendingRuns.size
|
||||
}
|
||||
}
|
||||
true
|
||||
if (ack.isTerminal) {
|
||||
clearPendingRun(actualRunId)
|
||||
removeOptimisticMessage(actualRunId)
|
||||
pendingToolCallsById.clear()
|
||||
publishPendingToolCalls()
|
||||
_streamingAssistantText.value = null
|
||||
if (ack.isTerminalSuccess) {
|
||||
refreshCurrentHistoryBestEffort()
|
||||
true
|
||||
} else {
|
||||
// Terminal timeout/error means the gateway did not accept a runnable turn.
|
||||
// Surface failed acceptance instead of letting a cleared composer look successful.
|
||||
_errorText.value = "Chat failed before the run started; try again."
|
||||
false
|
||||
}
|
||||
} else {
|
||||
true
|
||||
}
|
||||
} catch (err: Throwable) {
|
||||
clearPendingRun(runId)
|
||||
removeOptimisticMessage(runId)
|
||||
|
|
@ -303,7 +332,7 @@ class ChatController(
|
|||
put("sessionKey", JsonPrimitive(_sessionKey.value))
|
||||
put("runId", JsonPrimitive(runId))
|
||||
}
|
||||
session.request("chat.abort", params.toString())
|
||||
requestGateway("chat.abort", params.toString())
|
||||
} catch (_: Throwable) {
|
||||
// best-effort
|
||||
}
|
||||
|
|
@ -356,7 +385,7 @@ class ChatController(
|
|||
) {
|
||||
try {
|
||||
val historyJson =
|
||||
session.request(
|
||||
requestGateway(
|
||||
"chat.history",
|
||||
buildJsonObject { put("sessionKey", JsonPrimitive(sessionKey)) }.toString(),
|
||||
)
|
||||
|
|
@ -391,7 +420,7 @@ class ChatController(
|
|||
put("includeUnknown", JsonPrimitive(false))
|
||||
if (limit != null && limit > 0) put("limit", JsonPrimitive(limit))
|
||||
}
|
||||
val res = session.request("sessions.list", params.toString())
|
||||
val res = requestGateway("sessions.list", params.toString())
|
||||
_sessions.value = parseSessions(res)
|
||||
} catch (_: Throwable) {
|
||||
// best-effort
|
||||
|
|
@ -408,7 +437,7 @@ class ChatController(
|
|||
if (!force && last != null && now - last < 10_000) return
|
||||
lastHealthPollAtMs = now
|
||||
try {
|
||||
session.request("health", null)
|
||||
requestGateway("health", null)
|
||||
_healthOk.value = true
|
||||
} catch (_: Throwable) {
|
||||
_healthOk.value = false
|
||||
|
|
@ -451,7 +480,7 @@ class ChatController(
|
|||
val currentSessionKey = _sessionKey.value
|
||||
val currentGeneration = historyLoadGeneration.get()
|
||||
val historyJson =
|
||||
session.request(
|
||||
requestGateway(
|
||||
"chat.history",
|
||||
buildJsonObject { put("sessionKey", JsonPrimitive(currentSessionKey)) }.toString(),
|
||||
)
|
||||
|
|
@ -632,6 +661,45 @@ class ChatController(
|
|||
optimisticMessagesByRunId.entries.removeAll { entry -> entry.value !in retained }
|
||||
}
|
||||
|
||||
private fun refreshCurrentHistoryBestEffort() {
|
||||
scope.launch {
|
||||
try {
|
||||
val currentSessionKey = _sessionKey.value
|
||||
val currentGeneration = historyLoadGeneration.get()
|
||||
val historyJson =
|
||||
requestGateway(
|
||||
"chat.history",
|
||||
buildJsonObject { put("sessionKey", JsonPrimitive(currentSessionKey)) }.toString(),
|
||||
)
|
||||
if (
|
||||
!isCurrentHistoryLoad(
|
||||
currentSessionKey,
|
||||
_sessionKey.value,
|
||||
currentGeneration,
|
||||
historyLoadGeneration.get(),
|
||||
)
|
||||
) {
|
||||
return@launch
|
||||
}
|
||||
val history =
|
||||
parseHistory(
|
||||
historyJson,
|
||||
sessionKey = currentSessionKey,
|
||||
previousMessages = _messages.value,
|
||||
)
|
||||
prunePersistedOptimisticMessages(history.messages)
|
||||
_messages.value = mergeOptimisticMessages(incoming = history.messages, optimistic = optimisticMessagesByRunId.values)
|
||||
_sessionId.value = history.sessionId
|
||||
history.thinkingLevel
|
||||
?.trim()
|
||||
?.takeIf { it.isNotEmpty() }
|
||||
?.let { _thinkingLevel.value = it }
|
||||
} catch (_: Throwable) {
|
||||
// best-effort
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun parseHistory(
|
||||
historyJson: String,
|
||||
sessionKey: String,
|
||||
|
|
@ -728,17 +796,6 @@ class ChatController(
|
|||
_sessions.value = _sessions.value.filterNot { it.key == key }
|
||||
}
|
||||
|
||||
private fun parseRunId(resJson: String): String? =
|
||||
try {
|
||||
json
|
||||
.parseToJsonElement(resJson)
|
||||
.asObjectOrNull()
|
||||
?.get("runId")
|
||||
.asStringOrNull()
|
||||
} catch (_: Throwable) {
|
||||
null
|
||||
}
|
||||
|
||||
private fun normalizeThinking(raw: String): String =
|
||||
when (raw.trim().lowercase()) {
|
||||
"low" -> "low"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
package ai.openclaw.app.gateway
|
||||
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.JsonPrimitive
|
||||
|
||||
internal data class ChatSendAck(
|
||||
val runId: String?,
|
||||
val status: String?,
|
||||
) {
|
||||
val normalizedStatus: String
|
||||
get() = status?.trim()?.lowercase().orEmpty()
|
||||
|
||||
val isTerminalSuccess: Boolean
|
||||
get() = normalizedStatus == "ok"
|
||||
|
||||
val isTerminalFailure: Boolean
|
||||
get() = normalizedStatus == "timeout" || normalizedStatus == "error"
|
||||
|
||||
val isTerminal: Boolean
|
||||
get() = isTerminalSuccess || isTerminalFailure
|
||||
}
|
||||
|
||||
internal fun chatSendAckHistorySinceSeconds(
|
||||
ack: ChatSendAck,
|
||||
startedAtSeconds: Double,
|
||||
): Double? = if (ack.isTerminalSuccess) null else startedAtSeconds
|
||||
|
||||
internal fun parseChatSendAck(
|
||||
json: Json,
|
||||
responseJson: String,
|
||||
): ChatSendAck =
|
||||
try {
|
||||
val obj = json.parseToJsonElement(responseJson).asObjectOrNull()
|
||||
ChatSendAck(
|
||||
runId = obj?.get("runId").asStringOrNull(),
|
||||
status = obj?.get("status").asStringOrNull(),
|
||||
)
|
||||
} catch (_: Throwable) {
|
||||
ChatSendAck(runId = null, status = null)
|
||||
}
|
||||
|
||||
private fun JsonElement?.asObjectOrNull(): JsonObject? = this as? JsonObject
|
||||
|
||||
private fun JsonElement?.asStringOrNull(): String? = (this as? JsonPrimitive)?.takeIf { it.isString }?.content
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
package ai.openclaw.app.voice
|
||||
|
||||
import ai.openclaw.app.gateway.ChatSendAck
|
||||
import android.Manifest
|
||||
import android.annotation.SuppressLint
|
||||
import android.content.Context
|
||||
|
|
@ -43,7 +44,7 @@ data class VoiceConversationEntry(
|
|||
)
|
||||
|
||||
/** Coordinates live mic transcription, queued sends, and assistant audio replies. */
|
||||
class MicCaptureManager(
|
||||
internal class MicCaptureManager(
|
||||
private val context: Context,
|
||||
private val scope: CoroutineScope,
|
||||
private val createTranscriptionSession: suspend () -> String,
|
||||
|
|
@ -54,11 +55,12 @@ class MicCaptureManager(
|
|||
) -> Unit,
|
||||
private val closeTranscriptionSession: suspend (sessionId: String) -> Unit,
|
||||
/**
|
||||
* Send [message] to the gateway and return the run ID.
|
||||
* Send [message] to the gateway and return the full chat.send ACK.
|
||||
* [onRunIdKnown] is called with the idempotency key *before* the network
|
||||
* round-trip so [pendingRunId] is set before any chat events can arrive.
|
||||
*/
|
||||
private val sendToGateway: suspend (message: String, onRunIdKnown: (String) -> Unit) -> String?,
|
||||
private val sendToGateway: suspend (message: String, onRunIdKnown: (String) -> Unit) -> ChatSendAck,
|
||||
private val refreshAfterTerminalSuccess: suspend () -> Unit = {},
|
||||
private val speakAssistantReply: suspend (String) -> Unit = {},
|
||||
) {
|
||||
companion object {
|
||||
|
|
@ -483,24 +485,30 @@ class MicCaptureManager(
|
|||
|
||||
scope.launch {
|
||||
try {
|
||||
val runId =
|
||||
val ack =
|
||||
sendToGateway(next) { earlyRunId ->
|
||||
// Called with the idempotency key before chat.send fires so that
|
||||
// pendingRunId is populated before any chat events can arrive.
|
||||
pendingRunId = earlyRunId
|
||||
}
|
||||
val runId = ack.runId
|
||||
// Update to the real runId if the gateway returned a different one.
|
||||
if (runId != null && runId != pendingRunId) pendingRunId = runId
|
||||
if (runId == null) {
|
||||
pendingRunTimeoutJob?.cancel()
|
||||
pendingRunTimeoutJob = null
|
||||
removeFirstQueuedMessage()
|
||||
publishQueue()
|
||||
_isSending.value = false
|
||||
pendingAssistantEntryId = null
|
||||
sendQueuedIfIdle()
|
||||
} else {
|
||||
armPendingRunTimeout(runId)
|
||||
when {
|
||||
ack.isTerminalSuccess -> {
|
||||
completePendingTurn()
|
||||
refreshAfterTerminalSuccess()
|
||||
}
|
||||
ack.isTerminalFailure -> {
|
||||
completePendingTurn()
|
||||
_statusText.value = "Send failed: Chat failed before the run started; try again."
|
||||
}
|
||||
runId == null -> {
|
||||
completePendingTurn()
|
||||
}
|
||||
else -> {
|
||||
armPendingRunTimeout(runId)
|
||||
}
|
||||
}
|
||||
} catch (err: Throwable) {
|
||||
pendingRunTimeoutJob?.cancel()
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
package ai.openclaw.app.voice
|
||||
|
||||
import ai.openclaw.app.gateway.ChatSendAck
|
||||
import ai.openclaw.app.gateway.GatewaySession
|
||||
import ai.openclaw.app.gateway.chatSendAckHistorySinceSeconds
|
||||
import ai.openclaw.app.gateway.parseChatSendAck
|
||||
import android.Manifest
|
||||
import android.annotation.SuppressLint
|
||||
import android.content.Context
|
||||
|
|
@ -381,11 +384,20 @@ class TalkModeManager internal constructor(
|
|||
reloadConfig()
|
||||
val startedAt = System.currentTimeMillis().toDouble() / 1000.0
|
||||
val prompt = buildPrompt(command)
|
||||
val runId = sendChat(prompt, session)
|
||||
val ok = waitForChatFinal(runId)
|
||||
val ack = sendChat(prompt, session)
|
||||
val runId = ack.runId ?: throw IllegalStateException("chat.send returned no run id")
|
||||
if (ack.isTerminalFailure) {
|
||||
_statusText.value = if (ack.normalizedStatus == "error") "Chat error" else "Aborted"
|
||||
return@launch
|
||||
}
|
||||
val ok = if (ack.isTerminalSuccess) true else waitForChatFinal(runId)
|
||||
val assistant =
|
||||
consumeRunText(runId)
|
||||
?: waitForAssistantText(session, startedAt, if (ok) 12_000 else 25_000)
|
||||
?: waitForAssistantText(
|
||||
session,
|
||||
chatSendAckHistorySinceSeconds(ack, startedAt),
|
||||
if (ok) 12_000 else 25_000,
|
||||
)
|
||||
if (!assistant.isNullOrBlank()) {
|
||||
val playbackToken = playbackGeneration.incrementAndGet()
|
||||
cancelActivePlayback()
|
||||
|
|
@ -398,8 +410,9 @@ class TalkModeManager internal constructor(
|
|||
}
|
||||
} catch (err: Throwable) {
|
||||
Log.w(tag, "speakWakeCommand failed: ${err.message}")
|
||||
} finally {
|
||||
onComplete()
|
||||
}
|
||||
onComplete()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1604,16 +1617,26 @@ class TalkModeManager internal constructor(
|
|||
try {
|
||||
val startedAt = System.currentTimeMillis().toDouble() / 1000.0
|
||||
Log.d(tag, "chat.send start sessionKey=${mainSessionKey.ifBlank { "main" }} chars=${prompt.length}")
|
||||
val runId = sendChat(prompt, session)
|
||||
Log.d(tag, "chat.send ok runId=$runId")
|
||||
val ok = waitForChatFinal(runId)
|
||||
val ack = sendChat(prompt, session)
|
||||
val runId = ack.runId ?: throw IllegalStateException("chat.send returned no run id")
|
||||
Log.d(tag, "chat.send ok runId=$runId status=${ack.status}")
|
||||
if (ack.isTerminalFailure) {
|
||||
_statusText.value = if (ack.normalizedStatus == "error") "Chat error" else "Aborted"
|
||||
start()
|
||||
return
|
||||
}
|
||||
val ok = if (ack.isTerminalSuccess) true else waitForChatFinal(runId)
|
||||
if (!ok) {
|
||||
Log.w(tag, "chat final timeout runId=$runId; attempting history fallback")
|
||||
}
|
||||
// Use text cached from the final event first — avoids chat.history polling
|
||||
val assistant =
|
||||
consumeRunText(runId)
|
||||
?: waitForAssistantText(session, startedAt, if (ok) 12_000 else 25_000)
|
||||
?: waitForAssistantText(
|
||||
session,
|
||||
chatSendAckHistorySinceSeconds(ack, startedAt),
|
||||
if (ok) 12_000 else 25_000,
|
||||
)
|
||||
if (assistant.isNullOrBlank()) {
|
||||
_statusText.value = "No reply"
|
||||
Log.w(tag, "assistant text timeout runId=$runId")
|
||||
|
|
@ -1679,7 +1702,7 @@ class TalkModeManager internal constructor(
|
|||
private suspend fun sendChat(
|
||||
message: String,
|
||||
session: GatewaySession,
|
||||
): String {
|
||||
): ChatSendAck {
|
||||
val runId = UUID.randomUUID().toString()
|
||||
armPendingRun(runId)
|
||||
val params =
|
||||
|
|
@ -1692,11 +1715,15 @@ class TalkModeManager internal constructor(
|
|||
}
|
||||
try {
|
||||
val res = session.request("chat.send", params.toString())
|
||||
val parsed = parseRunId(res) ?: runId
|
||||
if (parsed != runId) {
|
||||
pendingRunId = parsed
|
||||
val parsed = parseChatSendAck(json, res)
|
||||
val actualRunId = parsed.runId ?: runId
|
||||
if (actualRunId != runId) {
|
||||
pendingRunId = actualRunId
|
||||
}
|
||||
return parsed
|
||||
if (parsed.isTerminal) {
|
||||
clearPendingRun(actualRunId)
|
||||
}
|
||||
return parsed.copy(runId = actualRunId)
|
||||
} catch (err: Throwable) {
|
||||
clearPendingRun(runId)
|
||||
throw err
|
||||
|
|
@ -1777,7 +1804,7 @@ class TalkModeManager internal constructor(
|
|||
|
||||
private suspend fun waitForAssistantText(
|
||||
session: GatewaySession,
|
||||
sinceSeconds: Double,
|
||||
sinceSeconds: Double?,
|
||||
timeoutMs: Long,
|
||||
): String? {
|
||||
val deadline = SystemClock.elapsedRealtime() + timeoutMs
|
||||
|
|
|
|||
|
|
@ -0,0 +1,144 @@
|
|||
package ai.openclaw.app.chat
|
||||
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.test.advanceUntilIdle
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.serialization.json.Json
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
||||
class ChatControllerTerminalAckTest {
|
||||
private val json = Json { ignoreUnknownKeys = true }
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun terminalTimeoutAckRemovesOptimisticUserEchoAndSurfacesFailedAcceptance() =
|
||||
runTest {
|
||||
var requestedMethod: String? = null
|
||||
val controller =
|
||||
ChatController(
|
||||
scope = this,
|
||||
json = json,
|
||||
requestGateway = { method, _ ->
|
||||
requestedMethod = method
|
||||
"""{"runId":"run-timeout","status":"timeout"}"""
|
||||
},
|
||||
)
|
||||
controller.handleGatewayEvent("health", null)
|
||||
|
||||
val accepted =
|
||||
controller.sendMessageAwaitAcceptance(
|
||||
message = "message that times out before start",
|
||||
thinkingLevel = "off",
|
||||
attachments = emptyList(),
|
||||
)
|
||||
|
||||
assertFalse(accepted)
|
||||
assertEquals("chat.send", requestedMethod)
|
||||
assertEquals(0, controller.pendingRunCount.value)
|
||||
assertEquals("Chat failed before the run started; try again.", controller.errorText.value)
|
||||
assertFalse(controller.messages.value.hasUserText("message that times out before start"))
|
||||
}
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun nonTerminalStartedAckRetainsOptimisticUserEchoAndPendingRun() =
|
||||
runTest {
|
||||
val controller =
|
||||
ChatController(
|
||||
scope = this,
|
||||
json = json,
|
||||
requestGateway = { _, _ -> """{"runId":"run-started","status":"started"}""" },
|
||||
)
|
||||
controller.handleGatewayEvent("health", null)
|
||||
|
||||
val accepted =
|
||||
controller.sendMessageAwaitAcceptance(
|
||||
message = "message that started",
|
||||
thinkingLevel = "off",
|
||||
attachments = emptyList(),
|
||||
)
|
||||
|
||||
assertTrue(accepted)
|
||||
assertEquals(1, controller.pendingRunCount.value)
|
||||
assertNull(controller.errorText.value)
|
||||
assertTrue(controller.messages.value.hasUserText("message that started"))
|
||||
}
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun terminalOkAckClearsOptimisticUserEchoAndRefreshesHistory() =
|
||||
runTest {
|
||||
val requestedMethods = mutableListOf<String>()
|
||||
val controller =
|
||||
ChatController(
|
||||
scope = this,
|
||||
json = json,
|
||||
requestGateway = { method, _ ->
|
||||
requestedMethods += method
|
||||
when (method) {
|
||||
"chat.send" -> """{"runId":"run-ok","status":"ok"}"""
|
||||
"chat.history" ->
|
||||
"""
|
||||
{
|
||||
"sessionId": "session-1",
|
||||
"messages": [
|
||||
{ "role": "assistant", "content": "cached success reply", "timestamp": 1 }
|
||||
]
|
||||
}
|
||||
""".trimIndent()
|
||||
else -> "{}"
|
||||
}
|
||||
},
|
||||
)
|
||||
controller.handleGatewayEvent("health", null)
|
||||
|
||||
val accepted =
|
||||
controller.sendMessageAwaitAcceptance(
|
||||
message = "message that already completed",
|
||||
thinkingLevel = "off",
|
||||
attachments = emptyList(),
|
||||
)
|
||||
advanceUntilIdle()
|
||||
|
||||
assertTrue(accepted)
|
||||
assertEquals(listOf("chat.send", "chat.history"), requestedMethods)
|
||||
assertEquals(0, controller.pendingRunCount.value)
|
||||
assertNull(controller.errorText.value)
|
||||
assertFalse(controller.messages.value.hasUserText("message that already completed"))
|
||||
assertTrue(controller.messages.value.any { message -> message.role == "assistant" && message.content.any { part -> part.text == "cached success reply" } })
|
||||
}
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun terminalErrorAckRemovesOptimisticUserEchoAndSurfacesErrorText() =
|
||||
runTest {
|
||||
val controller =
|
||||
ChatController(
|
||||
scope = this,
|
||||
json = json,
|
||||
requestGateway = { _, _ -> """{"runId":"run-error","status":"error"}""" },
|
||||
)
|
||||
controller.handleGatewayEvent("health", null)
|
||||
|
||||
val accepted =
|
||||
controller.sendMessageAwaitAcceptance(
|
||||
message = "message that errors before start",
|
||||
thinkingLevel = "off",
|
||||
attachments = emptyList(),
|
||||
)
|
||||
|
||||
assertFalse(accepted)
|
||||
assertEquals(0, controller.pendingRunCount.value)
|
||||
assertEquals("Chat failed before the run started; try again.", controller.errorText.value)
|
||||
assertFalse(controller.messages.value.hasUserText("message that errors before start"))
|
||||
}
|
||||
|
||||
private fun List<ChatMessage>.hasUserText(text: String): Boolean =
|
||||
any { message ->
|
||||
message.role == "user" && message.content.any { part -> part.text == text }
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
package ai.openclaw.app.gateway
|
||||
|
||||
import kotlinx.serialization.json.Json
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
||||
class ChatSendAckTest {
|
||||
private val json = Json { ignoreUnknownKeys = true }
|
||||
|
||||
@Test
|
||||
fun parseChatSendAckPreservesNonTerminalStartedStatus() {
|
||||
val ack = parseChatSendAck(json, """{"runId":"run-1","status":"started"}""")
|
||||
|
||||
assertEquals("run-1", ack.runId)
|
||||
assertEquals("started", ack.normalizedStatus)
|
||||
assertFalse(ack.isTerminal)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun parseChatSendAckMarksOkAsTerminalSuccess() {
|
||||
val ack = parseChatSendAck(json, """{"runId":"run-ok","status":" ok "}""")
|
||||
|
||||
assertEquals("run-ok", ack.runId)
|
||||
assertEquals("ok", ack.normalizedStatus)
|
||||
assertTrue(ack.isTerminal)
|
||||
assertTrue(ack.isTerminalSuccess)
|
||||
assertFalse(ack.isTerminalFailure)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun parseChatSendAckMarksTimeoutAndErrorAsTerminalFailures() {
|
||||
val timeout = parseChatSendAck(json, """{"runId":"run-timeout","status":"timeout"}""")
|
||||
val error = parseChatSendAck(json, """{"runId":"run-error","status":" error "}""")
|
||||
|
||||
assertEquals("run-timeout", timeout.runId)
|
||||
assertTrue(timeout.isTerminal)
|
||||
assertFalse(timeout.isTerminalSuccess)
|
||||
assertTrue(timeout.isTerminalFailure)
|
||||
assertEquals("run-error", error.runId)
|
||||
assertTrue(error.isTerminal)
|
||||
assertFalse(error.isTerminalSuccess)
|
||||
assertTrue(error.isTerminalFailure)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun cachedOkAckUsesUnfilteredHistoryFallback() {
|
||||
val startedAt = 123.0
|
||||
val ok = parseChatSendAck(json, """{"runId":"run-ok","status":"ok"}""")
|
||||
val started = parseChatSendAck(json, """{"runId":"run-started","status":"started"}""")
|
||||
|
||||
assertNull(chatSendAckHistorySinceSeconds(ok, startedAt))
|
||||
assertEquals(startedAt, chatSendAckHistorySinceSeconds(started, startedAt) ?: -1.0, 0.0)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun parseChatSendAckToleratesMalformedPayloads() {
|
||||
val ack = parseChatSendAck(json, "not-json")
|
||||
|
||||
assertNull(ack.runId)
|
||||
assertEquals("", ack.normalizedStatus)
|
||||
assertFalse(ack.isTerminal)
|
||||
assertFalse(ack.isTerminalSuccess)
|
||||
assertFalse(ack.isTerminalFailure)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
package ai.openclaw.app.voice
|
||||
|
||||
import ai.openclaw.app.gateway.ChatSendAck
|
||||
import android.Manifest
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
|
|
@ -34,7 +35,7 @@ class MicCaptureManagerTest {
|
|||
sendToGateway = { message, onRunIdKnown ->
|
||||
sentMessages += message
|
||||
onRunIdKnown("run-1")
|
||||
null
|
||||
ChatSendAck(runId = "run-1", status = "started")
|
||||
},
|
||||
)
|
||||
|
||||
|
|
@ -84,7 +85,7 @@ class MicCaptureManagerTest {
|
|||
sendToGateway = { message, onRunIdKnown ->
|
||||
sentMessages += message
|
||||
onRunIdKnown("run-1")
|
||||
"run-1"
|
||||
ChatSendAck(runId = "run-1", status = "started")
|
||||
},
|
||||
)
|
||||
|
||||
|
|
@ -111,7 +112,7 @@ class MicCaptureManagerTest {
|
|||
sendToGateway = { message, onRunIdKnown ->
|
||||
sentMessages += message
|
||||
onRunIdKnown("run-voice-e2e")
|
||||
"run-voice-e2e"
|
||||
ChatSendAck(runId = "run-voice-e2e", status = "started")
|
||||
},
|
||||
)
|
||||
|
||||
|
|
@ -134,6 +135,88 @@ class MicCaptureManagerTest {
|
|||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun terminalGatewayTimeoutSendDoesNotAcceptDelayedOldRunEvents() =
|
||||
runTest {
|
||||
val manager =
|
||||
createManager(
|
||||
scope = this,
|
||||
sendToGateway = { _, onRunIdKnown ->
|
||||
onRunIdKnown("run-terminal")
|
||||
ChatSendAck(runId = "run-terminal", status = "timeout")
|
||||
},
|
||||
)
|
||||
|
||||
manager.onGatewayConnectionChanged(true)
|
||||
manager.submitTranscribedMessage("terminal ack message")
|
||||
runCurrent()
|
||||
|
||||
assertNull(privateField<String?>(manager, "pendingRunId"))
|
||||
assertEquals(false, manager.isSending.value)
|
||||
assertEquals("Send failed: Chat failed before the run started; try again.", manager.statusText.value)
|
||||
|
||||
manager.handleGatewayEvent("chat", chatFinalPayload(runId = "run-terminal", text = "stale reply"))
|
||||
advanceUntilIdle()
|
||||
|
||||
assertEquals(
|
||||
listOf(VoiceConversationRole.User),
|
||||
manager.conversation.value.map { it.role },
|
||||
)
|
||||
assertEquals(
|
||||
"terminal ack message",
|
||||
manager.conversation.value
|
||||
.single()
|
||||
.text,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun terminalGatewayErrorSurfacesFailureWithoutWaitingForRunEvents() =
|
||||
runTest {
|
||||
val manager =
|
||||
createManager(
|
||||
scope = this,
|
||||
sendToGateway = { _, onRunIdKnown ->
|
||||
onRunIdKnown("run-error")
|
||||
ChatSendAck(runId = "run-error", status = "error")
|
||||
},
|
||||
)
|
||||
|
||||
manager.onGatewayConnectionChanged(true)
|
||||
manager.submitTranscribedMessage("terminal error message")
|
||||
runCurrent()
|
||||
|
||||
assertNull(privateField<String?>(manager, "pendingRunId"))
|
||||
assertEquals(false, manager.isSending.value)
|
||||
assertEquals("Send failed: Chat failed before the run started; try again.", manager.statusText.value)
|
||||
}
|
||||
|
||||
@Test
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
fun terminalGatewayOkRefreshesHistoryWithoutWaitingForRunEvents() =
|
||||
runTest {
|
||||
var refreshCalls = 0
|
||||
val manager =
|
||||
createManager(
|
||||
scope = this,
|
||||
sendToGateway = { _, onRunIdKnown ->
|
||||
onRunIdKnown("run-ok")
|
||||
ChatSendAck(runId = "run-ok", status = "ok")
|
||||
},
|
||||
refreshAfterTerminalSuccess = { refreshCalls += 1 },
|
||||
)
|
||||
|
||||
manager.onGatewayConnectionChanged(true)
|
||||
manager.submitTranscribedMessage("terminal ok message")
|
||||
runCurrent()
|
||||
|
||||
assertNull(privateField<String?>(manager, "pendingRunId"))
|
||||
assertEquals(false, manager.isSending.value)
|
||||
assertEquals(1, refreshCalls)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun pcm16FramesAreEncodedAsPcmuFrames() {
|
||||
val manager = createManager()
|
||||
|
|
@ -230,10 +313,11 @@ class MicCaptureManagerTest {
|
|||
scope: CoroutineScope = CoroutineScope(Dispatchers.Unconfined),
|
||||
createTranscriptionSession: suspend () -> String = { "transcription-1" },
|
||||
closeTranscriptionSession: suspend (String) -> Unit = { _ -> },
|
||||
sendToGateway: suspend (String, (String) -> Unit) -> String? = { _, onRunIdKnown ->
|
||||
sendToGateway: suspend (String, (String) -> Unit) -> ChatSendAck = { _, onRunIdKnown ->
|
||||
onRunIdKnown("run-1")
|
||||
"run-1"
|
||||
ChatSendAck(runId = "run-1", status = "started")
|
||||
},
|
||||
refreshAfterTerminalSuccess: suspend () -> Unit = {},
|
||||
): MicCaptureManager =
|
||||
MicCaptureManager(
|
||||
context =
|
||||
|
|
@ -245,6 +329,7 @@ class MicCaptureManagerTest {
|
|||
appendTranscriptionAudio = { _, _, _ -> },
|
||||
closeTranscriptionSession = closeTranscriptionSession,
|
||||
sendToGateway = sendToGateway,
|
||||
refreshAfterTerminalSuccess = refreshAfterTerminalSuccess,
|
||||
)
|
||||
|
||||
private fun setPrivateField(
|
||||
|
|
|
|||
|
|
@ -1008,39 +1008,60 @@ final class TalkModeManager: NSObject {
|
|||
self.logger.info(
|
||||
"chat.send start sessionKey=\(sessionKey, privacy: .public) chars=\(prompt.count, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat.send start sessionKey=\(sessionKey) chars=\(prompt.count)")
|
||||
let runId = try await self.sendChat(prompt, gateway: gateway)
|
||||
self.logger.info("chat.send ok runId=\(runId, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat.send ok runId=\(runId)")
|
||||
let ack = try await self.sendChat(prompt, gateway: gateway)
|
||||
let runId = ack.runId
|
||||
let normalizedStatus = Self.normalizedChatSendStatus(ack.status)
|
||||
self.logger.info(
|
||||
"chat.send ok runId=\(runId, privacy: .public) status=\(normalizedStatus, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat.send ok runId=\(runId) status=\(normalizedStatus)")
|
||||
if Self.isTerminalChatSendFailure(ack.status) {
|
||||
self.statusText = normalizedStatus == "error" ? "Chat error" : "Aborted"
|
||||
self.logger.warning(
|
||||
"chat.send terminal ack runId=\(runId, privacy: .public) status=\(normalizedStatus, privacy: .public)")
|
||||
GatewayDiagnostics.log(
|
||||
"talk: chat.send terminal ack runId=\(runId) status=\(normalizedStatus)")
|
||||
if restartAfter {
|
||||
await self.start()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
let shouldIncremental = self.shouldUseIncrementalTTS()
|
||||
var streamingTask: Task<Void, Never>?
|
||||
if shouldIncremental {
|
||||
self.resetIncrementalSpeech()
|
||||
streamingTask = Task { @MainActor [weak self] in
|
||||
guard let self else { return }
|
||||
await self.streamAssistant(runId: runId, gateway: gateway)
|
||||
let completion: ChatCompletionResult
|
||||
if Self.isTerminalChatSendSuccess(ack.status) {
|
||||
GatewayDiagnostics.log("talk: chat.send terminal ok runId=\(runId); using history fallback")
|
||||
completion = ChatCompletionResult(state: .final, assistantText: nil)
|
||||
} else {
|
||||
if shouldIncremental {
|
||||
self.resetIncrementalSpeech()
|
||||
streamingTask = Task { @MainActor [weak self] in
|
||||
guard let self else { return }
|
||||
await self.streamAssistant(runId: runId, gateway: gateway)
|
||||
}
|
||||
}
|
||||
completion = await self.waitForChatCompletion(runId: runId, gateway: gateway, timeoutSeconds: 120)
|
||||
if completion.state == .timeout {
|
||||
self.logger.warning(
|
||||
"chat completion timeout runId=\(runId, privacy: .public); attempting history fallback")
|
||||
GatewayDiagnostics.log("talk: chat completion timeout runId=\(runId)")
|
||||
} else if completion.state == .aborted {
|
||||
self.statusText = "Aborted"
|
||||
self.logger.warning("chat completion aborted runId=\(runId, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat completion aborted runId=\(runId)")
|
||||
streamingTask?.cancel()
|
||||
await self.finishIncrementalSpeech()
|
||||
await self.start()
|
||||
return
|
||||
} else if completion.state == .error {
|
||||
self.statusText = "Chat error"
|
||||
self.logger.warning("chat completion error runId=\(runId, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat completion error runId=\(runId)")
|
||||
streamingTask?.cancel()
|
||||
await self.finishIncrementalSpeech()
|
||||
await self.start()
|
||||
return
|
||||
}
|
||||
}
|
||||
let completion = await waitForChatCompletion(runId: runId, gateway: gateway, timeoutSeconds: 120)
|
||||
if completion.state == .timeout {
|
||||
self.logger.warning(
|
||||
"chat completion timeout runId=\(runId, privacy: .public); attempting history fallback")
|
||||
GatewayDiagnostics.log("talk: chat completion timeout runId=\(runId)")
|
||||
} else if completion.state == .aborted {
|
||||
self.statusText = "Aborted"
|
||||
self.logger.warning("chat completion aborted runId=\(runId, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat completion aborted runId=\(runId)")
|
||||
streamingTask?.cancel()
|
||||
await self.finishIncrementalSpeech()
|
||||
await self.start()
|
||||
return
|
||||
} else if completion.state == .error {
|
||||
self.statusText = "Chat error"
|
||||
self.logger.warning("chat completion error runId=\(runId, privacy: .public)")
|
||||
GatewayDiagnostics.log("talk: chat completion error runId=\(runId)")
|
||||
streamingTask?.cancel()
|
||||
await self.finishIncrementalSpeech()
|
||||
await self.start()
|
||||
return
|
||||
}
|
||||
|
||||
var assistantText = completion.assistantText
|
||||
|
|
@ -1053,7 +1074,7 @@ final class TalkModeManager: NSObject {
|
|||
if assistantText == nil {
|
||||
assistantText = try await self.waitForAssistantTextFromHistory(
|
||||
gateway: gateway,
|
||||
since: startedAt,
|
||||
since: Self.chatSendHistorySince(response: ack, startedAt: startedAt),
|
||||
timeoutSeconds: completion.state == .final ? 12 : 25)
|
||||
}
|
||||
guard let assistantText else {
|
||||
|
|
@ -1343,8 +1364,27 @@ final class TalkModeManager: NSObject {
|
|||
var assistantText: String?
|
||||
}
|
||||
|
||||
private func sendChat(_ message: String, gateway: GatewayNodeSession) async throws -> String {
|
||||
struct SendResponse: Decodable { let runId: String }
|
||||
private static func normalizedChatSendStatus(_ status: String) -> String {
|
||||
status.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
|
||||
}
|
||||
|
||||
private static func isTerminalChatSendSuccess(_ status: String) -> Bool {
|
||||
self.normalizedChatSendStatus(status) == "ok"
|
||||
}
|
||||
|
||||
private static func isTerminalChatSendFailure(_ status: String) -> Bool {
|
||||
let normalized = self.normalizedChatSendStatus(status)
|
||||
return normalized == "timeout" || normalized == "error"
|
||||
}
|
||||
|
||||
private static func chatSendHistorySince(
|
||||
response: OpenClawChatSendResponse,
|
||||
startedAt: Double) -> Double?
|
||||
{
|
||||
self.isTerminalChatSendSuccess(response.status) ? nil : startedAt
|
||||
}
|
||||
|
||||
private func sendChat(_ message: String, gateway: GatewayNodeSession) async throws -> OpenClawChatSendResponse {
|
||||
let payload: [String: Any] = [
|
||||
"sessionKey": mainSessionKey,
|
||||
"message": message,
|
||||
|
|
@ -1360,8 +1400,7 @@ final class TalkModeManager: NSObject {
|
|||
userInfo: [NSLocalizedDescriptionKey: "Failed to encode chat payload"])
|
||||
}
|
||||
let res = try await gateway.request(method: "chat.send", paramsJSON: json, timeoutSeconds: 30)
|
||||
let decoded = try JSONDecoder().decode(SendResponse.self, from: res)
|
||||
return decoded.runId
|
||||
return try JSONDecoder().decode(OpenClawChatSendResponse.self, from: res)
|
||||
}
|
||||
|
||||
private func waitForChatCompletion(
|
||||
|
|
@ -1440,7 +1479,7 @@ final class TalkModeManager: NSObject {
|
|||
|
||||
private func waitForAssistantTextFromHistory(
|
||||
gateway: GatewayNodeSession,
|
||||
since: Double,
|
||||
since: Double?,
|
||||
timeoutSeconds: Int) async throws -> String?
|
||||
{
|
||||
let deadline = Date().addingTimeInterval(TimeInterval(timeoutSeconds))
|
||||
|
|
|
|||
|
|
@ -391,20 +391,40 @@ actor TalkModeRuntime {
|
|||
idempotencyKey: runId,
|
||||
attachments: [])
|
||||
guard self.isCurrent(gen) else { return }
|
||||
let normalizedStatus = Self.normalizedChatSendStatus(response.status)
|
||||
self.logger.info(
|
||||
"talk chat.send ok runId=\(response.runId, privacy: .public) " +
|
||||
"status=\(normalizedStatus, privacy: .public) " +
|
||||
"session=\(sessionKey, privacy: .public)")
|
||||
if Self.isTerminalChatSendFailure(response.status) {
|
||||
self.logger.warning(
|
||||
"talk chat.send terminal ack runId=\(response.runId, privacy: .public) " +
|
||||
"status=\(normalizedStatus, privacy: .public)")
|
||||
await self.resumeListeningIfNeeded()
|
||||
return
|
||||
}
|
||||
|
||||
var assistantText = await self.waitForAssistantEventText(
|
||||
sessionKey: sessionKey,
|
||||
runId: response.runId,
|
||||
timeoutSeconds: 45)
|
||||
if assistantText == nil {
|
||||
self.logger.warning("talk assistant event text missing; using history fallback")
|
||||
var assistantText: String?
|
||||
if Self.isTerminalChatSendSuccess(response.status) {
|
||||
self.logger.info(
|
||||
"talk chat.send terminal ok runId=\(response.runId, privacy: .public); " +
|
||||
"using history fallback")
|
||||
assistantText = await self.waitForAssistantTextFromHistory(
|
||||
sessionKey: sessionKey,
|
||||
since: startedAt,
|
||||
since: nil,
|
||||
timeoutSeconds: 12)
|
||||
} else {
|
||||
assistantText = await self.waitForAssistantEventText(
|
||||
sessionKey: sessionKey,
|
||||
runId: response.runId,
|
||||
timeoutSeconds: 45)
|
||||
if assistantText == nil {
|
||||
self.logger.warning("talk assistant event text missing; using history fallback")
|
||||
assistantText = await self.waitForAssistantTextFromHistory(
|
||||
sessionKey: sessionKey,
|
||||
since: startedAt,
|
||||
timeoutSeconds: 12)
|
||||
}
|
||||
}
|
||||
guard let assistantText
|
||||
else {
|
||||
|
|
@ -499,6 +519,19 @@ actor TalkModeRuntime {
|
|||
}
|
||||
}
|
||||
|
||||
private static func normalizedChatSendStatus(_ status: String) -> String {
|
||||
status.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
|
||||
}
|
||||
|
||||
private static func isTerminalChatSendSuccess(_ status: String) -> Bool {
|
||||
self.normalizedChatSendStatus(status) == "ok"
|
||||
}
|
||||
|
||||
private static func isTerminalChatSendFailure(_ status: String) -> Bool {
|
||||
let normalized = self.normalizedChatSendStatus(status)
|
||||
return normalized == "timeout" || normalized == "error"
|
||||
}
|
||||
|
||||
private static func matchesSessionKey(_ incoming: String, _ current: String) -> Bool {
|
||||
let incoming = incoming.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
|
||||
let current = current.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
|
||||
|
|
@ -509,7 +542,7 @@ actor TalkModeRuntime {
|
|||
|
||||
private func waitForAssistantTextFromHistory(
|
||||
sessionKey: String,
|
||||
since: Double,
|
||||
since: Double?,
|
||||
timeoutSeconds: Int) async -> String?
|
||||
{
|
||||
let deadline = Date().addingTimeInterval(TimeInterval(timeoutSeconds))
|
||||
|
|
|
|||
|
|
@ -868,25 +868,32 @@ public final class OpenClawChatViewModel {
|
|||
self.pendingLocalUserEchoMessageIDsByRunID[response.runId] = pendingUserMessageID
|
||||
self.armPendingRunTimeout(runId: response.runId)
|
||||
}
|
||||
let historyContext = self.beginHistoryRequest(for: sessionSnapshot)
|
||||
await self.refreshHistoryAfterRun(historyRequest: historyContext)
|
||||
guard self.isCurrentSession(sessionSnapshot) else { return }
|
||||
if !self.clearPendingRunIfAssistantMessagePresent(
|
||||
runId: response.runId,
|
||||
after: userMessageTimestamp)
|
||||
{
|
||||
self.armPostSendRefreshFallback(
|
||||
if response.status == "ok" {
|
||||
let historyContext = self.beginHistoryRequest(for: sessionSnapshot)
|
||||
await self.refreshHistoryAfterRun(historyRequest: historyContext)
|
||||
guard self.isCurrentSession(sessionSnapshot) else { return }
|
||||
self.finishPendingRunAfterTerminalOkSendAck(response)
|
||||
} else if !self.finishPendingRunIfTerminalSendAck(response) {
|
||||
let historyContext = self.beginHistoryRequest(for: sessionSnapshot)
|
||||
await self.refreshHistoryAfterRun(historyRequest: historyContext)
|
||||
guard self.isCurrentSession(sessionSnapshot) else { return }
|
||||
if !self.clearPendingRunIfAssistantMessagePresent(
|
||||
runId: response.runId,
|
||||
sessionSnapshot: sessionSnapshot,
|
||||
userMessageTimestamp: userMessageTimestamp)
|
||||
self.armRunCompletionRefresh(
|
||||
runId: response.runId,
|
||||
sessionSnapshot: sessionSnapshot,
|
||||
userMessageTimestamp: userMessageTimestamp)
|
||||
after: userMessageTimestamp)
|
||||
{
|
||||
self.armPostSendRefreshFallback(
|
||||
runId: response.runId,
|
||||
sessionSnapshot: sessionSnapshot,
|
||||
userMessageTimestamp: userMessageTimestamp)
|
||||
self.armRunCompletionRefresh(
|
||||
runId: response.runId,
|
||||
sessionSnapshot: sessionSnapshot,
|
||||
userMessageTimestamp: userMessageTimestamp)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
guard self.isCurrentSession(sessionSnapshot) else { return }
|
||||
self.pendingLocalUserEchoMessageIDsByRunID[runId] = nil
|
||||
self.removePendingLocalUserEcho(for: runId)
|
||||
self.clearPendingRun(runId)
|
||||
self.errorText = error.localizedDescription
|
||||
self.logDiagnostic(
|
||||
|
|
@ -1717,6 +1724,48 @@ public final class OpenClawChatViewModel {
|
|||
return "Chat failed"
|
||||
}
|
||||
|
||||
private func finishPendingRunAfterTerminalOkSendAck(_ response: OpenClawChatSendResponse) {
|
||||
self.clearPendingRun(response.runId)
|
||||
self.pendingToolCallsById = [:]
|
||||
self.streamingAssistantText = nil
|
||||
self.logDiagnostic(
|
||||
"chat.ui send terminal ack sessionKey=\(self.sessionKey) "
|
||||
+ "runId=\(response.runId) status=ok")
|
||||
}
|
||||
|
||||
private func finishPendingRunIfTerminalSendAck(_ response: OpenClawChatSendResponse) -> Bool {
|
||||
switch response.status {
|
||||
case "timeout":
|
||||
self.removePendingLocalUserEcho(for: response.runId)
|
||||
self.clearPendingRun(response.runId)
|
||||
self.pendingToolCallsById = [:]
|
||||
self.streamingAssistantText = nil
|
||||
self.errorText = "Chat failed before the run started; try again."
|
||||
self.logDiagnostic(
|
||||
"chat.ui send terminal ack sessionKey=\(self.sessionKey) "
|
||||
+ "runId=\(response.runId) status=timeout")
|
||||
return true
|
||||
case "error":
|
||||
self.removePendingLocalUserEcho(for: response.runId)
|
||||
self.clearPendingRun(response.runId)
|
||||
self.pendingToolCallsById = [:]
|
||||
self.streamingAssistantText = nil
|
||||
self.errorText = "Chat failed before the run started; try again."
|
||||
self.logDiagnostic(
|
||||
"chat.ui send terminal ack sessionKey=\(self.sessionKey) "
|
||||
+ "runId=\(response.runId) status=error")
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
private func removePendingLocalUserEcho(for runId: String) {
|
||||
guard let messageID = self.pendingLocalUserEchoMessageIDsByRunID[runId] else { return }
|
||||
self.messages.removeAll { $0.id == messageID }
|
||||
self.pendingLocalUserEchoMessageIDsByRunID[runId] = nil
|
||||
}
|
||||
|
||||
private func armPostSendRefreshFallback(
|
||||
runId: String,
|
||||
sessionSnapshot: SessionSnapshot,
|
||||
|
|
|
|||
|
|
@ -21,6 +21,15 @@ private func chatErrorMessage(role: String, errorMessage: String, timestamp: Dou
|
|||
])
|
||||
}
|
||||
|
||||
fileprivate extension Array where Element == OpenClawChatMessage {
|
||||
func containsUserText(_ text: String) -> Bool {
|
||||
self.contains { message in
|
||||
message.role == "user" &&
|
||||
message.content.contains { $0.text == text }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func historyPayload(
|
||||
sessionKey: String = "main",
|
||||
sessionId: String? = "sess-main",
|
||||
|
|
@ -104,6 +113,7 @@ private func makeViewModel(
|
|||
compactSessionHook: (@Sendable (String) async throws -> Void)? = nil,
|
||||
setSessionModelHook: (@Sendable (String?) async throws -> Void)? = nil,
|
||||
setSessionThinkingHook: (@Sendable (String) async throws -> Void)? = nil,
|
||||
sendMessageHook: (@Sendable (String) async throws -> OpenClawChatSendResponse)? = nil,
|
||||
waitForRunCompletionHook: (@Sendable (String, Int) async -> Bool)? = nil,
|
||||
healthResponses: [Bool] = [true],
|
||||
initialThinkingLevel: String? = nil,
|
||||
|
|
@ -122,6 +132,7 @@ private func makeViewModel(
|
|||
compactSessionHook: compactSessionHook,
|
||||
setSessionModelHook: setSessionModelHook,
|
||||
setSessionThinkingHook: setSessionThinkingHook,
|
||||
sendMessageHook: sendMessageHook,
|
||||
waitForRunCompletionHook: waitForRunCompletionHook,
|
||||
healthResponses: healthResponses)
|
||||
let vm = await MainActor.run {
|
||||
|
|
@ -347,6 +358,7 @@ private final class TestChatTransport: @unchecked Sendable, OpenClawChatTranspor
|
|||
private let compactSessionHook: (@Sendable (String) async throws -> Void)?
|
||||
private let setSessionModelHook: (@Sendable (String?) async throws -> Void)?
|
||||
private let setSessionThinkingHook: (@Sendable (String) async throws -> Void)?
|
||||
private let sendMessageHook: (@Sendable (String) async throws -> OpenClawChatSendResponse)?
|
||||
private let waitForRunCompletionHook: (@Sendable (String, Int) async -> Bool)?
|
||||
private let healthResponses: [Bool]
|
||||
|
||||
|
|
@ -364,6 +376,7 @@ private final class TestChatTransport: @unchecked Sendable, OpenClawChatTranspor
|
|||
compactSessionHook: (@Sendable (String) async throws -> Void)? = nil,
|
||||
setSessionModelHook: (@Sendable (String?) async throws -> Void)? = nil,
|
||||
setSessionThinkingHook: (@Sendable (String) async throws -> Void)? = nil,
|
||||
sendMessageHook: (@Sendable (String) async throws -> OpenClawChatSendResponse)? = nil,
|
||||
waitForRunCompletionHook: (@Sendable (String, Int) async -> Bool)? = nil,
|
||||
healthResponses: [Bool] = [true])
|
||||
{
|
||||
|
|
@ -377,6 +390,7 @@ private final class TestChatTransport: @unchecked Sendable, OpenClawChatTranspor
|
|||
self.compactSessionHook = compactSessionHook
|
||||
self.setSessionModelHook = setSessionModelHook
|
||||
self.setSessionThinkingHook = setSessionThinkingHook
|
||||
self.sendMessageHook = sendMessageHook
|
||||
self.waitForRunCompletionHook = waitForRunCompletionHook
|
||||
self.healthResponses = healthResponses
|
||||
var cont: AsyncStream<OpenClawChatTransportEvent>.Continuation!
|
||||
|
|
@ -435,6 +449,9 @@ private final class TestChatTransport: @unchecked Sendable, OpenClawChatTranspor
|
|||
await self.state.sentSessionKeysAppend(sessionKey)
|
||||
await self.state.sentRunIdsAppend(idempotencyKey)
|
||||
await self.state.sentThinkingLevelsAppend(thinking)
|
||||
if let sendMessageHook {
|
||||
return try await sendMessageHook(idempotencyKey)
|
||||
}
|
||||
return OpenClawChatSendResponse(runId: idempotencyKey, status: "ok")
|
||||
}
|
||||
|
||||
|
|
@ -880,6 +897,50 @@ struct ChatViewModelTests {
|
|||
#expect(await MainActor.run { vm.input } == "second")
|
||||
}
|
||||
|
||||
@Test func terminalOkSendAckClearsPendingRunWithoutWaitingForCompletion() async throws {
|
||||
let sessionId = "sess-main"
|
||||
let history = historyPayload(sessionId: sessionId, messages: [])
|
||||
let (transport, vm) = await makeViewModel(historyResponses: [history, history])
|
||||
try await loadAndWaitBootstrap(vm: vm, sessionId: sessionId)
|
||||
|
||||
await sendUserMessage(vm, text: "cached")
|
||||
try await waitUntil("terminal ok ack clears pending run") {
|
||||
await MainActor.run { vm.pendingRunCount == 0 && !vm.isSending }
|
||||
}
|
||||
|
||||
#expect(await MainActor.run { vm.errorText } == nil)
|
||||
#expect(await transport.waitCompletionRunIds().isEmpty)
|
||||
#expect(await MainActor.run { vm.messages.containsUserText("cached") })
|
||||
}
|
||||
|
||||
@Test func terminalTimeoutSendAckSurfacesErrorAndAllowsNextSend() async throws {
|
||||
let sessionId = "sess-main"
|
||||
let history = historyPayload(sessionId: sessionId, messages: [])
|
||||
let sendCount = AsyncCounter()
|
||||
let (transport, vm) = await makeViewModel(
|
||||
historyResponses: [history],
|
||||
sendMessageHook: { runId in
|
||||
let count = await sendCount.increment()
|
||||
return OpenClawChatSendResponse(
|
||||
runId: runId,
|
||||
status: count == 1 ? "timeout" : "ok")
|
||||
})
|
||||
try await loadAndWaitBootstrap(vm: vm, sessionId: sessionId)
|
||||
|
||||
await sendUserMessage(vm, text: "first")
|
||||
try await waitUntil("timeout ack clears pending run") {
|
||||
await MainActor.run { vm.pendingRunCount == 0 && !vm.isSending }
|
||||
}
|
||||
#expect(await transport.sentRunIds().count == 1)
|
||||
#expect(await MainActor.run { vm.errorText } == "Chat failed before the run started; try again.")
|
||||
#expect(await MainActor.run { !vm.messages.containsUserText("first") })
|
||||
|
||||
await sendUserMessage(vm, text: "second")
|
||||
try await waitUntil("second send is accepted after timeout ack") {
|
||||
await transport.sentRunIds().count == 2
|
||||
}
|
||||
}
|
||||
|
||||
@Test func `keeps optimistic user message when final refresh returns only assistant history`() async throws {
|
||||
let sessionId = "sess-main"
|
||||
let now = Date().timeIntervalSince1970 * 1000
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { describe, expect, it } from "vitest";
|
|||
import {
|
||||
AgentsListResultSchema,
|
||||
SkillsProposalInspectResultSchema,
|
||||
SkillsProposalRequestRevisionResultSchema,
|
||||
ToolsEffectiveResultSchema,
|
||||
} from "./agents-models-skills.js";
|
||||
|
||||
|
|
@ -149,3 +150,26 @@ describe("SkillsProposalInspectResultSchema", () => {
|
|||
expect(Value.Check(SkillsProposalInspectResultSchema, result)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("SkillsProposalRequestRevisionResultSchema", () => {
|
||||
it.each(["started", "in_flight", "ok", "timeout", "error"])(
|
||||
"accepts forwarded chat.send ack status %s",
|
||||
(status) => {
|
||||
expect(
|
||||
Value.Check(SkillsProposalRequestRevisionResultSchema, {
|
||||
runId: "run-revision",
|
||||
status,
|
||||
}),
|
||||
).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects unknown forwarded chat.send ack statuses", () => {
|
||||
expect(
|
||||
Value.Check(SkillsProposalRequestRevisionResultSchema, {
|
||||
runId: "run-revision",
|
||||
status: "queued",
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -787,7 +787,13 @@ export const SkillsProposalRequestRevisionParamsSchema = Type.Object(
|
|||
export const SkillsProposalRequestRevisionResultSchema = Type.Object(
|
||||
{
|
||||
runId: NonEmptyString,
|
||||
status: Type.Union([Type.Literal("started"), Type.Literal("in_flight"), Type.Literal("ok")]),
|
||||
status: Type.Union([
|
||||
Type.Literal("started"),
|
||||
Type.Literal("in_flight"),
|
||||
Type.Literal("ok"),
|
||||
Type.Literal("timeout"),
|
||||
Type.Literal("error"),
|
||||
]),
|
||||
},
|
||||
{ additionalProperties: true },
|
||||
);
|
||||
|
|
|
|||
|
|
@ -111,6 +111,15 @@ function createSessionRow(params: {
|
|||
};
|
||||
}
|
||||
|
||||
async function settlePromptQuickly<T>(promise: Promise<T>): Promise<T | "pending"> {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<"pending">((resolve) => {
|
||||
setTimeout(() => resolve("pending"), 50);
|
||||
}),
|
||||
]);
|
||||
}
|
||||
|
||||
async function startPendingPrompt(params: {
|
||||
agent: AcpGatewayAgent;
|
||||
sentRunIds: string[];
|
||||
|
|
@ -409,6 +418,81 @@ describe("acp translator stable lifecycle handlers", () => {
|
|||
sessionStore.clearAllSessionsForTest();
|
||||
});
|
||||
|
||||
it("resolves prompts when chat send returns a terminal timeout ack", async () => {
|
||||
const request = vi.fn(async (method: string, params?: Record<string, unknown>) => {
|
||||
if (method === "chat.send") {
|
||||
return { runId: params?.idempotencyKey, status: "timeout" };
|
||||
}
|
||||
if (method === "sessions.list") {
|
||||
return createGatewaySessions([createSessionRow({ key: "agent:main:work" })]);
|
||||
}
|
||||
return { ok: true };
|
||||
}) as GatewayClient["request"];
|
||||
const sessionStore = createInMemorySessionStore();
|
||||
sessionStore.createSession({
|
||||
sessionId: "session-1",
|
||||
sessionKey: "agent:main:work",
|
||||
cwd: "/tmp/openclaw",
|
||||
});
|
||||
const agent = new AcpGatewayAgent(createAcpConnection(), createAcpGateway(request), {
|
||||
sessionStore,
|
||||
});
|
||||
|
||||
await expect(
|
||||
settlePromptQuickly(agent.prompt(createPromptRequest("session-1"))),
|
||||
).resolves.toEqual({
|
||||
stopReason: "cancelled",
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects prompts when chat send returns a terminal error ack", async () => {
|
||||
const request = vi.fn(async (method: string, params?: Record<string, unknown>) => {
|
||||
if (method === "chat.send") {
|
||||
return { runId: params?.idempotencyKey, status: "error" };
|
||||
}
|
||||
return { ok: true };
|
||||
}) as GatewayClient["request"];
|
||||
const sessionStore = createInMemorySessionStore();
|
||||
sessionStore.createSession({
|
||||
sessionId: "session-1",
|
||||
sessionKey: "agent:main:work",
|
||||
cwd: "/tmp/openclaw",
|
||||
});
|
||||
const agent = new AcpGatewayAgent(createAcpConnection(), createAcpGateway(request), {
|
||||
sessionStore,
|
||||
});
|
||||
|
||||
await expect(
|
||||
settlePromptQuickly(agent.prompt(createPromptRequest("session-1"))),
|
||||
).rejects.toThrow("Chat failed before the run started; try again.");
|
||||
});
|
||||
|
||||
it("resolves prompts when chat send returns a terminal ok ack", async () => {
|
||||
const requestMock = vi.fn(async (method: string, params?: Record<string, unknown>) => {
|
||||
if (method === "chat.send") {
|
||||
return { runId: params?.idempotencyKey, status: "ok" };
|
||||
}
|
||||
return { ok: true };
|
||||
});
|
||||
const request = requestMock as GatewayClient["request"];
|
||||
const sessionStore = createInMemorySessionStore();
|
||||
sessionStore.createSession({
|
||||
sessionId: "session-1",
|
||||
sessionKey: "agent:main:work",
|
||||
cwd: "/tmp/openclaw",
|
||||
});
|
||||
const agent = new AcpGatewayAgent(createAcpConnection(), createAcpGateway(request), {
|
||||
sessionStore,
|
||||
});
|
||||
|
||||
await expect(
|
||||
settlePromptQuickly(agent.prompt(createPromptRequest("session-1"))),
|
||||
).resolves.toEqual({
|
||||
stopReason: "end_turn",
|
||||
});
|
||||
expect(requestMock.mock.calls.filter(([method]) => method === "chat.send")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("closes sessions by aborting active work, resolving pending prompts, and deleting bridge state", async () => {
|
||||
const sentRunIds: string[] = [];
|
||||
const request = vi.fn(async (method: string, params?: Record<string, unknown>) => {
|
||||
|
|
|
|||
|
|
@ -562,6 +562,64 @@ describe("acp translator stop reason mapping", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("ignores stale send close errors while reconnect finish is settling the same prompt", async () => {
|
||||
let rejectChatSend: ((err: Error) => void) | undefined;
|
||||
const chatSendPromise = new Promise<never>((_, reject) => {
|
||||
rejectChatSend = reject;
|
||||
});
|
||||
const request = vi.fn((method: string) => {
|
||||
if (method === "chat.send") {
|
||||
return chatSendPromise;
|
||||
}
|
||||
if (method === "agent.wait") {
|
||||
return Promise.resolve({ status: "ok" });
|
||||
}
|
||||
return Promise.resolve({});
|
||||
}) as GatewayClient["request"];
|
||||
let releaseSessionUpdate: (() => void) | undefined;
|
||||
let blockNextSessionUpdate = true;
|
||||
const sessionUpdate = vi.fn(() => {
|
||||
if (!blockNextSessionUpdate) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
blockNextSessionUpdate = false;
|
||||
return new Promise<void>((resolve) => {
|
||||
releaseSessionUpdate = resolve;
|
||||
});
|
||||
});
|
||||
const connection = createAcpConnection();
|
||||
connection.sessionUpdate = sessionUpdate as typeof connection.sessionUpdate;
|
||||
const sessionStore = createInMemorySessionStore();
|
||||
const sessionId = "session-1";
|
||||
sessionStore.createSession({
|
||||
sessionId,
|
||||
sessionKey: "agent:main:main",
|
||||
cwd: "/tmp",
|
||||
});
|
||||
const agent = new AcpGatewayAgent(connection, createAcpGateway(request), {
|
||||
sessionStore,
|
||||
});
|
||||
|
||||
agent.handleGatewayDisconnect("1006: connection lost");
|
||||
const promptPromise = promptAgent(agent, sessionId);
|
||||
await Promise.resolve();
|
||||
agent.handleGatewayReconnect();
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(sessionUpdate).toHaveBeenCalled();
|
||||
});
|
||||
rejectChatSend?.(new Error("gateway closed (1006): connection lost"));
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
await expect(Promise.race([promptPromise, Promise.resolve("pending")])).resolves.toBe(
|
||||
"pending",
|
||||
);
|
||||
|
||||
releaseSessionUpdate?.();
|
||||
await expect(promptPromise).resolves.toEqual({ stopReason: "end_turn" });
|
||||
});
|
||||
|
||||
it("does not let a stale disconnect deadline reject a newer prompt on the same session", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -109,6 +109,24 @@ const MAX_PROMPT_BYTES = 2 * 1024 * 1024;
|
|||
const ACP_LOAD_SESSION_REPLAY_LIMIT = 1_000_000;
|
||||
const ACP_GATEWAY_DISCONNECT_GRACE_MS = 5_000;
|
||||
|
||||
type ChatSendAck = {
|
||||
runId?: unknown;
|
||||
status?: unknown;
|
||||
};
|
||||
|
||||
function normalizedChatSendAckStatus(status: unknown): string {
|
||||
return typeof status === "string" ? status.trim().toLowerCase() : "";
|
||||
}
|
||||
|
||||
function isTerminalChatSendAckFailure(status: unknown): boolean {
|
||||
const normalized = normalizedChatSendAckStatus(status);
|
||||
return normalized === "timeout" || normalized === "error";
|
||||
}
|
||||
|
||||
function isTerminalChatSendAckSuccess(status: unknown): boolean {
|
||||
return normalizedChatSendAckStatus(status) === "ok";
|
||||
}
|
||||
|
||||
let acpCommandsModulePromise: Promise<typeof import("./commands.js")> | undefined;
|
||||
let acpSdkModulePromise: Promise<typeof import("@agentclientprotocol/sdk")> | undefined;
|
||||
|
||||
|
|
@ -181,7 +199,8 @@ function isAdminScopeProvenanceRejection(err: unknown): boolean {
|
|||
}
|
||||
|
||||
function isGatewayCloseError(err: unknown): boolean {
|
||||
return err instanceof Error && err.message.startsWith("gateway closed (");
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
return message.startsWith("gateway closed (");
|
||||
}
|
||||
|
||||
type AgentWaitResult = {
|
||||
|
|
@ -237,6 +256,7 @@ export class AcpGatewayAgent implements Agent {
|
|||
private sessionUpdates: AcpTranslatorSessionUpdates;
|
||||
private sessionCreateRateLimiter: FixedWindowRateLimiter;
|
||||
private pendingPrompts = new Map<string, PendingPrompt>();
|
||||
private settlingPromptKeys = new Set<string>();
|
||||
private approvalRelays = new Map<string, PendingApprovalRelay>();
|
||||
private clientCapabilities: ClientCapabilityState = normalizeClientCapabilities(undefined);
|
||||
private clientInfo: InitializeRequest["clientInfo"] = null;
|
||||
|
|
@ -244,6 +264,10 @@ export class AcpGatewayAgent implements Agent {
|
|||
private activeDisconnectContext: DisconnectContext | null = null;
|
||||
private disconnectGeneration = 0;
|
||||
|
||||
private pendingPromptKey(sessionId: string, runId: string): string {
|
||||
return `${sessionId}\u0000${runId}`;
|
||||
}
|
||||
|
||||
private getPendingPrompt(sessionId: string, runId: string): PendingPrompt | undefined {
|
||||
const pending = this.pendingPrompts.get(sessionId);
|
||||
if (pending?.idempotencyKey !== runId) {
|
||||
|
|
@ -706,16 +730,54 @@ export class AcpGatewayAgent implements Agent {
|
|||
pending.sendAccepted = true;
|
||||
}
|
||||
};
|
||||
const applyTerminalAck = async (ack: ChatSendAck | undefined): Promise<boolean> => {
|
||||
const status = normalizedChatSendAckStatus(ack?.status);
|
||||
const pending = () => this.getPendingPrompt(params.sessionId, runId);
|
||||
if (status === "timeout") {
|
||||
const current = pending();
|
||||
if (current) {
|
||||
await this.finishPrompt(params.sessionId, current, "cancelled");
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (status === "error") {
|
||||
const current = pending();
|
||||
if (current) {
|
||||
this.rejectPendingPrompt(
|
||||
current,
|
||||
new Error("Chat failed before the run started; try again."),
|
||||
);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (status === "ok") {
|
||||
markSendAccepted();
|
||||
await this.sessionUpdates.recordUserPrompt(session, runId, params.prompt);
|
||||
const current = pending();
|
||||
if (current) {
|
||||
await this.finishPrompt(params.sessionId, current, "end_turn");
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return isTerminalChatSendAckFailure(status) || isTerminalChatSendAckSuccess(status);
|
||||
};
|
||||
|
||||
const sendChat = async (payload: Record<string, unknown>): Promise<boolean> => {
|
||||
const ack = await this.gateway.request<ChatSendAck>("chat.send", payload, {
|
||||
timeoutMs: null,
|
||||
});
|
||||
return await applyTerminalAck(ack);
|
||||
};
|
||||
|
||||
try {
|
||||
await this.gateway.request(
|
||||
"chat.send",
|
||||
{
|
||||
...requestParams,
|
||||
systemInputProvenance,
|
||||
systemProvenanceReceipt,
|
||||
},
|
||||
{ timeoutMs: null },
|
||||
);
|
||||
const terminal = await sendChat({
|
||||
...requestParams,
|
||||
systemInputProvenance,
|
||||
systemProvenanceReceipt,
|
||||
});
|
||||
if (terminal) {
|
||||
return;
|
||||
}
|
||||
markSendAccepted();
|
||||
await this.sessionUpdates.recordUserPrompt(session, runId, params.prompt);
|
||||
} catch (err) {
|
||||
|
|
@ -723,7 +785,10 @@ export class AcpGatewayAgent implements Agent {
|
|||
(systemInputProvenance || systemProvenanceReceipt) &&
|
||||
isAdminScopeProvenanceRejection(err)
|
||||
) {
|
||||
await this.gateway.request("chat.send", requestParams, { timeoutMs: null });
|
||||
const terminal = await sendChat(requestParams);
|
||||
if (terminal) {
|
||||
return;
|
||||
}
|
||||
markSendAccepted();
|
||||
await this.sessionUpdates.recordUserPrompt(session, runId, params.prompt);
|
||||
return;
|
||||
|
|
@ -733,7 +798,12 @@ export class AcpGatewayAgent implements Agent {
|
|||
};
|
||||
|
||||
void sendWithProvenanceFallback().catch((err: unknown) => {
|
||||
if (isGatewayCloseError(err) && this.getPendingPrompt(params.sessionId, runId)) {
|
||||
const promptKey = this.pendingPromptKey(params.sessionId, runId);
|
||||
if (
|
||||
isGatewayCloseError(err) &&
|
||||
(this.getPendingPrompt(params.sessionId, runId) ||
|
||||
this.settlingPromptKeys.has(promptKey))
|
||||
) {
|
||||
return;
|
||||
}
|
||||
this.clearApprovalRelaysForPrompt(params.sessionId, runId, { denyActive: true });
|
||||
|
|
@ -1154,31 +1224,37 @@ export class AcpGatewayAgent implements Agent {
|
|||
pending: PendingPrompt,
|
||||
stopReason: StopReason,
|
||||
): Promise<void> {
|
||||
this.clearApprovalRelaysForPrompt(sessionId, pending.idempotencyKey, { denyActive: true });
|
||||
this.pendingPrompts.delete(sessionId);
|
||||
this.sessionStore.clearActiveRun(sessionId);
|
||||
if (this.pendingPrompts.size === 0) {
|
||||
this.clearDisconnectTimer();
|
||||
}
|
||||
const sessionSnapshot = await this.getSessionSnapshot(pending.sessionKey);
|
||||
const promptKey = this.pendingPromptKey(sessionId, pending.idempotencyKey);
|
||||
this.settlingPromptKeys.add(promptKey);
|
||||
try {
|
||||
await this.sendSessionSnapshotUpdate(
|
||||
{
|
||||
sessionId,
|
||||
sessionKey: pending.sessionKey,
|
||||
...(pending.ledgerSessionId ? { ledgerSessionId: pending.ledgerSessionId } : {}),
|
||||
},
|
||||
sessionSnapshot,
|
||||
{
|
||||
includeControls: false,
|
||||
record: true,
|
||||
runId: pending.idempotencyKey,
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
this.log(`session snapshot update failed for ${sessionId}: ${String(err)}`);
|
||||
this.clearApprovalRelaysForPrompt(sessionId, pending.idempotencyKey, { denyActive: true });
|
||||
this.pendingPrompts.delete(sessionId);
|
||||
this.sessionStore.clearActiveRun(sessionId);
|
||||
if (this.pendingPrompts.size === 0) {
|
||||
this.clearDisconnectTimer();
|
||||
}
|
||||
const sessionSnapshot = await this.getSessionSnapshot(pending.sessionKey);
|
||||
try {
|
||||
await this.sendSessionSnapshotUpdate(
|
||||
{
|
||||
sessionId,
|
||||
sessionKey: pending.sessionKey,
|
||||
...(pending.ledgerSessionId ? { ledgerSessionId: pending.ledgerSessionId } : {}),
|
||||
},
|
||||
sessionSnapshot,
|
||||
{
|
||||
includeControls: false,
|
||||
record: true,
|
||||
runId: pending.idempotencyKey,
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
this.log(`session snapshot update failed for ${sessionId}: ${String(err)}`);
|
||||
}
|
||||
pending.resolve({ stopReason });
|
||||
} finally {
|
||||
this.settlingPromptKeys.delete(promptKey);
|
||||
}
|
||||
pending.resolve({ stopReason });
|
||||
}
|
||||
|
||||
private findPendingBySessionKey(sessionKey: string, runId?: string): PendingPrompt | undefined {
|
||||
|
|
|
|||
|
|
@ -1408,6 +1408,49 @@ describe("talk.client.toolCall handler", () => {
|
|||
expectRespondOk(respond, { runId: "run-voice-1" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["timeout", "Realtime agent consult ended before the run started."],
|
||||
["error", "Realtime agent consult failed before the run started."],
|
||||
["ok", "Realtime agent consult completed before the tool result subscription started."],
|
||||
] as const)(
|
||||
"rejects terminal agent consult chat.send ACKs with status %s",
|
||||
async (status, message) => {
|
||||
mocks.chatSend.mockImplementationOnce(
|
||||
async ({
|
||||
respond,
|
||||
}: {
|
||||
respond: (ok: boolean, result?: unknown, error?: unknown) => void;
|
||||
}) => {
|
||||
respond(true, { runId: `run-${status}`, status }, undefined);
|
||||
},
|
||||
);
|
||||
const respond = vi.fn();
|
||||
|
||||
await talkHandlers["talk.client.toolCall"]({
|
||||
req: { type: "req", id: "1", method: "talk.client.toolCall" },
|
||||
params: {
|
||||
sessionKey: "main",
|
||||
relaySessionId: "relay-1",
|
||||
callId: "call-1",
|
||||
name: "openclaw_agent_consult",
|
||||
args: { question: "What now?" },
|
||||
},
|
||||
client: { connId: "conn-1" } as never,
|
||||
isWebchatConnect: () => false,
|
||||
respond: respond as never,
|
||||
context: {
|
||||
getRuntimeConfig: () => ({}) as OpenClawConfig,
|
||||
} as never,
|
||||
});
|
||||
|
||||
expect(mocks.registerTalkRealtimeRelayAgentRun).not.toHaveBeenCalled();
|
||||
expectRespondError(respond, {
|
||||
code: ErrorCodes.UNAVAILABLE,
|
||||
message,
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects client tool calls that are not the agent consult tool", async () => {
|
||||
const respond = vi.fn();
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,40 @@ import type {
|
|||
import { registerTalkRealtimeRelayAgentRun } from "./talk-realtime-relay.js";
|
||||
import { formatForLog } from "./ws-log.js";
|
||||
|
||||
type TalkChatSendAckStatus = "started" | "in_flight" | "ok" | "timeout" | "error";
|
||||
|
||||
function normalizeTalkChatSendAckStatus(result: unknown): TalkChatSendAckStatus {
|
||||
if (!result || typeof result !== "object" || Array.isArray(result)) {
|
||||
return "started";
|
||||
}
|
||||
const status = (result as Record<string, unknown>).status;
|
||||
return status === "in_flight" || status === "ok" || status === "timeout" || status === "error"
|
||||
? status
|
||||
: "started";
|
||||
}
|
||||
|
||||
function terminalTalkChatSendAckError(status: TalkChatSendAckStatus): ErrorShape | undefined {
|
||||
if (status === "timeout") {
|
||||
return errorShape(
|
||||
ErrorCodes.UNAVAILABLE,
|
||||
"Realtime agent consult ended before the run started.",
|
||||
);
|
||||
}
|
||||
if (status === "error") {
|
||||
return errorShape(
|
||||
ErrorCodes.UNAVAILABLE,
|
||||
"Realtime agent consult failed before the run started.",
|
||||
);
|
||||
}
|
||||
if (status === "ok") {
|
||||
return errorShape(
|
||||
ErrorCodes.UNAVAILABLE,
|
||||
"Realtime agent consult completed before the tool result subscription started.",
|
||||
);
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts the agent-consult chat run that backs realtime Talk tool calls.
|
||||
*/
|
||||
|
|
@ -83,6 +117,10 @@ export async function startTalkRealtimeAgentConsult(params: {
|
|||
return { ok: false, error: chatResponse.error };
|
||||
}
|
||||
const result = chatResponse.result;
|
||||
const terminalAckError = terminalTalkChatSendAckError(normalizeTalkChatSendAckStatus(result));
|
||||
if (terminalAckError) {
|
||||
return { ok: false, error: terminalAckError };
|
||||
}
|
||||
const runId =
|
||||
result && typeof result === "object" && !Array.isArray(result)
|
||||
? typeof (result as Record<string, unknown>).runId === "string"
|
||||
|
|
|
|||
|
|
@ -75,6 +75,7 @@ import type {
|
|||
ChatSendOptions,
|
||||
TuiAgentsList,
|
||||
TuiBackend,
|
||||
TuiChatSendResult,
|
||||
TuiEvent,
|
||||
TuiModelChoice,
|
||||
TuiSessionList,
|
||||
|
|
@ -355,7 +356,7 @@ export class EmbeddedTuiBackend implements TuiBackend {
|
|||
setEmbeddedMode(false);
|
||||
}
|
||||
|
||||
async sendChat(opts: ChatSendOptions): Promise<{ runId: string }> {
|
||||
async sendChat(opts: ChatSendOptions): Promise<TuiChatSendResult> {
|
||||
const runId = opts.runId ?? randomUUID();
|
||||
const question = resolveBtwQuestion(opts.message);
|
||||
const runScope = {
|
||||
|
|
|
|||
|
|
@ -655,6 +655,24 @@ describe("GatewayChatClient", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("returns the actual chat send ack status from the gateway", async () => {
|
||||
const client = new GatewayChatClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
token: "test-token",
|
||||
allowInsecureLocalOperatorUi: true,
|
||||
});
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-gateway", status: "timeout" });
|
||||
(client as unknown as { client: { request: typeof request } }).client.request = request;
|
||||
|
||||
const result = await client.sendChat({
|
||||
sessionKey: "main",
|
||||
message: "hello",
|
||||
runId: "run-local",
|
||||
});
|
||||
|
||||
expect(result).toEqual({ runId: "run-gateway", status: "timeout" });
|
||||
});
|
||||
|
||||
it("lists gateway commands through commands.list", async () => {
|
||||
const client = new GatewayChatClient({
|
||||
url: "ws://127.0.0.1:18789",
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ import type {
|
|||
TuiModelChoice,
|
||||
TuiSessionList,
|
||||
TuiSessionMutationResult,
|
||||
TuiChatSendResult,
|
||||
} from "./tui-backend.js";
|
||||
|
||||
export type GatewayConnectionOptions = {
|
||||
|
|
@ -100,6 +101,10 @@ function sleep(ms: number): Promise<void> {
|
|||
});
|
||||
}
|
||||
|
||||
function nonEmptyString(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
|
||||
}
|
||||
|
||||
export type GatewaySessionList = TuiSessionList;
|
||||
export type GatewayAgentsList = TuiAgentsList;
|
||||
export type GatewayModelChoice = TuiModelChoice;
|
||||
|
|
@ -194,9 +199,9 @@ export class GatewayChatClient implements TuiBackend {
|
|||
await this.readyPromise;
|
||||
}
|
||||
|
||||
async sendChat(opts: ChatSendOptions): Promise<{ runId: string }> {
|
||||
async sendChat(opts: ChatSendOptions): Promise<TuiChatSendResult> {
|
||||
const runId = opts.runId ?? randomUUID();
|
||||
await this.client.request("chat.send", {
|
||||
const response = await this.client.request<{ runId?: unknown; status?: unknown }>("chat.send", {
|
||||
sessionKey: opts.sessionKey,
|
||||
...(opts.agentId ? { agentId: opts.agentId } : {}),
|
||||
...(opts.sessionId ? { sessionId: opts.sessionId } : {}),
|
||||
|
|
@ -206,7 +211,9 @@ export class GatewayChatClient implements TuiBackend {
|
|||
timeoutMs: opts.timeoutMs,
|
||||
idempotencyKey: runId,
|
||||
});
|
||||
return { runId };
|
||||
const acceptedRunId = nonEmptyString(response?.runId) ?? runId;
|
||||
const status = nonEmptyString(response?.status);
|
||||
return status ? { runId: acceptedRunId, status } : { runId: acceptedRunId };
|
||||
}
|
||||
|
||||
async abortChat(opts: { sessionKey: string; agentId?: string; runId: string }) {
|
||||
|
|
|
|||
|
|
@ -22,6 +22,11 @@ export type ChatSendOptions = {
|
|||
runId?: string;
|
||||
};
|
||||
|
||||
export type TuiChatSendResult = {
|
||||
runId: string;
|
||||
status?: string;
|
||||
};
|
||||
|
||||
/** Options for forwarding a goal command to a backend session. */
|
||||
export type TuiGoalCommandOptions = {
|
||||
sessionKey: string;
|
||||
|
|
@ -143,7 +148,7 @@ export type TuiBackend = {
|
|||
start: () => void;
|
||||
stop: () => void | Promise<void>;
|
||||
subscribeSessionEvents?: () => Promise<unknown>;
|
||||
sendChat: (opts: ChatSendOptions) => Promise<{ runId: string }>;
|
||||
sendChat: (opts: ChatSendOptions) => Promise<TuiChatSendResult>;
|
||||
abortChat: (opts: {
|
||||
sessionKey: string;
|
||||
agentId?: string;
|
||||
|
|
|
|||
|
|
@ -93,7 +93,9 @@ function createHarness(params?: {
|
|||
isRunObserved?: (runId: string) => boolean;
|
||||
flushPendingHistoryRefreshIfIdle?: FlushPendingHistoryRefreshMock;
|
||||
}) {
|
||||
const sendChat = params?.sendChat ?? vi.fn().mockResolvedValue({ runId: "r1" });
|
||||
const sendChat =
|
||||
params?.sendChat ??
|
||||
vi.fn().mockImplementation(async (opts: { runId?: string }) => ({ runId: opts.runId ?? "r1" }));
|
||||
const getGatewayStatus = params?.getGatewayStatus ?? vi.fn().mockResolvedValue({});
|
||||
const listSessions = params?.listSessions ?? vi.fn().mockResolvedValue({ sessions: [] });
|
||||
const listModels = params?.listModels ?? vi.fn().mockResolvedValue([]);
|
||||
|
|
@ -120,6 +122,7 @@ function createHarness(params?: {
|
|||
const applySessionMutationResult = params?.applySessionMutationResult ?? vi.fn();
|
||||
const setActivityStatus = params?.setActivityStatus ?? (vi.fn() as SetActivityStatusMock);
|
||||
const forgetLocalRunId = vi.fn();
|
||||
const forgetLocalBtwRunId = vi.fn();
|
||||
const openOverlay = vi.fn();
|
||||
const closeOverlay = vi.fn();
|
||||
const requestExit = vi.fn();
|
||||
|
|
@ -181,7 +184,7 @@ function createHarness(params?: {
|
|||
noteLocalRunId,
|
||||
noteLocalBtwRunId,
|
||||
forgetLocalRunId,
|
||||
forgetLocalBtwRunId: vi.fn(),
|
||||
forgetLocalBtwRunId,
|
||||
consumeCompletedRunForPendingSend: params?.consumeCompletedRunForPendingSend,
|
||||
isRunObserved: params?.isRunObserved,
|
||||
flushPendingHistoryRefreshIfIdle: params?.flushPendingHistoryRefreshIfIdle,
|
||||
|
|
@ -220,6 +223,7 @@ function createHarness(params?: {
|
|||
noteLocalRunId,
|
||||
noteLocalBtwRunId,
|
||||
forgetLocalRunId,
|
||||
forgetLocalBtwRunId,
|
||||
requestExit,
|
||||
abortActive,
|
||||
state,
|
||||
|
|
@ -644,6 +648,174 @@ describe("tui command handlers", () => {
|
|||
expect(noteLocalRunId).toHaveBeenCalledWith("run-accepted");
|
||||
});
|
||||
|
||||
it("clears optimistic state when chat send returns a terminal timeout ack", async () => {
|
||||
const sendChat = vi.fn().mockImplementation(async (opts: { runId: string }) => ({
|
||||
runId: opts.runId,
|
||||
status: "timeout",
|
||||
}));
|
||||
const historyReload = { clearSystemMessages: undefined as (() => void) | undefined };
|
||||
const loadHistory = vi.fn().mockImplementation(async () => {
|
||||
historyReload.clearSystemMessages?.();
|
||||
}) as LoadHistoryMock;
|
||||
const { handleCommand, state, dropPendingUser, addSystem, setActivityStatus } = createHarness({
|
||||
sendChat,
|
||||
loadHistory,
|
||||
});
|
||||
historyReload.clearSystemMessages = () => addSystem.mockClear();
|
||||
|
||||
await handleCommand("hello");
|
||||
|
||||
const sentRunId = (firstMockArg(sendChat, "sendChat") as { runId: string }).runId;
|
||||
expect(dropPendingUser).toHaveBeenCalledWith(sentRunId);
|
||||
expect(state.pendingSubmitDraft).toBeNull();
|
||||
expect(state.pendingChatRunId).toBeNull();
|
||||
expect(state.pendingOptimisticUserMessage).toBe(false);
|
||||
expect(addSystem).toHaveBeenCalledWith(
|
||||
"send failed: Chat failed before the run started; try again.",
|
||||
);
|
||||
expect(setActivityStatus).toHaveBeenLastCalledWith("error");
|
||||
expect(loadHistory).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("reports failure when chat send returns a terminal error ack", async () => {
|
||||
const sendChat = vi.fn().mockImplementation(async (opts: { runId: string }) => ({
|
||||
runId: opts.runId,
|
||||
status: "error",
|
||||
}));
|
||||
const loadHistory = vi.fn().mockResolvedValue(undefined) as LoadHistoryMock;
|
||||
const { handleCommand, state, dropPendingUser, addSystem, setActivityStatus } = createHarness({
|
||||
sendChat,
|
||||
loadHistory,
|
||||
});
|
||||
|
||||
await handleCommand("hello");
|
||||
|
||||
const sentRunId = (firstMockArg(sendChat, "sendChat") as { runId: string }).runId;
|
||||
expect(dropPendingUser).toHaveBeenCalledWith(sentRunId);
|
||||
expect(addSystem).toHaveBeenCalledWith(
|
||||
"send failed: Chat failed before the run started; try again.",
|
||||
);
|
||||
expect(state.pendingSubmitDraft).toBeNull();
|
||||
expect(state.pendingChatRunId).toBeNull();
|
||||
expect(state.pendingOptimisticUserMessage).toBe(false);
|
||||
expect(setActivityStatus).toHaveBeenLastCalledWith("error");
|
||||
expect(loadHistory).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("refreshes history without waiting when chat send returns a terminal ok ack", async () => {
|
||||
const sendChat = vi.fn().mockImplementation(async (opts: { runId: string }) => ({
|
||||
runId: opts.runId,
|
||||
status: "ok",
|
||||
}));
|
||||
const loadHistory = vi.fn().mockResolvedValue(undefined) as LoadHistoryMock;
|
||||
const { handleCommand, state, dropPendingUser, setActivityStatus } = createHarness({
|
||||
sendChat,
|
||||
loadHistory,
|
||||
});
|
||||
|
||||
await handleCommand("hello");
|
||||
|
||||
expect(dropPendingUser).not.toHaveBeenCalled();
|
||||
expect(state.pendingSubmitDraft).toBeNull();
|
||||
expect(state.pendingChatRunId).toBeNull();
|
||||
expect(state.pendingOptimisticUserMessage).toBe(false);
|
||||
expect(setActivityStatus).toHaveBeenLastCalledWith("idle");
|
||||
expect(loadHistory).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["/btw", "timeout", undefined],
|
||||
["/side", "error", "run-accepted-side-error"],
|
||||
])(
|
||||
"clears local BTW tracking and reports failure when %s receives a terminal %s ack",
|
||||
async (command, status, acceptedRunId) => {
|
||||
const sendChat = vi.fn().mockImplementation(async (opts: { runId: string }) => ({
|
||||
runId: acceptedRunId ?? opts.runId,
|
||||
status,
|
||||
}));
|
||||
const {
|
||||
handleCommand,
|
||||
sendChat: sendChatMock,
|
||||
forgetLocalBtwRunId,
|
||||
addSystem,
|
||||
state,
|
||||
} = createHarness({
|
||||
sendChat,
|
||||
activeChatRunId: "run-main",
|
||||
});
|
||||
|
||||
await handleCommand(`${command} check terminal ack`);
|
||||
|
||||
const sentRunId = (firstMockArg(sendChatMock, "sendChat") as { runId: string }).runId;
|
||||
expect(forgetLocalBtwRunId).toHaveBeenCalledWith(sentRunId);
|
||||
if (acceptedRunId) {
|
||||
expect(forgetLocalBtwRunId).toHaveBeenCalledWith(acceptedRunId);
|
||||
}
|
||||
expect(addSystem).toHaveBeenCalledWith(
|
||||
"btw failed: Chat failed before the run started; try again.",
|
||||
);
|
||||
expect(state.activeChatRunId).toBe("run-main");
|
||||
expect(state.pendingChatRunId).toBeNull();
|
||||
expect(state.pendingOptimisticUserMessage).toBe(false);
|
||||
},
|
||||
);
|
||||
|
||||
it("clears local BTW tracking when a detached send receives a terminal ok ack", async () => {
|
||||
const sendChat = vi.fn().mockImplementation(async () => ({
|
||||
runId: "run-accepted-btw",
|
||||
status: "ok",
|
||||
}));
|
||||
const {
|
||||
handleCommand,
|
||||
sendChat: sendChatMock,
|
||||
forgetLocalBtwRunId,
|
||||
addSystem,
|
||||
state,
|
||||
} = createHarness({
|
||||
sendChat,
|
||||
activeChatRunId: "run-main",
|
||||
});
|
||||
|
||||
await handleCommand("/side finish detached");
|
||||
|
||||
const sentRunId = (firstMockArg(sendChatMock, "sendChat") as { runId: string }).runId;
|
||||
expect(forgetLocalBtwRunId).toHaveBeenCalledWith(sentRunId);
|
||||
expect(forgetLocalBtwRunId).toHaveBeenCalledWith("run-accepted-btw");
|
||||
expect(addSystem).not.toHaveBeenCalled();
|
||||
expect(state.activeChatRunId).toBe("run-main");
|
||||
expect(state.pendingChatRunId).toBeNull();
|
||||
expect(state.pendingOptimisticUserMessage).toBe(false);
|
||||
});
|
||||
|
||||
it("tracks the backend-accepted runId for a detached non-terminal ack", async () => {
|
||||
const sendChat = vi.fn().mockImplementation(async () => ({
|
||||
runId: "run-accepted-btw",
|
||||
status: "in_flight",
|
||||
}));
|
||||
const {
|
||||
handleCommand,
|
||||
sendChat: sendChatMock,
|
||||
noteLocalBtwRunId,
|
||||
forgetLocalBtwRunId,
|
||||
addSystem,
|
||||
state,
|
||||
} = createHarness({
|
||||
sendChat,
|
||||
activeChatRunId: "run-main",
|
||||
});
|
||||
|
||||
await handleCommand("/btw continue detached");
|
||||
|
||||
const sentRunId = (firstMockArg(sendChatMock, "sendChat") as { runId: string }).runId;
|
||||
expect(noteLocalBtwRunId).toHaveBeenCalledWith(sentRunId);
|
||||
expect(forgetLocalBtwRunId).toHaveBeenCalledWith(sentRunId);
|
||||
expect(noteLocalBtwRunId).toHaveBeenCalledWith("run-accepted-btw");
|
||||
expect(addSystem).not.toHaveBeenCalled();
|
||||
expect(state.activeChatRunId).toBe("run-main");
|
||||
expect(state.pendingChatRunId).toBeNull();
|
||||
expect(state.pendingOptimisticUserMessage).toBe(false);
|
||||
});
|
||||
|
||||
it("does not reintroduce a backend-accepted runId after an early terminal event", async () => {
|
||||
const sendChat = vi.fn().mockResolvedValue({ runId: "run-accepted" });
|
||||
const consumeCompletedRunForPendingSend = vi.fn((runId: string) => runId === "run-accepted");
|
||||
|
|
|
|||
|
|
@ -84,6 +84,21 @@ function isSlashStopCommand(text: string): boolean {
|
|||
return trimmed.startsWith("/") && isChatStopCommandText(trimmed);
|
||||
}
|
||||
|
||||
function normalizedChatSendAckStatus(status: unknown): string {
|
||||
return typeof status === "string" ? status.trim().toLowerCase() : "";
|
||||
}
|
||||
|
||||
function isTerminalChatSendAckFailure(status: unknown): boolean {
|
||||
const normalized = normalizedChatSendAckStatus(status);
|
||||
return normalized === "timeout" || normalized === "error";
|
||||
}
|
||||
|
||||
function isTerminalChatSendAckSuccess(status: unknown): boolean {
|
||||
return normalizedChatSendAckStatus(status) === "ok";
|
||||
}
|
||||
|
||||
const TERMINAL_CHAT_SEND_FAILURE_MESSAGE = "Chat failed before the run started; try again.";
|
||||
|
||||
function goalContinuationPrompt(text: string): string | null {
|
||||
const parsed = parseGoalCommand(text);
|
||||
if (!parsed) {
|
||||
|
|
@ -782,25 +797,69 @@ export function createCommandHandlers(context: CommandHandlerContext) {
|
|||
timeoutMs: opts.timeoutMs,
|
||||
runId,
|
||||
});
|
||||
const acceptedRunId = sendResult.runId || runId;
|
||||
const terminalAckFailure = isTerminalChatSendAckFailure(sendResult.status);
|
||||
const terminalAckSuccess = isTerminalChatSendAckSuccess(sendResult.status);
|
||||
const terminalAck = terminalAckFailure || terminalAckSuccess;
|
||||
if (isBtw && terminalAck) {
|
||||
forgetLocalBtwRunId?.(runId);
|
||||
if (acceptedRunId !== runId) {
|
||||
forgetLocalBtwRunId?.(acceptedRunId);
|
||||
}
|
||||
if (terminalAckFailure) {
|
||||
chatLog.addSystem(`btw failed: ${TERMINAL_CHAT_SEND_FAILURE_MESSAGE}`);
|
||||
}
|
||||
tui.requestRender();
|
||||
return;
|
||||
}
|
||||
if (isBtw) {
|
||||
if (acceptedRunId !== runId) {
|
||||
forgetLocalBtwRunId?.(runId);
|
||||
noteLocalBtwRunId?.(acceptedRunId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (!isBtw) {
|
||||
const acceptedRunId = sendResult.runId || runId;
|
||||
const acceptedRunAlreadyCompleted =
|
||||
acceptedRunId !== runId && (consumeCompletedRunForPendingSend?.(acceptedRunId) ?? false);
|
||||
acceptedRunId !== runId &&
|
||||
!terminalAck &&
|
||||
(consumeCompletedRunForPendingSend?.(acceptedRunId) ?? false);
|
||||
if (acceptedRunId !== runId) {
|
||||
forgetLocalRunId?.(runId);
|
||||
if (!acceptedRunAlreadyCompleted) {
|
||||
if (!acceptedRunAlreadyCompleted && !terminalAck) {
|
||||
noteLocalRunId?.(acceptedRunId);
|
||||
}
|
||||
if (state.pendingSubmitDraft?.runId === runId) {
|
||||
// If the accepted run already emitted events, it is registered;
|
||||
// re-arming the draft would let a later abort drop a row whose
|
||||
// reply already rendered.
|
||||
state.pendingSubmitDraft = isRunObserved?.(acceptedRunId)
|
||||
? null
|
||||
: { runId: acceptedRunId, text };
|
||||
// If the accepted run already emitted events or the ACK is already terminal,
|
||||
// re-arming the draft would let a later abort drop a row whose lifecycle ended.
|
||||
state.pendingSubmitDraft =
|
||||
isRunObserved?.(acceptedRunId) || terminalAck ? null : { runId: acceptedRunId, text };
|
||||
}
|
||||
chatLog.rekeyPendingUser(runId, acceptedRunId);
|
||||
}
|
||||
if (terminalAck) {
|
||||
if (state.pendingSubmitDraft?.runId === acceptedRunId) {
|
||||
state.pendingSubmitDraft = null;
|
||||
}
|
||||
forgetLocalRunId?.(acceptedRunId);
|
||||
if (terminalAckFailure) {
|
||||
chatLog.dropPendingUser(acceptedRunId);
|
||||
}
|
||||
if (state.activeChatRunId === acceptedRunId) {
|
||||
state.activeChatRunId = null;
|
||||
}
|
||||
state.pendingOptimisticUserMessage = false;
|
||||
state.pendingChatRunId = null;
|
||||
await loadHistory();
|
||||
if (terminalAckFailure) {
|
||||
chatLog.addSystem(`send failed: ${TERMINAL_CHAT_SEND_FAILURE_MESSAGE}`);
|
||||
setActivityStatus("error");
|
||||
} else {
|
||||
setActivityStatus("idle");
|
||||
}
|
||||
tui.requestRender();
|
||||
return;
|
||||
}
|
||||
if (state.pendingOptimisticUserMessage) {
|
||||
if (acceptedRunAlreadyCompleted) {
|
||||
if (state.pendingSubmitDraft?.runId === acceptedRunId) {
|
||||
|
|
|
|||
|
|
@ -1452,6 +1452,71 @@ describe("handleSendChat", () => {
|
|||
expect(host.chatMessage).toBe("keep this draft");
|
||||
});
|
||||
|
||||
it("does not seed refreshSessionsAfterChat for a terminal timeout ack on a refreshing send", async () => {
|
||||
const request = vi.fn(async (method: string) => {
|
||||
if (method === "chat.send") {
|
||||
return { status: "timeout" };
|
||||
}
|
||||
throw new Error(`Unexpected request: ${method}`);
|
||||
});
|
||||
const host = makeHost({
|
||||
client: { request } as unknown as ChatHost["client"],
|
||||
chatMessage: "/reset",
|
||||
sessionKey: "agent:main",
|
||||
});
|
||||
|
||||
await handleSendChat(host);
|
||||
|
||||
const payload = findRequestPayload(
|
||||
request as unknown as MockCallSource,
|
||||
"chat.send",
|
||||
"chat send payload",
|
||||
);
|
||||
const runId = String(payload.idempotencyKey);
|
||||
const runState = host as ChatHost & {
|
||||
chatStreamStartedAt?: number | null;
|
||||
lastLocalTerminalReconcile?: unknown;
|
||||
};
|
||||
expect(host.chatRunId).toBeNull();
|
||||
expect(host.chatStream).toBeNull();
|
||||
expect(runState.chatStreamStartedAt).toBeNull();
|
||||
expect(runState.lastLocalTerminalReconcile).toMatchObject({
|
||||
phase: "interrupted",
|
||||
runId,
|
||||
sessionKey: "agent:main",
|
||||
sessionStatus: "killed",
|
||||
});
|
||||
expect(host.refreshSessionsAfterChat.size).toBe(0);
|
||||
});
|
||||
|
||||
it("marks terminal error ACK sends failed instead of accepting the queued message", async () => {
|
||||
const request = vi.fn(async (method: string, params?: unknown) => {
|
||||
if (method === "chat.send") {
|
||||
const payload = requireRecord(params, "chat send payload");
|
||||
return { runId: payload.idempotencyKey, status: "error" };
|
||||
}
|
||||
throw new Error(`Unexpected request: ${method}`);
|
||||
});
|
||||
const host = makeHost({
|
||||
client: { request } as unknown as ChatHost["client"],
|
||||
chatMessage: "send before failing",
|
||||
sessionKey: "agent:main",
|
||||
});
|
||||
|
||||
await handleSendChat(host);
|
||||
|
||||
expect(host.chatMessages).toStrictEqual([]);
|
||||
expect(host.chatMessage).toBe("send before failing");
|
||||
expect(host.chatQueue).toHaveLength(1);
|
||||
expect(host.chatQueue[0]).toMatchObject({
|
||||
text: "send before failing",
|
||||
sendState: "failed",
|
||||
sendError: "Chat failed before the run started; try again.",
|
||||
});
|
||||
expect(host.lastError).toBe("Chat failed before the run started; try again.");
|
||||
expect(host.chatRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("records visible send timing phases for a normal chat send", async () => {
|
||||
const request = vi.fn(async (method: string) => {
|
||||
if (method === "chat.send") {
|
||||
|
|
@ -2763,6 +2828,30 @@ describe("handleSendChat", () => {
|
|||
expect(host.lastError).toBe("network down");
|
||||
});
|
||||
|
||||
it("restores the BTW draft when detached send returns a terminal timeout ACK", async () => {
|
||||
const host = makeHost({
|
||||
client: {
|
||||
request: vi.fn(async (method: string) => {
|
||||
if (method === "chat.send") {
|
||||
return { runId: "btw-terminal", status: "timeout" };
|
||||
}
|
||||
throw new Error(`Unexpected request: ${method}`);
|
||||
}),
|
||||
} as unknown as ChatHost["client"],
|
||||
chatRunId: "run-main",
|
||||
chatStream: "Working...",
|
||||
chatMessage: "/btw what changed?",
|
||||
});
|
||||
|
||||
await handleSendChat(host);
|
||||
|
||||
expect(host.chatQueue).toStrictEqual([]);
|
||||
expect(host.chatRunId).toBe("run-main");
|
||||
expect(host.chatStream).toBe("Working...");
|
||||
expect(host.chatMessage).toBe("/btw what changed?");
|
||||
expect(host.lastError).toBe("The active run ended before the detached message was accepted.");
|
||||
});
|
||||
|
||||
it("clears BTW side results when /clear resets chat history", async () => {
|
||||
const request = vi.fn(async (method: string) => {
|
||||
if (method === "sessions.reset") {
|
||||
|
|
@ -2903,6 +2992,54 @@ describe("handleSendChat", () => {
|
|||
expect(host.chatQueue[0]?.pendingRunId).toBe("run-1");
|
||||
});
|
||||
|
||||
it("removes queued steer indicators when chat.send returns terminal ok", async () => {
|
||||
const request = vi.fn(async (method: string) => {
|
||||
if (method === "chat.send") {
|
||||
return { status: "ok", runId: "steer-ok" };
|
||||
}
|
||||
throw new Error(`Unexpected request: ${method}`);
|
||||
});
|
||||
const host = makeHost({
|
||||
client: { request } as unknown as ChatHost["client"],
|
||||
chatRunId: "run-1",
|
||||
chatStream: "Working...",
|
||||
chatQueue: [{ id: "queued-1", text: "tighten the plan", createdAt: 1 }],
|
||||
sessionKey: "agent:main:main",
|
||||
});
|
||||
|
||||
await steerQueuedChatMessage(host, "queued-1");
|
||||
|
||||
expect(host.chatRunId).toBe("run-1");
|
||||
expect(host.chatStream).toBe("Working...");
|
||||
expect(host.chatQueue).toStrictEqual([]);
|
||||
expect(setLastActiveSessionKeyMock).toHaveBeenCalledWith(expect.anything(), "agent:main:main");
|
||||
});
|
||||
|
||||
it("restores queued steer items when chat.send returns terminal error", async () => {
|
||||
const request = vi.fn(async (method: string) => {
|
||||
if (method === "chat.send") {
|
||||
return { status: "error", runId: "steer-error" };
|
||||
}
|
||||
throw new Error(`Unexpected request: ${method}`);
|
||||
});
|
||||
const original = { id: "queued-1", text: "tighten the plan", createdAt: 1 };
|
||||
const host = makeHost({
|
||||
client: { request } as unknown as ChatHost["client"],
|
||||
chatRunId: "run-1",
|
||||
chatStream: "Working...",
|
||||
chatQueue: [original],
|
||||
sessionKey: "agent:main:main",
|
||||
});
|
||||
|
||||
await steerQueuedChatMessage(host, "queued-1");
|
||||
|
||||
expect(host.chatRunId).toBe("run-1");
|
||||
expect(host.chatStream).toBe("Working...");
|
||||
expect(host.chatQueue).toStrictEqual([original]);
|
||||
expect(host.lastError).toBe("Steer failed before it reached the run; try again.");
|
||||
expect(setLastActiveSessionKeyMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("removes pending steer indicators when the run finishes", () => {
|
||||
const host = makeHost({
|
||||
chatQueue: [
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
// Control UI module implements app chat behavior.
|
||||
import { isNonTerminalAgentRunStatus } from "../../../src/shared/agent-run-status.js";
|
||||
import { setLastActiveSessionKey } from "./app-last-active-session.ts";
|
||||
import { scheduleChatScroll, resetChatScroll } from "./app-scroll.ts";
|
||||
import { resetToolStream } from "./app-tool-stream.ts";
|
||||
|
|
@ -152,6 +153,36 @@ function setChatError(host: ChatHost, error: string | null) {
|
|||
host.chatError = error;
|
||||
}
|
||||
|
||||
type AcceptedChatSendAck = ChatSendAck & { status: "started" | "in_flight" | "ok" };
|
||||
type TerminalFailureChatSendAck = ChatSendAck & { status: "timeout" | "error" };
|
||||
|
||||
function isAcceptedChatSendAck(ack: ChatSendAck | null): ack is AcceptedChatSendAck {
|
||||
return ack != null && (ack.status === "ok" || isNonTerminalAgentRunStatus(ack.status));
|
||||
}
|
||||
|
||||
function isTerminalFailureChatSendAck(ack: ChatSendAck | null): ack is TerminalFailureChatSendAck {
|
||||
return ack?.status === "timeout" || ack?.status === "error";
|
||||
}
|
||||
|
||||
function formatTerminalChatSendAckError(
|
||||
ack: TerminalFailureChatSendAck,
|
||||
context: "chat" | "detached" | "steer",
|
||||
): string {
|
||||
if (ack.status === "error") {
|
||||
if (context === "steer") {
|
||||
return "Steer failed before it reached the run; try again.";
|
||||
}
|
||||
return "Chat failed before the run started; try again.";
|
||||
}
|
||||
if (context === "detached") {
|
||||
return "The active run ended before the detached message was accepted.";
|
||||
}
|
||||
if (context === "steer") {
|
||||
return "The active run ended before the steer message was accepted.";
|
||||
}
|
||||
return "The run ended before the message was accepted.";
|
||||
}
|
||||
|
||||
export type ChatSendOptions = {
|
||||
confirmReset?: boolean;
|
||||
restoreDraft?: boolean;
|
||||
|
|
@ -1077,6 +1108,38 @@ async function sendQueuedChatMessage(
|
|||
requestDurationMs: roundedControlUiDurationMs(controlUiNowMs() - requestStartedAtMs),
|
||||
...chatSendAckServerTimingEventFields(ack),
|
||||
});
|
||||
if (isTerminalFailureChatSendAck(ack)) {
|
||||
const error = formatTerminalChatSendAckError(ack, "chat");
|
||||
updateQueuedMessageForSession(host, sessionKey, id, (item) => ({
|
||||
...item,
|
||||
sendError: error,
|
||||
sendState: "failed",
|
||||
}));
|
||||
if (isVisibleSession()) {
|
||||
reconcileChatRunLifecycle(
|
||||
host as unknown as Parameters<typeof reconcileChatRunLifecycle>[0],
|
||||
{
|
||||
outcome: "interrupted",
|
||||
sessionStatus: ack.status === "error" ? "failed" : "killed",
|
||||
runId: ack.runId,
|
||||
sessionKey,
|
||||
clearLocalRun: true,
|
||||
clearChatStream: true,
|
||||
clearToolStream: true,
|
||||
clearSideResultTerminalRuns: true,
|
||||
publishRunStatus: false,
|
||||
armLocalTerminalReconcile: ack.runId === runId,
|
||||
},
|
||||
);
|
||||
setChatError(host, error);
|
||||
restoreComposerAfterFailedSend(host, opts ?? {});
|
||||
}
|
||||
recordChatSendTiming(host, sendingItem, "failed", sendingItem.sendSubmittedAtMs, {
|
||||
error,
|
||||
ackStatus: ack.status,
|
||||
});
|
||||
return "failed";
|
||||
}
|
||||
removeQueuedMessageWithoutReleasing(host, id, sessionKey);
|
||||
if (isVisibleSession()) {
|
||||
appendUserChatMessage(
|
||||
|
|
@ -1102,7 +1165,7 @@ async function sendQueuedChatMessage(
|
|||
},
|
||||
);
|
||||
void loadChatHistory(host as unknown as ChatState);
|
||||
} else {
|
||||
} else if (isNonTerminalAgentRunStatus(ack.status)) {
|
||||
const hasAlreadyAdoptedRunStream =
|
||||
host.chatRunId === ack.runId && typeof host.chatStream === "string";
|
||||
host.chatRunId = ack.runId;
|
||||
|
|
@ -1114,6 +1177,22 @@ async function sendQueuedChatMessage(
|
|||
(host as ChatHost & { chatStreamStartedAt?: number | null }).chatStreamStartedAt =
|
||||
startedAt;
|
||||
}
|
||||
} else {
|
||||
reconcileChatRunLifecycle(
|
||||
host as unknown as Parameters<typeof reconcileChatRunLifecycle>[0],
|
||||
{
|
||||
outcome: "interrupted",
|
||||
sessionStatus: ack.status === "error" ? "failed" : "killed",
|
||||
runId: ack.runId,
|
||||
sessionKey,
|
||||
clearLocalRun: true,
|
||||
clearChatStream: true,
|
||||
clearToolStream: true,
|
||||
clearSideResultTerminalRuns: true,
|
||||
publishRunStatus: false,
|
||||
armLocalTerminalReconcile: ack.runId === runId,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
if (prepared.refreshSessions) {
|
||||
|
|
@ -1126,7 +1205,7 @@ async function sendQueuedChatMessage(
|
|||
...createChatSessionsLoadOverrides(host),
|
||||
...scopedAgentListParamsForRefreshTarget(host, refreshTarget),
|
||||
});
|
||||
} else {
|
||||
} else if (isNonTerminalAgentRunStatus(ack.status)) {
|
||||
host.refreshSessionsAfterChat.set(ack.runId, refreshTarget);
|
||||
}
|
||||
}
|
||||
|
|
@ -1359,18 +1438,21 @@ async function sendDetachedBtwMessage(
|
|||
previousAttachments?: ChatAttachment[];
|
||||
},
|
||||
) {
|
||||
const runId = await sendDetachedChatMessage(
|
||||
const ack = await sendDetachedChatMessage(
|
||||
host as unknown as ChatState,
|
||||
message,
|
||||
opts?.attachments,
|
||||
);
|
||||
const ok = Boolean(runId);
|
||||
const ok = isAcceptedChatSendAck(ack);
|
||||
if (!ok && opts?.previousDraft != null) {
|
||||
host.chatMessage = opts.previousDraft;
|
||||
}
|
||||
if (!ok && opts?.previousAttachments) {
|
||||
host.chatAttachments = opts.previousAttachments;
|
||||
}
|
||||
if (isTerminalFailureChatSendAck(ack)) {
|
||||
setChatError(host, formatTerminalChatSendAckError(ack, "detached"));
|
||||
}
|
||||
if (ok) {
|
||||
setLastActiveSessionKey(
|
||||
host as unknown as Parameters<typeof setLastActiveSessionKey>[0],
|
||||
|
|
@ -1402,15 +1484,21 @@ export async function steerQueuedChatMessage(host: ChatHost, id: string) {
|
|||
host.chatQueue = host.chatQueue.map((entry) =>
|
||||
entry.id === id ? { ...entry, kind: "steered", pendingRunId: activeRunId } : entry,
|
||||
);
|
||||
const runId = await sendSteerChatMessage(
|
||||
const ack = await sendSteerChatMessage(
|
||||
host as unknown as ChatState,
|
||||
message,
|
||||
hasAttachments ? attachments : undefined,
|
||||
);
|
||||
if (!runId) {
|
||||
if (!ack || isTerminalFailureChatSendAck(ack)) {
|
||||
host.chatQueue = host.chatQueue.map((entry) => (entry.id === id ? item : entry));
|
||||
if (isTerminalFailureChatSendAck(ack)) {
|
||||
setChatError(host, formatTerminalChatSendAckError(ack, "steer"));
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (ack.status === "ok") {
|
||||
removeQueuedMessageWithoutReleasing(host, id, host.sessionKey);
|
||||
}
|
||||
releaseChatAttachmentPayloads(attachments);
|
||||
setLastActiveSessionKey(
|
||||
host as unknown as Parameters<typeof setLastActiveSessionKey>[0],
|
||||
|
|
|
|||
|
|
@ -922,12 +922,68 @@ describe("executeSlashCommand /steer (soft inject)", () => {
|
|||
);
|
||||
|
||||
expect(result.content).toBe("Steered.");
|
||||
expect(result.pendingCurrentRun).toBe(true);
|
||||
const chatSend = requireRequestCall(request, "chat.send");
|
||||
expect(chatSend.payload.sessionKey).toBe("agent:main:main");
|
||||
expect(chatSend.payload.message).toBe("try a different approach");
|
||||
expect(chatSend.payload.deliver).toBe(false);
|
||||
});
|
||||
|
||||
it("does not mark the current run pending when chat.send returns terminal ok", async () => {
|
||||
const request = vi.fn(async (method: string, _payload?: unknown) => {
|
||||
if (method === "sessions.list") {
|
||||
return { sessions: [row("agent:main:main", { status: "running" })] };
|
||||
}
|
||||
if (method === "chat.send") {
|
||||
return { status: "ok", runId: "run-ok", messageSeq: 2 };
|
||||
}
|
||||
throw new Error(`unexpected method: ${method}`);
|
||||
});
|
||||
|
||||
const result = await executeSlashCommand(
|
||||
{ request } as unknown as GatewayBrowserClient,
|
||||
"agent:main:main",
|
||||
"steer",
|
||||
"try a different approach",
|
||||
);
|
||||
|
||||
expect(result.content).toBe("Steered.");
|
||||
expect(result.pendingCurrentRun).toBeUndefined();
|
||||
const chatSend = requireRequestCall(request, "chat.send");
|
||||
expect(chatSend.payload.deliver).toBe(false);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["timeout", "The active run ended before the steer message was accepted."],
|
||||
["error", "Steer failed before it reached the run; try again."],
|
||||
] as const)(
|
||||
"reports terminal %s ACK without marking the current run pending",
|
||||
async (status, expectedContent) => {
|
||||
const request = vi.fn(async (method: string, _payload?: unknown) => {
|
||||
if (method === "sessions.list") {
|
||||
return { sessions: [row("agent:main:main", { status: "running" })] };
|
||||
}
|
||||
if (method === "chat.send") {
|
||||
return { status, runId: `run-${status}`, summary: "aborted" };
|
||||
}
|
||||
throw new Error(`unexpected method: ${method}`);
|
||||
});
|
||||
|
||||
const result = await executeSlashCommand(
|
||||
{ request } as unknown as GatewayBrowserClient,
|
||||
"agent:main:main",
|
||||
"steer",
|
||||
"try a different approach",
|
||||
);
|
||||
|
||||
expect(result.content).toBe(expectedContent);
|
||||
expect(result.content).not.toBe("Steered.");
|
||||
expect(result.pendingCurrentRun).toBeUndefined();
|
||||
const chatSend = requireRequestCall(request, "chat.send");
|
||||
expect(chatSend.payload.deliver).toBe(false);
|
||||
},
|
||||
);
|
||||
|
||||
it("passes selected-agent scope when steering the selected global session", async () => {
|
||||
const request = vi.fn(async (method: string, _payload?: unknown) => {
|
||||
if (method === "sessions.list") {
|
||||
|
|
@ -1189,6 +1245,47 @@ describe("executeSlashCommand /redirect (hard kill-and-restart)", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("does not track a pending run when sessions.steer returns terminal ok", async () => {
|
||||
const request = vi.fn(async (method: string, _payload?: unknown) => {
|
||||
if (method === "sessions.steer") {
|
||||
return { status: "ok", runId: "run-ok", messageSeq: 2 };
|
||||
}
|
||||
throw new Error(`unexpected method: ${method}`);
|
||||
});
|
||||
|
||||
const result = await executeSlashCommand(
|
||||
{ request } as unknown as GatewayBrowserClient,
|
||||
"agent:main:main",
|
||||
"redirect",
|
||||
"start over with a new plan",
|
||||
);
|
||||
|
||||
expect(result.content).toBe("Redirected.");
|
||||
expect(result.trackRunId).toBeUndefined();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["timeout", "The active run ended before the redirect message was accepted."],
|
||||
["error", "Redirect failed before it reached the run; try again."],
|
||||
] as const)("reports terminal %s ACK from sessions.steer", async (status, expectedContent) => {
|
||||
const request = vi.fn(async (method: string, _payload?: unknown) => {
|
||||
if (method === "sessions.steer") {
|
||||
return { status, runId: `run-${status}`, summary: "aborted" };
|
||||
}
|
||||
throw new Error(`unexpected method: ${method}`);
|
||||
});
|
||||
|
||||
const result = await executeSlashCommand(
|
||||
{ request } as unknown as GatewayBrowserClient,
|
||||
"agent:main:main",
|
||||
"redirect",
|
||||
"start over with a new plan",
|
||||
);
|
||||
|
||||
expect(result.content).toBe(expectedContent);
|
||||
expect(result.trackRunId).toBeUndefined();
|
||||
});
|
||||
|
||||
it("passes selected-agent scope when redirecting the selected global session", async () => {
|
||||
const request = vi.fn(async (method: string, _payload?: unknown) => {
|
||||
if (method === "sessions.steer") {
|
||||
|
|
|
|||
|
|
@ -745,6 +745,38 @@ function isActiveSteerSession(session: GatewaySessionRow | undefined): boolean {
|
|||
return session?.status === "running" && session.endedAt == null;
|
||||
}
|
||||
|
||||
type SteerChatSendAckStatus = "started" | "in_flight" | "ok" | "timeout" | "error";
|
||||
|
||||
function normalizeSteerChatSendAckStatus(payload: unknown): SteerChatSendAckStatus {
|
||||
if (!payload || typeof payload !== "object") {
|
||||
return "started";
|
||||
}
|
||||
const status = (payload as Record<string, unknown>).status;
|
||||
return status === "in_flight" || status === "ok" || status === "timeout" || status === "error"
|
||||
? status
|
||||
: "started";
|
||||
}
|
||||
|
||||
function formatTerminalSteerAckContent(status: SteerChatSendAckStatus): string | undefined {
|
||||
if (status === "timeout") {
|
||||
return "The active run ended before the steer message was accepted.";
|
||||
}
|
||||
if (status === "error") {
|
||||
return "Steer failed before it reached the run; try again.";
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function formatTerminalRedirectAckContent(status: SteerChatSendAckStatus): string | undefined {
|
||||
if (status === "timeout") {
|
||||
return "The active run ended before the redirect message was accepted.";
|
||||
}
|
||||
if (status === "error") {
|
||||
return "Redirect failed before it reached the run; try again.";
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** Soft inject — queues a message into the active run via chat.send (deliver: false). */
|
||||
async function executeSteer(
|
||||
client: GatewayBrowserClient,
|
||||
|
|
@ -771,17 +803,24 @@ async function executeSteer(
|
|||
content: "No active run. Use the chat input or `/redirect` instead.",
|
||||
};
|
||||
}
|
||||
await client.request("chat.send", {
|
||||
sessionKey: resolved.key,
|
||||
...selectedGlobalScope(resolved.key, context),
|
||||
message: resolved.message,
|
||||
deliver: false,
|
||||
idempotencyKey: generateUUID(),
|
||||
});
|
||||
return {
|
||||
content: "Steered.",
|
||||
pendingCurrentRun: resolved.key === sessionKey,
|
||||
};
|
||||
const ackStatus = normalizeSteerChatSendAckStatus(
|
||||
await client.request("chat.send", {
|
||||
sessionKey: resolved.key,
|
||||
...selectedGlobalScope(resolved.key, context),
|
||||
message: resolved.message,
|
||||
deliver: false,
|
||||
idempotencyKey: generateUUID(),
|
||||
}),
|
||||
);
|
||||
const terminalAckContent = formatTerminalSteerAckContent(ackStatus);
|
||||
if (terminalAckContent) {
|
||||
return { content: terminalAckContent };
|
||||
}
|
||||
const result: SlashCommandResult = { content: "Steered." };
|
||||
if (ackStatus === "started" || ackStatus === "in_flight") {
|
||||
result.pendingCurrentRun = resolved.key === sessionKey;
|
||||
}
|
||||
return result;
|
||||
} catch (err) {
|
||||
return { content: `Failed to steer: ${String(err)}` };
|
||||
}
|
||||
|
|
@ -801,15 +840,20 @@ async function executeRedirect(
|
|||
content: resolved.error === "empty" ? "Usage: `/redirect <message>`" : resolved.error,
|
||||
};
|
||||
}
|
||||
const resp = await client.request<{ runId?: string }>("sessions.steer", {
|
||||
const resp = await client.request<{ runId?: string; status?: unknown }>("sessions.steer", {
|
||||
key: resolved.key,
|
||||
...selectedGlobalScope(resolved.key, context),
|
||||
message: resolved.message,
|
||||
});
|
||||
const ackStatus = normalizeSteerChatSendAckStatus(resp);
|
||||
const terminalAckContent = formatTerminalRedirectAckContent(ackStatus);
|
||||
if (terminalAckContent) {
|
||||
return { content: terminalAckContent };
|
||||
}
|
||||
const runId = typeof resp?.runId === "string" ? resp.runId : undefined;
|
||||
return {
|
||||
content: "Redirected.",
|
||||
trackRunId: runId,
|
||||
...(ackStatus === "started" || ackStatus === "in_flight" ? { trackRunId: runId } : {}),
|
||||
};
|
||||
} catch (err) {
|
||||
return { content: `Failed to redirect: ${String(err)}` };
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ import {
|
|||
requestChatSend,
|
||||
requestSkillWorkshopRevisionChatSend,
|
||||
sendChatMessage,
|
||||
sendDetachedChatMessage,
|
||||
sendSteerChatMessage,
|
||||
type ChatEventPayload,
|
||||
type ChatState,
|
||||
} from "./chat.ts";
|
||||
|
|
@ -69,6 +71,15 @@ function requireFirstRequestCall(request: ReturnType<typeof vi.fn>): unknown[] {
|
|||
return call;
|
||||
}
|
||||
|
||||
function createStartedChatSendAck(params: unknown) {
|
||||
const requestParams = requireRecord(params);
|
||||
const runId = requestParams.idempotencyKey;
|
||||
if (typeof runId !== "string") {
|
||||
throw new Error("Expected chat.send idempotencyKey");
|
||||
}
|
||||
return { runId, status: "started" as const };
|
||||
}
|
||||
|
||||
function expectTextChatMessage(message: unknown, role: string, text: string): void {
|
||||
const record = requireRecord(message);
|
||||
expect(record.role).toBe(role);
|
||||
|
|
@ -1875,7 +1886,9 @@ describe("sendChatMessage", () => {
|
|||
sessionId: "session-before-reconnect",
|
||||
messages: [],
|
||||
})
|
||||
.mockResolvedValueOnce({ runId: "run-1", status: "started" });
|
||||
.mockImplementationOnce((_method: string, params?: unknown) =>
|
||||
createStartedChatSendAck(params),
|
||||
);
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
|
|
@ -2060,8 +2073,88 @@ describe("sendChatMessage", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("clears the local run and rejects acceptance when the send acks a terminal timeout", async () => {
|
||||
const request = vi.fn((_method: string, params: { idempotencyKey: string }) =>
|
||||
Promise.resolve({ runId: params.idempotencyKey, status: "timeout" }),
|
||||
);
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
});
|
||||
|
||||
const result = await sendChatMessage(state, "aborted before dispatch");
|
||||
|
||||
expect(result).toBeNull();
|
||||
expect(state.chatMessages).toStrictEqual([]);
|
||||
expect(state.lastError).toBe("The run ended before the message was accepted.");
|
||||
expect(state.chatRunId).toBeNull();
|
||||
expect(state.chatStream).toBeNull();
|
||||
expect(state.chatStreamStartedAt).toBeNull();
|
||||
const runState = state as ChatState & {
|
||||
chatRunStatus?: unknown;
|
||||
lastLocalTerminalReconcile?: unknown;
|
||||
};
|
||||
expect(runState.chatRunStatus).toMatchObject({
|
||||
phase: "interrupted",
|
||||
runId: expect.stringMatching(UUID_V4_RE),
|
||||
sessionKey: "main",
|
||||
});
|
||||
expect(runState.lastLocalTerminalReconcile).toMatchObject({
|
||||
phase: "interrupted",
|
||||
runId: expect.stringMatching(UUID_V4_RE),
|
||||
sessionKey: "main",
|
||||
sessionStatus: "killed",
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves terminal timeout acks from Skill Workshop revision sends", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-revision", status: "timeout" });
|
||||
const state = createState({
|
||||
sessionKey: "global",
|
||||
currentSessionId: "session-visible",
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
});
|
||||
|
||||
const result = await requestSkillWorkshopRevisionChatSend(state, {
|
||||
proposalId: "support-file-sampler-20260531-68207b7b7f",
|
||||
instructions: "Make the support files 5",
|
||||
runId: "run-revision",
|
||||
});
|
||||
|
||||
expect(result).toEqual({ runId: "run-revision", status: "timeout" });
|
||||
});
|
||||
|
||||
it("preserves terminal failure acks from generated-run sends", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-detached", status: "error" });
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
});
|
||||
|
||||
await expect(sendDetachedChatMessage(state, "/btw summarize this")).resolves.toEqual({
|
||||
runId: "run-detached",
|
||||
status: "error",
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves terminal ok acks from generated-run steer sends", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-steer-ok", status: "ok" });
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
});
|
||||
|
||||
await expect(sendSteerChatMessage(state, "tighten the plan")).resolves.toEqual({
|
||||
runId: "run-steer-ok",
|
||||
status: "ok",
|
||||
});
|
||||
});
|
||||
|
||||
it("serializes non-image chat attachments as files", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-1", status: "started" });
|
||||
const request = vi.fn((_method: string, params?: unknown) =>
|
||||
Promise.resolve(createStartedChatSendAck(params)),
|
||||
);
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
|
|
@ -2106,7 +2199,9 @@ describe("sendChatMessage", () => {
|
|||
});
|
||||
|
||||
it("serializes attachments from the side payload store without copying data URLs into chat state", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-1", status: "started" });
|
||||
const request = vi.fn((_method: string, params?: unknown) =>
|
||||
Promise.resolve(createStartedChatSendAck(params)),
|
||||
);
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
|
|
@ -2160,7 +2255,9 @@ describe("sendChatMessage", () => {
|
|||
});
|
||||
|
||||
it("sends inline image payloads without copying data URLs into optimistic chat state", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ runId: "run-1", status: "started" });
|
||||
const request = vi.fn((_method: string, params?: unknown) =>
|
||||
Promise.resolve(createStartedChatSendAck(params)),
|
||||
);
|
||||
const state = createState({
|
||||
connected: true,
|
||||
client: { request } as unknown as ChatState["client"],
|
||||
|
|
@ -2200,7 +2297,9 @@ describe("sendChatMessage", () => {
|
|||
]);
|
||||
expect(JSON.stringify(state.chatMessages)).not.toContain("data:image/png;base64");
|
||||
|
||||
const captionedRequest = vi.fn().mockResolvedValue({ runId: "run-2", status: "started" });
|
||||
const captionedRequest = vi.fn((_method: string, params?: unknown) =>
|
||||
Promise.resolve(createStartedChatSendAck(params)),
|
||||
);
|
||||
const captionedState = createState({
|
||||
connected: true,
|
||||
client: { request: captionedRequest } as unknown as ChatState["client"],
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
// Control UI controller manages chat gateway state.
|
||||
import type { CommandsListResult } from "../../../../packages/gateway-protocol/src/index.js";
|
||||
import { isNonTerminalAgentRunStatus } from "../../../../src/shared/agent-run-status.js";
|
||||
import { getChatAttachmentDataUrl } from "../chat/attachment-payload-store.ts";
|
||||
import {
|
||||
isAssistantHeartbeatAckForDisplay,
|
||||
|
|
@ -891,7 +892,7 @@ function buildApiAttachments(attachments?: ChatAttachment[]) {
|
|||
: undefined;
|
||||
}
|
||||
|
||||
export type ChatSendAckStatus = "started" | "in_flight" | "ok";
|
||||
export type ChatSendAckStatus = "started" | "in_flight" | "ok" | "timeout" | "error";
|
||||
|
||||
export type ChatSendAckServerTiming = {
|
||||
receivedToAckMs?: number;
|
||||
|
|
@ -936,7 +937,10 @@ function normalizeChatSendAck(payload: unknown, fallbackRunId: string): ChatSend
|
|||
const serverTiming = normalizeChatSendAckServerTiming(record.serverTiming);
|
||||
return {
|
||||
runId,
|
||||
status: status === "in_flight" || status === "ok" ? status : "started",
|
||||
status:
|
||||
status === "in_flight" || status === "ok" || status === "timeout" || status === "error"
|
||||
? status
|
||||
: "started",
|
||||
...(serverTiming ? { serverTiming } : {}),
|
||||
};
|
||||
}
|
||||
|
|
@ -1108,7 +1112,7 @@ export async function sendChatMessage(
|
|||
}
|
||||
|
||||
const now = Date.now();
|
||||
appendUserChatMessage(state, msg, attachments, now);
|
||||
const optimisticMessage = appendUserChatMessage(state, msg, attachments, now);
|
||||
|
||||
state.chatSending = true;
|
||||
setChatError(state, null);
|
||||
|
|
@ -1135,10 +1139,33 @@ export async function sendChatMessage(
|
|||
armLocalTerminalReconcile: true,
|
||||
},
|
||||
);
|
||||
} else {
|
||||
} else if (isNonTerminalAgentRunStatus(ack.status)) {
|
||||
state.chatRunId = ack.runId;
|
||||
} else {
|
||||
state.chatMessages = state.chatMessages.filter(
|
||||
(messageEntry) => messageEntry !== optimisticMessage,
|
||||
);
|
||||
reconcileChatRunLifecycle(
|
||||
state as unknown as Parameters<typeof reconcileChatRunLifecycle>[0],
|
||||
{
|
||||
outcome: "interrupted",
|
||||
sessionStatus: ack.status === "error" ? "failed" : "killed",
|
||||
runId: ack.runId,
|
||||
sessionKey: state.sessionKey,
|
||||
clearLocalRun: true,
|
||||
clearChatStream: true,
|
||||
armLocalTerminalReconcile: ack.runId === runId,
|
||||
},
|
||||
);
|
||||
setChatError(
|
||||
state,
|
||||
ack.status === "error"
|
||||
? "Chat failed before the run started; try again."
|
||||
: "The run ended before the message was accepted.",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
return ack.status === "ok" ? ack.runId : runId;
|
||||
return ack.runId;
|
||||
} catch (err) {
|
||||
const error = formatConnectError(err);
|
||||
reconcileChatRunLifecycle(state as unknown as Parameters<typeof reconcileChatRunLifecycle>[0], {
|
||||
|
|
@ -1170,21 +1197,20 @@ export function appendUserChatMessage(
|
|||
attachments?: ChatAttachment[],
|
||||
timestamp = Date.now(),
|
||||
) {
|
||||
state.chatMessages = [
|
||||
...state.chatMessages,
|
||||
{
|
||||
role: "user",
|
||||
content: buildUserChatMessageContentBlocks(message, attachments),
|
||||
timestamp,
|
||||
},
|
||||
];
|
||||
const entry = {
|
||||
role: "user" as const,
|
||||
content: buildUserChatMessageContentBlocks(message, attachments),
|
||||
timestamp,
|
||||
};
|
||||
state.chatMessages = [...state.chatMessages, entry];
|
||||
return entry;
|
||||
}
|
||||
|
||||
async function sendChatMessageWithGeneratedRunId(
|
||||
state: ChatState,
|
||||
message: string,
|
||||
attachments?: ChatAttachment[],
|
||||
): Promise<string | null> {
|
||||
): Promise<ChatSendAck | null> {
|
||||
if (!state.client || !state.connected) {
|
||||
return null;
|
||||
}
|
||||
|
|
@ -1196,8 +1222,7 @@ async function sendChatMessageWithGeneratedRunId(
|
|||
setChatError(state, null);
|
||||
const runId = generateUUID();
|
||||
try {
|
||||
const ack = await requestChatSend(state, { message: msg, attachments, runId });
|
||||
return ack.runId;
|
||||
return await requestChatSend(state, { message: msg, attachments, runId });
|
||||
} catch (err) {
|
||||
setChatError(state, formatConnectError(err));
|
||||
return null;
|
||||
|
|
@ -1208,7 +1233,7 @@ export async function sendDetachedChatMessage(
|
|||
state: ChatState,
|
||||
message: string,
|
||||
attachments?: ChatAttachment[],
|
||||
): Promise<string | null> {
|
||||
): Promise<ChatSendAck | null> {
|
||||
return sendChatMessageWithGeneratedRunId(state, message, attachments);
|
||||
}
|
||||
|
||||
|
|
@ -1216,7 +1241,7 @@ export async function sendSteerChatMessage(
|
|||
state: ChatState,
|
||||
message: string,
|
||||
attachments?: ChatAttachment[],
|
||||
): Promise<string | null> {
|
||||
): Promise<ChatSendAck | null> {
|
||||
return sendChatMessageWithGeneratedRunId(state, message, attachments);
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue