diff --git a/apps/.i18n/native-source.json b/apps/.i18n/native-source.json index ff8e6b16da8d..7c2c0084f98f 100644 --- a/apps/.i18n/native-source.json +++ b/apps/.i18n/native-source.json @@ -1723,7 +1723,7 @@ }, { "kind": "ui-call", - "line": 1016, + "line": 1017, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Wait for the current response to finish before starting a new chat.", "surface": "android", @@ -1731,7 +1731,7 @@ }, { "kind": "ui-call", - "line": 1148, + "line": 1149, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not update model.", "surface": "android", @@ -1739,7 +1739,7 @@ }, { "kind": "ui-call", - "line": 1213, + "line": 1214, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not update thinking level.", "surface": "android", @@ -1747,7 +1747,7 @@ }, { "kind": "ui-call", - "line": 1706, + "line": 1707, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Chat failed before the run started; try again.", "surface": "android", @@ -1755,7 +1755,7 @@ }, { "kind": "ui-call", - "line": 2738, + "line": 2743, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not stage an attachment for sending.", "surface": "android", @@ -1763,7 +1763,7 @@ }, { "kind": "ui-call", - "line": 2771, + "line": 2776, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Offline queue is full ($OUTBOX_MAX_QUEUED messages); delete queued items first.", "surface": "android", @@ -1771,7 +1771,7 @@ }, { "kind": "ui-call", - "line": 2777, + "line": 2782, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Attachments are too large to queue for one message; remove some and try again.", "surface": "android", @@ -1779,7 +1779,7 @@ }, { "kind": "ui-call", - "line": 2783, + "line": 2788, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Offline attachment storage is full; delete queued items first.", "surface": "android", @@ -1787,7 +1787,7 @@ }, { "kind": "ui-call", - "line": 2788, + "line": 2793, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Gateway health not OK; cannot send", "surface": "android", @@ -1795,7 +1795,7 @@ }, { "kind": "ui-call", - "line": 2795, + "line": 2800, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Could not queue message for later delivery.", "surface": "android", @@ -1803,7 +1803,7 @@ }, { "kind": "ui-call", - "line": 3512, + "line": 3517, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Chat failed", "surface": "android", @@ -1811,7 +1811,7 @@ }, { "kind": "ui-call", - "line": 3650, + "line": 3655, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Event stream interrupted; try refreshing.", "surface": "android", @@ -1819,7 +1819,7 @@ }, { "kind": "ui-call", - "line": 3762, + "line": 3792, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Timed out waiting for a reply; try again or refresh.", "surface": "android", @@ -1827,7 +1827,7 @@ }, { "kind": "ui-call", - "line": 3972, + "line": 4002, "path": "apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt", "source": "Timed out confirming the sent message; refresh to check delivery.", "surface": "android", 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 4f805243fd39..62922c2668e1 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 @@ -428,7 +428,8 @@ class ChatController internal constructor( pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null - clearPlanSteps() + // Older gateways cannot restate plan state, so reconnect retains it until + // recovery proves another run, a terminal state, or an explicit empty snapshot. _historyLoading.value = false _sessionId.value = null // Failed connect attempts pass through onGatewayScopeChanging, which empties the published @@ -534,8 +535,8 @@ class ChatController internal constructor( pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null - clearPlanSteps() } + clearPlanSteps() appliedMainSessionKey = "main" beginHistoryLoad( key = "main", @@ -2042,7 +2043,6 @@ class ChatController internal constructor( pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null - clearPlanSteps() refreshHistoryForRecovery() } "chat" -> { @@ -2174,6 +2174,7 @@ class ChatController internal constructor( runIdsToReconcile: Set = emptySet(), ): HistoryRefreshResult { val requestSequence = historyRequestSequence.incrementAndGet() + val runIdsOwnedAtRequest = synchronized(pendingRuns) { pendingRuns.toSet() } val requestModelSelectionGeneration = modelSelectionGeneration.get() val requestCacheScope = currentCacheScope() val requestTracksDefaultAgent = activeSessionTracksDefaultAgent(sessionKey) @@ -2229,6 +2230,10 @@ class ChatController internal constructor( ) { return@synchronized false } + val runIdsOwnedAfterRequest = + synchronized(pendingRuns) { + pendingRuns.filterNotTo(mutableSetOf()) { it in runIdsOwnedAtRequest } + } latestAppliedHistoryRequest = requestSequence if (updateSessionInfo) { updateSessionFromHistory(history) @@ -2277,11 +2282,11 @@ class ChatController internal constructor( unknownOutcomeRunIds.contains(runId) && unresolvedRepliesByRunId.containsKey(runId) }.forEach(::clearPendingRun) } - clearTransientRunUiIfIdle() + clearTransientRunUiIfIdle(preservePlan = true) // All live history paths (bootstrap, reconnect recovery, cache-first // replace) adopt the gateway's in-flight run snapshot so restored // runs keep their pending state and streaming text. - adoptInFlightRun(history.inFlightRun) + adoptInFlightRun(history, runIdsOwnedAfterRequest) history.thinkingLevel ?.trim() ?.takeIf { it.isNotEmpty() } @@ -3690,14 +3695,30 @@ class ChatController internal constructor( /** * Adopts the run the gateway reports still streaming for this session so reconnect, - * cold start, and seq-gap recovery restore pending/streaming UI state. Snapshot absence - * never clears local state: live terminal events and the pending-run timeout own - * completion, and a snapshot fetched before our own send must not cancel that run. + * cold start, and seq-gap recovery restore pending/streaming UI state. */ - private fun adoptInFlightRun(run: ChatInFlightRun?) { - if (run == null) return - val runId = run.runId.trim() - if (runId.isEmpty()) return + private fun adoptInFlightRun( + history: ChatHistory, + runIdsOwnedAfterRequest: Set, + ) { + // Plan reconciliation shares run adoption: rejected history cannot clobber newer live state. + // A missing plan is version-skew unknown; replacement or explicit terminal evidence clears it. + // Snapshots predating a locally owned run are rejected unless they name that newer run. + val run = history.inFlightRun + val runId = run?.runId?.trim()?.takeIf { it.isNotEmpty() } + if (runIdsOwnedAfterRequest.isNotEmpty() && (runId == null || runId !in runIdsOwnedAfterRequest)) return + if (run == null) { + val retainedRunId = planRunId ?: return + val activeRunIds = history.sessionInfo?.activeRunIds + if ( + history.sessionInfo?.hasActiveRun == false || + (activeRunIds != null && retainedRunId !in activeRunIds) + ) { + clearPlanSteps() + } + return + } + if (runId == null) return synchronized(pendingRuns) { // A different locally-owned run means this snapshot predates it; ignore. if (pendingRuns.isNotEmpty() && runId !in pendingRuns) return @@ -3709,6 +3730,15 @@ class ChatController internal constructor( if (run.text.isNotEmpty()) { _streamingAssistantText.value = run.text } + val plan = run.plan + if (plan == null) { + if (planRunId != null && planRunId != runId) clearPlanSteps() + } else if (plan.steps.isEmpty()) { + clearPlanSteps() + } else { + planRunId = runId + _planSteps.value = plan.steps + } } private fun armPendingRunTimeout(runId: String) { @@ -3795,12 +3825,12 @@ class ChatController internal constructor( } } - private fun clearTransientRunUiIfIdle() { + private fun clearTransientRunUiIfIdle(preservePlan: Boolean = false) { if (synchronized(pendingRuns) { pendingRuns.isNotEmpty() }) return pendingToolCallsById.clear() publishPendingToolCalls() _streamingAssistantText.value = null - clearPlanSteps() + if (!preservePlan) clearPlanSteps() } private fun clearPendingRuns( @@ -4059,7 +4089,18 @@ class ChatController internal constructor( private fun parseInFlightRun(root: JsonObject): ChatInFlightRun? { val obj = root["inFlightRun"].asObjectOrNull() ?: return null val runId = obj["runId"].asStringOrNull()?.trim()?.takeIf { it.isNotEmpty() } ?: return null - return ChatInFlightRun(runId = runId, text = obj["text"].asStringOrNull().orEmpty()) + val plan = + obj["plan"].asObjectOrNull()?.let { plan -> + ChatPlanSnapshot( + steps = parseChatPlanSteps(plan["steps"]), + explanation = plan["explanation"].asStringOrNull()?.trim()?.takeIf { it.isNotEmpty() }, + ) + } + return ChatInFlightRun( + runId = runId, + text = obj["text"].asStringOrNull().orEmpty(), + plan = plan, + ) } private data class SessionListResult( @@ -4131,6 +4172,11 @@ class ChatController internal constructor( "totalTokens" in obj || "totalTokensFresh" in obj || "contextTokens" in obj, + hasActiveRun = obj["hasActiveRun"].asBooleanOrNull(), + activeRunIds = + obj["activeRunIds"] + .asArrayOrNull() + ?.mapNotNull { it.asStringOrNull()?.trim()?.takeIf(String::isNotEmpty) }, ) } diff --git a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatModels.kt b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatModels.kt index 32b59fb9e5a6..849575f01bfc 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatModels.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/chat/ChatModels.kt @@ -163,6 +163,8 @@ data class ChatSessionEntry( val thinkingDefault: String? = null, val contextTokens: Long? = null, val hasContextUsageMetadata: Boolean = totalTokens != null || totalTokensFresh != null || contextTokens != null, + val hasActiveRun: Boolean? = null, + val activeRunIds: List? = null, ) /** Local fallback for server-side `sessions.list` search over cached entries. */ @@ -196,6 +198,12 @@ data class ChatCommandEntry( data class ChatInFlightRun( val runId: String, val text: String, + val plan: ChatPlanSnapshot? = null, +) + +data class ChatPlanSnapshot( + val steps: List, + val explanation: String? = null, ) /** diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt index b1215eedbff0..e036db31560b 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt @@ -409,6 +409,249 @@ class ChatControllerReconnectRestoreTest { assertEquals(2, controller.messages.value.size) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun reconnectRestoresInFlightPlanSnapshot() = + runTest { + val gateway = ScriptedGateway(json) + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + val controller = newController(gateway) + controller.load("main") + runCurrent() + + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-active" to "working", + inFlightPlan = + ChatPlanSnapshot( + steps = + listOf( + ChatPlanStep("Inspect", ChatPlanStepStatus.Completed), + ChatPlanStep("Reconnect", ChatPlanStepStatus.InProgress), + ), + explanation = "Restore checklist", + ), + ), + ) + controller.onDisconnected("Reconnecting…") + controller.onGatewayConnected() + runCurrent() + + assertEquals( + listOf( + ChatPlanStep("Inspect", ChatPlanStepStatus.Completed), + ChatPlanStep("Reconnect", ChatPlanStepStatus.InProgress), + ), + controller.planSteps.value, + ) + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun historyPlanReconciliationContract() = + runTest { + val retainedSteps = listOf(ChatPlanStep("Retained", ChatPlanStepStatus.InProgress)) + + data class Case( + val name: String, + val history: String, + val expectedSteps: List, + val staleAfterLivePlan: Boolean = false, + val snapshotForNewLiveRun: ChatPlanSnapshot? = null, + val gatewayScopeChange: Boolean = false, + ) + + val cases = + listOf( + Case( + name = "replace", + history = + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-retained" to "working", + inFlightPlan = + ChatPlanSnapshot( + steps = listOf(ChatPlanStep("Replacement", ChatPlanStepStatus.Completed)), + ), + ), + expectedSteps = listOf(ChatPlanStep("Replacement", ChatPlanStepStatus.Completed)), + ), + Case( + name = "legacy-preserve", + history = + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-retained" to "working", + ), + expectedSteps = retainedSteps, + ), + Case( + name = "superseded", + history = + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-next" to "next", + inFlightPlan = + ChatPlanSnapshot( + steps = listOf(ChatPlanStep("Next run", ChatPlanStepStatus.InProgress)), + ), + ), + expectedSteps = listOf(ChatPlanStep("Next run", ChatPlanStepStatus.InProgress)), + ), + Case( + name = "active-preserve", + history = + historyResponse( + "session-1", + emptyList(), + hasActiveRun = true, + activeRunIds = listOf("run-retained"), + ), + expectedSteps = retainedSteps, + ), + Case( + name = "terminal-clear", + history = + historyResponse( + "session-1", + emptyList(), + hasActiveRun = false, + activeRunIds = emptyList(), + ), + expectedSteps = emptyList(), + ), + Case( + name = "no-evidence-preserve", + history = + historyResponse( + "session-1", + emptyList(), + hasActiveRun = null, + activeRunIds = null, + ), + expectedSteps = retainedSteps, + ), + Case( + name = "stale-response-does-not-clobber-newer-live-plan", + history = + historyResponse( + "session-1", + emptyList(), + hasActiveRun = false, + activeRunIds = emptyList(), + ), + expectedSteps = listOf(ChatPlanStep("New live plan", ChatPlanStepStatus.InProgress)), + staleAfterLivePlan = true, + ), + Case( + name = "stale-previous-run-snapshot-does-not-clobber-newer-live-plan", + history = + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-previous" to "stale", + inFlightPlan = ChatPlanSnapshot(steps = emptyList()), + ), + expectedSteps = listOf(ChatPlanStep("New live plan", ChatPlanStepStatus.InProgress)), + staleAfterLivePlan = true, + ), + Case( + name = "snapshot-for-newer-owned-run-is-accepted", + history = historyResponse("session-1", emptyList()), + expectedSteps = listOf(ChatPlanStep("Matching snapshot", ChatPlanStepStatus.Completed)), + staleAfterLivePlan = true, + snapshotForNewLiveRun = + ChatPlanSnapshot( + steps = listOf(ChatPlanStep("Matching snapshot", ChatPlanStepStatus.Completed)), + ), + ), + Case( + name = "explicit-empty-clears", + history = + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-retained" to "working", + inFlightPlan = ChatPlanSnapshot(steps = emptyList()), + ), + expectedSteps = emptyList(), + ), + Case( + name = "gateway-scope-change-clears", + history = historyResponse("session-1", emptyList()), + expectedSteps = emptyList(), + gatewayScopeChange = true, + ), + ) + + for (testCase in cases) { + val gateway = ScriptedGateway(json) + if (testCase.staleAfterLivePlan) { + gateway.respondWith("chat.history", historyResponse("session-1", emptyList())) + } else { + gateway.respondWith( + "chat.history", + historyResponse( + "session-1", + emptyList(), + inFlightRun = "run-retained" to "working", + inFlightPlan = ChatPlanSnapshot(steps = retainedSteps), + ), + ) + } + val controller = newController(gateway) + controller.load("main") + runCurrent() + + if (testCase.staleAfterLivePlan) { + val historyStarted = CompletableDeferred() + val releaseHistory = CompletableDeferred() + gateway.respond("chat.history") { + historyStarted.complete(Unit) + releaseHistory.await() + } + gateway.respondChatSend(status = "started") + controller.refresh() + runCurrent() + historyStarted.await() + assertTrue(controller.sendMessageAwaitAcceptance("new work", "off", emptyList())) + val runId = requireNotNull(gateway.lastRunId) + controller.handleGatewayEvent( + "agent", + """{"sessionKey":"main","runId":"$runId","seq":1,"ts":10,"stream":"plan","data":{"phase":"update","steps":[{"step":"New live plan","status":"in_progress"}]}}""", + ) + releaseHistory.complete( + testCase.snapshotForNewLiveRun?.let { plan -> + historyResponse( + "session-1", + emptyList(), + inFlightRun = runId to "matching", + inFlightPlan = plan, + ) + } ?: testCase.history, + ) + runCurrent() + assertEquals(testCase.name, 1, controller.pendingRunCount.value) + } else if (testCase.gatewayScopeChange) { + controller.onGatewayScopeChanging() + runCurrent() + } else { + gateway.respondWith("chat.history", testCase.history) + controller.onDisconnected("Reconnecting…") + controller.onGatewayConnected() + runCurrent() + } + + assertEquals(testCase.name, testCase.expectedSteps, controller.planSteps.value) + } + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun reconnectWithoutInFlightRunStaysClean() = diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt index d07fa4bbddc3..aa11df4d8004 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatReplayHarness.kt @@ -104,6 +104,9 @@ internal fun historyResponse( sessionId: String, messages: List, inFlightRun: Pair? = null, + inFlightPlan: ChatPlanSnapshot? = null, + hasActiveRun: Boolean? = inFlightRun?.let { true }, + activeRunIds: List? = inFlightRun?.let { listOf(it.first) }, ): String = buildJsonObject { put("sessionId", JsonPrimitive(sessionId)) @@ -113,9 +116,47 @@ internal fun historyResponse( buildJsonObject { put("runId", JsonPrimitive(inFlightRun.first)) put("text", JsonPrimitive(inFlightRun.second)) + if (inFlightPlan != null) { + put( + "plan", + buildJsonObject { + put( + "steps", + JsonArray( + inFlightPlan.steps.map { step -> + buildJsonObject { + put("step", JsonPrimitive(step.step)) + put( + "status", + JsonPrimitive( + when (step.status) { + ChatPlanStepStatus.Pending -> "pending" + ChatPlanStepStatus.InProgress -> "in_progress" + ChatPlanStepStatus.Completed -> "completed" + }, + ), + ) + } + }, + ), + ) + inFlightPlan.explanation?.let { put("explanation", JsonPrimitive(it)) } + }, + ) + } + }, + ) + } + if (hasActiveRun != null || activeRunIds != null) { + put( + "sessionInfo", + buildJsonObject { + hasActiveRun?.let { put("hasActiveRun", JsonPrimitive(it)) } + activeRunIds?.let { ids -> + put("activeRunIds", JsonArray(ids.map(::JsonPrimitive))) + } }, ) - put("sessionInfo", buildJsonObject { put("hasActiveRun", JsonPrimitive(true)) }) } put( "messages", diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift index b162d6f64e71..08675b3f438c 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift @@ -406,20 +406,35 @@ public struct OpenClawChatMessage: Codable, Hashable, Identifiable, Sendable { public struct OpenClawChatInFlightRun: Codable, Sendable { public let runId: String public let text: String + public let plan: OpenClawChatPlanSnapshot? // periphery:ignore - package tests construct history fixtures; app consumers decode this payload. - public init(runId: String, text: String) { + public init(runId: String, text: String, plan: OpenClawChatPlanSnapshot? = nil) { self.runId = runId self.text = text + self.plan = plan + } +} + +public struct OpenClawChatPlanSnapshot: Codable, Sendable { + public let steps: [OpenClawChatPlanStep] + public let explanation: String? + + // periphery:ignore - package tests construct history fixtures; app consumers decode this payload. + public init(steps: [OpenClawChatPlanStep], explanation: String? = nil) { + self.steps = steps + self.explanation = explanation } } public struct OpenClawChatSessionInfo: Codable, Sendable { public let hasActiveRun: Bool? + public let activeRunIds: [String]? // periphery:ignore - package tests construct history fixtures; app consumers decode this payload. - public init(hasActiveRun: Bool?) { + public init(hasActiveRun: Bool?, activeRunIds: [String]? = nil) { self.hasActiveRun = hasActiveRun + self.activeRunIds = activeRunIds } } @@ -541,8 +556,8 @@ public struct OpenClawAgentEventPayload: Codable, Sendable, Identifiable { public let data: [String: AnyCodable] } -public struct OpenClawChatPlanStep: Hashable, Sendable { - public enum Status: String, Hashable, Sendable { +public struct OpenClawChatPlanStep: Codable, Hashable, Sendable { + public enum Status: String, Codable, Hashable, Sendable { case pending case inProgress = "in_progress" case completed diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+HistoryReconciliation.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+HistoryReconciliation.swift index 4964d744491e..978d96c4ab1b 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+HistoryReconciliation.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+HistoryReconciliation.swift @@ -755,7 +755,7 @@ extension OpenClawChatViewModel { self.pruneRunMessageScopes() self.rescopeRunsAdoptedAfterHistoryRequest(request) self.sessionId = payload.sessionId - self.applyInFlightRunSnapshot(payload.inFlightRun, for: request) + self.applyInFlightRunSnapshot(payload, for: request) // Incomplete refreshes can arrive before durable assistant history. // The latest visible user turn must survive answered before it can reject older replies. let canInvalidateOlderHistory = if let latestUserTurn = request.latestUserTurn { @@ -826,51 +826,4 @@ extension OpenClawChatViewModel { } } } - - private func applyInFlightRunSnapshot( - _ snapshot: OpenClawChatInFlightRun?, - for request: HistoryRequest) - { - guard request.runOwnershipGeneration == self.runOwnershipGeneration, - request.id >= self.latestAppliedRunSnapshotRequestID - else { - return - } - self.latestAppliedRunSnapshotRequestID = request.id - guard let snapshot, - let runId = Self.normalizedRunID(snapshot.runId) - else { - return - } - - self.isApplyingRunSnapshot = true - defer { self.isApplyingRunSnapshot = false } - self.updateActiveSessionRunWithoutChatSnapshot(false) - self.adoptRun(runId: runId, bufferedText: snapshot.text) - } - - func adoptRun(runId: String, bufferedText: String) { - let canonicalPendingRuns = Set([runId]) - if self.pendingRuns != canonicalPendingRuns { - // Gateway snapshots and live deltas are canonical for this session. - // Replace stale local ownership so only that run consumes later events. - clearPendingRuns(reason: nil) - self.pendingRuns.insert(runId) - self.pendingToolCallsById = [:] - self.updateStreamingAssistantText(nil) - clearPlan() - } - if self.runMessageScopesByRunID[runId] == nil { - self.runMessageScopesByRunID[runId] = self.currentRunMessageScope() - } - if self.pendingRunOwnerArmIDs[runId] == nil { - armPendingRunOwner(runId: runId) - } - if !bufferedText.isEmpty { - self.updateStreamingAssistantText(bufferedText) - } - self.logDiagnostic( - "chat.ui adopted in-flight run sessionKey=\(self.sessionKey) " - + "runId=\(runId) bufferedTextLen=\(bufferedText.count)") - } } diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+Plan.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+Plan.swift index 34713ede16f6..d96a7ea1e222 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+Plan.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+Plan.swift @@ -5,13 +5,21 @@ import OpenClawKit extension OpenClawChatViewModel { func applyPlanSnapshot(runId: String, data: [String: AnyCodable]) { let steps = OpenClawChatPlanStep.parseSteps(data["steps"]) + let explanation = data["explanation"]?.value as? String + self.applyPlanSnapshot(runId: runId, steps: steps, explanation: explanation) + } + + func applyPlanSnapshot( + runId: String, + steps: [OpenClawChatPlanStep], + explanation: String?) + { guard !steps.isEmpty else { self.clearPlan(for: runId) return } - let explanation = (data["explanation"]?.value as? String)? - .trimmingCharacters(in: .whitespacesAndNewlines) - let normalizedExplanation = explanation?.isEmpty == false ? explanation : nil + let trimmedExplanation = explanation?.trimmingCharacters(in: .whitespacesAndNewlines) + let normalizedExplanation = trimmedExplanation?.isEmpty == false ? trimmedExplanation : nil guard planRunId != runId || planSteps != steps || planExplanation != normalizedExplanation diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift new file mode 100644 index 000000000000..502971d74603 --- /dev/null +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift @@ -0,0 +1,72 @@ +import OpenClawKit + +/// In-flight run adoption shared by history replay and live transport events. +extension OpenClawChatViewModel { + func applyInFlightRunSnapshot( + _ payload: OpenClawChatHistoryPayload, + for request: HistoryRequest) + { + guard request.runOwnershipGeneration == self.runOwnershipGeneration, + request.id >= self.latestAppliedRunSnapshotRequestID + else { + return + } + self.latestAppliedRunSnapshotRequestID = request.id + // Plan reconciliation shares run adoption: rejected history cannot clobber newer live state. + // A missing plan is version-skew unknown; replacement or explicit terminal evidence clears it. + guard let snapshot = payload.inFlightRun, + let runId = Self.normalizedRunID(snapshot.runId) + else { + guard let retainedRunId = self.planRunId else { return } + let activeRunIds = payload.sessionInfo?.activeRunIds + let confirmsAnotherRun = activeRunIds.map { !$0.contains(retainedRunId) } == true + if payload.sessionInfo?.hasActiveRun == false || confirmsAnotherRun { + self.clearPlan(for: retainedRunId) + } + return + } + + self.isApplyingRunSnapshot = true + defer { self.isApplyingRunSnapshot = false } + self.updateActiveSessionRunWithoutChatSnapshot(false) + self.adoptRunState(runId: runId, bufferedText: snapshot.text, preservePlan: true) + if self.planRunId != nil, self.planRunId != runId { + self.clearPlan() + } + if let planSnapshot = snapshot.plan { + self.applyPlanSnapshot( + runId: runId, + steps: planSnapshot.steps, + explanation: planSnapshot.explanation) + } + } + + func adoptRun(runId: String, bufferedText: String) { + self.adoptRunState(runId: runId, bufferedText: bufferedText, preservePlan: false) + } + + private func adoptRunState(runId: String, bufferedText: String, preservePlan: Bool) { + let canonicalPendingRuns = Set([runId]) + let replacedRun = self.pendingRuns != canonicalPendingRuns + if replacedRun { + // Gateway snapshots and live deltas are canonical for this session. + // Replace stale local ownership so only that run consumes later events. + clearPendingRuns(reason: nil, preservePlan: preservePlan) + self.pendingRuns.insert(runId) + self.pendingToolCallsById = [:] + self.updateStreamingAssistantText(nil) + } + if self.runMessageScopesByRunID[runId] == nil { + self.runMessageScopesByRunID[runId] = currentRunMessageScope() + } + if self.pendingRunOwnerArmIDs[runId] == nil { + armPendingRunOwner(runId: runId) + } + if !bufferedText.isEmpty { + self.updateStreamingAssistantText(bufferedText) + } + self.logDiagnostic( + "chat.ui adopted in-flight run sessionKey=\(self.sessionKey) " + + "runId=\(runId) bufferedTextLen=\(bufferedText.count)") + } +} diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift index 38ba6b165763..e80f15f89b1f 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift @@ -448,7 +448,6 @@ extension OpenClawChatViewModel { // but it is enough to retain the run ID this client already owns. self.pendingToolCallsById = [:] self.updateStreamingAssistantText(nil) - self.clearPlan(for: runId) return true } if let timestamp, @@ -972,7 +971,8 @@ extension OpenClawChatViewModel { func clearPendingRuns( reason: String?, - hapticEvent: OpenClawChatHaptics.Event? = nil) + hapticEvent: OpenClawChatHaptics.Event? = nil, + preservePlan: Bool = false) { let runIds = Array(pendingRuns) for runId in self.pendingRuns { @@ -981,7 +981,9 @@ extension OpenClawChatViewModel { self.pendingRunOwnerTasks.removeAll() self.pendingRunOwnerArmIDs.removeAll() self.pendingRuns.removeAll() - self.clearPlan() + if !preservePlan { + self.clearPlan() + } self.pendingLocalUserEchoMessageIDsByRunID.removeAll() if !runIds.isEmpty, let hapticEvent { self.haptics.perform(hapticEvent) diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift index 24ace216dc80..9f1c0a585387 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift @@ -70,6 +70,7 @@ private func historyPayload( messages: [AnyCodable] = [], supportsActiveRunState: Bool = true, hasActiveRun: Bool? = nil, + activeRunIds: [String]? = nil, inFlightRun: OpenClawChatInFlightRun? = nil) -> OpenClawChatHistoryPayload { OpenClawChatHistoryPayload( @@ -78,7 +79,9 @@ private func historyPayload( messages: messages, thinkingLevel: "off", sessionInfo: supportsActiveRunState - ? OpenClawChatSessionInfo(hasActiveRun: hasActiveRun ?? (inFlightRun != nil)) + ? OpenClawChatSessionInfo( + hasActiveRun: hasActiveRun ?? (inFlightRun != nil), + activeRunIds: activeRunIds ?? inFlightRun.map { [$0.runId] }) : nil, inFlightRun: inFlightRun) } @@ -1206,13 +1209,17 @@ struct ChatViewModelTests { } @Test func `decodes in-flight run from chat history`() throws { - let data = #"{"sessionKey":"main","messages":[],"inFlightRun":{"runId":"run-active","text":"partial"}}"# + let data = #"{"sessionKey":"main","messages":[],"inFlightRun":{"runId":"run-active","text":"partial","plan":{"steps":[{"step":"Reconnect","status":"in_progress"}],"explanation":"Current work"}}}"# .data(using: .utf8)! let payload = try JSONDecoder().decode(OpenClawChatHistoryPayload.self, from: data) #expect(payload.inFlightRun?.runId == "run-active") #expect(payload.inFlightRun?.text == "partial") + #expect(payload.inFlightRun?.plan?.steps == [ + OpenClawChatPlanStep(step: "Reconnect", status: .inProgress), + ]) + #expect(payload.inFlightRun?.plan?.explanation == "Current work") } @Test func `decodes agent scope from chat event`() throws { @@ -1267,6 +1274,128 @@ struct ChatViewModelTests { #expect(await MainActor.run { !vm.canSend }) } + @Test func `bootstrap adopts in-flight plan snapshot`() async throws { + let history = historyPayload( + inFlightRun: OpenClawChatInFlightRun( + runId: "run-plan", + text: "working", + plan: OpenClawChatPlanSnapshot( + steps: [ + OpenClawChatPlanStep(step: "Inspect", status: .completed), + OpenClawChatPlanStep(step: "Reconnect", status: .inProgress), + ], + explanation: "Restore checklist"))) + let (_, vm) = await makeViewModel(historyResponses: [history]) + + try await loadAndWaitBootstrap(vm: vm) + + #expect(await MainActor.run { vm.planRunId } == "run-plan") + #expect(await MainActor.run { vm.planSteps } == [ + OpenClawChatPlanStep(step: "Inspect", status: .completed), + OpenClawChatPlanStep(step: "Reconnect", status: .inProgress), + ]) + #expect(await MainActor.run { vm.planExplanation } == "Restore checklist") + } + + @Test func `history plan reconciliation contract`() async { + let retainedSteps = [OpenClawChatPlanStep(step: "Retained", status: .inProgress)] + let liveSteps = [OpenClawChatPlanStep(step: "New live plan", status: .inProgress)] + let cases: [( + name: String, + payload: OpenClawChatHistoryPayload, + expectedRunId: String?, + expectedSteps: [OpenClawChatPlanStep], + staleAfterLivePlan: Bool)] = [ + ( + "replace", + historyPayload( + inFlightRun: OpenClawChatInFlightRun( + runId: "run-retained", + text: "working", + plan: OpenClawChatPlanSnapshot( + steps: [OpenClawChatPlanStep(step: "Replacement", status: .completed)]))), + "run-retained", + [OpenClawChatPlanStep(step: "Replacement", status: .completed)], + false), + ( + "legacy-preserve", + historyPayload( + inFlightRun: OpenClawChatInFlightRun(runId: "run-retained", text: "working")), + "run-retained", + retainedSteps, + false), + ( + "superseded", + historyPayload( + inFlightRun: OpenClawChatInFlightRun( + runId: "run-next", + text: "next", + plan: OpenClawChatPlanSnapshot( + steps: [OpenClawChatPlanStep(step: "Next run", status: .inProgress)]))), + "run-next", + [OpenClawChatPlanStep(step: "Next run", status: .inProgress)], + false), + ( + "active-preserve", + historyPayload(hasActiveRun: true, activeRunIds: ["run-retained"]), + "run-retained", + retainedSteps, + false), + ( + "terminal-clear", + historyPayload(hasActiveRun: false, activeRunIds: []), + nil, + [], + false), + ( + "no-evidence-preserve", + historyPayload(supportsActiveRunState: false), + "run-retained", + retainedSteps, + false), + ( + "stale-response-does-not-clobber-newer-live-plan", + historyPayload(hasActiveRun: false, activeRunIds: []), + "run-live", + liveSteps, + true), + ( + "explicit-empty-clears", + historyPayload( + inFlightRun: OpenClawChatInFlightRun( + runId: "run-retained", + text: "working", + plan: OpenClawChatPlanSnapshot(steps: []))), + nil, + [], + false), + ] + + for testCase in cases { + let (_, vm) = await makeViewModel(historyResponses: []) + await MainActor.run { + vm.applyPlanSnapshot( + runId: "run-retained", + steps: retainedSteps, + explanation: nil) + let request = vm.beginHistoryRequest() + if testCase.staleAfterLivePlan { + vm.invalidateRunSnapshots() + vm.adoptRun(runId: "run-live", bufferedText: "live") + vm.applyPlanSnapshot(runId: "run-live", steps: liveSteps, explanation: nil) + } + #expect( + vm.applyHistoryPayload( + testCase.payload, + for: request, + preservingOptimisticLocalMessages: true), + "\(testCase.name): history applies") + #expect(vm.planRunId == testCase.expectedRunId, "\(testCase.name): run owner") + #expect(vm.planSteps == testCase.expectedSteps, "\(testCase.name): steps") + } + } + } + @Test func `foreground history refreshes adopted run snapshot`() async throws { let firstHistory = historyPayload( inFlightRun: OpenClawChatInFlightRun(runId: "run-active", text: "first partial")) @@ -1475,7 +1604,11 @@ struct ChatViewModelTests { @Test func `foreground clears completed run without assistant output`() async throws { let activeHistory = historyPayload( messages: [chatTextMessage(role: "user", text: "quiet task", timestamp: 1)], - inFlightRun: OpenClawChatInFlightRun(runId: "run-quiet", text: "")) + inFlightRun: OpenClawChatInFlightRun( + runId: "run-quiet", + text: "", + plan: OpenClawChatPlanSnapshot( + steps: [OpenClawChatPlanStep(step: "Finish", status: .inProgress)]))) let completedHistory = historyPayload( messages: [chatTextMessage(role: "user", text: "quiet task", timestamp: 1)], hasActiveRun: false) @@ -1483,11 +1616,13 @@ struct ChatViewModelTests { try await loadAndWaitBootstrap(vm: vm) #expect(await MainActor.run { vm.pendingRunCount == 1 }) + #expect(await MainActor.run { vm.planRunId == "run-quiet" }) await MainActor.run { vm.resumeFromForeground() } try await waitUntil("silent completed run clears") { await MainActor.run { vm.pendingRunCount == 0 } } #expect(await MainActor.run { !vm.hasActiveSessionRunWithoutChatSnapshot }) + #expect(await MainActor.run { vm.planSteps.isEmpty && vm.planRunId == nil }) } @Test func `foreground active session with answered chat does not show activity indicator`() async throws { @@ -1558,6 +1693,39 @@ struct ChatViewModelTests { } } + @Test func `post-send stale inactive history preserves newer live plan`() async throws { + let historyGate = AsyncGate() + let historyCalls = AsyncCounter() + let inactiveHistory = historyPayload(hasActiveRun: false) + let (transport, vm) = await makeViewModel( + historyResponses: [historyPayload(), inactiveHistory], + requestHistoryHook: { _ in + if await historyCalls.increment() == 2 { + await historyGate.wait() + } + }, + sendMessageStatus: "pending") + + try await loadAndWaitBootstrap(vm: vm) + await sendUserMessage(vm, text: "finish while disconnected") + let runId = try await waitForLastSentRunId(transport) + try await waitUntil("post-send history starts") { await historyCalls.current() == 2 } + emitPlan( + transport: transport, + runId: runId, + steps: [planStep("Finish", status: "in_progress")]) + try await waitUntil("plan applies before inactive history") { + await MainActor.run { vm.planRunId == runId } + } + + await historyGate.open() + try await Task.sleep(for: .milliseconds(50)) + #expect(await MainActor.run { vm.planRunId == runId }) + #expect(await MainActor.run { vm.planSteps == [ + OpenClawChatPlanStep(step: "Finish", status: .inProgress), + ] }) + } + @Test func `legacy history omission does not clear pending run`() async throws { let legacyHistory = historyPayload(supportsActiveRunState: false) let (_, vm) = await makeViewModel( @@ -3059,7 +3227,11 @@ struct ChatViewModelTests { let activeRunId = "active-run" let initialHistory = historyPayload() let activeHistory = historyPayload( - inFlightRun: OpenClawChatInFlightRun(runId: activeRunId, text: "")) + inFlightRun: OpenClawChatInFlightRun( + runId: activeRunId, + text: "", + plan: OpenClawChatPlanSnapshot( + steps: [OpenClawChatPlanStep(step: "Keep working", status: .inProgress)]))) let (transport, vm) = await makeViewModel( historyResponses: [initialHistory, activeHistory, activeHistory], sendMessageHook: { _ in diff --git a/src/gateway/chat-abort.test.ts b/src/gateway/chat-abort.test.ts index 3d72503d882a..9a0373f2f8ff 100644 --- a/src/gateway/chat-abort.test.ts +++ b/src/gateway/chat-abort.test.ts @@ -602,6 +602,7 @@ describe("resolveInFlightRunSnapshot", () => { const snap = (p: { chatAbortControllers: Map; chatRunBuffers: Map; + chatRunPlanSnapshots?: Parameters[0]["chatRunPlanSnapshots"]; sessionKey: string; canonicalSessionKey?: string; agentId?: string; @@ -610,6 +611,7 @@ describe("resolveInFlightRunSnapshot", () => { resolveInFlightRunSnapshot({ chatAbortControllers: p.chatAbortControllers, chatRunBuffers: p.chatRunBuffers, + chatRunPlanSnapshots: p.chatRunPlanSnapshots, requestedSessionKey: p.sessionKey, canonicalSessionKey: p.canonicalSessionKey ?? p.sessionKey, agentId: p.agentId, @@ -625,6 +627,32 @@ describe("resolveInFlightRunSnapshot", () => { expect(result).toEqual({ runId: "run-1", text: "partial answer so far" }); }); + it("returns the active run plan snapshot with buffered text", () => { + const plan = { + explanation: "Current work", + steps: [{ step: "Implement replay", status: "in_progress" as const }], + }; + expect( + snap({ + chatAbortControllers: new Map([["run-1", inFlightEntry("agent:main:s")]]), + chatRunBuffers: new Map([["run-1", "partial"]]), + chatRunPlanSnapshots: new Map([["run-1", plan]]), + sessionKey: "agent:main:s", + }), + ).toEqual({ runId: "run-1", text: "partial", plan }); + }); + + it("returns an explicit empty plan snapshot for dismissal", () => { + expect( + snap({ + chatAbortControllers: new Map([["run-1", inFlightEntry("agent:main:s")]]), + chatRunBuffers: new Map(), + chatRunPlanSnapshots: new Map([["run-1", { steps: [] }]]), + sessionKey: "agent:main:s", + }), + ).toEqual({ runId: "run-1", text: "", plan: { steps: [] } }); + }); + it("is a no-op when chatAbortControllers is not a Map (unpopulated context)", () => { expect( snap({ @@ -807,23 +835,63 @@ describe("resolveInFlightRunSnapshot", () => { ).toEqual({ runId: "run-b", text: "b" }); }); - it("keeps in-flight text when it fits the chat history budget", () => { + it("keeps in-flight text and plan when they fit the chat history budget", () => { + const plan = { + steps: [{ step: "Keep this", status: "pending" as const }], + }; expect( boundInFlightRunSnapshotForChatHistory({ - snapshot: { runId: "run-1", text: "partial" }, + snapshot: { runId: "run-1", text: "partial", plan }, messages: [], maxBytes: 1_000, }), - ).toEqual({ runId: "run-1", text: "partial" }); + ).toEqual({ runId: "run-1", text: "partial", plan }); }); it("drops oversized in-flight text but keeps the run id for adoption", () => { + const plan = { + steps: [{ step: "Keep this", status: "pending" as const }], + }; expect( boundInFlightRunSnapshotForChatHistory({ - snapshot: { runId: "run-1", text: "x".repeat(1_000) }, + snapshot: { runId: "run-1", text: "x".repeat(1_000), plan }, messages: [], - maxBytes: 100, + maxBytes: 200, }), - ).toEqual({ runId: "run-1", text: "" }); + ).toEqual({ runId: "run-1", text: "", plan }); + }); + + it("drops an oversized plan after dropping text", () => { + expect( + boundInFlightRunSnapshotForChatHistory({ + snapshot: { + runId: "run-1", + text: "", + plan: { + steps: [{ step: "x".repeat(500), status: "pending" }], + }, + }, + messages: [{ role: "user", content: "near budget" }], + maxBytes: 160, + }), + ).toEqual({ runId: "run-1", text: "", plan: { steps: [] } }); + }); + + it("keeps small buffered text and clears an oversized plan explicitly", () => { + // Absence means legacy-gateway unknown to clients; a budget drop must send + // an explicit empty plan so retained stale checklists cannot survive. + expect( + boundInFlightRunSnapshotForChatHistory({ + snapshot: { + runId: "run-1", + text: "short answer", + plan: { + steps: [{ step: "x".repeat(500), status: "pending" }], + }, + }, + messages: [], + maxBytes: 200, + }), + ).toEqual({ runId: "run-1", text: "short answer", plan: { steps: [] } }); }); }); diff --git a/src/gateway/chat-abort.ts b/src/gateway/chat-abort.ts index f7a7ccf32b98..d1d4e3ce253a 100644 --- a/src/gateway/chat-abort.ts +++ b/src/gateway/chat-abort.ts @@ -13,7 +13,11 @@ import type { OpenClawConfig } from "../config/types.openclaw.js"; import { emitAgentEvent, getAgentEventLifecycleGeneration } from "../infra/agent-events.js"; import { jsonUtf8Bytes } from "../infra/json-utf8-bytes.js"; import { projectLiveAssistantBufferedText } from "./live-chat-projector.js"; -import { createChatAbortMarker, type ChatAbortMarker } from "./server-chat-state.js"; +import { + createChatAbortMarker, + type ChatAbortMarker, + type ChatRunPlanSnapshot, +} from "./server-chat-state.js"; const DEFAULT_CHAT_RUN_ABORT_GRACE_MS = 60_000; @@ -260,11 +264,12 @@ function normalizeActiveAgentId(agentId: string | undefined): string | undefined export function resolveInFlightRunSnapshot(params: { chatAbortControllers: Map; chatRunBuffers: Map; + chatRunPlanSnapshots?: Map; requestedSessionKey: string; canonicalSessionKey: string; agentId?: string; defaultAgentId?: string; -}): { runId: string; text: string } | undefined { +}): { runId: string; text: string; plan?: ChatRunPlanSnapshot } | undefined { const matchesKey = (entry: ChatAbortControllerEntry, key: string): boolean => { if (entry.sessionKey !== key) { return false; @@ -328,26 +333,51 @@ export function resolveInFlightRunSnapshot(params: { const projected = projectLiveAssistantBufferedText(bufferedText, { suppressLeadFragments: true, }); - return { runId: best.runId, text: projected.suppress ? "" : projected.text }; + const plan = params.chatRunPlanSnapshots?.get(best.runId); + return { + runId: best.runId, + text: projected.suppress ? "" : projected.text, + ...(plan ? { plan } : {}), + }; } export function boundInFlightRunSnapshotForChatHistory(params: { - snapshot: { runId: string; text: string } | undefined; + snapshot: { runId: string; text: string; plan?: ChatRunPlanSnapshot } | undefined; messages: unknown[]; maxBytes: number; -}): { runId: string; text: string } | undefined { - if (!params.snapshot?.text) { - return params.snapshot; +}): { runId: string; text: string; plan?: ChatRunPlanSnapshot } | undefined { + if (!params.snapshot) { + return undefined; } const messagesBytes = jsonUtf8Bytes(params.messages); const snapshotBytes = jsonUtf8Bytes(params.snapshot); if (messagesBytes + snapshotBytes <= params.maxBytes) { return params.snapshot; } - // The run id is the recovery contract; buffered partial text is opportunistic. - // If it would break the history payload budget, keep adoption and wait for the - // next live delta/final instead of sending an oversized chat.history response. - return { runId: params.snapshot.runId, text: "" }; + // Recovery priority is run adoption, then plan replay, then opportunistic text. + const withoutText = { + runId: params.snapshot.runId, + text: "", + ...(params.snapshot.plan ? { plan: params.snapshot.plan } : {}), + }; + if (params.snapshot.plan && messagesBytes + jsonUtf8Bytes(withoutText) <= params.maxBytes) { + return withoutText; + } + // An oversized plan must not also cost the deliverable buffered text. Clients + // treat an ABSENT plan as legacy-gateway unknown and preserve retained state, + // so a budget-dropped plan is sent as an explicit empty snapshot (authoritative + // clear) — accepted tradeoff: the checklist blanks until the next live plan + // event instead of showing a possibly obsolete retained plan indefinitely. + const droppedPlan = params.snapshot.plan ? { plan: { steps: [] } } : {}; + const withoutPlan = { + runId: params.snapshot.runId, + text: params.snapshot.text, + ...droppedPlan, + }; + if (params.snapshot.text && messagesBytes + jsonUtf8Bytes(withoutPlan) <= params.maxBytes) { + return withoutPlan; + } + return { runId: params.snapshot.runId, text: "", ...droppedPlan }; } export type ChatAbortOps = { diff --git a/src/gateway/local-request-context.ts b/src/gateway/local-request-context.ts index cf60d3eef2d0..e5515c67f607 100644 --- a/src/gateway/local-request-context.ts +++ b/src/gateway/local-request-context.ts @@ -56,6 +56,8 @@ function createLocalGatewayRequestContext( const sessionEvents = new Set(); const chatRuns = new Map(); const chatRunBuffers: GatewayRequestContext["chatRunBuffers"] = new Map(); + const chatRunPlanSnapshots: NonNullable = + new Map(); const chatDeltaSentAt: GatewayRequestContext["chatDeltaSentAt"] = new Map(); const chatDeltaLastBroadcastLen: GatewayRequestContext["chatDeltaLastBroadcastLen"] = new Map(); const chatDeltaLastBroadcastText: GatewayRequestContext["chatDeltaLastBroadcastText"] = new Map(); @@ -65,6 +67,7 @@ function createLocalGatewayRequestContext( // deltas share the client run id prefix but are tracked under separate keys. const clearChatRunState = (runId: string) => { chatRunBuffers.delete(runId); + chatRunPlanSnapshots.delete(runId); chatDeltaSentAt.delete(runId); chatDeltaLastBroadcastLen.delete(runId); chatDeltaLastBroadcastText.delete(runId); @@ -106,6 +109,7 @@ function createLocalGatewayRequestContext( chatQueuedTurns: new Map(), chatAbortedRuns: new Map(), chatRunBuffers, + chatRunPlanSnapshots, chatDeltaSentAt, chatDeltaLastBroadcastLen, chatDeltaLastBroadcastText, diff --git a/src/gateway/server-chat-state.ts b/src/gateway/server-chat-state.ts index 28b62a21852b..5d19cea89f28 100644 --- a/src/gateway/server-chat-state.ts +++ b/src/gateway/server-chat-state.ts @@ -1,3 +1,4 @@ +import type { AgentPlanStep } from "../channels/streaming.js"; // Gateway chat run state registries. // Tracks active runs, delta buffers, tool recipients, and session subscribers. import type { AgentEventPayload } from "../infra/agent-events.js"; @@ -81,6 +82,11 @@ export type BufferedAgentEvent = { payload: AgentEventPayload & { spawnedBy?: string }; }; +export type ChatRunPlanSnapshot = { + steps: AgentPlanStep[]; + explanation?: string; +}; + export type ChatRunRegistry = { add: (sessionId: string, entry: ChatRunRegistration) => void; peek: (sessionId: string) => ChatRunEntry | undefined; @@ -147,6 +153,7 @@ export type ChatRunState = { registry: ChatRunRegistry; rawBuffers: Map; buffers: Map; + planSnapshots: Map; /** Last time any buffered assistant text changed, including suppressed raw buffers. */ bufferUpdatedAt: Map; deltaSentAt: Map; @@ -165,6 +172,7 @@ export function createChatRunState(): ChatRunState { const registry = createChatRunRegistry(); const rawBuffers = new Map(); const buffers = new Map(); + const planSnapshots = new Map(); const bufferUpdatedAt = new Map(); const deltaSentAt = new Map(); const deltaLastBroadcastLen = new Map(); @@ -176,6 +184,7 @@ export function createChatRunState(): ChatRunState { const clearRun = (runId: string) => { rawBuffers.delete(runId); buffers.delete(runId); + planSnapshots.delete(runId); bufferUpdatedAt.delete(runId); deltaSentAt.delete(runId); deltaLastBroadcastLen.delete(runId); @@ -190,6 +199,7 @@ export function createChatRunState(): ChatRunState { registry.clear(); rawBuffers.clear(); buffers.clear(); + planSnapshots.clear(); bufferUpdatedAt.clear(); deltaSentAt.clear(); deltaLastBroadcastLen.clear(); @@ -203,6 +213,7 @@ export function createChatRunState(): ChatRunState { registry, rawBuffers, buffers, + planSnapshots, bufferUpdatedAt, deltaSentAt, deltaLastBroadcastLen, diff --git a/src/gateway/server-chat.agent-events.test.ts b/src/gateway/server-chat.agent-events.test.ts index c53ba2b31012..d1544b63b047 100644 --- a/src/gateway/server-chat.agent-events.test.ts +++ b/src/gateway/server-chat.agent-events.test.ts @@ -225,6 +225,62 @@ describe("agent event handler", () => { }); }); + it("records, replaces, dismisses, and clears normalized plan snapshots", () => { + const { chatRunState, handler } = createHarness(); + chatRunState.registry.add("provider-run", { + sessionKey: "session-1", + clientRunId: "client-run", + }); + + handler({ + runId: "provider-run", + seq: 1, + stream: "plan", + ts: 1_000, + data: { + phase: "update", + explanation: " Initial plan ", + steps: ["Legacy step", { step: "Active step", status: "in_progress" }], + }, + }); + expect(chatRunState.planSnapshots.get("client-run")).toEqual({ + explanation: "Initial plan", + steps: [ + { step: "Legacy step", status: "pending" }, + { step: "Active step", status: "in_progress" }, + ], + }); + + handler({ + runId: "provider-run", + seq: 2, + stream: "plan", + ts: 1_100, + data: { + phase: "update", + steps: [{ step: "Replacement", status: "completed" }], + }, + }); + expect(chatRunState.planSnapshots.get("client-run")).toEqual({ + steps: [{ step: "Replacement", status: "completed" }], + }); + + handler({ + runId: "provider-run", + seq: 3, + stream: "plan", + ts: 1_200, + data: { phase: "update", steps: [] }, + }); + expect(chatRunState.planSnapshots.get("client-run")).toEqual({ steps: [] }); + + chatRunState.planSnapshots.set("client-run", { + steps: [{ step: "Temporary", status: "pending" }], + }); + chatRunState.clearRun("client-run"); + expect(chatRunState.planSnapshots.has("client-run")).toBe(false); + }); + it.each([ { stream: "assistant", data: { text: "Recovered" } }, { stream: "tool", data: { phase: "start", name: "read" } }, diff --git a/src/gateway/server-chat.ts b/src/gateway/server-chat.ts index e35c09a42ad2..5377be7ecf3c 100644 --- a/src/gateway/server-chat.ts +++ b/src/gateway/server-chat.ts @@ -10,6 +10,7 @@ import { resolveToolSearchCodeDisplayTarget } from "../agents/tool-display-commo import { readToolValidationErrorSummary } from "../agents/tool-error-summary.js"; import { DEFAULT_HEARTBEAT_ACK_MAX_CHARS, stripHeartbeatToken } from "../auto-reply/heartbeat.js"; import { normalizeVerboseLevel } from "../auto-reply/thinking.js"; +import { normalizeAgentPlanSteps } from "../channels/streaming.js"; import { getRuntimeConfig } from "../config/io.js"; import { type AgentEventPayload, @@ -1393,6 +1394,15 @@ export function createAgentEventHandler({ if (evt.stream === "assistant") { updateRunToolErrorSummary?.({ runId: evt.runId, clientRunId, summary: undefined }); } + if (evt.stream === "plan" && evt.data?.phase === "update") { + const steps = normalizeAgentPlanSteps(evt.data.steps) ?? []; + const explanation = + typeof evt.data.explanation === "string" ? evt.data.explanation.trim() : ""; + chatRunState.planSnapshots.set(clientRunId, { + steps, + ...(explanation ? { explanation } : {}), + }); + } if (isToolEvent) { const toolPhase = typeof evt.data?.phase === "string" ? evt.data.phase : ""; if (toolPhase === "start") { diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 004b7b12807a..e36c8ad5ee7d 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -712,6 +712,7 @@ async function handleChatHistoryRequest({ const inFlightRun = resolveInFlightRunSnapshot({ chatAbortControllers: context.chatAbortControllers, chatRunBuffers: context.chatRunBuffers, + chatRunPlanSnapshots: context.chatRunPlanSnapshots, requestedSessionKey: sessionKey, canonicalSessionKey: resolveSessionStoreKey({ cfg, sessionKey }), agentId: activeRunAgentId, diff --git a/src/gateway/server-methods/shared-types.ts b/src/gateway/server-methods/shared-types.ts index 5a47e00de3fa..9c168256462f 100644 --- a/src/gateway/server-methods/shared-types.ts +++ b/src/gateway/server-methods/shared-types.ts @@ -36,6 +36,7 @@ import type { BufferedAgentEvent, ChatAbortMarker, ChatRunEntry, + ChatRunPlanSnapshot, ChatRunRegistration, } from "../server-chat-state.js"; import type { GatewayCronServiceContract } from "../server-cron-contract.js"; @@ -194,6 +195,7 @@ export type GatewayRequestContext = { chatQueuedTurns: Map; chatAbortedRuns: Map; chatRunBuffers: Map; + chatRunPlanSnapshots?: Map; chatDeltaSentAt: Map; chatDeltaLastBroadcastLen: Map; chatDeltaLastBroadcastText: Map; diff --git a/src/gateway/server-request-context.ts b/src/gateway/server-request-context.ts index 19040c38a541..a25904db0c3c 100644 --- a/src/gateway/server-request-context.ts +++ b/src/gateway/server-request-context.ts @@ -64,6 +64,7 @@ type GatewayRequestContextParams = { chatQueuedTurns: GatewayRequestContext["chatQueuedTurns"]; chatAbortedRuns: GatewayRequestContext["chatAbortedRuns"]; chatRunBuffers: GatewayRequestContext["chatRunBuffers"]; + chatRunPlanSnapshots?: GatewayRequestContext["chatRunPlanSnapshots"]; chatDeltaSentAt: GatewayRequestContext["chatDeltaSentAt"]; chatDeltaLastBroadcastLen: GatewayRequestContext["chatDeltaLastBroadcastLen"]; chatDeltaLastBroadcastText: GatewayRequestContext["chatDeltaLastBroadcastText"]; @@ -292,6 +293,7 @@ export function createGatewayRequestContext( chatQueuedTurns: params.chatQueuedTurns, chatAbortedRuns: params.chatAbortedRuns, chatRunBuffers: params.chatRunBuffers, + chatRunPlanSnapshots: params.chatRunPlanSnapshots, chatDeltaSentAt: params.chatDeltaSentAt, chatDeltaLastBroadcastLen: params.chatDeltaLastBroadcastLen, chatDeltaLastBroadcastText: params.chatDeltaLastBroadcastText, diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts index d30af5cb280b..b22566ebe0b6 100644 --- a/src/gateway/server.chat.gateway-server-chat-b.test.ts +++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts @@ -206,6 +206,10 @@ async function removeTempDir(dir: string): Promise { function createDirectChatContext(): GatewayRequestContext { return { loadGatewayModelCatalog: vi.fn().mockResolvedValue([]), + loadGatewayModelCatalogSnapshot: vi.fn().mockResolvedValue({ + entries: [], + routeVariants: [], + }), logGateway: { info: vi.fn(), warn: vi.fn(), @@ -216,6 +220,7 @@ function createDirectChatContext(): GatewayRequestContext { chatAbortControllers: new Map(), chatAbortedRuns: new Map(), chatRunBuffers: new Map(), + chatRunPlanSnapshots: new Map(), chatDeltaSentAt: new Map(), chatDeltaLastBroadcastLen: new Map(), chatDeltaLastBroadcastText: new Map(), @@ -424,6 +429,63 @@ async function prepareMainHistoryHarness(params: { } describe("gateway server chat", () => { + test.each(["chat.history", "chat.startup"] as const)( + "%s replays the active plan snapshot in inFlightRun", + async (method) => { + const sessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gw-")); + try { + testState.sessionStorePath = path.join(sessionDir, "sessions.json"); + await writeMainSessionStore(sessionDir); + const context = createDirectChatContext(); + const controller = new AbortController(); + context.chatAbortControllers.set("run-active", { + controller, + sessionId: "sess-main", + sessionKey: "main", + startedAtMs: 1_000, + expiresAtMs: 10_000, + projectSessionActive: true, + }); + context.chatRunBuffers.set("run-active", "partial reply"); + context.chatRunPlanSnapshots?.set("run-active", { + explanation: "Replay on reconnect", + steps: [{ step: "Reconnect clients", status: "in_progress" }], + }); + const responses: Array<{ ok: boolean; payload?: unknown }> = []; + const { chatHandlers } = await import("./server-methods/chat.js"); + + await expectDefined( + chatHandlers[method], + `${method} test invariant`, + )({ + req: { type: "req", id: method, method, params: { sessionKey: "main" } }, + params: { sessionKey: "main" }, + client: null, + isWebchatConnect: () => false, + respond: ((ok, payload) => responses.push({ ok, payload })) as RespondFn, + context, + }); + + expect(responses).toHaveLength(1); + expect(responses[0]?.ok).toBe(true); + expect( + (responses[0]?.payload as { inFlightRun?: unknown } | undefined)?.inFlightRun, + ).toEqual({ + runId: "run-active", + text: "partial reply", + plan: { + explanation: "Replay on reconnect", + steps: [{ step: "Reconnect clients", status: "in_progress" }], + }, + }); + } finally { + testState.sessionStorePath = undefined; + clearConfigCache(); + await removeTempDir(sessionDir); + } + }, + ); + test("chat.history returns catalog-backed session metadata with history", async () => { const sessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gw-")); try { diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index ef8082ebe4fb..0675debca868 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -1896,6 +1896,7 @@ export async function startGatewayServer( chatQueuedTurns, chatAbortedRuns: chatRunState.abortedRuns, chatRunBuffers: chatRunState.buffers, + chatRunPlanSnapshots: chatRunState.planSnapshots, chatDeltaSentAt: chatRunState.deltaSentAt, chatDeltaLastBroadcastLen: chatRunState.deltaLastBroadcastLen, chatDeltaLastBroadcastText: chatRunState.deltaLastBroadcastText, diff --git a/ui/src/pages/chat/chat-history.test.ts b/ui/src/pages/chat/chat-history.test.ts new file mode 100644 index 000000000000..308cda299049 --- /dev/null +++ b/ui/src/pages/chat/chat-history.test.ts @@ -0,0 +1,209 @@ +// @vitest-environment node +import { describe, expect, it, vi } from "vitest"; +import type { GatewayBrowserClient } from "../../api/gateway.ts"; +import { loadChatHistory, type ChatHistoryResult, type ChatState } from "./chat-history.ts"; +import { handleAgentEvent, type PlanStatus, type ToolStreamEntry } from "./tool-stream.ts"; + +type TestState = ChatState & + Parameters[0] & { + requestUpdate: () => void; + }; + +function createState(result: ChatHistoryResult): TestState { + const client = { + request: vi.fn().mockResolvedValue(result), + } as unknown as GatewayBrowserClient; + return { + client, + connected: true, + connectionEpoch: 1, + sessionKey: "main", + chatLoading: false, + chatMessages: [], + chatThinkingLevel: null, + chatVerboseLevel: null, + chatSending: false, + chatMessage: "", + chatAttachments: [], + chatQueue: [], + chatRunId: null, + chatStream: null, + chatStreamStartedAt: null, + chatStreamSegments: [], + toolStreamById: new Map(), + toolStreamOrder: [], + chatToolMessages: [], + toolStreamSyncTimer: null, + planStatus: { + runId: "stale-run", + steps: [{ step: "Reset me", status: "in_progress" }], + }, + lastError: null, + hello: null, + sessions: { + setModelOverride: vi.fn(), + }, + requestUpdate: vi.fn(), + }; +} + +function activeHistory( + runId: string, + plan?: NonNullable["plan"], +): ChatHistoryResult { + return { + messages: [], + sessionInfo: { + key: "main", + kind: "direct", + updatedAt: 1, + hasActiveRun: true, + activeRunIds: [runId], + status: "running", + }, + inFlightRun: { + runId, + text: "intentionally ignored on web", + ...(plan !== undefined ? { plan } : {}), + }, + } satisfies ChatHistoryResult; +} + +describe("chat history plan replay", () => { + const retainedPlan = { + runId: "run-retained", + steps: [{ step: "Retained", status: "in_progress" }], + } satisfies PlanStatus; + const livePlan = { + runId: "run-live", + steps: [{ step: "New live plan", status: "in_progress" }], + } satisfies PlanStatus; + const cases: Array<{ + name: string; + history: ChatHistoryResult; + expected: PlanStatus | null; + staleAfterLivePlan?: boolean; + }> = [ + { + name: "replace", + history: activeHistory("run-retained", { + explanation: " Reconnected work ", + steps: [ + { step: "First active", status: "in_progress" }, + { step: "Second active", status: "in_progress" }, + "Legacy step", + ], + }), + expected: { + runId: "run-retained", + explanation: "Reconnected work", + steps: [ + { step: "First active", status: "in_progress" }, + { step: "Second active", status: "pending" }, + { step: "Legacy step", status: "pending" }, + ], + }, + }, + { + name: "legacy-preserve", + history: activeHistory("run-retained"), + expected: retainedPlan, + }, + { + name: "superseded", + history: activeHistory("run-next", { + steps: [{ step: "Next run", status: "in_progress" }], + }), + expected: { + runId: "run-next", + steps: [{ step: "Next run", status: "in_progress" }], + }, + }, + { + name: "active-preserve", + history: { + messages: [], + sessionInfo: { + key: "main", + kind: "direct", + updatedAt: 1, + hasActiveRun: true, + activeRunIds: ["run-retained"], + }, + }, + expected: retainedPlan, + }, + { + name: "terminal-clear", + history: { + messages: [], + sessionInfo: { + key: "main", + kind: "direct", + updatedAt: 1, + hasActiveRun: false, + activeRunIds: [], + }, + }, + expected: null, + }, + { + name: "no-evidence-preserve", + history: { messages: [] }, + expected: retainedPlan, + }, + { + name: "stale-response-does-not-clobber-newer-live-plan", + history: { + messages: [], + sessionInfo: { + key: "main", + kind: "direct", + updatedAt: 1, + hasActiveRun: false, + activeRunIds: [], + }, + }, + expected: livePlan, + staleAfterLivePlan: true, + }, + { + name: "explicit-empty-clears", + history: activeHistory("run-retained", { steps: [] }), + expected: null, + }, + ]; + + it.each(cases)("$name", async (testCase) => { + let resolveHistory!: (result: ChatHistoryResult) => void; + const historyPromise = new Promise((resolve) => { + resolveHistory = resolve; + }); + const state = createState(testCase.history); + state.planStatus = retainedPlan; + if (testCase.staleAfterLivePlan) { + const request = vi.fn().mockReturnValue(historyPromise); + state.client = { request } as unknown as GatewayBrowserClient; + const loadPromise = loadChatHistory(state); + await vi.waitFor(() => expect(request).toHaveBeenCalledOnce()); + state.chatRunId = "run-live"; + handleAgentEvent(state, { + runId: "run-live", + seq: 2, + stream: "plan", + ts: 2, + sessionKey: "main", + data: { + phase: "update", + steps: [{ step: "New live plan", status: "in_progress" }], + }, + }); + resolveHistory(testCase.history); + await loadPromise; + } else { + await loadChatHistory(state); + } + + expect(state.planStatus).toEqual(testCase.expected); + }); +}); diff --git a/ui/src/pages/chat/chat-history.ts b/ui/src/pages/chat/chat-history.ts index 24a818b3bd13..1c64083c5ebc 100644 --- a/ui/src/pages/chat/chat-history.ts +++ b/ui/src/pages/chat/chat-history.ts @@ -81,6 +81,7 @@ import { visibleCurrentAssistantStreamTail, } from "./stream-reconciliation.ts"; import { reconcileAuthoritativeTerminalHistory } from "./terminal-message-identity.ts"; +import { normalizePlanSnapshot, type PlanStatus } from "./tool-stream.ts"; const SILENT_REPLY_PATTERN = /^\s*NO_REPLY\s*$/; const SYNTHETIC_TRANSCRIPT_REPAIR_RESULT = @@ -334,6 +335,7 @@ export type ChatState = { chatRunId: string | null; chatStream: string | null; chatStreamStartedAt: number | null; + planStatus?: PlanStatus | null; lastError: string | null; chatError?: string | null; /** Completed side-chat turns (oldest first); follow-ups accumulate here. */ @@ -380,8 +382,44 @@ export type ChatHistoryResult = { sessionInfo?: GatewaySessionRow; agentsList?: AgentsListResult; metadata?: ChatMetadataResult; + inFlightRun?: { + runId: string; + text?: string; + plan?: { + steps: Array; + explanation?: string; + }; + }; }; +function reconcileHistoryPlanStatus(params: { + canAdoptRunSnapshot: boolean; + inFlightRun: ChatHistoryResult["inFlightRun"]; + retainedPlan: PlanStatus | null; + sessionInfo: GatewaySessionRow | undefined; +}): PlanStatus | null { + if (!params.canAdoptRunSnapshot) { + return params.retainedPlan; + } + const run = params.inFlightRun; + const runId = run?.runId?.trim(); + if (run && runId) { + if (Object.hasOwn(run, "plan")) { + return run.plan ? normalizePlanSnapshot(run.plan, runId) : null; + } + return params.retainedPlan?.runId === runId ? params.retainedPlan : null; + } + const retainedRunId = params.retainedPlan?.runId; + if (!retainedRunId) { + return params.retainedPlan; + } + const activeRunIds = params.sessionInfo?.activeRunIds; + const confirmsTerminal = + params.sessionInfo?.hasActiveRun === false || + (Array.isArray(activeRunIds) && !activeRunIds.includes(retainedRunId)); + return confirmsTerminal ? null : params.retainedPlan; +} + export function resolveChatHistoryPagination( result: ChatHistoryResult | undefined, ): ChatHistoryPagination { @@ -1097,6 +1135,7 @@ async function loadChatHistoryUncached( state.chatVerboseLevel = res.verboseLevel ?? null; state.chatQueueModeOverride = res.sessionInfo?.queueMode; state.chatEffectiveQueueMode = res.sessionInfo?.effectiveQueueMode; + const planStatusBeforeStreamReset = state.planStatus ?? null; const resetStream = !state.chatRunId || state.chatRunId === previousRunId; if (resetStream) { const streamReconciliation = { @@ -1167,6 +1206,14 @@ async function loadChatHistoryUncached( prunePersistedToolStreamMessages(state, persistedToolStreamIds); } } + // Plan reconciliation shares stream adoption: rejected history cannot clobber newer live state. + // A missing plan is version-skew unknown; replacement or explicit terminal evidence clears it. + state.planStatus = reconcileHistoryPlanStatus({ + canAdoptRunSnapshot: resetStream, + inFlightRun: res.inFlightRun, + retainedPlan: planStatusBeforeStreamReset, + sessionInfo: res.sessionInfo, + }); recordChatHistoryTiming(state, "applied", startedAtMs, { requestSessionKey: sessionKey, requestAgentId, diff --git a/ui/src/pages/chat/tool-stream.ts b/ui/src/pages/chat/tool-stream.ts index cd9456101e3d..143ddf12afa3 100644 --- a/ui/src/pages/chat/tool-stream.ts +++ b/ui/src/pages/chat/tool-stream.ts @@ -740,6 +740,23 @@ function parsePlanSteps(value: unknown): PlanStatus["steps"] { return steps; } +export function normalizePlanSnapshot( + snapshot: { steps?: unknown; explanation?: unknown }, + runIdValue?: unknown, +): PlanStatus | null { + const steps = parsePlanSteps(snapshot.steps); + if (steps.length === 0) { + return null; + } + const explanation = toTrimmedString(snapshot.explanation); + const runId = toTrimmedString(runIdValue); + return { + ...(runId ? { runId } : {}), + ...(explanation ? { explanation } : {}), + steps, + }; +} + function handlePlanEvent(host: PlanHost, payload: AgentEventPayload) { // Plan snapshots are run-owned: a stale or spawned-run event in the same // session must not overwrite (or clear) the active run's checklist. Mirrors @@ -751,17 +768,7 @@ function handlePlanEvent(host: PlanHost, payload: AgentEventPayload) { if (data.phase !== "update") { return; } - const steps = parsePlanSteps(data.steps); - const explanation = toTrimmedString(data.explanation); - const runId = toTrimmedString(payload.runId); - host.planStatus = - steps.length > 0 - ? { - ...(runId ? { runId } : {}), - ...(explanation ? { explanation } : {}), - steps, - } - : null; + host.planStatus = normalizePlanSnapshot(data, payload.runId); host.requestUpdate?.(); }