ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-25 11:55:47 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): make activeRunIds presence mean a complete exact run set (#126106)

Browse Source 
* 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
This commit is contained in:
Peter Steinberger
2026-08-19 02:08:57 -07:00
committed by GitHub
parent c695b70a57
commit e71fc902ee
29 changed files with 607 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/android/app/src/main/java/ai/openclaw/app/chat/ChatController.kt
+13 -1
View File
@@ -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<String> = 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,
apps/android/app/src/main/java/ai/openclaw/app/chat/ChatModels.kt
+1
View File
@@ -332,6 +332,7 @@ data class ChatSessionEntry(
val hasActiveRun: Boolean? = null,
val activeRunIds: List<String>? = 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,
apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerReconnectRestoreTest.kt
+23
View File
@@ -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<String>()
gateway.respond("sessions.list") { pendingSessionList.await() }
reconnect(controller)
assertEquals(1, controller.selectedActiveRunPresentation.value.count)
assertNull(controller.selectedActiveRunPresentation.value.runId)
}
@Test
fun reconnectStaysUnhealthyUntilRecoveryHistoryApplies() =
runTest {
apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerUsageStreamTest.kt
+23
View File
@@ -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() =
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatGatewayPayloadCodec.swift
+4 -1
View File
@@ -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":
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatModels.swift
+59 -1
View File
@@ -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<T: Decodable>(_ 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
}
}
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatSessionSidebarModel.swift
+2 -2
View File
@@ -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
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTransport.swift
+13 -3
View File
@@ -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)
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+RunSnapshot.swift
+11 -4
View File
@@ -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),
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+SessionKeys.swift
+1 -5
View File
@@ -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 {
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+TransportEvents.swift
+26 -15
View File
@@ -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.
apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatSessionSidebarModelTests.swift
+26
View File
@@ -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",
apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelTests.swift
+125 -1
View File
@@ -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",
docs/gateway/clients.md
+26 -7
View File
@@ -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,
docs/gateway/protocol.md
+5 -4
View File
@@ -643,7 +643,7 @@ methods. Treat this as feature discovery, not a full enumeration of
</Accordion>
<Accordion title="Session control">
- `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 (`<tool_call>...</tool_call>`, `<function_call>...</function_call>`, `<tool_calls>...</tool_calls>`, `<function_calls>...</function_calls>`, 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 (`<tool_call>...</tool_call>`, `<function_call>...</function_call>`, `<tool_calls>...</tool_calls>`, `<function_calls>...</function_calls>`, 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.
src/gateway/server-chat.agent-events.test.ts
+44
View File
@@ -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<NonNullable<AgentEventHandlerOptions["resolveSessionActiveRunState"]>>()
.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();
src/gateway/server-chat.ts
+11 -5
View File
@@ -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;
src/gateway/server-methods/chat-history-handler.ts
+3 -1
View File
@@ -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";
}
src/gateway/server-methods/session-active-runs.test.ts
+9 -9
View File
@@ -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);
src/gateway/server-methods/session-active-runs.ts
+14 -2
View File
@@ -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<Pick<GatewayRequestContext, "chatAbortControllers">>,
): 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 } : {}),
};
}
src/gateway/server-methods/session-change-event.test.ts
+63 -11
View File
@@ -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<typeof import("../session-event-payload.js")>();
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);
src/gateway/server-methods/session-change-event.ts
+5 -2
View File
@@ -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,
src/gateway/server-methods/sessions-read.ts
+1 -1
View File
@@ -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 }
: {}),
});
src/gateway/server-session-events.test.ts
+7 -1
View File
@@ -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");
});
src/gateway/server-session-events.ts
+5 -4
View File
@@ -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<string, unknown> {
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<string, unknown> | 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
? {
src/gateway/server.sessions.list-changed.test.ts
+20 -1
View File
@@ -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({
src/gateway/session-event-payload.ts
+8 -1
View File
@@ -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<string, unknown> {
const { sessionRow } = params;
const omitUnscopedGlobalGoal = sessionRow.key === "global" && !params.agentId;
src/gateway/session-utils.types.ts
+1
View File
@@ -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;
ui/src/lib/sessions/reconcile.test.ts
+58
View File
@@ -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([