From 4bfaccafd62ac2ff2e70ca1decc40fb1297ab438 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 17 Jul 2026 17:10:07 +0100 Subject: [PATCH] feat(chat): replay the active plan snapshot to reconnecting clients (#109966) * feat(chat): replay the active plan snapshot to reconnecting clients Plan checklists were live-event-only: a client that connected or refreshed mid-run saw nothing until the next update_plan call, and iOS adoptRun actively wiped retained plan state on reconnect. The gateway now records the latest normalized plan snapshot per run beside the existing chat text buffers and replays it through the established chat.history/chat.startup inFlightRun recovery contract (runId > plan > text under the payload budget; a budget-dropped plan is sent as an explicit empty snapshot so clients never preserve a stale checklist). Web, iOS, and Android seed and reconcile the snapshot through one shared contract riding each client's existing inFlightRun adoption gate: present plan replaces, absent plan preserves (legacy gateway), a different or terminated run clears, and stale history responses cannot clobber a newer live plan. Verified live gap: openclaw/openclaw#108675 comment 5001475066. * fix(ios): extract run-snapshot adoption below the file-length lint ceiling ChatViewModel.swift crossed SwiftLint's 1500-line limit; the in-flight snapshot adoption and plan reconciliation move to a focused extension. Also refresh the native i18n inventory for shifted line anchors. * fixup: regenerate i18n anchors and test formatting after rebase * ci: retrigger after wedged queued runs * fix(ui): use Object.hasOwn for the plan field presence probe * fix(ios): annotate the plan-snapshot fixture init for periphery --- apps/.i18n/native-source.json | 28 +- .../ai/openclaw/app/chat/ChatController.kt | 76 ++++-- .../java/ai/openclaw/app/chat/ChatModels.kt | 8 + .../ChatControllerReconnectRestoreTest.kt | 243 ++++++++++++++++++ .../ai/openclaw/app/chat/ChatReplayHarness.kt | 43 +++- .../Sources/OpenClawChatUI/ChatModels.swift | 23 +- .../ChatViewModel+HistoryReconciliation.swift | 49 +--- .../OpenClawChatUI/ChatViewModel+Plan.swift | 14 +- .../ChatViewModel+RunSnapshot.swift | 72 ++++++ .../ChatViewModel+TransportEvents.swift | 8 +- .../OpenClawKitTests/ChatViewModelTests.swift | 180 ++++++++++++- src/gateway/chat-abort.test.ts | 80 +++++- src/gateway/chat-abort.ts | 52 +++- src/gateway/local-request-context.ts | 4 + src/gateway/server-chat-state.ts | 11 + src/gateway/server-chat.agent-events.test.ts | 56 ++++ src/gateway/server-chat.ts | 10 + src/gateway/server-methods/chat.ts | 1 + src/gateway/server-methods/shared-types.ts | 2 + src/gateway/server-request-context.ts | 2 + .../server.chat.gateway-server-chat-b.test.ts | 62 +++++ src/gateway/server.impl.ts | 1 + ui/src/pages/chat/chat-history.test.ts | 209 +++++++++++++++ ui/src/pages/chat/chat-history.ts | 47 ++++ ui/src/pages/chat/tool-stream.ts | 29 ++- 25 files changed, 1190 insertions(+), 120 deletions(-) create mode 100644 apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift create mode 100644 ui/src/pages/chat/chat-history.test.ts 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?.(); }