diff --git a/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt b/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt index f900b1b0b3e..53e2ec422dc 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/NodeRuntime.kt @@ -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 = diff --git a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt index f513232afef..08fc44000b2 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt @@ -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 = _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" diff --git a/apps/android/app/src/main/java/ai/openclaw/app/gateway/ChatSendAck.kt b/apps/android/app/src/main/java/ai/openclaw/app/gateway/ChatSendAck.kt new file mode 100644 index 00000000000..be0d6c9ad8a --- /dev/null +++ b/apps/android/app/src/main/java/ai/openclaw/app/gateway/ChatSendAck.kt @@ -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 diff --git a/apps/android/app/src/main/java/ai/openclaw/app/voice/MicCaptureManager.kt b/apps/android/app/src/main/java/ai/openclaw/app/voice/MicCaptureManager.kt index a7e4fcfa63e..c58706bc387 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/voice/MicCaptureManager.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/voice/MicCaptureManager.kt @@ -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() diff --git a/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt b/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt index 022965c1a1d..1acafc200a0 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/voice/TalkModeManager.kt @@ -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 diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt new file mode 100644 index 00000000000..216a7e57e4e --- /dev/null +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerTerminalAckTest.kt @@ -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() + 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.hasUserText(text: String): Boolean = + any { message -> + message.role == "user" && message.content.any { part -> part.text == text } + } +} diff --git a/apps/android/app/src/test/java/ai/openclaw/app/gateway/ChatSendAckTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/gateway/ChatSendAckTest.kt new file mode 100644 index 00000000000..519acfa1244 --- /dev/null +++ b/apps/android/app/src/test/java/ai/openclaw/app/gateway/ChatSendAckTest.kt @@ -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) + } +} diff --git a/apps/android/app/src/test/java/ai/openclaw/app/voice/MicCaptureManagerTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/voice/MicCaptureManagerTest.kt index d23ac4c4464..9031d6e4f0f 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/voice/MicCaptureManagerTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/voice/MicCaptureManagerTest.kt @@ -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(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(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(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( diff --git a/apps/ios/Sources/Voice/TalkModeManager.swift b/apps/ios/Sources/Voice/TalkModeManager.swift index a501723befc..2272192e605 100644 --- a/apps/ios/Sources/Voice/TalkModeManager.swift +++ b/apps/ios/Sources/Voice/TalkModeManager.swift @@ -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? - 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)) diff --git a/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift b/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift index 4edfd4b2759..fc7e552fc11 100644 --- a/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift +++ b/apps/macos/Sources/OpenClaw/TalkModeRuntime.swift @@ -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)) diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel.swift index d37f8616479..85e1cd6f5fd 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel.swift @@ -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, diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift index 91950ab8000..0c57aa7884f 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift @@ -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.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 diff --git a/packages/gateway-protocol/src/schema/agents-models-skills.test.ts b/packages/gateway-protocol/src/schema/agents-models-skills.test.ts index 9dd96354645..e783efb8281 100644 --- a/packages/gateway-protocol/src/schema/agents-models-skills.test.ts +++ b/packages/gateway-protocol/src/schema/agents-models-skills.test.ts @@ -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); + }); +}); diff --git a/packages/gateway-protocol/src/schema/agents-models-skills.ts b/packages/gateway-protocol/src/schema/agents-models-skills.ts index 8fd8548d0a9..8132bf6d575 100644 --- a/packages/gateway-protocol/src/schema/agents-models-skills.ts +++ b/packages/gateway-protocol/src/schema/agents-models-skills.ts @@ -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 }, ); diff --git a/src/acp/translator.lifecycle.test.ts b/src/acp/translator.lifecycle.test.ts index cc948bfacdc..71cbb5d708c 100644 --- a/src/acp/translator.lifecycle.test.ts +++ b/src/acp/translator.lifecycle.test.ts @@ -111,6 +111,15 @@ function createSessionRow(params: { }; } +async function settlePromptQuickly(promise: Promise): Promise { + 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) => { + 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) => { + 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) => { + 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) => { diff --git a/src/acp/translator.stop-reason.test.ts b/src/acp/translator.stop-reason.test.ts index 3a2e1e69810..1f845158ffe 100644 --- a/src/acp/translator.stop-reason.test.ts +++ b/src/acp/translator.stop-reason.test.ts @@ -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((_, 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((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 { diff --git a/src/acp/translator.ts b/src/acp/translator.ts index 35f47e6c106..1a8de32d1dd 100644 --- a/src/acp/translator.ts +++ b/src/acp/translator.ts @@ -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 | undefined; let acpSdkModulePromise: Promise | 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(); + private settlingPromptKeys = new Set(); private approvalRelays = new Map(); 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 => { + 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): Promise => { + const ack = await this.gateway.request("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 { - 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 { diff --git a/src/gateway/server-methods/talk.test.ts b/src/gateway/server-methods/talk.test.ts index c316acbb3f8..60730385204 100644 --- a/src/gateway/server-methods/talk.test.ts +++ b/src/gateway/server-methods/talk.test.ts @@ -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(); diff --git a/src/gateway/talk-agent-consult.ts b/src/gateway/talk-agent-consult.ts index 0ea56e5835f..70d092d2894 100644 --- a/src/gateway/talk-agent-consult.ts +++ b/src/gateway/talk-agent-consult.ts @@ -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).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).runId === "string" diff --git a/src/tui/embedded-backend.ts b/src/tui/embedded-backend.ts index fa11145a445..b8d2854c343 100644 --- a/src/tui/embedded-backend.ts +++ b/src/tui/embedded-backend.ts @@ -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 { const runId = opts.runId ?? randomUUID(); const question = resolveBtwQuestion(opts.message); const runScope = { diff --git a/src/tui/gateway-chat.test.ts b/src/tui/gateway-chat.test.ts index 4eb4ebf6c07..2c16ee968a2 100644 --- a/src/tui/gateway-chat.test.ts +++ b/src/tui/gateway-chat.test.ts @@ -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", diff --git a/src/tui/gateway-chat.ts b/src/tui/gateway-chat.ts index 14ebf2f63e1..ad22981c9ea 100644 --- a/src/tui/gateway-chat.ts +++ b/src/tui/gateway-chat.ts @@ -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 { }); } +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 { 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 }) { diff --git a/src/tui/tui-backend.ts b/src/tui/tui-backend.ts index c1d5201107a..4b427c0eca2 100644 --- a/src/tui/tui-backend.ts +++ b/src/tui/tui-backend.ts @@ -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; subscribeSessionEvents?: () => Promise; - sendChat: (opts: ChatSendOptions) => Promise<{ runId: string }>; + sendChat: (opts: ChatSendOptions) => Promise; abortChat: (opts: { sessionKey: string; agentId?: string; diff --git a/src/tui/tui-command-handlers.test.ts b/src/tui/tui-command-handlers.test.ts index 203cb3f66c2..9e7a1d38ba7 100644 --- a/src/tui/tui-command-handlers.test.ts +++ b/src/tui/tui-command-handlers.test.ts @@ -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"); diff --git a/src/tui/tui-command-handlers.ts b/src/tui/tui-command-handlers.ts index 00d4eedc2c0..3245171957d 100644 --- a/src/tui/tui-command-handlers.ts +++ b/src/tui/tui-command-handlers.ts @@ -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) { diff --git a/ui/src/ui/app-chat.test.ts b/ui/src/ui/app-chat.test.ts index 0b6b44c4a85..427c4b7a6f7 100644 --- a/ui/src/ui/app-chat.test.ts +++ b/ui/src/ui/app-chat.test.ts @@ -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: [ diff --git a/ui/src/ui/app-chat.ts b/ui/src/ui/app-chat.ts index 370b6b205e3..ad51ef18779 100644 --- a/ui/src/ui/app-chat.ts +++ b/ui/src/ui/app-chat.ts @@ -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[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[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[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[0], diff --git a/ui/src/ui/chat/slash-command-executor.node.test.ts b/ui/src/ui/chat/slash-command-executor.node.test.ts index 4aa523a910d..685928744cb 100644 --- a/ui/src/ui/chat/slash-command-executor.node.test.ts +++ b/ui/src/ui/chat/slash-command-executor.node.test.ts @@ -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") { diff --git a/ui/src/ui/chat/slash-command-executor.ts b/ui/src/ui/chat/slash-command-executor.ts index e10b0d28126..39b0b0ea1a2 100644 --- a/ui/src/ui/chat/slash-command-executor.ts +++ b/ui/src/ui/chat/slash-command-executor.ts @@ -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).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 `" : 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)}` }; diff --git a/ui/src/ui/controllers/chat.test.ts b/ui/src/ui/controllers/chat.test.ts index a2ddad71731..04a4a48629d 100644 --- a/ui/src/ui/controllers/chat.test.ts +++ b/ui/src/ui/controllers/chat.test.ts @@ -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): 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"], diff --git a/ui/src/ui/controllers/chat.ts b/ui/src/ui/controllers/chat.ts index 103d0a9d30c..0ab28494679 100644 --- a/ui/src/ui/controllers/chat.ts +++ b/ui/src/ui/controllers/chat.ts @@ -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[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[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 { +): Promise { 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 { +): Promise { return sendChatMessageWithGeneratedRunId(state, message, attachments); } @@ -1216,7 +1241,7 @@ export async function sendSteerChatMessage( state: ChatState, message: string, attachments?: ChatAttachment[], -): Promise { +): Promise { return sendChatMessageWithGeneratedRunId(state, message, attachments); }