furyhawk/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-21 10:16:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(apps): lease expiry demotes scope to needs-reconciliation; document send-race tradeoff

Browse Source
This commit is contained in:
Peter Steinberger
2026-07-19 21:04:59 -07:00
parent 1287c293e8
commit d2d6be04b2
5 changed files with 240 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTranscriptCache.swift
+102 -25
View File
@@ -167,6 +167,7 @@ public actor OpenClawChatSQLiteTranscriptCache: OpenClawChatTranscriptCache,
branch_epoch INTEGER NOT NULL DEFAULT 0,
last_active_leaf_id TEXT,
switch_pending_since REAL,
needs_reconciliation INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY(gateway_id, session_key, agent_id)
)
"""
@@ -619,13 +620,19 @@ extension OpenClawChatSQLiteTranscriptCache {
guard !self.isRetired, let db = await handle() else { return nil }
guard self.execute(db, sql: "BEGIN IMMEDIATE", bindings: []) else { return nil }
var committed = false
var invalidatedScopes: [OpenClawChatOutboxScope] = []
defer {
if !committed {
_ = self.execute(db, sql: "ROLLBACK", bindings: [])
} else {
for scope in invalidatedScopes {
self.emitOutboxChange(.invalidated(scope: scope))
}
}
}
guard self.applyOutboxStaleness(db) else { return nil }
guard self.clearExpiredBranchSwitches(db) else { return nil }
guard let expiredScopes = self.expireBranchSwitchLeases(db) else { return nil }
invalidatedScopes = expiredScopes
guard let activeClaimCount = selectInt(
db,
sql: "SELECT COUNT(*) FROM outbox_commands WHERE gateway_id = ?1 AND status = 'sending'",
@@ -645,6 +652,7 @@ extension OpenClawChatSQLiteTranscriptCache {
AND s.agent_id = c.agent_id
WHERE c.gateway_id = ?1 AND c.status = 'queued'
AND s.switch_pending_since IS NULL
AND COALESCE(s.needs_reconciliation, 0) = 0
ORDER BY c.created_at ASC, c.id ASC LIMIT 1
""",
bindings: [self.gatewayID]),
@@ -760,7 +768,8 @@ extension OpenClawChatSQLiteTranscriptCache {
epoch: state.epoch,
lastActiveLeafEntryID: state.lastActiveLeafEntryID,
hadPendingCommands: pendingCount > 0,
switchPendingSince: state.switchPendingSince)
switchPendingSince: state.switchPendingSince,
needsReconciliation: state.needsReconciliation)
}
public func beginBranchSwitch(_ scope: OpenClawChatOutboxScope) async -> Bool {
@@ -773,9 +782,20 @@ extension OpenClawChatSQLiteTranscriptCache {
if !committed { _ = self.execute(db, sql: "ROLLBACK", bindings: []) }
}
guard self.ensureBranchScopeRow(db, scope: scope),
self.clearExpiredBranchSwitches(db, scope: scope),
let state = self.readBranchState(db, scope: scope),
let expiredScopes = self.expireBranchSwitchLeases(db, scope: scope)
else { return false }
if !expiredScopes.isEmpty {
committed = self.execute(db, sql: "COMMIT", bindings: [])
if committed {
for scope in expiredScopes {
self.emitOutboxChange(.invalidated(scope: scope))
}
}
return false
}
guard let state = self.readBranchState(db, scope: scope),
state.switchPendingSince == nil,
!state.needsReconciliation,
self.unconfirmedCommandCount(db, scope: scope) == 0,
self.execute(
db,
@@ -841,13 +861,16 @@ extension OpenClawChatSQLiteTranscriptCache {
var state = self.readBranchState(db, scope: scope),
state.epoch == previousState.epoch
else { return nil }
guard self.clearExpiredBranchSwitches(db, scope: scope),
guard self.expireBranchSwitchLeases(db, scope: scope) != nil,
let refreshedState = self.readBranchState(db, scope: scope),
let currentUnconfirmedCommandCount = self.unconfirmedCommandCount(db, scope: scope)
else { return nil }
state = refreshedState
guard state.switchPendingSince == nil else { return nil }
// A cross-client switch can win after this reconciliation but before send reaches the gateway.
// Sends, like the web client, carry no branch precondition and land on the active branch at arrival.
// Close this only with a protocol-level expectedActiveLeaf precondition.
if let activeLeafEntryID,
let lastLeaf = previousState.lastActiveLeafEntryID,
lastLeaf != activeLeafEntryID,
@@ -943,6 +966,7 @@ extension OpenClawChatSQLiteTranscriptCache {
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
AND branch_epoch = ?5
AND switch_pending_since IS NULL
AND needs_reconciliation = 0
AND NOT EXISTS (
SELECT 1 FROM outbox_commands c
WHERE c.gateway_id = ?1 AND c.session_key = ?2 AND c.agent_id = ?3
@@ -976,28 +1000,58 @@ extension OpenClawChatSQLiteTranscriptCache {
bindings: self.scopeBindings(scope))
}
private func clearExpiredBranchSwitches(
private func expireBranchSwitchLeases(
_ db: OpaquePointer,
scope: OpenClawChatOutboxScope? = nil) -> Bool
scope: OpenClawChatOutboxScope? = nil) -> [OpenClawChatOutboxScope]?
{
let cutoff = Date().timeIntervalSince1970 - Self.outboxBranchSwitchLeaseMaxAge
if let scope {
return self.execute(
guard self.execute(
db,
sql: """
UPDATE outbox_branch_scopes SET switch_pending_since = NULL
UPDATE outbox_branch_scopes
SET switch_pending_since = NULL, needs_reconciliation = 1
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
AND switch_pending_since <= ?4
""",
bindings: self.scopeBindings(scope) + [cutoff])
else { return nil }
return sqlite3_changes(db) > 0 ? [scope] : []
}
return self.execute(
db,
sql: """
UPDATE outbox_branch_scopes SET switch_pending_since = NULL
WHERE gateway_id = ?1 AND switch_pending_since <= ?2
""",
bindings: [self.gatewayID, cutoff])
guard let scopes = self.selectExpiredBranchSwitchScopes(db, cutoff: cutoff),
self.execute(
db,
sql: """
UPDATE outbox_branch_scopes
SET switch_pending_since = NULL, needs_reconciliation = 1
WHERE gateway_id = ?1 AND switch_pending_since <= ?2
""",
bindings: [self.gatewayID, cutoff])
else { return nil }
return scopes
}
private func selectExpiredBranchSwitchScopes(
_ db: OpaquePointer,
cutoff: TimeInterval) -> [OpenClawChatOutboxScope]?
{
var statement: OpaquePointer?
let sql = """
SELECT session_key, agent_id FROM outbox_branch_scopes
WHERE gateway_id = ?1 AND switch_pending_since <= ?2
"""
guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return nil }
defer { sqlite3_finalize(statement) }
guard self.bind(statement, bindings: [self.gatewayID, cutoff]) else { return nil }
var scopes: [OpenClawChatOutboxScope] = []
while sqlite3_step(statement) == SQLITE_ROW {
guard let sessionKey = sqlite3_column_text(statement, 0) else { return nil }
let agentID = sqlite3_column_text(statement, 1).map { String(cString: $0) }
scopes.append(OpenClawChatOutboxScope(
sessionKey: String(cString: sessionKey),
agentID: agentID))
}
return scopes
}
private func ensureBranchScopeRow(_ db: OpaquePointer, scope: OpenClawChatOutboxScope) -> Bool {
@@ -1005,8 +1059,8 @@ extension OpenClawChatSQLiteTranscriptCache {
db,
sql: """
INSERT OR IGNORE INTO outbox_branch_scopes(
gateway_id, session_key, agent_id, branch_epoch, last_active_leaf_id
) VALUES (?1, ?2, ?3, 0, NULL)
gateway_id, session_key, agent_id, branch_epoch, last_active_leaf_id, needs_reconciliation
) VALUES (?1, ?2, ?3, 0, NULL, 0)
""",
bindings: self.scopeBindings(scope))
}
@@ -1017,7 +1071,8 @@ extension OpenClawChatSQLiteTranscriptCache {
{
var statement: OpaquePointer?
let sql = """
SELECT branch_epoch, last_active_leaf_id, switch_pending_since FROM outbox_branch_scopes
SELECT branch_epoch, last_active_leaf_id, switch_pending_since, needs_reconciliation
FROM outbox_branch_scopes
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
"""
guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return nil }
@@ -1030,14 +1085,16 @@ extension OpenClawChatSQLiteTranscriptCache {
lastActiveLeafEntryID: sqlite3_column_text(statement, 1).map { String(cString: $0) },
switchPendingSince: sqlite3_column_type(statement, 2) == SQLITE_NULL
? nil
: sqlite3_column_double(statement, 2))
: sqlite3_column_double(statement, 2),
needsReconciliation: sqlite3_column_int(statement, 3) != 0)
}
private func clearBranchSwitch(_ db: OpaquePointer, scope: OpenClawChatOutboxScope) -> Bool {
self.execute(
db,
sql: """
UPDATE outbox_branch_scopes SET switch_pending_since = NULL
UPDATE outbox_branch_scopes
SET switch_pending_since = NULL, needs_reconciliation = 0
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
""",
bindings: self.scopeBindings(scope))
@@ -1053,7 +1110,8 @@ extension OpenClawChatSQLiteTranscriptCache {
return self.execute(
db,
sql: """
UPDATE outbox_branch_scopes SET branch_epoch = ?4, last_active_leaf_id = NULL
UPDATE outbox_branch_scopes
SET branch_epoch = ?4, last_active_leaf_id = NULL, needs_reconciliation = 0
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
""",
bindings: self.scopeBindings(scope) + [epoch])
@@ -1061,7 +1119,8 @@ extension OpenClawChatSQLiteTranscriptCache {
return self.execute(
db,
sql: """
UPDATE outbox_branch_scopes SET branch_epoch = ?4, last_active_leaf_id = ?5
UPDATE outbox_branch_scopes
SET branch_epoch = ?4, last_active_leaf_id = ?5, needs_reconciliation = 0
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
""",
bindings: self.scopeBindings(scope) + [epoch, lastActiveLeafEntryID])
@@ -1079,7 +1138,8 @@ extension OpenClawChatSQLiteTranscriptCache {
db,
sql: """
UPDATE outbox_branch_scopes
SET branch_epoch = ?4, last_active_leaf_id = ?5, switch_pending_since = NULL
SET branch_epoch = ?4, last_active_leaf_id = ?5,
switch_pending_since = NULL, needs_reconciliation = 0
WHERE gateway_id = ?1 AND session_key = ?2 AND agent_id = ?3
""",
bindings: self.scopeBindings(scope) + [nextEpoch, activeLeafEntryID])
@@ -1695,7 +1755,8 @@ extension OpenClawChatSQLiteTranscriptCache {
self.ensureOutboxBranchEpochColumn(opened),
self.ensureOutboxParkedWasAcceptedColumn(opened),
sqlite3_exec(opened, Self.createOutboxBranchScopeTableSQL, nil, nil, nil) == SQLITE_OK,
self.ensureOutboxSwitchPendingSinceColumn(opened)
self.ensureOutboxSwitchPendingSinceColumn(opened),
self.ensureOutboxNeedsReconciliationColumn(opened)
else {
sqlite3_close_v2(opened)
return nil
@@ -1969,6 +2030,22 @@ extension OpenClawChatSQLiteTranscriptCache {
return self.table(db, hasColumn: "switch_pending_since", tableName: "outbox_branch_scopes")
}
private func ensureOutboxNeedsReconciliationColumn(_ db: OpaquePointer) -> Bool {
if self.table(db, hasColumn: "needs_reconciliation", tableName: "outbox_branch_scopes") {
return true
}
if sqlite3_exec(
db,
"ALTER TABLE outbox_branch_scopes ADD COLUMN needs_reconciliation INTEGER NOT NULL DEFAULT 0",
nil,
nil,
nil) == SQLITE_OK
{
return true
}
return self.table(db, hasColumn: "needs_reconciliation", tableName: "outbox_branch_scopes")
}
private func migrateTranscriptTableToV3(_ db: OpaquePointer) -> Bool {
let hadAgentID = self.table(db, hasColumn: "agent_id", tableName: "cached_transcripts")
guard sqlite3_exec(
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatTranscriptCacheContracts.swift
+4 -1
View File
@@ -97,17 +97,20 @@ public struct OpenClawChatOutboxBranchState: Equatable, Sendable {
public let lastActiveLeafEntryID: String?
public let hadPendingCommands: Bool
public let switchPendingSince: TimeInterval?
public let needsReconciliation: Bool
public init(
epoch: Int,
lastActiveLeafEntryID: String?,
hadPendingCommands: Bool = false,
switchPendingSince: TimeInterval? = nil)
switchPendingSince: TimeInterval? = nil,
needsReconciliation: Bool = false)
{
self.epoch = epoch
self.lastActiveLeafEntryID = lastActiveLeafEntryID
self.hadPendingCommands = hadPendingCommands
self.switchPendingSince = switchPendingSince
self.needsReconciliation = needsReconciliation
}
}
apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+Outbox.swift
+9
View File
@@ -1463,6 +1463,15 @@ extension OpenClawChatViewModel {
let session = self.currentSessionSnapshot()
guard self.outboxBranchScope(for: session) == scope else { return }
self.restoreOutboxMessages(session: session)
guard let outbox = self.outbox else { return }
Task { [weak self] in
guard let self,
await outbox.branchState(for: scope)?.needsReconciliation == true,
self.outboxBranchScope(for: self.currentSessionSnapshot()) == scope
else { return }
self.reconciledOutboxBranchScopes.remove(scope)
self.restoreOutboxMessages(session: self.currentSessionSnapshot())
}
}
}
apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatTranscriptCacheStoreTests.swift
+77 -2
View File
@@ -1424,11 +1424,12 @@ struct ChatCommandOutboxStoreTests {
#expect(await !switchStore.beginBranchSwitch(scope))
}
@Test func `expired branch switch lease is reclaimed by claim`() async throws {
@Test func `expired branch switch lease blocks claims until matching reconciliation`() async throws {
let url = try makeDatabaseURL()
defer { try? FileManager.default.removeItem(at: url.deletingLastPathComponent()) }
let store = OpenClawChatSQLiteTranscriptCache(databaseURL: url, gatewayID: "gw-a")
let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "agent-a")
#expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope))
#expect(await store.beginBranchSwitch(scope))
#expect(await store.enqueueCommand(outboxCommand(
id: "after-crash",
@@ -1453,8 +1454,82 @@ struct ChatCommandOutboxStoreTests {
sqlite3_finalize(statement)
sqlite3_close(raw)
#expect(await store.claimNextCommand() == nil)
let expiredState = try #require(await store.branchState(for: scope))
#expect(expiredState.switchPendingSince == nil)
#expect(expiredState.needsReconciliation)
_ = try #require(await store.reconcileBranchScope(
scope,
previousState: expiredState,
activeLeafEntryID: "leaf-a",
branchLeafEntryIDs: ["leaf-a"],
lastError: "branch changed"))
#expect(await store.branchState(for: scope)?.needsReconciliation == false)
#expect(await store.claimNextCommand()?.id == "after-crash")
#expect(await store.branchState(for: scope)?.switchPendingSince == nil)
}
@Test func `expired branch switch lease parks rows after changed-leaf reconciliation`() async throws {
let url = try makeDatabaseURL()
defer { try? FileManager.default.removeItem(at: url.deletingLastPathComponent()) }
let store = OpenClawChatSQLiteTranscriptCache(databaseURL: url, gatewayID: "gw-a")
let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "agent-a")
#expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope))
#expect(await store.beginBranchSwitch(scope))
#expect(await store.enqueueCommand(outboxCommand(
id: "old-branch-row",
sessionKey: "main",
agentID: "agent-a",
text: "park after expiry")))
var raw: OpaquePointer?
#expect(sqlite3_open(url.path, &raw) == SQLITE_OK)
let expired = Date().timeIntervalSince1970 -
OpenClawChatSQLiteTranscriptCache.outboxBranchSwitchLeaseMaxAge - 1
var statement: OpaquePointer?
#expect(sqlite3_prepare_v2(
raw,
"UPDATE outbox_branch_scopes SET switch_pending_since = ?1",
-1,
&statement,
nil) == SQLITE_OK)
sqlite3_bind_double(statement, 1, expired)
#expect(sqlite3_step(statement) == SQLITE_DONE)
sqlite3_finalize(statement)
sqlite3_close(raw)
#expect(await store.claimNextCommand() == nil)
let expiredState = try #require(await store.branchState(for: scope))
_ = try #require(await store.reconcileBranchScope(
scope,
previousState: expiredState,
activeLeafEntryID: "leaf-b",
branchLeafEntryIDs: ["leaf-a", "leaf-b"],
lastError: "branch changed"))
#expect(await store.branchState(for: scope)?.needsReconciliation == false)
#expect(await store.loadCommands().first?.status == .failed)
}
@Test func `expired branch switch lease blocks a new mutation lease until reconciliation`() async throws {
let url = try makeDatabaseURL()
defer { try? FileManager.default.removeItem(at: url.deletingLastPathComponent()) }
let store = OpenClawChatSQLiteTranscriptCache(databaseURL: url, gatewayID: "gw-a")
let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "agent-a")
#expect(await store.beginBranchSwitch(scope))
var raw: OpaquePointer?
#expect(sqlite3_open(url.path, &raw) == SQLITE_OK)
let expired = Date().timeIntervalSince1970 -
OpenClawChatSQLiteTranscriptCache.outboxBranchSwitchLeaseMaxAge - 1
#expect(sqlite3_exec(
raw,
"UPDATE outbox_branch_scopes SET switch_pending_since = \(expired)",
nil,
nil,
nil) == SQLITE_OK)
sqlite3_close(raw)
#expect(await !store.beginBranchSwitch(scope))
#expect(await store.branchState(for: scope)?.needsReconciliation == true)
}
@Test func `branch observation preserves evidence until queued work reconciles`() async throws {
apps/shared/OpenClawKit/Tests/OpenClawKitTests/ChatViewModelOutboxTests.swift
+48
View File
@@ -1,5 +1,6 @@
import Foundation
import OpenClawKit
import SQLite3
import Testing
@testable import OpenClawChatUI
@@ -1470,6 +1471,53 @@ struct ChatViewModelOutboxTests {
}
}
@Test func `expired branch lease invalidation makes a sibling reconcile again`() async throws {
let url = try makeOutboxDatabaseURL()
defer { try? FileManager.default.removeItem(at: url.deletingLastPathComponent()) }
let store = OpenClawChatSQLiteTranscriptCache(databaseURL: url, gatewayID: "gw-test")
let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "main")
#expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope))
let transport = OutboxTestTransport(healthy: false, activeBranchLeafEntryID: "leaf-a")
let initiatingVM = await makeOutboxViewModel(transport: transport, outbox: store)
let siblingVM = await makeOutboxViewModel(transport: transport, outbox: store)
await MainActor.run {
initiatingVM.reconciledOutboxBranchScopes.insert(scope)
siblingVM.reconciledOutboxBranchScopes.insert(scope)
}
#expect(await store.beginBranchSwitch(scope))
#expect(await store.enqueueCommand(OpenClawChatOutboxCommand(
id: "expired-lease-row",
sessionKey: "main",
deliverySessionKey: "agent:main:main",
routingContract: "per-sender|main|main",
agentID: "main",
text: "wait for reconciliation",
thinking: "off",
createdAt: Date().timeIntervalSince1970,
status: .queued,
retryCount: 0,
lastError: nil)))
var raw: OpaquePointer?
#expect(sqlite3_open(url.path, &raw) == SQLITE_OK)
let expired = Date().timeIntervalSince1970 -
OpenClawChatSQLiteTranscriptCache.outboxBranchSwitchLeaseMaxAge - 1
#expect(sqlite3_exec(
raw,
"UPDATE outbox_branch_scopes SET switch_pending_since = \(expired)",
nil,
nil,
nil) == SQLITE_OK)
sqlite3_close(raw)
#expect(await store.claimNextCommand() == nil)
try await waitUntil("sibling observes the reconciliation invalidation") {
await MainActor.run {
!siblingVM.reconciledOutboxBranchScopes.contains(scope)
}
}
}
@Test func `matching epoch flush uses no branch list lookup`() async throws {
let url = try makeOutboxDatabaseURL()
defer { try? FileManager.default.removeItem(at: url.deletingLastPathComponent()) }