From e71fc902eece0f795d593e5d4e03ad6e7496d9a7 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 19 Aug 2026 02:08:57 -0700 Subject: [PATCH] fix(gateway): make activeRunIds presence mean a complete exact run set (#126106) * fix(gateway): make activeRunIds presence mean a complete exact run set Session rows no longer emit activeRunIds: [] while hasActiveRun is true. Presence now means the complete exact set of direct run ids; omission means identities are unavailable (projected/embedded owners); [] only ever represents proven idle. Consumers stop guessing: soleActiveSessionRunId() replaces the arbitrary [0] fallbacks in the observer digest, transcript cache key, activity inspector, and stale-terminal reconciliation, each falling back to its owner fact. Follows the maintainer direction from #125983: the field stays as Gateway-owned exact facts; producer-side liveness/observer projections are a named follow-up. * fix(gateway): clear unavailable active run ids in events * fix(gateway): preserve idle active run sets * fix(clients): close active run id cache gaps * test(android): isolate history run snapshot --- .../ai/openclaw/app/chat/ChatController.kt | 14 +- .../java/ai/openclaw/app/chat/ChatModels.kt | 1 + .../ChatControllerReconnectRestoreTest.kt | 23 ++++ .../app/chat/ChatControllerUsageStreamTest.kt | 23 ++++ .../ChatGatewayPayloadCodec.swift | 5 +- .../Sources/OpenClawChatUI/ChatModels.swift | 60 ++++++++- .../ChatSessionSidebarModel.swift | 4 +- .../OpenClawChatUI/ChatTransport.swift | 16 ++- .../ChatViewModel+RunSnapshot.swift | 15 ++- .../ChatViewModel+SessionKeys.swift | 6 +- .../ChatViewModel+TransportEvents.swift | 41 +++--- .../ChatSessionSidebarModelTests.swift | 26 ++++ .../OpenClawKitTests/ChatViewModelTests.swift | 126 +++++++++++++++++- docs/gateway/clients.md | 33 ++++- docs/gateway/protocol.md | 9 +- src/gateway/server-chat.agent-events.test.ts | 44 ++++++ src/gateway/server-chat.ts | 16 ++- .../server-methods/chat-history-handler.ts | 4 +- .../session-active-runs.test.ts | 18 +-- .../server-methods/session-active-runs.ts | 16 ++- .../session-change-event.test.ts | 74 ++++++++-- .../server-methods/session-change-event.ts | 7 +- src/gateway/server-methods/sessions-read.ts | 2 +- src/gateway/server-session-events.test.ts | 8 +- src/gateway/server-session-events.ts | 9 +- .../server.sessions.list-changed.test.ts | 21 ++- src/gateway/session-event-payload.ts | 9 +- src/gateway/session-utils.types.ts | 1 + ui/src/lib/sessions/reconcile.test.ts | 58 ++++++++ 29 files changed, 607 insertions(+), 82 deletions(-) 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 bbeefd5c7330..11079f336283 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 @@ -6658,6 +6658,7 @@ class ChatController internal constructor( .asArrayOrNull() ?.mapNotNull { it.asStringOrNull()?.trim()?.takeIf(String::isNotEmpty) }, hasActiveRunMetadata = "hasActiveRun" in obj || "activeRunIds" in obj, + hasActiveRunIdsMetadata = "activeRunIds" in obj, parentSessionKey = obj["parentSessionKey"].asStringOrNull()?.trim(), spawnedBy = obj["spawnedBy"].asStringOrNull()?.trim(), hasActiveSubagentRun = obj["hasActiveSubagentRun"].asBooleanOrNull(), @@ -6877,6 +6878,7 @@ class ChatController internal constructor( upsertSessionEntry( info, preserveExistingContextUsageWithoutTotal = true, + replaceActiveRunIds = true, publishRunState = publishRunState, ) } @@ -6884,6 +6886,7 @@ class ChatController internal constructor( private fun upsertSessionEntry( entry: ChatSessionEntry, preserveExistingContextUsageWithoutTotal: Boolean = false, + replaceActiveRunIds: Boolean = false, clearedFields: Set = emptySet(), publishRunState: Boolean = true, ) { @@ -6898,6 +6901,7 @@ class ChatController internal constructor( existing = it[index], next = entry, preserveExistingContextUsageWithoutTotal = preserveExistingContextUsageWithoutTotal, + replaceActiveRunIds = replaceActiveRunIds, ) if (clearedFields.isNotEmpty()) { applied = @@ -7532,10 +7536,12 @@ internal fun mergeChatSessionEntry( existing: ChatSessionEntry, next: ChatSessionEntry, preserveExistingContextUsageWithoutTotal: Boolean = false, + replaceActiveRunIds: Boolean = false, ): ChatSessionEntry { val preserveExistingContextUsage = preserveExistingContextUsageWithoutTotal && next.totalTokens == null val hasActiveRun = if (next.hasActiveRunMetadata) next.hasActiveRun else existing.hasActiveRun - val activeRunIds = if (next.hasActiveRunMetadata) next.activeRunIds else existing.activeRunIds + val activeRunIds = + if (replaceActiveRunIds || next.hasActiveRunIdsMetadata) next.activeRunIds else existing.activeRunIds val observerDigest = reconcileSessionObserverDigest( existing = existing.observerDigest, @@ -7600,6 +7606,12 @@ internal fun mergeChatSessionEntry( hasActiveRun = hasActiveRun, activeRunIds = activeRunIds, hasActiveRunMetadata = existing.hasActiveRunMetadata || next.hasActiveRunMetadata, + hasActiveRunIdsMetadata = + if (replaceActiveRunIds) { + next.hasActiveRunIdsMetadata + } else { + existing.hasActiveRunIdsMetadata || next.hasActiveRunIdsMetadata + }, parentSessionKey = next.parentSessionKey ?: existing.parentSessionKey, spawnedBy = next.spawnedBy ?: existing.spawnedBy, hasActiveSubagentRun = next.hasActiveSubagentRun ?: existing.hasActiveSubagentRun, 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 d76d3591b247..8d1881747317 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 @@ -332,6 +332,7 @@ data class ChatSessionEntry( val hasActiveRun: Boolean? = null, val activeRunIds: List? = null, val hasActiveRunMetadata: Boolean = hasActiveRun != null || activeRunIds != null, + val hasActiveRunIdsMetadata: Boolean = activeRunIds != null, val parentSessionKey: String? = null, val spawnedBy: String? = null, val hasActiveSubagentRun: Boolean? = 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 925622bfcabb..3c143f09e8ca 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 @@ -482,6 +482,29 @@ class ChatControllerReconnectRestoreTest { assertEquals(1, controller.messages.value.size) } + @Test + fun reconnectHistoryOmissionClearsStaleExactRunIds() = + runTest { + val gateway = ScriptedGateway(json) + val controller = loadController(gateway, history(emptyList())) + controller.handleGatewayEvent( + "sessions.changed", + """{"reason":"patch","session":{"key":"main","agentId":"main","hasActiveRun":true,"activeRunIds":["run-stale"]}}""", + ) + assertEquals("run-stale", controller.selectedActiveRunPresentation.value.runId) + + gateway.respondWith( + "chat.history", + history(emptyList(), hasActiveRun = true, activeRunIds = null), + ) + val pendingSessionList = CompletableDeferred() + gateway.respond("sessions.list") { pendingSessionList.await() } + reconnect(controller) + + assertEquals(1, controller.selectedActiveRunPresentation.value.count) + assertNull(controller.selectedActiveRunPresentation.value.runId) + } + @Test fun reconnectStaysUnhealthyUntilRecoveryHistoryApplies() = runTest { diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerUsageStreamTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerUsageStreamTest.kt index e19b7637e5a3..0b7430cc1958 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerUsageStreamTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerUsageStreamTest.kt @@ -110,6 +110,29 @@ class ChatControllerUsageStreamTest { assertEquals("main:active", presentation.clockKey) } + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun activeRunIdTombstoneClearsExactIdsWhileOmissionPreservesThem() = + runTest { + val gateway = ScriptedGateway(json) + val controller = ChatController(scope = backgroundScope, json = json, requestGateway = gateway::request) + controller.handleGatewayEvent("sessions.changed", advertise("run-exact")) + assertEquals("run-exact", controller.selectedActiveRunPresentation.value.runId) + + controller.handleGatewayEvent( + "sessions.changed", + """{"reason":"patch","session":{"key":"main","agentId":"main","hasActiveRun":true}}""", + ) + assertEquals("run-exact", controller.selectedActiveRunPresentation.value.runId) + + controller.handleGatewayEvent( + "sessions.changed", + """{"reason":"patch","session":{"key":"main","agentId":"main","hasActiveRun":true,"activeRunIds":null}}""", + ) + assertEquals(1, controller.selectedActiveRunPresentation.value.count) + assertNull(controller.selectedActiveRunPresentation.value.runId) + } + @Test @OptIn(ExperimentalCoroutinesApi::class) fun idlessReplacementRunGetsANewStartedAtClockKey() = diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatGatewayPayloadCodec.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatGatewayPayloadCodec.swift index cdd57a3d0f0b..816b9d16b7bc 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatGatewayPayloadCodec.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatGatewayPayloadCodec.swift @@ -152,7 +152,10 @@ public enum OpenClawChatGatewayPayloadCodec { agentId: message.agentId, message: canonicalMessage, messageId: message.messageId, - messageSeq: message.messageSeq)) + messageSeq: message.messageSeq, + hasActiveRun: message.hasActiveRun, + activeRunIds: message.activeRunIds, + activeRunIdsPresent: message.activeRunIdsPresent)) } return .sessionMessage(message) case "agent": diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift index a201a5038e32..0c75565d2de8 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift @@ -809,6 +809,9 @@ public struct OpenClawSessionMessageEventPayload: Codable, Sendable { public let message: OpenClawChatMessage? public let messageId: String? public let messageSeq: Int? + public let hasActiveRun: Bool? + public let activeRunIds: [String]? + let activeRunIdsPresent: Bool // periphery:ignore - package tests construct transport events; app consumers decode them. public init( @@ -816,13 +819,68 @@ public struct OpenClawSessionMessageEventPayload: Codable, Sendable { agentId: String? = nil, message: OpenClawChatMessage?, messageId: String?, - messageSeq: Int?) + messageSeq: Int?, + hasActiveRun: Bool? = nil, + activeRunIds: [String]? = nil, + activeRunIdsPresent: Bool? = nil) { self.sessionKey = sessionKey self.agentId = agentId self.message = message self.messageId = messageId self.messageSeq = messageSeq + self.hasActiveRun = hasActiveRun + self.activeRunIds = activeRunIds + self.activeRunIdsPresent = activeRunIdsPresent ?? (activeRunIds != nil) + } + + public init(from decoder: Decoder) throws { + let container = try decoder.container(keyedBy: CodingKeys.self) + let nested = try? container.nestedContainer(keyedBy: CodingKeys.self, forKey: .session) + + func decode(_ type: T.Type, forKey key: CodingKeys) throws -> T? { + if container.contains(key) { + return try container.decodeIfPresent(type, forKey: key) + } + return try nested?.decodeIfPresent(type, forKey: key) + } + + self.sessionKey = try decode(String.self, forKey: .sessionKey) + self.agentId = try decode(String.self, forKey: .agentId) + self.message = try container.decodeIfPresent(OpenClawChatMessage.self, forKey: .message) + self.messageId = try container.decodeIfPresent(String.self, forKey: .messageId) + self.messageSeq = try container.decodeIfPresent(Int.self, forKey: .messageSeq) + self.hasActiveRun = try decode(Bool.self, forKey: .hasActiveRun) + self.activeRunIds = try decode([String].self, forKey: .activeRunIds) + self.activeRunIdsPresent = container.contains(.activeRunIds) || nested?.contains(.activeRunIds) == true + } + + public func encode(to encoder: Encoder) throws { + var container = encoder.container(keyedBy: CodingKeys.self) + try container.encodeIfPresent(self.sessionKey, forKey: .sessionKey) + try container.encodeIfPresent(self.agentId, forKey: .agentId) + try container.encodeIfPresent(self.message, forKey: .message) + try container.encodeIfPresent(self.messageId, forKey: .messageId) + try container.encodeIfPresent(self.messageSeq, forKey: .messageSeq) + try container.encodeIfPresent(self.hasActiveRun, forKey: .hasActiveRun) + if self.activeRunIdsPresent { + if let activeRunIds { + try container.encode(activeRunIds, forKey: .activeRunIds) + } else { + try container.encodeNil(forKey: .activeRunIds) + } + } + } + + private enum CodingKeys: String, CodingKey { + case session + case sessionKey + case agentId + case message + case messageId + case messageSeq + case hasActiveRun + case activeRunIds } } diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatSessionSidebarModel.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatSessionSidebarModel.swift index 616b3d21b446..b92ff86fd904 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatSessionSidebarModel.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatSessionSidebarModel.swift @@ -384,8 +384,8 @@ public enum ChatSessionSidebarModel { if let hasActiveRun = change.hasActiveRun { session.hasActiveRun = hasActiveRun } - if let activeRunIds = change.activeRunIds { - session.activeRunIds = activeRunIds + if change.activeRunIdsPresent { + session.activeRunIds = change.activeRunIds } if let startedAt = change.startedAt { session.startedAt = startedAt diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTransport.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTransport.swift index cb352e3a5f00..9b38011c60e5 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTransport.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTransport.swift @@ -88,6 +88,7 @@ public struct OpenClawChatSessionsChangedEvent: Codable, Sendable, Equatable { let observerDigestPresent: Bool let statusPresent: Bool let lastRunErrorPresent: Bool + let activeRunIdsPresent: Bool public init( sessionKey: String?, @@ -115,7 +116,8 @@ public struct OpenClawChatSessionsChangedEvent: Codable, Sendable, Equatable { agentStatusPresent: Bool? = nil, observerDigestPresent: Bool? = nil, statusPresent: Bool? = nil, - lastRunErrorPresent: Bool? = nil) + lastRunErrorPresent: Bool? = nil, + activeRunIdsPresent: Bool? = nil) { self.sessionKey = sessionKey self.agentId = agentId @@ -132,7 +134,7 @@ public struct OpenClawChatSessionsChangedEvent: Codable, Sendable, Equatable { self.status = status self.lastRunError = lastRunError self.hasActiveRun = hasActiveRun - self.activeRunIds = activeRunIds + self.activeRunIds = activeRunIds ?? session?.activeRunIds self.startedAt = startedAt self.endedAt = endedAt self.swarmGroupId = swarmGroupId @@ -143,6 +145,7 @@ public struct OpenClawChatSessionsChangedEvent: Codable, Sendable, Equatable { self.observerDigestPresent = observerDigestPresent ?? (observerDigest != nil) self.statusPresent = statusPresent ?? (status != nil) self.lastRunErrorPresent = lastRunErrorPresent ?? (lastRunError != nil) + self.activeRunIdsPresent = activeRunIdsPresent ?? (activeRunIds != nil || session?.activeRunIds != nil) } public init(from decoder: Decoder) throws { @@ -190,6 +193,7 @@ public struct OpenClawChatSessionsChangedEvent: Codable, Sendable, Equatable { self.observerDigestPresent = container.contains(.observerDigest) || nested?.contains(.observerDigest) == true self.statusPresent = container.contains(.status) || nested?.contains(.status) == true self.lastRunErrorPresent = container.contains(.lastRunError) || nested?.contains(.lastRunError) == true + self.activeRunIdsPresent = container.contains(.activeRunIds) || nested?.contains(.activeRunIds) == true } public func encode(to encoder: Encoder) throws { @@ -209,7 +213,13 @@ public struct OpenClawChatSessionsChangedEvent: Codable, Sendable, Equatable { try container.encodeIfPresent(self.status, forKey: .status) try container.encodeIfPresent(self.lastRunError, forKey: .lastRunError) try container.encodeIfPresent(self.hasActiveRun, forKey: .hasActiveRun) - try container.encodeIfPresent(self.activeRunIds, forKey: .activeRunIds) + if self.activeRunIdsPresent { + if let activeRunIds { + try container.encode(activeRunIds, forKey: .activeRunIds) + } else { + try container.encodeNil(forKey: .activeRunIds) + } + } try container.encodeIfPresent(self.startedAt, forKey: .startedAt) try container.encodeIfPresent(self.endedAt, forKey: .endedAt) try container.encodeIfPresent(self.swarmGroupId, forKey: .swarmGroupId) diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift index 2449da8b1374..f117edfd069a 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift @@ -12,10 +12,17 @@ extension OpenClawChatViewModel { return } self.latestAppliedRunSnapshotRequestID = request.id - if let activeRunIDs = payload.sessionInfo?.activeRunIds { - self.updateActiveSessionRunIDs(activeRunIDs) - } else if payload.sessionInfo?.hasActiveRun == false { - self.updateActiveSessionRunIDs([]) + if let sessionInfo = payload.sessionInfo { + if let index = self.sessions.firstIndex(where: { + self.matchesCurrentSessionKey(incoming: $0.key, current: request.session.key) + }) { + var updated = self.sessions + updated[index].hasActiveRun = sessionInfo.hasActiveRun + updated[index].activeRunIds = sessionInfo.activeRunIds + self.sessions = updated + } else { + self.updateActiveSessionRunIDs(sessionInfo.activeRunIds ?? []) + } } guard let snapshot = payload.inFlightRun, let runId = Self.normalizedRunID(snapshot.runId), diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+SessionKeys.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+SessionKeys.swift index 32691b6d4581..d2496a8be8f3 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+SessionKeys.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+SessionKeys.swift @@ -102,11 +102,7 @@ extension OpenClawChatViewModel { self.updateActiveSessionRunIDs([]) return } - if let activeRunIDs = session.activeRunIds { - self.updateActiveSessionRunIDs(activeRunIDs) - } else if session.hasActiveRun == false { - self.updateActiveSessionRunIDs([]) - } + self.updateActiveSessionRunIDs(session.activeRunIds ?? []) } func ownsLiveTelemetryRun(_ runID: String) -> Bool { diff --git a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift index 7d084b31033f..7563816d4783 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift @@ -258,7 +258,8 @@ extension OpenClawChatViewModel { existing: existing, snapshot: snapshot, phase: phase, - runID: runID) + activeRunIDs: change.activeRunIds, + activeRunIDsPresent: change.activeRunIdsPresent) self.sessions = OpenClawChatSessionListOrganizer.organize(updated) self.persistSessionsToCache(self.sessions) return .merged @@ -325,10 +326,9 @@ extension OpenClawChatViewModel { existing: OpenClawChatSessionEntry, snapshot: OpenClawChatSessionEntry, phase: String, - runID: String?) -> OpenClawChatSessionEntry + activeRunIDs: [String]?, + activeRunIDsPresent: Bool) -> OpenClawChatSessionEntry { - let isTerminal = phase == "end" || phase == "error" - let existingActiveRunIDs = existing.activeRunIds?.compactMap { Self.normalizedRunID($0) } ?? [] var merged = existing merged.updatedAt = snapshot.updatedAt ?? existing.updatedAt merged.status = snapshot.status ?? existing.status @@ -339,16 +339,8 @@ extension OpenClawChatViewModel { merged.lastRunError = snapshot.lastRunError ?? existing.lastRunError } - if let activeRunIDs = snapshot.activeRunIds { + if activeRunIDsPresent { merged.activeRunIds = activeRunIDs - } else if phase == "start", let runID { - merged.activeRunIds = existingActiveRunIDs.contains(runID) - ? existingActiveRunIDs - : existingActiveRunIDs + [runID] - merged.hasActiveRun = true - } else if isTerminal, let runID { - merged.activeRunIds = existingActiveRunIDs.filter { $0 != runID } - merged.hasActiveRun = merged.activeRunIds?.isEmpty == false } switch phase { @@ -403,11 +395,30 @@ extension OpenClawChatViewModel { } private func handleSessionMessageEvent(_ payload: OpenClawSessionMessageEventPayload) { - guard let message = payload.message else { return } - let sanitized = Self.stripInboundMetadata(from: message) let isCurrentSession = payload.sessionKey.map { self.matchesCurrentSessionKey(incoming: $0, agentId: payload.agentId, current: self.sessionKey) } ?? true + if isCurrentSession, payload.hasActiveRun != nil || payload.activeRunIdsPresent { + let change = OpenClawChatSessionsChangedEvent( + sessionKey: payload.sessionKey, + agentId: payload.agentId, + reason: "message", + hasActiveRun: payload.hasActiveRun, + activeRunIds: payload.activeRunIds, + activeRunIdsPresent: payload.activeRunIdsPresent) + if let projected = ChatSessionSidebarModel.applying( + sessionChange: change, + to: self.sessions, + activeAgentId: self.activeAgentId) + { + self.sessions = projected + } + if payload.activeRunIdsPresent { + self.updateActiveSessionRunIDs(payload.activeRunIds ?? []) + } + } + guard let message = payload.message else { return } + let sanitized = Self.stripInboundMetadata(from: message) // Confirmation is gateway-scoped, not presentation-scoped. A flush // can drain session A while session B is visible, and A's event must // still retire its durable row before this handler returns early. diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatSessionSidebarModelTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatSessionSidebarModelTests.swift index 7413f86d109a..52631d9a02be 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatSessionSidebarModelTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatSessionSidebarModelTests.swift @@ -776,6 +776,32 @@ struct ChatSessionSidebarModelTests { #expect(cleared.lastRunError == nil) } + @Test func `active run id tombstone clears exact ids while omission is inert`() throws { + let existing = self.entry( + key: "agent:main:work", + updatedAt: 100, + status: "running", + hasActiveRun: true, + activeRunIds: ["run-exact"]) + let decoder = JSONDecoder() + + let omitted = try decoder.decode( + OpenClawChatSessionsChangedEvent.self, + from: Data(#"{"reason":"run-progress","session":{"key":"agent:main:work","updatedAt":200,"hasActiveRun":true}}"#.utf8)) + let retained = try #require(ChatSessionSidebarModel.applying( + sessionChange: omitted, + to: [existing])) + #expect(retained[0].activeRunIds == ["run-exact"]) + + let tombstoned = try decoder.decode( + OpenClawChatSessionsChangedEvent.self, + from: Data(#"{"reason":"run-progress","session":{"key":"agent:main:work","updatedAt":300,"hasActiveRun":true,"activeRunIds":null}}"#.utf8)) + let cleared = try #require(ChatSessionSidebarModel.applying( + sessionChange: tombstoned, + to: retained)) + #expect(cleared[0].activeRunIds == nil) + } + @Test func `subtitle precedence keeps attention and status above observer digest`() { let digest = OpenClawChatSessionObserverDigest( runId: "run-1", diff --git a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift index db4e4f7cfe85..7bd323fd2bc4 100644 --- a/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift +++ b/apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift @@ -149,7 +149,7 @@ private func lifecycleSessionEntry( updatedAt: Double, status: String, hasActiveRun: Bool, - activeRunIds: [String], + activeRunIds: [String]?, startedAt: Double? = nil, endedAt: Double? = nil, runtimeMs: Double? = nil, @@ -2575,6 +2575,130 @@ struct ChatViewModelTests { #expect(!viewModel.hasAdvertisedLiveRun) } + @Test @MainActor func `snapshot row omission clears stale exact run projection`() { + let viewModel = OpenClawChatViewModel( + sessionKey: "main", + transport: TestChatTransport(historyResponses: [])) + var running = sessionEntry(key: "main", updatedAt: 1) + running.hasActiveRun = true + running.activeRunIds = ["run-stale"] + viewModel.sessions = [running] + #expect(viewModel.activeSessionRunIDs == ["run-stale"]) + + var unavailable = sessionEntry(key: "main", updatedAt: 2) + unavailable.hasActiveRun = true + unavailable.activeRunIds = nil + viewModel.sessions = [unavailable] + + #expect(viewModel.activeSessionRunIDs.isEmpty) + } + + @Test @MainActor func `history snapshot omission clears stale exact run ids`() { + let viewModel = OpenClawChatViewModel( + sessionKey: "main", + transport: TestChatTransport(historyResponses: [])) + var running = sessionEntry(key: "main", updatedAt: 1) + running.hasActiveRun = true + running.activeRunIds = ["run-stale"] + viewModel.sessions = [running] + let request = viewModel.beginHistoryRequest() + + #expect(viewModel.applyHistoryPayload( + historyPayload(hasActiveRun: true, activeRunIds: nil), + for: request, + preservingOptimisticLocalMessages: true)) + + #expect(viewModel.currentSessionEntry()?.activeRunIds == nil) + #expect(viewModel.activeSessionRunIDs.isEmpty) + } + + @Test @MainActor func `event tombstone clears stale exact run projection`() { + let viewModel = OpenClawChatViewModel( + sessionKey: "main", + transport: TestChatTransport(historyResponses: [])) + var running = sessionEntry(key: "main", updatedAt: 1) + running.hasActiveRun = true + running.activeRunIds = ["run-stale"] + viewModel.sessions = [running] + + viewModel.handleTransportEvent(.sessionsChanged(.init( + sessionKey: "main", + reason: "run-progress", + updatedAt: 2, + hasActiveRun: true, + activeRunIds: nil, + activeRunIdsPresent: true))) + + #expect(viewModel.currentSessionEntry()?.activeRunIds == nil) + #expect(viewModel.activeSessionRunIDs.isEmpty) + } + + @Test @MainActor func `lifecycle tombstone clears instead of inferring an exact run id`() { + let viewModel = OpenClawChatViewModel( + sessionKey: "main", + transport: TestChatTransport(historyResponses: [])) + var running = sessionEntry(key: "main", updatedAt: 1) + running.status = "running" + running.hasActiveRun = true + running.activeRunIds = ["run-stale"] + viewModel.sessions = [running] + + viewModel.handleTransportEvent(.sessionsChanged(.init( + sessionKey: "main", + reason: "run-progress", + phase: "start", + runId: "run-hidden", + session: lifecycleSessionEntry( + key: "main", + updatedAt: 2, + status: "running", + hasActiveRun: true, + activeRunIds: nil)))) + #expect(viewModel.currentSessionEntry()?.activeRunIds == ["run-stale"]) + + viewModel.handleTransportEvent(.sessionsChanged(.init( + sessionKey: "main", + reason: "run-progress", + phase: "start", + runId: "run-hidden", + session: lifecycleSessionEntry( + key: "main", + updatedAt: 3, + status: "running", + hasActiveRun: true, + activeRunIds: nil), + hasActiveRun: true, + activeRunIds: nil, + activeRunIdsPresent: true))) + + #expect(viewModel.currentSessionEntry()?.activeRunIds == nil) + #expect(viewModel.activeSessionRunIDs.isEmpty) + } + + @Test @MainActor func `session message tombstone clears stale exact run ids`() throws { + let viewModel = OpenClawChatViewModel( + sessionKey: "main", + transport: TestChatTransport(historyResponses: [])) + var running = sessionEntry(key: "main", updatedAt: 1) + running.hasActiveRun = true + running.activeRunIds = ["run-stale"] + viewModel.sessions = [running] + let omitted = try JSONDecoder().decode( + OpenClawSessionMessageEventPayload.self, + from: Data(#"{"sessionKey":"main","hasActiveRun":true,"messageId":"message-1","message":{"role":"assistant","content":[{"type":"text","text":"working"}],"timestamp":2}}"#.utf8)) + viewModel.handleTransportEvent(.sessionMessage(omitted)) + #expect(viewModel.currentSessionEntry()?.activeRunIds == ["run-stale"]) + + let payload = try JSONDecoder().decode( + OpenClawSessionMessageEventPayload.self, + from: Data(#"{"sessionKey":"main","hasActiveRun":true,"activeRunIds":null,"messageId":"message-2","message":{"role":"assistant","content":[{"type":"text","text":"still working"}],"timestamp":3}}"#.utf8)) + + viewModel.handleTransportEvent(.sessionMessage(payload)) + + #expect(viewModel.currentSessionEntry()?.activeRunIds == nil) + #expect(viewModel.activeSessionRunIDs.isEmpty) + } + @Test @MainActor func `remote lifecycle merges terminal recap metadata`() { let viewModel = OpenClawChatViewModel( sessionKey: "main", diff --git a/docs/gateway/clients.md b/docs/gateway/clients.md index b318e3e3e3f9..fbc82cd77fd9 100644 --- a/docs/gateway/clients.md +++ b/docs/gateway/clients.md @@ -146,13 +146,15 @@ current in-memory run state: 3. If `inFlightRun` is present, adopt its `runId`, buffered `text`, and optional `plan`. Adopt the run even when `text` is empty. 4. Treat `sessionInfo.hasActiveRun` as aggregate direct-session activity. - `activeRunIds`, when present, contains known exact active run identities and - can be empty while aggregate activity is still true. Omission means the field - was not projected and provides no identity information. In incremental merge - events, replace the cached list when the field is present; an empty array is - the tombstone that clears prior exact identities. Correlate only a run ID the - client owns locally or received from a request, history response, or event, - and never select the first list entry as an owner. + `activeRunIds`, when present, is the complete exact active set; an empty array + therefore proves the session is idle. When `hasActiveRun` is true and + `activeRunIds` is omitted, another runtime owner is active but its exact run + identities are unavailable. In incremental merge events, omission means no + change, `null` is the event-only tombstone that clears cached exact IDs to + unavailable, and an array replaces the cache (including `[]` for proven + idle). Correlate only a run ID the client owns locally or received from a + request, history response, or event, and never select the first list entry as + an owner. 5. Show an observer headline or run-inspector link only when the observer digest's exact `runId` is present in `activeRunIds`. Aggregate activity alone does not make a retained digest current. @@ -161,6 +163,23 @@ current in-memory run state: already-seen or lower sequence, and treat a forward gap as a reason to reload authoritative history. +### Active-run cache matrix + +Classify the source before applying `activeRunIds`; the same omission has +different meaning in a full snapshot and an incremental delta. + +| Client cache | Read path | Class | Required behavior | +| ------------------------ | ---------------------------------------------------------- | -------- | -------------------------------------------------------------------------------------------------- | +| Web session roster | `sessions.list`, reconnect hydration | Snapshot | Replace the row; omission clears cached exact IDs to unavailable. | +| Web selected session | `chat.history.sessionInfo` | Snapshot | Replace the row projection; omission clears cached exact IDs. | +| Web session events | `sessions.changed`, `session.message`, lifecycle snapshots | Delta | Omission is inert; `null` clears; an array replaces. | +| Android session roster | `sessions.list`, reconnect hydration | Snapshot | Replace the list rows; omission clears cached exact IDs. | +| Android selected session | `chat.history.sessionInfo`, reconnect recovery | Snapshot | Replace `activeRunIds` even while other partial history fields merge. | +| Android session events | `sessions.changed`, `session.message`, lifecycle snapshots | Delta | Field presence controls replacement; `null` clears and omission is inert. | +| Apple session roster | `sessions.list`, reconnect hydration | Snapshot | Replace live rows; the offline cache strips transient active-run facts. | +| Apple selected session | `chat.history.sessionInfo`, reconnect recovery | Snapshot | Replace both the current row and its run-ID projection; omission clears both. | +| Apple session events | `sessions.changed`, `session.message`, lifecycle snapshots | Delta | Preserve field presence through decoding; omission is inert, `null` clears, and an array replaces. | + The outer event frame also has an optional `seq`, which orders events on the current WebSocket connection. It resets with a new connection. The `seq` inside an `agent` event payload is assigned per run and orders that run's lifecycle, diff --git a/docs/gateway/protocol.md b/docs/gateway/protocol.md index a22b075e4fc0..f173e0905d0a 100644 --- a/docs/gateway/protocol.md +++ b/docs/gateway/protocol.md @@ -643,7 +643,7 @@ methods. Treat this as feature discovery, not a full enumeration of - - `sessions.list` returns the current session index, including per-row `agentRuntime` metadata when an agent runtime backend is configured. `hasActiveRun` is the authoritative aggregate direct-session activity fact. When projected, `activeRunIds` contains known exact active run identities and may be empty while aggregate activity remains true. Omission means exact identities were not projected and is not a replacement value. Incremental merge events can send an empty array as the replacement tombstone that clears previously cached identities. Clients correlate only exact IDs they own locally or received from requests, history, or events and never select the first list entry as an owner. When cloud-worker placement is enabled or durable recovery state exists, session rows also include a closed `placement` state (`local`, `requested`, `provisioning`, `syncing`, `starting`, `active`, `draining`, `reconciling`, `reclaimed`, or `failed`) plus state-specific environment, owner-epoch, workspace, bundle, ACK-cursor, or recovery fields. Active placements may include an advisory `diskSpace` sample with `status` (`ok`, `warning`, or `critical`), `availableBytes`, `totalBytes`, and `observedAtMs`. Rows carry ownership projections — write-once `createdActor`, the mutable `owner` (actor plus `assignedBy`/`assignedAt`), a bounded `participants` list (owner excluded, up to 4 actors), and the full `participantCount`; actor display labels and avatars are resolved from current profiles and agent identities at read time. Pass `creatorId` to filter by immutable `createdActor.id`; pass `ownerId` to filter by the current assignable owner, falling back to `createdActor` when no owner is assigned. The complete `owners` facet is independent of pagination and remains unfiltered by either query, so clients can render the full owner picker. Authenticated callers can pass `involvingMe: true` to keep only sessions the caller owns or has prompted, evaluated against the full participant history (profile-backed human participants only). + - `sessions.list` returns the current session index, including per-row `agentRuntime` metadata when an agent runtime backend is configured. `hasActiveRun` is the authoritative aggregate direct-session activity fact. When projected, `activeRunIds` is the complete exact active set; an empty array proves the session is idle. If aggregate activity is true while the field is omitted, another runtime owner is active but its exact identities are unavailable. Snapshot omission means identities unavailable. On incremental events, omission means no change, `null` is the event-only tombstone that clears cached exact IDs to unavailable, and an array replaces the cache. Clients correlate only exact IDs they own locally or received from requests, history, or events and never select the first list entry as an owner. When cloud-worker placement is enabled or durable recovery state exists, session rows also include a closed `placement` state (`local`, `requested`, `provisioning`, `syncing`, `starting`, `active`, `draining`, `reconciling`, `reclaimed`, or `failed`) plus state-specific environment, owner-epoch, workspace, bundle, ACK-cursor, or recovery fields. Active placements may include an advisory `diskSpace` sample with `status` (`ok`, `warning`, or `critical`), `availableBytes`, `totalBytes`, and `observedAtMs`. Rows carry ownership projections — write-once `createdActor`, the mutable `owner` (actor plus `assignedBy`/`assignedAt`), a bounded `participants` list (owner excluded, up to 4 actors), and the full `participantCount`; actor display labels and avatars are resolved from current profiles and agent identities at read time. Pass `creatorId` to filter by immutable `createdActor.id`; pass `ownerId` to filter by the current assignable owner, falling back to `createdActor` when no owner is assigned. The complete `owners` facet is independent of pagination and remains unfiltered by either query, so clients can render the full owner picker. Authenticated callers can pass `involvingMe: true` to keep only sessions the caller owns or has prompted, evaluated against the full participant history (profile-backed human participants only). - `sessions.subscribe` enables session change events for the current WebSocket client. The subscription ends when that client disconnects. - `sessions.messages.subscribe` and `sessions.messages.unsubscribe` toggle transcript/message event subscriptions for one session. Pass `includeApprovals: true` to also receive sanitized `session.approval` lifecycle events for approvals whose persisted audience includes that exact session and whose reviewer binding authorizes the subscribing client. The subscribe response then includes a bounded pending `approvalReplay`; it is authoritative when `truncated` is false. The opt-in is per subscribe call, not sticky: re-subscribing to the same session without `includeApprovals: true` removes an existing approval subscription. In addition to normal session-read authority, this opt-in requires `operator.admin`, or `operator.approvals` on a paired device. - `sessions.preview` returns bounded transcript previews for specific session keys. @@ -661,7 +661,7 @@ methods. Treat this as feature discovery, not a full enumeration of - `sessions.assignOwner` (`operator.write`) reassigns the session's mutable owner to a person or configured agent (`{ key, owner: { type, id } }`). It requires an identified caller (authenticated profile or trusted agent identity), authorizes by session visibility, and records `assignedBy`/`assignedAt` on the row's `owner` field. The write-once `createdActor` and creator-anchored sharing authority are unchanged; see [Multi-user mode](/concepts/multi-user#assigning-an-owner). - `sessions.reset`, `sessions.delete`, and `sessions.compact` perform session maintenance. - `sessions.get` returns the full stored session row. - - Chat execution still uses `chat.history`, `chat.send`, `chat.abort`, and `chat.inject`. Its `sessionInfo` uses the same aggregate `hasActiveRun` and optional known-exact `activeRunIds` semantics as `sessions.list`. `chat.history` is display-normalized for UI clients: inline directive tags are stripped from visible text, plain-text tool-call XML payloads (`...`, `...`, `...`, `...`, and truncated tool-call blocks) and leaked ASCII/full-width model control tokens are stripped, pure silent-token assistant rows (exact `NO_REPLY` / `no_reply`) are omitted, and oversized rows can be replaced with placeholders. + - Chat execution still uses `chat.history`, `chat.send`, `chat.abort`, and `chat.inject`. Its `sessionInfo` uses the same aggregate `hasActiveRun` and optional complete-exact `activeRunIds` semantics as `sessions.list`. `chat.history` is display-normalized for UI clients: inline directive tags are stripped from visible text, plain-text tool-call XML payloads (`...`, `...`, `...`, `...`, and truncated tool-call blocks) and leaked ASCII/full-width model control tokens are stripped, pure silent-token assistant rows (exact `NO_REPLY` / `no_reply`) are omitted, and oversized rows can be replaced with placeholders. Tail responses can include an opaque `deltaCursor`. Pass it back as `cursor` to `chat.history` or `chat.startup` instead of `offset` or `messageId`. A successful catch-up returns `{ kind: "delta", messages, deltaCursor, sessionInfo }`; replay each `messages` entry through the same reducer as a live `session.message` payload. `{ kind: "reset" }` means the cursor is invalid, stale, belongs to another session, crossed a reset or compaction, or is too far behind; fetch a normal tail page. Catch-up never returns a partial page or continuation: more than 200 raw events or the 1 MB payload budget resets to a tail fetch. - `chat.message.get` is the additive bounded full-message reader for a single visible transcript entry. Pass `sessionKey`, optional `agentId` when session selection is agent-scoped, and a transcript `messageId` previously surfaced through `chat.history`; the gateway returns the same display-normalized projection without the lightweight history truncation cap when the stored entry is still available and not oversized. - `chat.toolTitles` returns short purpose titles for tool calls rendered in the Control UI (batched, max 24 items with bounded inputs). The feature is opt-in via `gateway.controlUi.toolTitles` (default off); disabled gateways answer `{ titles: {}, disabled: true }` with no model call so clients stop asking. When enabled, titles use standard utility-model routing: an explicitly configured `utilityModel` (an operator decision that, like all utility tasks, may send bounded task content to the chosen provider), else the session provider's declared small-model default so no new egress destination appears implicitly; an empty `utilityModel` disables them entirely. Titles never fall back to the primary model. Results cache in the per-agent state database keyed by tool name + input, so repeated views never re-bill the same calls. @@ -758,8 +758,9 @@ methods. Treat this as feature discovery, not a full enumeration of Clients show its headline or inspector link only while the digest's exact `runId` is present in `activeRunIds`. - `sessions.changed`: session index or metadata changed. Active-run fields use the - same aggregate and known-exact semantics as `sessions.list`; a present empty - `activeRunIds` replaces and clears the client's cached exact-identity list. + same aggregate and complete-exact semantics as `sessions.list`; `activeRunIds: null` + clears cached exact identities to unavailable, omission leaves the cache unchanged, + and an array replaces it. - `presence`: system presence snapshot updates. - `tick`: periodic keepalive/liveness event. - `health`: gateway health snapshot update. diff --git a/src/gateway/server-chat.agent-events.test.ts b/src/gateway/server-chat.agent-events.test.ts index 5381b01a6682..a5b6ec992c94 100644 --- a/src/gateway/server-chat.agent-events.test.ts +++ b/src/gateway/server-chat.agent-events.test.ts @@ -2271,6 +2271,50 @@ describe("agent event handler", () => { expect(requireRecord(persistEvent.data, "persist lifecycle event data").phase).toBe("end"); }); + it("tombstones exact run ids when lifecycle events expose only aggregate liveness", async () => { + vi.mocked(loadGatewaySessionRow).mockReturnValue({ + key: "session-projected", + kind: "direct", + sessionId: "session-id", + updatedAt: 1_000, + status: "running", + }); + const resolveSessionActiveRunState = vi + .fn>() + .mockReturnValue({ active: true }); + const { broadcastToConnIds, sessionEventSubscribers, handler } = createHarness({ + resolveSessionKeyForRun: () => "session-projected", + resolveSessionActiveRunState, + }); + sessionEventSubscribers.subscribe("conn-session"); + + emitAgentEvent( + handler, + "projected-run", + "lifecycle", + { phase: "start", startedAt: 1_000 }, + { sessionKey: "session-projected", sessionId: "session-id", ts: 1_000 }, + ); + + await waitForFast(() => { + expect( + broadcastToConnIds.mock.calls.filter(([event]) => event === "sessions.changed"), + ).toHaveLength(1); + }); + const payload = requireRecord( + broadcastToConnIds.mock.calls.find(([event]) => event === "sessions.changed")?.[1], + "sessions changed payload", + ); + expectRecordFields(payload, { + hasActiveRun: true, + activeRunIds: null, + }); + expectRecordFields(requireRecord(payload.session, "sessions changed session"), { + hasActiveRun: true, + activeRunIds: null, + }); + }); + it("publishes run lifecycle changes to plugins without websocket subscribers", async () => { const sessionKey = "agent:main:headless-run"; const received = vi.fn(); diff --git a/src/gateway/server-chat.ts b/src/gateway/server-chat.ts index 73358848ee39..ef23df6f4201 100644 --- a/src/gateway/server-chat.ts +++ b/src/gateway/server-chat.ts @@ -53,7 +53,10 @@ import type { ToolEventRecipientRegistry, } from "./server-chat-state.js"; import { hasSessionChangeReceivers } from "./session-change-receivers.js"; -import { buildGatewaySessionEventRow } from "./session-event-payload.js"; +import { + buildGatewaySessionEventRow, + projectSessionEventActiveRunIds, +} from "./session-event-payload.js"; import { deriveGatewaySessionLifecycleProjectionPatch, isRestartRecoveryLifecycleEvent, @@ -372,7 +375,7 @@ export type AgentEventHandlerOptions = { canonicalKey: string; sessionId?: string; agentId?: string; - }) => { active: boolean; runIds: string[] }; + }) => { active: boolean; runIds?: string[] }; }; type AgentEventHandler = ((event: AgentEventPayload) => void) & { @@ -602,10 +605,13 @@ export function createAgentEventHandler({ ...(agentId ? { agentId } : {}), }) : undefined; - // Agent lifecycle broadcasts merge into cached session rows in the UI. - // Always replace run identity so a newer start cannot inherit a completed run. + // Agent lifecycle broadcasts merge into cached session rows in the UI. Replace + // run identities only when the Gateway owns the complete exact set. const activeRunFields = activeRunState - ? { hasActiveRun: activeRunState.active, activeRunIds: activeRunState.runIds } + ? { + hasActiveRun: activeRunState.active, + activeRunIds: projectSessionEventActiveRunIds(activeRunState), + } : {}; const clearsLastRunError = Object.hasOwn(lifecyclePatch, "lastRunError") && lifecyclePatch.lastRunError === undefined; diff --git a/src/gateway/server-methods/chat-history-handler.ts b/src/gateway/server-methods/chat-history-handler.ts index 1ec45de3b954..27f17e012472 100644 --- a/src/gateway/server-methods/chat-history-handler.ts +++ b/src/gateway/server-methods/chat-history-handler.ts @@ -502,7 +502,9 @@ async function handleChatHistoryRequest({ defaultAgentId: compatibilityOwnerAgentId, }); sessionInfo.hasActiveRun = activeRunState.active; - sessionInfo.activeRunIds = activeRunState.runIds; + if (activeRunState.runIds !== undefined) { + sessionInfo.activeRunIds = activeRunState.runIds; + } if (activeRunState.active) { sessionInfo.status = activeRunState.status ?? "running"; } diff --git a/src/gateway/server-methods/session-active-runs.test.ts b/src/gateway/server-methods/session-active-runs.test.ts index 296913be2043..d840d4b15b99 100644 --- a/src/gateway/server-methods/session-active-runs.test.ts +++ b/src/gateway/server-methods/session-active-runs.test.ts @@ -199,7 +199,7 @@ it("projects a lifecycle-owned worker run without widening event visibility", () canonicalKey: "agent:main:worker", sessionId: "worker-session", }), - ).toEqual({ active: true, runIds: [] }); + ).toEqual({ active: true }); } finally { clearAgentRunContext("worker-run"); } @@ -224,7 +224,7 @@ it("projects reply lifecycle state without hiding independent embedded work", () canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [], status: "queued" }); + ).toEqual({ active: true, status: "queued" }); operation.markWaitingForGlobalLane(); expect( @@ -234,7 +234,7 @@ it("projects reply lifecycle state without hiding independent embedded work", () canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [], status: "queued" }); + ).toEqual({ active: true, status: "queued" }); operation.markGlobalLaneWaitEnded(); operation.setPhase("running"); @@ -246,7 +246,7 @@ it("projects reply lifecycle state without hiding independent embedded work", () canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [], status: "queued" }); + ).toEqual({ active: true, status: "queued" }); operation.markGlobalLaneWaitEnded(); markReplyOperationExecutionStarted(operation); expect( @@ -256,7 +256,7 @@ it("projects reply lifecycle state without hiding independent embedded work", () canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [] }); + ).toEqual({ active: true }); operation.markWaitingForGlobalLane(); expect( resolveVisibleActiveSessionRunState({ @@ -265,7 +265,7 @@ it("projects reply lifecycle state without hiding independent embedded work", () canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [] }); + ).toEqual({ active: true }); operation.markGlobalLaneWaitEnded(); expect(operation.abortByUser()).toBe(true); expect(isEmbeddedAgentRunActive(sessionId)).toBe(true); @@ -286,7 +286,7 @@ it("projects reply lifecycle state without hiding independent embedded work", () canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [] }); + ).toEqual({ active: true }); } finally { clearActiveEmbeddedRun(sessionId, replacementHandle, sessionKey); operation.complete(); @@ -310,7 +310,7 @@ it("preserves an independent lifecycle-owned worker while a reply operation sett canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [] }); + ).toEqual({ active: true }); } finally { operation.complete(); clearAgentRunContext("worker-overlap-run"); @@ -342,7 +342,7 @@ it("does not project an aborted embedded handle retained for cleanup as active", canonicalKey: sessionKey, sessionId, }), - ).toEqual({ active: true, runIds: [] }); + ).toEqual({ active: true }); expect(abortEmbeddedAgentRun(sessionId)).toBe(true); expect(isEmbeddedAgentRunActive(sessionId)).toBe(true); diff --git a/src/gateway/server-methods/session-active-runs.ts b/src/gateway/server-methods/session-active-runs.ts index 65f8c867b430..d2b3241bd3e2 100644 --- a/src/gateway/server-methods/session-active-runs.ts +++ b/src/gateway/server-methods/session-active-runs.ts @@ -16,6 +16,13 @@ type TrackedActiveSessionRun = { executionStarted: boolean; }; +type VisibleActiveSessionRunState = { + active: boolean; + /** Complete exact active set. Omitted when another active owner exposes only liveness. */ + runIds?: string[]; + status?: "queued"; +}; + export function collectTrackedActiveSessionRuns( context: Partial>, ): TrackedActiveSessionRun[] { @@ -149,7 +156,7 @@ export function resolveVisibleActiveSessionRunState(params: { defaultAgentId?: string; trackedActiveRuns?: readonly TrackedActiveSessionRun[]; projectedAgentRunIndex?: ProjectedAgentRunIndex; -}): { active: boolean; runIds: string[]; status?: "queued" } { +}): VisibleActiveSessionRunState { const sessionId = params.sessionId?.trim(); const resolvedAgentId = params.agentId ?? @@ -196,5 +203,10 @@ export function resolveVisibleActiveSessionRunState(params: { hasProjectedRun || embeddedRunState === "running"; const active = running || matchingTrackedRuns.length > 0 || embeddedRunState === "queued"; - return { active, runIds, ...(active && !running ? { status: "queued" as const } : {}) }; + const identitiesComplete = !hasProjectedRun && embeddedRunState === undefined; + return { + active, + ...(identitiesComplete ? { runIds } : {}), + ...(active && !running ? { status: "queued" as const } : {}), + }; } diff --git a/src/gateway/server-methods/session-change-event.test.ts b/src/gateway/server-methods/session-change-event.test.ts index b0c1288b9aea..1572417f4279 100644 --- a/src/gateway/server-methods/session-change-event.test.ts +++ b/src/gateway/server-methods/session-change-event.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { retainLegacyDefaultAgentId } from "../../config/legacy.default-agent-owner.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { clearAgentRunContext, registerAgentRunContext } from "../../infra/agent-run-registry.js"; import type { ChatAbortControllerEntry } from "../chat-abort.js"; import type { GatewayRequestContext } from "./types.js"; @@ -27,17 +28,26 @@ vi.mock("../session-utils.js", async (importOriginal) => { }; }); -vi.mock("../session-event-payload.js", () => ({ - buildGatewaySessionEventFields: ({ - sessionRow, - hasActiveRun, - activeRunIds, - }: { - sessionRow: { key: string; label: string }; - hasActiveRun?: boolean; - activeRunIds?: string[]; - }) => ({ key: sessionRow.key, label: sessionRow.label, hasActiveRun, activeRunIds }), -})); +vi.mock("../session-event-payload.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + buildGatewaySessionEventFields: ({ + sessionRow, + hasActiveRun, + activeRunIds, + }: { + sessionRow: { key: string; label: string }; + hasActiveRun?: boolean; + activeRunIds?: string[] | null; + }) => ({ + key: sessionRow.key, + label: sessionRow.label, + ...(hasActiveRun === undefined ? {} : { hasActiveRun }), + ...(activeRunIds === undefined ? {} : { activeRunIds }), + }), + }; +}); const { emitSessionsChanged, flushPendingSessionsChangedEvents, readSessionsMutationVersion } = await import("./session-change-event.js"); @@ -202,6 +212,48 @@ describe("sessions.changed coalescing", () => { ); }); + it("tombstones exact run ids when lifecycle projection takes ownership", () => { + const sessionKey = "agent:main:projected"; + const sessionId = `${sessionKey}-id`; + const chatAbortControllers = new Map([ + [ + "direct-run", + { + agentId: "main", + controller: new AbortController(), + expiresAtMs: 60_000, + sessionId, + sessionKey, + startedAtMs: 0, + } satisfies ChatAbortControllerEntry, + ], + ]); + const context = createContext(new Set(["conn-1"]), {}, chatAbortControllers); + + emitSessionsChanged(context, { reason: "update", sessionKey }); + expect(vi.mocked(context.broadcastToConnIds).mock.calls[0]?.[1]).toMatchObject({ + hasActiveRun: true, + activeRunIds: ["direct-run"], + }); + + chatAbortControllers.clear(); + registerAgentRunContext("hidden-worker-run", { + isControlUiVisible: false, + projectSessionActive: true, + sessionKey, + }); + try { + emitSessionsChanged(context, { reason: "update", sessionKey }); + flushPendingSessionsChangedEvents(context); + + const payload = vi.mocked(context.broadcastToConnIds).mock.calls[1]?.[1]; + expect(payload).toMatchObject({ hasActiveRun: true }); + expect(payload).toHaveProperty("activeRunIds", null); + } finally { + clearAgentRunContext("hidden-worker-run"); + } + }); + it("advances the mutation fence without loading rows when nobody receives events", () => { const context = createContext(new Set()); const initialVersion = readSessionsMutationVersion(context); diff --git a/src/gateway/server-methods/session-change-event.ts b/src/gateway/server-methods/session-change-event.ts index 3fbd20d9205d..8a50f415bdfd 100644 --- a/src/gateway/server-methods/session-change-event.ts +++ b/src/gateway/server-methods/session-change-event.ts @@ -1,7 +1,10 @@ // Shared sessions.changed broadcaster for gateway RPC and chat-command mutations. import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js"; import { hasSessionChangeReceivers } from "../session-change-receivers.js"; -import { buildGatewaySessionEventFields } from "../session-event-payload.js"; +import { + buildGatewaySessionEventFields, + projectSessionEventActiveRunIds, +} from "../session-event-payload.js"; import { tryResolveSessionCompatibilityOwnerAgentId } from "../session-request-agent.js"; import { invalidateSessionSharingSnapshot } from "../session-sharing.js"; import { loadGatewaySessionRow } from "../session-utils.js"; @@ -98,7 +101,7 @@ function broadcastSessionsChanged( agentId: effectiveAgentId, status: activeRunState?.active ? (activeRunState.status ?? "running") : undefined, hasActiveRun: activeRunState?.active, - activeRunIds: activeRunState?.runIds, + activeRunIds: projectSessionEventActiveRunIds(activeRunState), }), effectiveFastMode: sessionRow.effectiveFastMode, effectiveFastModeSource: sessionRow.effectiveFastModeSource, diff --git a/src/gateway/server-methods/sessions-read.ts b/src/gateway/server-methods/sessions-read.ts index a0f19e9d5def..28d4ac51bd20 100644 --- a/src/gateway/server-methods/sessions-read.ts +++ b/src/gateway/server-methods/sessions-read.ts @@ -405,7 +405,7 @@ export const sessionReadHandlers: GatewayRequestHandlers = { ? { status: activeRunState.status ?? ("running" as const) } : {}), ...projectPlacement(session.sessionId), - ...(activeRunState.runIds.length > 0 + ...(activeRunState.runIds !== undefined ? { activeRunIds: activeRunState.runIds } : {}), }); diff --git a/src/gateway/server-session-events.test.ts b/src/gateway/server-session-events.test.ts index 6f5fc8224df6..ff1e6b3a1177 100644 --- a/src/gateway/server-session-events.test.ts +++ b/src/gateway/server-session-events.test.ts @@ -483,7 +483,13 @@ describe("createTranscriptUpdateBroadcastHandler", () => { await expect(emitAssistantTranscriptUpdate(false)).resolves.toMatchObject({ sessionKey: "agent:main:main", hasActiveRun: true, - session: { key: "agent:main:main", sessionId: "sess-main", hasActiveRun: true }, + activeRunIds: null, + session: { + key: "agent:main:main", + sessionId: "sess-main", + hasActiveRun: true, + activeRunIds: null, + }, }); expect(resolveEmbeddedAgentRunProgressStateMock).toHaveBeenCalledWith("sess-main"); }); diff --git a/src/gateway/server-session-events.ts b/src/gateway/server-session-events.ts index 163300443ed5..8b9adf7ac589 100644 --- a/src/gateway/server-session-events.ts +++ b/src/gateway/server-session-events.ts @@ -27,6 +27,7 @@ import { hasSessionChangeReceivers } from "./session-change-receivers.js"; import { buildGatewaySessionEventFields, buildGatewaySessionEventRow, + projectSessionEventActiveRunIds, } from "./session-event-payload.js"; import { resolveSessionSubscriptionKeys } from "./session-subscription-keys.js"; import { projectSessionMessagePayload } from "./session-transcript-message.js"; @@ -83,14 +84,14 @@ export function buildGatewaySessionSnapshot(params: { parentSessionKey?: string; status?: GatewaySessionRow["status"]; hasActiveRun?: boolean; - activeRunIds?: string[]; + activeRunIds?: string[] | null; }): Record { const { sessionRow } = params; if (!sessionRow) { return {}; } // Nested snapshots are the UI merge source, so preserve explicit clear semantics there too. - const session = params.includeSession + const session: Record | undefined = params.includeSession ? { ...buildGatewaySessionEventRow(sessionRow), createdActor: sessionRow.createdActor ?? null, @@ -344,7 +345,7 @@ async function handleTranscriptUpdateBroadcast( includeSession: true, status: activeRunState?.active ? (activeRunState.status ?? "running") : undefined, hasActiveRun: activeRunState?.active, - activeRunIds: activeRunState?.runIds, + activeRunIds: projectSessionEventActiveRunIds(activeRunState), }); if (update.message === undefined) { // A committed batch without individually proven cursors must invalidate @@ -454,7 +455,7 @@ export function createLifecycleEventBroadcastHandler(params: { displayName: event.displayName, parentSessionKey: event.parentSessionKey, hasActiveRun: activeRunState?.active, - activeRunIds: activeRunState?.runIds, + activeRunIds: projectSessionEventActiveRunIds(activeRunState), }), ...(swarmEvent.swarmGroupId ? { diff --git a/src/gateway/server.sessions.list-changed.test.ts b/src/gateway/server.sessions.list-changed.test.ts index add54d23ef60..89d49d9c0a43 100644 --- a/src/gateway/server.sessions.list-changed.test.ts +++ b/src/gateway/server.sessions.list-changed.test.ts @@ -260,7 +260,7 @@ async function expectListedSessionActiveRun( const payload = expectRespondPayload(respond); const session = findSession(payload, "agent:main:main"); expect(session.hasActiveRun).toBe(expected); - expect(session.activeRunIds).toEqual(expected ? ["run-1"] : undefined); + expect(session.activeRunIds).toEqual(expected ? ["run-1"] : []); expect(session.status).toBe(expectedStatus); } @@ -743,6 +743,25 @@ test("sessions.list replaces a previous terminal status when execution starts", ); }); +test("sessions.list distinguishes proven idle from unavailable run identities", async () => { + await writeMainSessionStore(); + + const idle = await invokeSessionsList({ requestId: "req-sessions-list-idle-exact-runs" }); + const idleSession = findSession(expectRespondPayload(idle.respond), "agent:main:main"); + expect(idleSession).toMatchObject({ hasActiveRun: false, activeRunIds: [] }); + + embeddedRunMock.activeIds.add("sess-main"); + const unavailable = await invokeSessionsList({ + requestId: "req-sessions-list-unavailable-runs", + }); + const unavailableSession = findSession( + expectRespondPayload(unavailable.respond), + "agent:main:main", + ); + expect(unavailableSession).toMatchObject({ hasActiveRun: true }); + expect(unavailableSession).not.toHaveProperty("activeRunIds"); +}); + test("sessions.changed publishes visible active run ids", async () => { await writeMainSessionStore(); const result = await invokeSessionMutation({ diff --git a/src/gateway/session-event-payload.ts b/src/gateway/session-event-payload.ts index 46ffec4e315d..861af6b2e504 100644 --- a/src/gateway/session-event-payload.ts +++ b/src/gateway/session-event-payload.ts @@ -28,6 +28,13 @@ export function buildGatewaySessionEventRow( return session; } +/** Incremental events clear cached exact IDs when the current owner exposes only liveness. */ +export function projectSessionEventActiveRunIds( + state: { runIds?: string[] } | null | undefined, +): string[] | null | undefined { + return state ? (state.runIds ?? null) : undefined; +} + export function buildGatewaySessionEventFields(params: { sessionRow: GatewaySessionRow; agentId?: string; @@ -36,7 +43,7 @@ export function buildGatewaySessionEventFields(params: { parentSessionKey?: string; status?: GatewaySessionRow["status"]; hasActiveRun?: boolean; - activeRunIds?: string[]; + activeRunIds?: string[] | null; }): Record { const { sessionRow } = params; const omitUnscopedGlobalGoal = sessionRow.key === "global" && !params.agentId; diff --git a/src/gateway/session-utils.types.ts b/src/gateway/session-utils.types.ts index 6959d6b30641..79384e12c39e 100644 --- a/src/gateway/session-utils.types.ts +++ b/src/gateway/session-utils.types.ts @@ -157,6 +157,7 @@ export type GatewaySessionRow = { /** Compact user-facing reason for the latest failed or timed-out run. */ lastRunError?: string; hasActiveRun?: boolean; + /** Complete exact active set when present; omitted for active owners without exact identities. */ activeRunIds?: string[]; /** Active transcript-branch leaf for history rendered from this row. */ activeLeafEntryId?: string | null; diff --git a/ui/src/lib/sessions/reconcile.test.ts b/ui/src/lib/sessions/reconcile.test.ts index 6da00c8a0fb8..37c7b6dcffde 100644 --- a/ui/src/lib/sessions/reconcile.test.ts +++ b/ui/src/lib/sessions/reconcile.test.ts @@ -163,6 +163,64 @@ test("sessions.changed deletes every null-tombstoned field, not a hand-kept list expect(row?.updatedAt).toBe(2); }); +test("sessions.changed clears exact run ids only for an explicit tombstone", () => { + const key = "agent:main:main"; + const result = buildResult([ + { + key, + kind: "direct", + updatedAt: 1, + hasActiveRun: true, + activeRunIds: ["run-exact"], + }, + ]); + + const omitted = reconcileSessionChanged(result, { + sessionKey: key, + reason: "run-progress", + updatedAt: 2, + hasActiveRun: true, + }); + expect(omitted.row?.activeRunIds).toEqual(["run-exact"]); + + const tombstoned = reconcileSessionChanged(omitted.result, { + sessionKey: key, + reason: "run-progress", + updatedAt: 3, + hasActiveRun: true, + activeRunIds: null, + }); + expect(tombstoned.row?.activeRunIds).toBeUndefined(); +}); + +test("authoritative snapshot omission clears cached exact run ids", () => { + const key = "agent:main:main"; + const result = buildResult([ + { + key, + kind: "direct", + sessionId: "session-main", + updatedAt: 1, + hasActiveRun: true, + activeRunIds: ["run-stale"], + }, + ]); + + const reconciled = reconcileSessionHistory( + result, + { + key, + kind: "direct", + sessionId: "session-main", + updatedAt: 2, + hasActiveRun: true, + }, + undefined, + ); + + expect(reconciled?.sessions[0]?.activeRunIds).toBeUndefined(); +}); + test("sessions.changed invalidates the complete owner facet until canonical refresh", () => { const key = "agent:main:main"; const result = buildResult([