import Foundation import GRDB import OpenClawKit import SQLite3 import Testing @testable import OpenClawChatUI private func makeDatabaseDirectory() throws -> URL { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("chat-database-tests-\(UUID().uuidString)", isDirectory: true) try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) return directory } private func cacheMessage( role: String, text: String, timestamp: Double, idempotencyKey: String? = nil) -> OpenClawChatMessage { OpenClawChatMessage( role: role, content: [ OpenClawChatMessageContent( type: "text", text: text, mimeType: nil, fileName: nil, content: nil), ], timestamp: timestamp, idempotencyKey: idempotencyKey) } private func cacheSessionEntry(key: String, updatedAt: Double) -> OpenClawChatSessionEntry { OpenClawChatSessionEntry( key: key, kind: nil, displayName: nil, surface: nil, subject: nil, room: nil, space: nil, updatedAt: updatedAt, sessionId: nil, systemSent: nil, abortedLastRun: nil, thinkingLevel: nil, verboseLevel: nil, inputTokens: nil, outputTokens: nil, totalTokens: nil, modelProvider: nil, model: nil, contextTokens: nil) } private func messageTexts(_ messages: [OpenClawChatMessage]) -> [String] { messages.map { $0.content.compactMap(\.text).joined() } } extension OpenClawChatSQLiteTranscriptCache { fileprivate func storeTestTranscript( sessionKey: String, agentID: String? = nil, messages: [OpenClawChatMessage]) async { await storeCanonicalTranscript( sessionKey: sessionKey, agentID: agentID, messages: messages, canonicalMessageIdempotencyKeys: Set(messages.compactMap(\.idempotencyKey))) } } private struct CacheMessageRowProbe: Sendable { let position: Int let idempotencyKey: String? let payloadJSON: String } private func outboxCommand( id: String = UUID().uuidString, sessionKey: String = "main", text: String, attachments: [OpenClawChatOutboxAttachment] = [], thinking: String = "off", createdAt: Double = Date().timeIntervalSince1970, status: OpenClawChatOutboxCommand.Status = .queued) -> OpenClawChatOutboxCommand { OpenClawChatOutboxCommand( id: id, sessionKey: sessionKey, deliverySessionKey: "agent:main:main", routingContract: "per-sender|main|main", agentID: "main", text: text, attachments: attachments, thinking: thinking, createdAt: createdAt, status: status, retryCount: 0, lastError: nil) } private func withRawDatabase(at url: URL, _ body: (OpaquePointer) throws -> Void) throws { var raw: OpaquePointer? #expect(sqlite3_open(url.path, &raw) == SQLITE_OK) let database = try #require(raw) defer { sqlite3_close_v2(database) } try body(database) } private func execute(_ database: OpaquePointer, _ sql: String) { #expect(sqlite3_exec(database, sql, nil, nil, nil) == SQLITE_OK) } private func createLegacyV2Database( at url: URL, gatewayID: String, commandID: String, text: String = "preserve me") throws { try withRawDatabase(at: url) { raw in execute(raw, """ CREATE TABLE outbox_commands( id INTEGER PRIMARY KEY AUTOINCREMENT, client_uuid TEXT NOT NULL UNIQUE, gateway_id TEXT NOT NULL, session_key TEXT NOT NULL, text TEXT NOT NULL, thinking TEXT NOT NULL, created_at REAL NOT NULL, status TEXT NOT NULL, retry_count INTEGER NOT NULL DEFAULT 0, last_error TEXT NOT NULL DEFAULT '' ); INSERT INTO outbox_commands( client_uuid, gateway_id, session_key, text, thinking, created_at, status, retry_count, last_error ) VALUES ('\(commandID)', '\(gatewayID)', 'main', '\(text)', 'off', 1, 'queued', 2, ''); PRAGMA user_version = 2; """) } } struct ChatTranscriptCacheStoreTests { @Test func `one installation owns exactly the two named databases`() throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } _ = try OpenClawClientDatabases(directoryURL: directory) let sqliteFiles = try FileManager.default.contentsOfDirectory(atPath: directory.path) .filter { $0.hasSuffix(".sqlite") } .sorted() #expect(sqliteFiles == ["client-state.sqlite", "gateway-cache.sqlite"]) } @Test func `full removal deletes both databases legacy files and sidecars`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) await databases.store(gatewayID: "gw-a").storeSessions([ cacheSessionEntry(key: "main", updatedAt: 1), ]) let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") try withRawDatabase(at: legacyURL) { raw in execute(raw, "PRAGMA user_version = 99;") } try Data("sidecar".utf8).write(to: URL(fileURLWithPath: legacyURL.path + "-wal")) try databases.close() try OpenClawClientDatabases.removeDatabaseFiles(in: directory) for filename in [ OpenClawClientDatabases.gatewayCacheFilename, OpenClawClientDatabases.clientStateFilename, legacyURL.lastPathComponent, ] { for suffix in ["", "-wal", "-shm", "-journal"] { #expect(!FileManager.default.fileExists( atPath: directory.appendingPathComponent(filename).path + suffix)) } } } @Test func `transcript and sessions round trip as row JSON`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") let messages = [ cacheMessage(role: "user", text: "hello", timestamp: 1000, idempotencyKey: "run-1:user"), cacheMessage(role: "assistant", text: "hi", timestamp: 2000, idempotencyKey: "run-1"), ] await store.storeTestTranscript(sessionKey: "main", messages: messages) await store.storeSessions([cacheSessionEntry(key: "main", updatedAt: 2000)]) #expect(await messageTexts(store.loadTranscript(sessionKey: "main")) == ["hello", "hi"]) #expect(await store.loadSessions().map(\.key) == ["main"]) let messageRows = try await databases.cacheQueue.read { db in try Row.fetchAll(db, sql: """ SELECT position, timestamp_ms, idempotency_key, payload_json FROM cached_messages WHERE gateway_id = 'gw-a' ORDER BY position """).map { row in CacheMessageRowProbe( position: row["position"], idempotencyKey: row["idempotency_key"], payloadJSON: row["payload_json"]) } } #expect(messageRows.count == 2) #expect(messageRows.map(\.position) == [0, 1]) #expect(messageRows.map(\.idempotencyKey) == ["run-1:user", "run-1"]) #expect(messageRows[0].payloadJSON.contains("\"role\":\"user\"")) #expect(!messageRows[0].payloadJSON.hasPrefix("[")) } @Test func `cache format mismatch rebuilds without touching client state`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let stateIdentity = try #require(OpenClawChatSessionRoutingIdentity( scope: "per-sender", mainSessionKey: "main", defaultAgentID: "main")) do { let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") await store.storeSessionRoutingIdentity(stateIdentity) } let cacheURL = directory.appendingPathComponent("gateway-cache.sqlite") try withRawDatabase(at: cacheURL) { raw in execute(raw, "UPDATE cache_metadata SET format_version = 999 WHERE id = 1") execute(raw, "CREATE TABLE disposable_old_shape(value TEXT)") } let reopened = try OpenClawClientDatabases(directoryURL: directory) #expect(reopened.loadSessionRoutingIdentity(gatewayID: "gw-a") == stateIdentity) #expect(try await reopened.cacheQueue.read { db in try db.tableExists("disposable_old_shape") } == false) #expect(try await reopened.cacheQueue.read { db in try Int.fetchOne(db, sql: "SELECT format_version FROM cache_metadata WHERE id = 1") } == 1) } @Test func `corrupt cache rebuilds while corrupt client state is preserved`() async throws { let cacheDirectory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: cacheDirectory) } let cacheURL = cacheDirectory.appendingPathComponent("gateway-cache.sqlite") try Data("not sqlite".utf8).write(to: cacheURL) let repaired = try OpenClawClientDatabases(directoryURL: cacheDirectory) #expect(try await repaired.cacheQueue.read { db in try db.tableExists("cached_messages") }) let stateDirectory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: stateDirectory) } let stateURL = stateDirectory.appendingPathComponent("client-state.sqlite") let bytes = Data("durable bytes must survive".utf8) try bytes.write(to: stateURL) #expect(throws: (any Error).self) { _ = try OpenClawClientDatabases(directoryURL: stateDirectory) } #expect(try Data(contentsOf: stateURL) == bytes) } @Test func `transcripts are scoped by gateway and agent in one cache`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let storeA = databases.store(gatewayID: "gw-a") let storeB = databases.store(gatewayID: "gw-b") await storeA.storeTestTranscript( sessionKey: "global", agentID: "agent-a", messages: [cacheMessage(role: "user", text: "A", timestamp: 1)]) await storeA.storeTestTranscript( sessionKey: "global", agentID: "agent-b", messages: [cacheMessage(role: "user", text: "B", timestamp: 2)]) await storeB.storeTestTranscript( sessionKey: "global", agentID: "agent-a", messages: [cacheMessage(role: "user", text: "other gateway", timestamp: 3)]) #expect(await messageTexts(storeA.loadTranscript(sessionKey: "global", agentID: "agent-a")) == ["A"]) #expect(await messageTexts(storeA.loadTranscript(sessionKey: "global", agentID: "agent-b")) == ["B"]) #expect(await messageTexts(storeB.loadTranscript(sessionKey: "global", agentID: "agent-a")) == [ "other gateway", ]) #expect(await storeA.loadTranscript(sessionKey: "global").isEmpty) } @Test func `cache bounds sessions messages and transcript partitions`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") let sessions = (0..<(OpenClawChatSQLiteTranscriptCache.maxCachedSessions + 10)).map { cacheSessionEntry(key: "s\($0)", updatedAt: Double($0)) } await store.storeSessions(sessions) #expect(await store.loadSessions().count == OpenClawChatSQLiteTranscriptCache.maxCachedSessions) #expect(await store.loadSessions().contains(where: { $0.key == "s0" }) == false) let messages = (0..<(OpenClawChatSQLiteTranscriptCache.maxCachedMessagesPerSession + 20)).map { cacheMessage(role: "user", text: "m\($0)", timestamp: Double($0)) } await store.storeTestTranscript(sessionKey: "bounded", messages: messages) #expect(await store.loadTranscript(sessionKey: "bounded").count == OpenClawChatSQLiteTranscriptCache.maxCachedMessagesPerSession) #expect(await messageTexts(store.loadTranscript(sessionKey: "bounded")).first == "m20") for index in 0...OpenClawChatSQLiteTranscriptCache.maxCachedTranscripts { await store.storeTestTranscript( sessionKey: "partition-\(index)", messages: [cacheMessage(role: "user", text: "p\(index)", timestamp: Double(index))]) } #expect(await store.loadTranscript(sessionKey: "partition-0").isEmpty) #expect(await store.loadTranscript( sessionKey: "partition-\(OpenClawChatSQLiteTranscriptCache.maxCachedTranscripts)").isEmpty == false) } @Test func `empty transcript deletes its partition`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") await store.storeTestTranscript( sessionKey: "main", messages: [cacheMessage(role: "user", text: "old", timestamp: 1)]) await store.storeTestTranscript(sessionKey: "main", messages: []) #expect(await store.loadTranscript(sessionKey: "main").isEmpty) #expect(try await databases.cacheQueue.read { db in try Int.fetchOne(db, sql: "SELECT COUNT(*) FROM cached_transcripts") } == 0) } @Test func `canonical cache excludes optimistic outbox rows`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "queued", text: "local"))) let snapshot = [ cacheMessage(role: "user", text: "local", timestamp: 1, idempotencyKey: "queued:user"), cacheMessage(role: "assistant", text: "canonical", timestamp: 2, idempotencyKey: "other"), ] await store.storeCanonicalTranscript( sessionKey: "main", agentID: nil, messages: snapshot, canonicalMessageIdempotencyKeys: ["other"]) #expect(await messageTexts(store.loadTranscript(sessionKey: "main")) == ["canonical"]) #expect(await store.loadCommands().map(\.id) == ["queued"]) await store.storeCanonicalTranscript( sessionKey: "main", agentID: nil, messages: snapshot, canonicalMessageIdempotencyKeys: ["queued:user", "other"]) #expect(await messageTexts(store.loadTranscript(sessionKey: "main")) == ["local", "canonical"]) } @Test func `canceled optimistic row cannot reenter cache from a stale snapshot`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "canceled", text: "local"))) let capturedBeforeCancellation = [ cacheMessage(role: "user", text: "local", timestamp: 1, idempotencyKey: "canceled:user"), ] #expect(await store.cancelCommand(id: "canceled") == .updated) await store.storeCanonicalTranscript( sessionKey: "main", agentID: nil, messages: capturedBeforeCancellation, canonicalMessageIdempotencyKeys: []) #expect(await store.loadTranscript(sessionKey: "main").isEmpty) } @Test func `malformed cache partitions are discarded atomically`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") await store.storeSessions([cacheSessionEntry(key: "main", updatedAt: 1)]) await store.storeTestTranscript( sessionKey: "main", messages: [cacheMessage(role: "assistant", text: "cached", timestamp: 1)]) try await databases.cacheQueue.write { db in try db.execute( sql: "UPDATE cached_sessions SET payload_json = 'not-json' WHERE gateway_id = 'gw-a'") try db.execute( sql: "UPDATE cached_messages SET payload_json = 'not-json' WHERE gateway_id = 'gw-a'") } #expect(await store.loadSessions().isEmpty) #expect(await store.loadTranscript(sessionKey: "main").isEmpty) #expect(try await databases.cacheQueue.read { db in try Int.fetchOne(db, sql: "SELECT COUNT(*) FROM cached_sessions WHERE gateway_id = 'gw-a'") } == 0) #expect(try await databases.cacheQueue.read { db in try Int.fetchOne(db, sql: "SELECT COUNT(*) FROM cached_transcripts WHERE gateway_id = 'gw-a'") } == 0) } @Test func `canonical merge preserves newer cache rows`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") await store.storeTestTranscript(sessionKey: "main", messages: [ cacheMessage(role: "assistant", text: "newer", timestamp: 2, idempotencyKey: "newer"), ]) await store.mergeCanonicalTranscriptMessage( sessionKey: "main", agentID: nil, message: cacheMessage(role: "user", text: "confirmed", timestamp: 1, idempotencyKey: "confirmed:user"), canonicalMessageIdempotencyKey: "confirmed:user") #expect(await messageTexts(store.loadTranscript(sessionKey: "main")) == ["confirmed", "newer"]) } @Test func `concurrent canonical merges do not lose messages`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") await withTaskGroup(of: Void.self) { group in for index in 0..<20 { group.addTask { let key = "merge-\(index)" await store.mergeCanonicalTranscriptMessage( sessionKey: "main", agentID: nil, message: cacheMessage( role: "assistant", text: key, timestamp: Double(index), idempotencyKey: key), canonicalMessageIdempotencyKey: key) } } } #expect(await Set(messageTexts(store.loadTranscript(sessionKey: "main"))) == Set((0..<20).map { "merge-\($0)" })) } @Test func `cache projection strips payloads and keeps bounded diffs`() throws { let oversizedDiff = "+1 " + String(repeating: "x", count: 64100) let message = OpenClawChatMessage( role: "toolResult", content: [ OpenClawChatMessageContent( type: "toolCall", text: "done", mimeType: "image/jpeg", fileName: "photo.jpg", content: AnyCodable(String(repeating: "payload", count: 1000)), name: "apply_patch", arguments: AnyCodable([ "input": AnyCodable(oversizedDiff), "ignored": AnyCodable("drop"), ]), details: AnyCodable(["diff": AnyCodable(oversizedDiff), "ignored": AnyCodable("drop")])), ], timestamp: 1, details: AnyCodable(["diff": AnyCodable(oversizedDiff), "ignored": AnyCodable("drop")])) let cached = try #require(OpenClawChatSQLiteTranscriptCache.cacheableMessages([message]).first) #expect(cached.content[0].content == nil) #expect(cached.content[0].thinkingSignature == nil) #expect(Set(cached.content[0].arguments?.dictionaryValue?.keys.map(\.self) ?? []) == ["input"]) #expect(cached.content[0].arguments?.dictionaryValue?["input"]?.stringValue?.utf16.count == 64000) #expect(Set(cached.details?.dictionaryValue?.keys.map(\.self) ?? []) == ["diff"]) } @Test func `gateway removal deletes only that gateways cache and state`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let storeA = databases.store(gatewayID: "gw-a") let storeB = databases.store(gatewayID: "gw-b") await storeA.storeSessions([cacheSessionEntry(key: "a", updatedAt: 1)]) await storeB.storeSessions([cacheSessionEntry(key: "b", updatedAt: 2)]) #expect(await storeA.enqueueCommand(outboxCommand(id: "a", text: "A"))) #expect(await storeB.enqueueCommand(outboxCommand(id: "b", text: "B"))) try databases.removeGatewayData(gatewayID: "gw-a") #expect(await storeA.loadSessions().isEmpty) #expect(await storeA.loadCommands().isEmpty) #expect(await storeB.loadSessions().map(\.key) == ["b"]) #expect(await storeB.loadCommands().map(\.id) == ["b"]) } @Test func `staged gateway removal reconciles against the pairing registry`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } do { let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") await store.storeSessions([cacheSessionEntry(key: "main", updatedAt: 1)]) #expect(await store.enqueueCommand(outboxCommand(id: "keep", text: "pending"))) try databases.stageGatewayRemoval(gatewayID: "gw-a") #expect(await store.loadSessions().map(\.key) == ["main"]) #expect(await store.loadCommands().map(\.id) == ["keep"]) try databases.close() } do { let registered = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: ["gw-a"]) #expect(await registered.store(gatewayID: "gw-a").loadCommands().map(\.id) == ["keep"]) try registered.stageGatewayRemoval(gatewayID: "gw-a") try registered.close() } let forgotten = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: []) #expect(await forgotten.store(gatewayID: "gw-a").loadSessions().isEmpty) #expect(await forgotten.store(gatewayID: "gw-a").loadCommands().isEmpty) #expect(try await forgotten.stateQueue.read { db in try String.fetchOne( db, sql: "SELECT gateway_hash FROM forgotten_gateways WHERE gateway_id IS NULL") } == OpenClawClientDatabases.gatewayIdentityHash("gw-a")) } @Test func `staged gateway removal uses exact registry identifier bytes`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let composedGatewayID = "gateway-\u{00E9}" let decomposedGatewayID = "gateway-e\u{0301}" #expect(composedGatewayID == decomposedGatewayID) do { let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: composedGatewayID) #expect(await store.enqueueCommand(outboxCommand(id: "remove", text: "pending"))) try databases.stageGatewayRemoval(gatewayID: composedGatewayID) try databases.close() } let recovered = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: [decomposedGatewayID]) #expect(await recovered.store(gatewayID: composedGatewayID).loadCommands().isEmpty) #expect(try await recovered.stateQueue.read { db in try String.fetchOne( db, sql: "SELECT gateway_hash FROM forgotten_gateways WHERE gateway_id IS NULL") } == OpenClawClientDatabases.gatewayIdentityHash(composedGatewayID)) } @Test func `commit started recovery finishes even while gateway remains registered`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } do { let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") await store.storeSessions([cacheSessionEntry(key: "main", updatedAt: 1)]) #expect(await store.enqueueCommand(outboxCommand(id: "remove", text: "pending"))) try databases.stageGatewayRemoval(gatewayID: "gw-a") // Simulate termination immediately after the irreversible state // transaction but before cache cleanup and tombstone finalization. try await databases.stateQueue.write { db in try db.execute( sql: "UPDATE forgotten_gateways SET cleanup_phase = 2 WHERE gateway_id = ?", arguments: ["gw-a"]) try db.execute( sql: "DELETE FROM outbox_commands WHERE gateway_id = ?", arguments: ["gw-a"]) } try databases.close() } let recovered = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: ["gw-a"]) #expect(await recovered.store(gatewayID: "gw-a").loadSessions().isEmpty) #expect(await recovered.store(gatewayID: "gw-a").loadCommands().isEmpty) #expect(try await recovered.stateQueue.read { db in try Int.fetchOne( db, sql: "SELECT cleanup_phase FROM forgotten_gateways WHERE gateway_hash = ?", arguments: [OpenClawClientDatabases.gatewayIdentityHash("gw-a")]) } == 0) } @Test func `hash only scrub marker remains recoverable`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } do { let databases = try OpenClawClientDatabases(directoryURL: directory) try databases.stageGatewayRemoval(gatewayID: "gw-a") try await databases.stateQueue.write { db in try db.execute( sql: """ UPDATE forgotten_gateways SET gateway_id = NULL, cleanup_phase = 3, restore_finalized = 0 WHERE gateway_id = ? """, arguments: ["gw-a"]) } try databases.close() } let recovered = try OpenClawClientDatabases(directoryURL: directory) #expect(try await recovered.stateQueue.read { db in try Int.fetchOne( db, sql: "SELECT cleanup_phase FROM forgotten_gateways WHERE gateway_hash = ?", arguments: [OpenClawClientDatabases.gatewayIdentityHash("gw-a")]) } == 0) } @Test func `unknown registry preserves a cancelable staged removal`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } do { let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "keep", text: "pending"))) try databases.stageGatewayRemoval(gatewayID: "gw-a") try databases.close() } let reopened = try OpenClawClientDatabases(directoryURL: directory) #expect(await reopened.store(gatewayID: "gw-a").loadCommands().map(\.id) == ["keep"]) #expect(try await reopened.stateQueue.read { db in try Int.fetchOne( db, sql: "SELECT cleanup_phase FROM forgotten_gateways WHERE gateway_hash = ?", arguments: [OpenClawClientDatabases.gatewayIdentityHash("gw-a")]) } == 1) } @Test func `pending removal marker gates writable facade recreation`() throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) #expect(!databases.hasPendingGatewayRemoval(gatewayID: "gw-a")) try databases.stageGatewayRemoval(gatewayID: "gw-a") #expect(databases.hasPendingGatewayRemoval(gatewayID: "gw-a")) try databases.cancelGatewayRemoval(gatewayID: "gw-a") #expect(!databases.hasPendingGatewayRemoval(gatewayID: "gw-a")) } @Test func `cancelable stage does not suppress legacy state import`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) try databases.stageGatewayRemoval(gatewayID: "gw-a") let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") try createLegacyV2Database(at: legacyURL, gatewayID: "gw-a", commandID: "restore-on-cancel") databases.retryLegacyImport() #expect(await databases.store(gatewayID: "gw-a").loadCommands().map(\.id) == ["restore-on-cancel"]) #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) } @Test func `one broken pending removal does not block another gateway`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } do { let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "z-good") await store.storeSessions([cacheSessionEntry(key: "main", updatedAt: 1)]) #expect(await store.enqueueCommand(outboxCommand(id: "remove", text: "pending"))) try databases.stageGatewayRemoval(gatewayID: "z-good") try await databases.stateQueue.write { db in try db.execute( sql: "UPDATE forgotten_gateways SET cleanup_phase = 2 WHERE gateway_id = ?", arguments: ["z-good"]) try db.execute( sql: """ INSERT INTO forgotten_gateways( gateway_hash, gateway_id, forgotten_at, cleanup_phase, restore_finalized ) VALUES (?, ?, 0, 2, 0) """, arguments: [String(repeating: "0", count: 64), "a-broken"]) } try databases.close() } let recovered = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: []) #expect(await recovered.store(gatewayID: "z-good").loadSessions().isEmpty) #expect(await recovered.store(gatewayID: "z-good").loadCommands().isEmpty) #expect(try await recovered.stateQueue.read { db in try Int.fetchOne( db, sql: "SELECT cleanup_phase FROM forgotten_gateways WHERE gateway_id = ?", arguments: ["a-broken"]) } == 2) } @Test func `canceling a repeated forget preserves the finalized tombstone`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) try databases.removeGatewayData(gatewayID: "gw-a") try databases.stageGatewayRemoval(gatewayID: "gw-a") try databases.cancelGatewayRemoval(gatewayID: "gw-a") let tombstoneGatewayID: String? = try await databases.stateQueue.read { db in try String.fetchOne( db, sql: """ SELECT gateway_id FROM forgotten_gateways WHERE gateway_hash = ? """, arguments: [OpenClawClientDatabases.gatewayIdentityHash("gw-a")]) } let tombstonePhase = try await databases.stateQueue.read { db in try Int.fetchOne( db, sql: "SELECT cleanup_phase FROM forgotten_gateways WHERE gateway_hash = ?", arguments: [OpenClawClientDatabases.gatewayIdentityHash("gw-a")]) } #expect(tombstoneGatewayID == nil) #expect(tombstonePhase == 0) let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") try createLegacyV2Database(at: legacyURL, gatewayID: "gw-a", commandID: "must-not-return") databases.retryLegacyImport(registeredGatewayIDs: []) #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) #expect(await databases.store(gatewayID: "gw-a").loadCommands().isEmpty) } @Test func `gateway removal scrubs its payloads from shared database files`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let storeA = databases.store(gatewayID: "gw-a") let storeB = databases.store(gatewayID: "gw-b") let sensitiveText = "forgotten-sensitive-\(UUID().uuidString)" let sensitiveBytes = Data(sensitiveText.utf8) let attachment = OpenClawChatOutboxAttachment( type: "file", mimeType: "application/octet-stream", fileName: "secret.bin", data: sensitiveBytes) await storeA.storeTestTranscript( sessionKey: "main", messages: [cacheMessage(role: "user", text: sensitiveText, timestamp: 1)]) #expect(await storeA.enqueueCommand(outboxCommand( id: "sensitive", text: sensitiveText, attachments: [attachment]))) await storeB.storeSessions([cacheSessionEntry(key: "keep", updatedAt: 1)]) try databases.removeGatewayData(gatewayID: "gw-a") try databases.close() let files = try FileManager.default.contentsOfDirectory( at: directory, includingPropertiesForKeys: nil) for file in files where file.lastPathComponent.contains(".sqlite") { #expect(try Data(contentsOf: file).range(of: sensitiveBytes) == nil) } let reopened = try OpenClawClientDatabases(directoryURL: directory) #expect(await reopened.store(gatewayID: "gw-b").loadSessions().map(\.key) == ["keep"]) } @Test func `routing identity survives a cold container reopen`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let identity = try #require(OpenClawChatSessionRoutingIdentity( scope: " Per-Sender ", mainSessionKey: " Work ", defaultAgentID: " Main ")) do { let databases = try OpenClawClientDatabases(directoryURL: directory) await databases.store(gatewayID: "gw-a").storeSessionRoutingIdentity(identity) } let reopened = try OpenClawClientDatabases(directoryURL: directory) #expect(reopened.loadSessionRoutingIdentity(gatewayID: "gw-a") == identity) #expect(identity.contract == "per-sender|work|main") #expect(try await reopened.stateQueue.read { db in try String.fetchAll(db, sql: "SELECT identifier FROM grdb_migrations") } == [ "client-state-v1", "client-state-branch-ownership-v2", "client-state-branch-revision-v3", "client-state-agent-id-v4", "client-state-outbox-attempt-scope-v5", ]) } } struct ClientDatabaseLegacyImportTests { @Test func `legacy v1 cache is discarded`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") try withRawDatabase(at: legacyURL) { raw in execute(raw, """ CREATE TABLE cached_sessions( gateway_id TEXT PRIMARY KEY, payload TEXT NOT NULL, updated_at REAL NOT NULL ); PRAGMA user_version = 1; """) } let databases = try OpenClawClientDatabases(directoryURL: directory) #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) #expect(try await databases.stateQueue.read { db in try Int.fetchOne(db, sql: "SELECT COUNT(*) FROM outbox_commands") } == 0) } @Test func `legacy v2 outbox imports parked into client state`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let legacyURL = directory.appendingPathComponent( String(repeating: "a", count: 64) + ".sqlite") try createLegacyV2Database(at: legacyURL, gatewayID: "gw-a", commandID: "legacy-v2") let databases = try OpenClawClientDatabases(directoryURL: directory) let commands = await databases.store(gatewayID: "gw-a").loadCommands() #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) #expect(commands.map(\.id) == ["legacy-v2"]) #expect(commands.map(\.text) == ["preserve me"]) #expect(commands.map(\.status) == [.failed]) #expect(commands.map(\.lastError) == [OpenClawChatSQLiteTranscriptCache.outboxUnknownTargetError]) #expect(commands.map(\.routingContract) == [nil]) } @Test func `foreground retry imports a legacy database discovered after startup`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") try createLegacyV2Database(at: legacyURL, gatewayID: "gw-a", commandID: "late-legacy") databases.retryLegacyImport() #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) #expect(await databases.store(gatewayID: "gw-a").loadCommands().map(\.id) == ["late-legacy"]) } @Test func `forgotten gateway is never resurrected by a late legacy import`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) try databases.removeGatewayData(gatewayID: "gw-forgotten") let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") try createLegacyV2Database( at: legacyURL, gatewayID: "gw-forgotten", commandID: "must-not-return") databases.retryLegacyImport(registeredGatewayIDs: []) #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) let store = databases.store(gatewayID: "gw-forgotten") #expect(await store.loadCommands().isEmpty) #expect(try await databases.stateQueue.read { db in try Int.fetchOne( db, sql: "SELECT cleanup_phase FROM forgotten_gateways WHERE gateway_hash = ?", arguments: [OpenClawClientDatabases.gatewayIdentityHash("gw-forgotten")]) } == 0) #expect(await store.enqueueCommand(outboxCommand(id: "after-repair", text: "new pairing"))) databases.retryLegacyImport() #expect(await store.loadCommands().map(\.id) == ["after-repair"]) } @Test func `forget removes an unreadable per gateway legacy database and sidecars`() throws { let root = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: root) } let databaseDirectory = root.appendingPathComponent("databases", isDirectory: true) let legacyDirectory = root.appendingPathComponent("chat-cache", isDirectory: true) try FileManager.default.createDirectory(at: legacyDirectory, withIntermediateDirectories: true) let gatewayID = "manual|forgotten-secret.example|443" let legacyURL = OpenClawClientDatabases.legacyPerGatewayDatabaseURL( gatewayID: gatewayID, directoryURL: legacyDirectory) try withRawDatabase(at: legacyURL) { raw in execute(raw, "PRAGMA user_version = 99;") } for suffix in ["-wal", "-shm", "-journal"] { try Data("legacy-sensitive-bytes".utf8).write(to: URL(fileURLWithPath: legacyURL.path + suffix)) } let databases = try OpenClawClientDatabases( directoryURL: databaseDirectory, legacyDirectoryURLs: [legacyDirectory]) #expect(FileManager.default.fileExists(atPath: legacyURL.path)) try databases.removeGatewayData(gatewayID: gatewayID) for suffix in ["", "-wal", "-shm", "-journal"] { #expect(!FileManager.default.fileExists(atPath: legacyURL.path + suffix)) } #expect(try databases.stateQueue.read { db in try String.fetchOne( db, sql: "SELECT gateway_hash FROM forgotten_gateways WHERE gateway_id IS NULL") } == OpenClawClientDatabases.gatewayIdentityHash(gatewayID)) try databases.close() for suffix in ["", "-wal", "-shm"] { let url = URL(fileURLWithPath: databaseDirectory .appendingPathComponent(OpenClawClientDatabases.clientStateFilename).path + suffix) if FileManager.default.fileExists(atPath: url.path) { #expect(try Data(contentsOf: url).range(of: Data(gatewayID.utf8)) == nil) } } } @Test func `legacy import accepts only gateways in the pairing registry`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let keptURL = OpenClawClientDatabases.legacyPerGatewayDatabaseURL( gatewayID: "gw-kept", directoryURL: directory) let orphanedURL = OpenClawClientDatabases.legacyPerGatewayDatabaseURL( gatewayID: "gw-orphaned", directoryURL: directory) try createLegacyV2Database(at: keptURL, gatewayID: "gw-kept", commandID: "kept") try createLegacyV2Database(at: orphanedURL, gatewayID: "gw-orphaned", commandID: "orphaned") let databases = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: ["gw-kept"]) #expect(await databases.store(gatewayID: "gw-kept").loadCommands().map(\.id) == ["kept"]) #expect(await databases.store(gatewayID: "gw-orphaned").loadCommands().isEmpty) #expect(!FileManager.default.fileExists(atPath: keptURL.path)) #expect(FileManager.default.fileExists(atPath: orphanedURL.path)) } @Test func `legacy import uses exact registry identifier bytes`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let composedGatewayID = "gateway-\u{00E9}" let decomposedGatewayID = "gateway-e\u{0301}" #expect(composedGatewayID == decomposedGatewayID) let legacyURL = OpenClawClientDatabases.legacyPerGatewayDatabaseURL( gatewayID: composedGatewayID, directoryURL: directory) try createLegacyV2Database( at: legacyURL, gatewayID: composedGatewayID, commandID: "unowned") let databases = try OpenClawClientDatabases( directoryURL: directory, registeredGatewayIDs: [decomposedGatewayID]) #expect(await databases.store(gatewayID: composedGatewayID).loadCommands().isEmpty) #expect(FileManager.default.fileExists(atPath: legacyURL.path)) } @Test func `preserved shared legacy database blocks targeted forget`() async throws { let root = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: root) } let databaseDirectory = root.appendingPathComponent("databases", isDirectory: true) let legacyDirectory = root.appendingPathComponent("legacy", isDirectory: true) try FileManager.default.createDirectory(at: legacyDirectory, withIntermediateDirectories: true) let legacyURL = legacyDirectory.appendingPathComponent("chat-cache.sqlite") try withRawDatabase(at: legacyURL) { raw in execute(raw, "PRAGMA user_version = 99;") } let databases = try OpenClawClientDatabases( directoryURL: databaseDirectory, legacyDirectoryURLs: [legacyDirectory]) let store = databases.store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "keep", text: "not forgotten"))) #expect(throws: (any Error).self) { try databases.removeGatewayData(gatewayID: "gw-a") } #expect(FileManager.default.fileExists(atPath: legacyURL.path)) #expect(await store.loadCommands().map(\.id) == ["keep"]) } @Test func `legacy v6 imports attachments and routing identity`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let legacyURL = directory.appendingPathComponent("chat-cache.sqlite") let attachment = OpenClawChatOutboxAttachment( type: "image", mimeType: "image/jpeg", fileName: "photo.jpg", data: Data([1, 2, 3]), durationSeconds: nil) let attachmentsJSON = try #require(String( data: JSONEncoder().encode([attachment]), encoding: .utf8)) try withRawDatabase(at: legacyURL) { raw in execute(raw, """ CREATE TABLE outbox_commands( id INTEGER PRIMARY KEY AUTOINCREMENT, client_uuid TEXT NOT NULL UNIQUE, gateway_id TEXT NOT NULL, session_key TEXT NOT NULL, delivery_session_key TEXT NOT NULL DEFAULT '', routing_contract TEXT NOT NULL DEFAULT '', agent_id TEXT NOT NULL DEFAULT '', text TEXT NOT NULL, attachments TEXT NOT NULL DEFAULT '[]', attachment_bytes INTEGER NOT NULL DEFAULT 0, thinking TEXT NOT NULL DEFAULT '', created_at REAL NOT NULL, status TEXT NOT NULL, retry_count INTEGER NOT NULL DEFAULT 0, last_error TEXT NOT NULL DEFAULT '' ); CREATE TABLE gateway_routing_identity( gateway_id TEXT PRIMARY KEY, scope TEXT NOT NULL, main_session_key TEXT NOT NULL, default_agent_id TEXT NOT NULL, updated_at REAL NOT NULL ); """) var statement: OpaquePointer? #expect(sqlite3_prepare_v2(raw, """ INSERT INTO outbox_commands( client_uuid, gateway_id, session_key, delivery_session_key, routing_contract, agent_id, text, attachments, attachment_bytes, thinking, created_at, status, retry_count, last_error ) VALUES (?, 'gw-a', 'main', 'agent:main:main', 'per-sender|main|main', 'main', 'with image', ?, 3, 'off', 1, 'queued', 0, '') """, -1, &statement, nil) == SQLITE_OK) let transient = unsafeBitCast(-1, to: sqlite3_destructor_type.self) sqlite3_bind_text(statement, 1, "legacy-v6", -1, transient) sqlite3_bind_text(statement, 2, attachmentsJSON, -1, transient) #expect(sqlite3_step(statement) == SQLITE_DONE) sqlite3_finalize(statement) execute(raw, """ INSERT INTO gateway_routing_identity( gateway_id, scope, main_session_key, default_agent_id, updated_at ) VALUES ('gw-a', 'per-sender', 'main', 'main', 10); PRAGMA user_version = 6; """) } let databases = try OpenClawClientDatabases(directoryURL: directory) let command = try #require(await databases.store(gatewayID: "gw-a").loadCommands().first) #expect(command.id == "legacy-v6") #expect(command.attachments == [attachment]) #expect(databases.loadSessionRoutingIdentity(gatewayID: "gw-a")?.contract == "per-sender|main|main") #expect(!FileManager.default.fileExists(atPath: legacyURL.path)) } @Test func `unknown or corrupt legacy files remain untouched`() async throws { let unknownDirectory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: unknownDirectory) } let unknownURL = unknownDirectory.appendingPathComponent("chat-cache.sqlite") try withRawDatabase(at: unknownURL) { raw in execute(raw, "PRAGMA user_version = 999") } _ = try OpenClawClientDatabases(directoryURL: unknownDirectory) #expect(FileManager.default.fileExists(atPath: unknownURL.path)) let corruptDirectory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: corruptDirectory) } let corruptURL = corruptDirectory.appendingPathComponent("chat-cache.sqlite") let bytes = Data("unknown durable format".utf8) try bytes.write(to: corruptURL) let databases = try OpenClawClientDatabases(directoryURL: corruptDirectory) #expect(try Data(contentsOf: corruptURL) == bytes) #expect(try await databases.stateQueue.read { db in try db.tableExists("outbox_commands") }) } } struct ChatCommandOutboxStoreTests { @Test func `nil agent rows use the canonical empty scope owner`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: nil) #expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope)) #expect(await store.enqueueCommand(OpenClawChatOutboxCommand( id: "nil-agent", sessionKey: "main", deliverySessionKey: "main", routingContract: "legacy-unbound", agentID: nil, text: "deliver", thinking: "off", createdAt: Date().timeIntervalSince1970, status: .queued, retryCount: 0, lastError: nil))) let state = try #require(await store.branchState(for: scope)) #expect(state.hadPendingCommands) #expect(await store.claimNextCommand()?.id == "nil-agent") #expect(await store.loadCommands().first?.agentID == nil) } @Test func `parked accepted rows mint retry identity while queued rows keep it`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") let sendingScope = OpenClawChatOutboxScope(sessionKey: "sending", agentID: "main") #expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: sendingScope)) let sendingState = try #require(await store.branchState(for: sendingScope)) #expect(await store.enqueueCommand(outboxCommand(id: "sending", sessionKey: "sending", text: "maybe accepted"))) let sending = try #require(await store.claimNextCommand()) _ = try #require(await store.reconcileBranchScope( sendingScope, previousState: sendingState, activeLeafEntryID: "leaf-b", branchLeafEntryIDs: ["leaf-b"], lastError: "branch changed")) let parkedSending = try #require(await store.loadCommands().first(where: { $0.id == sending.id })) #expect(await store.markCommandRetriedIfPresent( id: parkedSending.id, expectation: OpenClawChatOutboxRetryExpectation( attemptVersion: parkedSending.attemptVersion, retryCount: parkedSending.retryCount, lastError: parkedSending.lastError), agentID: "main", deliverySessionKey: "sending", routingContract: "per-sender|sending|main", replacementID: "sending-retry") == .updated) let retriedSending = try #require(await store.loadCommands().first(where: { $0.id == "sending-retry" })) #expect(retriedSending.attemptVersion == 1) let queuedScope = OpenClawChatOutboxScope(sessionKey: "queued", agentID: "main") #expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: queuedScope)) let queuedState = try #require(await store.branchState(for: queuedScope)) #expect(await store.enqueueCommand(outboxCommand(id: "queued", sessionKey: "queued", text: "not sent"))) _ = try #require(await store.reconcileBranchScope( queuedScope, previousState: queuedState, activeLeafEntryID: "leaf-b", branchLeafEntryIDs: ["leaf-b"], lastError: "branch changed")) let parkedQueued = try #require(await store.loadCommands().first(where: { $0.id == "queued" })) #expect(await store.markCommandRetriedIfPresent( id: parkedQueued.id, expectation: OpenClawChatOutboxRetryExpectation( attemptVersion: parkedQueued.attemptVersion, retryCount: parkedQueued.retryCount, lastError: parkedQueued.lastError), agentID: "main", deliverySessionKey: "queued", routingContract: "per-sender|queued|main", replacementID: "unused") == .updated) #expect(await store.loadCommands().contains(where: { $0.id == "queued" && $0.attemptVersion == 2 })) let stickyDirectory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: stickyDirectory) } let stickyStore = try OpenClawClientDatabases(directoryURL: stickyDirectory).store(gatewayID: "gw-a") let requeuedScope = OpenClawChatOutboxScope(sessionKey: "requeued", agentID: "main") #expect(await stickyStore.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: requeuedScope)) let requeuedState = try #require(await stickyStore.branchState(for: requeuedScope)) #expect(await stickyStore.enqueueCommand(outboxCommand( id: "requeued", sessionKey: "requeued", text: "uncertain"))) let claimed = try #require(await stickyStore.claimNextCommand()) #expect(await stickyStore.markCommandQueued( id: claimed.id, attemptVersion: claimed.attemptVersion, retryCount: 1, lastError: "transport") == .updated) _ = try #require(await stickyStore.reconcileBranchScope( requeuedScope, previousState: requeuedState, activeLeafEntryID: "leaf-b", branchLeafEntryIDs: ["leaf-b"], activeTranscriptEntryIDs: [], lastError: "branch changed")) let parkedRequeued = try #require(await stickyStore.loadCommands().first(where: { $0.id == "requeued" })) #expect(await stickyStore.markCommandRetriedIfPresent( id: parkedRequeued.id, expectation: OpenClawChatOutboxRetryExpectation( attemptVersion: parkedRequeued.attemptVersion, retryCount: parkedRequeued.retryCount, lastError: parkedRequeued.lastError), agentID: "main", deliverySessionKey: "requeued", routingContract: "per-sender|requeued|main", replacementID: "requeued-retry") == .updated) #expect(await stickyStore.loadCommands() .contains(where: { $0.id == "requeued-retry" && $0.attemptVersion == 1 })) } @Test func `empty root reconcile permits first row and parks a wiped scope`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") let root = OpenClawChatOutboxScope(sessionKey: "root", agentID: "main") let rootState = try #require(await store.branchState(for: root)) _ = try #require(await store.reconcileBranchScope( root, previousState: rootState, activeLeafEntryID: nil, branchLeafEntryIDs: [], lastError: "branch changed")) #expect(await store.branchState(for: root)?.lastActiveLeafEntryID == nil) #expect(await store.enqueueCommand(outboxCommand(id: "first", sessionKey: "root", text: "first"))) #expect(await store.claimNextCommand()?.id == "first") let wiped = OpenClawChatOutboxScope(sessionKey: "wiped", agentID: "main") #expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: wiped)) let wipedState = try #require(await store.branchState(for: wiped)) #expect(await store.enqueueCommand(outboxCommand(id: "wiped", sessionKey: "wiped", text: "park"))) _ = try #require(await store.reconcileBranchScope( wiped, previousState: wipedState, activeLeafEntryID: nil, branchLeafEntryIDs: [], lastError: "branch changed")) #expect(await store.loadCommands().first(where: { $0.id == "wiped" })?.status == .failed) } @Test func `nonancestral parking gives failed uncertain rows a fresh retry identity`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "main") #expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope)) let state = try #require(await store.branchState(for: scope)) var failed = outboxCommand(id: "failed", text: "retry") failed.status = .failed failed.lastError = "transport" #expect(await store.enqueueCommand(failed)) try await databases.stateQueue.write { db in try db.execute( sql: "UPDATE outbox_commands SET had_unacknowledged_send = 1 WHERE client_uuid = ?", arguments: ["failed"]) } _ = try #require(await store.reconcileBranchScope( scope, previousState: state, activeLeafEntryID: "leaf-b", branchLeafEntryIDs: ["leaf-b"], activeTranscriptEntryIDs: [], lastError: "branch changed")) let parked = try #require(await store.loadCommands().first) #expect(await store.markCommandRetriedIfPresent( id: parked.id, expectation: OpenClawChatOutboxRetryExpectation( attemptVersion: parked.attemptVersion, retryCount: parked.retryCount, lastError: parked.lastError), agentID: "main", deliverySessionKey: "main", routingContract: "per-sender|main|main", replacementID: "failed-retry") == .updated) #expect(await store.loadCommands().first?.id == "failed-retry") } @Test func `bulk branch parking invalidates sibling subscribers`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") let sibling = databases.store(gatewayID: "gw-a") let otherGateway = databases.store(gatewayID: "gw-b") let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "main") #expect(await store.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope)) #expect(await store.enqueueCommand(outboxCommand(id: "shared", text: "park me"))) let changes = sibling.changes() let otherChanges = otherGateway.changes() var iterator = changes.makeAsyncIterator() _ = try #require(await store.confirmBranchChange( scope, activeLeafEntryID: "leaf-b", lastError: "branch changed")) #expect(await iterator.next() == .invalidated(gatewayID: "gw-a", scope: scope)) let crossGatewayDelivered = await withTaskGroup(of: Bool.self) { group in group.addTask { var iterator = otherChanges.makeAsyncIterator() return await iterator.next() != nil } group.addTask { try? await Task.sleep(for: .milliseconds(25)) return false } let result = await group.next() ?? true group.cancelAll() return result } #expect(crossGatewayDelivered == false) #expect(await sibling.loadCommands().first?.status == .failed) } @Test func `retiring a store ends only its forwarded change streams`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let retired = databases.store(gatewayID: "gw-a") let sibling = databases.store(gatewayID: "gw-a") var retiredIterator = retired.changes().makeAsyncIterator() var siblingIterator = sibling.changes().makeAsyncIterator() await retired.retire() #expect(await retiredIterator.next() == nil) let scope = OpenClawChatOutboxScope(sessionKey: "main", agentID: "main") #expect(await sibling.updateLastActiveLeafEntryID("leaf-a", expectedEpoch: 0, for: scope)) #expect(await sibling.enqueueCommand(outboxCommand(id: "sibling", text: "park"))) _ = try #require(await sibling.confirmBranchChange( scope, activeLeafEntryID: "leaf-b", lastError: "branch changed")) #expect(await siblingIterator.next() == .invalidated(gatewayID: "gw-a", scope: scope)) } @Test func `commands and attachment blobs round trip in order`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") let attachment = OpenClawChatOutboxAttachment( type: "audio", mimeType: "audio/m4a", fileName: "note.m4a", data: Data([4, 5, 6]), durationSeconds: 1.5) #expect(await store.enqueueCommand(outboxCommand( id: "later", text: "two", attachments: [attachment], createdAt: 2))) #expect(await store.enqueueCommand(outboxCommand(id: "earlier", text: "one", createdAt: 1))) let commands = await store.loadCommands() #expect(commands.map(\.id) == ["earlier", "later"]) #expect(commands[1].attachments == [attachment]) #expect(try await databases.stateQueue.read { db in try Data.fetchOne(db, sql: "SELECT payload FROM outbox_attachments WHERE command_id = 'later'") } == attachment.data) } @Test func `claims are insertion FIFO when timestamps tie and exclusive`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") let now = Date().timeIntervalSince1970 #expect(await store.enqueueCommand(outboxCommand(id: "z-first", text: "one", createdAt: now))) #expect(await store.enqueueCommand(outboxCommand(id: "a-second", text: "two", createdAt: now))) let first = try #require(await store.claimNextCommand()) #expect(first.id == "z-first") #expect(await store.claimNextCommand() == nil) #expect(await store.markCommandAwaitingConfirmation( id: "z-first", attemptVersion: first.attemptVersion) == .updated) #expect(await store.claimNextCommand()?.id == "a-second") } @Test func `cancellation stops only unclaimed client state`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "queued", text: "delete"))) #expect(await store.cancelCommand(id: "queued") == .updated) #expect(await store.loadCommands().isEmpty) #expect(await store.enqueueCommand(outboxCommand(id: "claimed", text: "send"))) #expect(await store.claimNextCommand()?.id == "claimed") #expect(await store.cancelCommand(id: "claimed") == .missing) #expect(await store.loadCommands().map(\.status) == [.sending]) } @Test func `canonical proof wins a cancellation race`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "landed", text: "sent"))) store.observeCanonicalMessageIdempotencyKeys(["landed:user"]) #expect(await store.cancelCommand(id: "landed") == .confirmed) #expect(await store.loadCommands().isEmpty) } @Test func `interrupted sends fail closed once`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "interrupted", text: "maybe sent"))) #expect(await store.claimNextCommand()?.status == .sending) #expect(await store.recoverInterruptedSends()) let recovered = try #require(await store.loadCommands().first) #expect(recovered.status == .failed) #expect(recovered.lastError == OpenClawChatSQLiteTranscriptCache.outboxUnconfirmedError) #expect(await store.recoverInterruptedSends()) } @Test func `retired facade cannot recover a replacement facades sends`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let retired = databases.store(gatewayID: "gw-a") #expect(await retired.enqueueCommand(outboxCommand(id: "live", text: "sending"))) #expect(await retired.claimNextCommand()?.status == .sending) await retired.retire() #expect(await retired.recoverInterruptedSends() == false) let replacement = databases.store(gatewayID: "gw-a") #expect(await replacement.loadCommands().map(\.status) == [.sending]) } @Test func `retry adopts a fresh verified route`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let store = try OpenClawClientDatabases(directoryURL: directory).store(gatewayID: "gw-a") #expect(await store.enqueueCommand(outboxCommand(id: "retry", text: "again", status: .failed))) let failed = try #require(await store.loadCommands().first) #expect(await store.markCommandRetriedIfPresent( id: "retry", expectation: OpenClawChatOutboxRetryExpectation( attemptVersion: failed.attemptVersion, retryCount: failed.retryCount, lastError: failed.lastError), agentID: "Agent-B", deliverySessionKey: "agent:agent-b:main", routingContract: "per-sender|main|agent-b", replacementID: nil) == .updated) let command = try #require(await store.loadCommands().first) #expect(command.status == .queued) #expect(command.agentID == "agent-b") #expect(command.deliverySessionKey == "agent:agent-b:main") #expect(command.routingContract == "per-sender|main|agent-b") } @Test func `stale queued and acknowledged commands require user action`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let store = databases.store(gatewayID: "gw-a") let old = Date().timeIntervalSince1970 - OpenClawChatSQLiteTranscriptCache.outboxCommandMaxAge - 1 #expect(await store.enqueueCommand(outboxCommand(id: "old-queued", text: "old", createdAt: old))) #expect(await store.enqueueCommand(outboxCommand(id: "old-ack", text: "old ack"))) let acknowledged = try #require(await store.claimNextCommand()) #expect(acknowledged.id == "old-ack") #expect(await store.markCommandAwaitingConfirmation( id: "old-ack", attemptVersion: acknowledged.attemptVersion) == .updated) try await databases.stateQueue.write { db in try db.execute( sql: "UPDATE outbox_commands SET created_at = ? WHERE gateway_id = ? AND client_uuid = ?", arguments: [old, "gw-a", "old-ack"]) } let commands = await store.loadCommands() let commandsByID = Dictionary(uniqueKeysWithValues: commands.map { ($0.id, $0) }) #expect(commandsByID["old-queued"]?.status == .failed) #expect(commandsByID["old-queued"]?.lastError == OpenClawChatSQLiteTranscriptCache.outboxExpiredError) #expect(commandsByID["old-ack"]?.status == .failed) #expect(commandsByID["old-ack"]?.lastError == OpenClawChatSQLiteTranscriptCache.outboxUnconfirmedError) } @Test func `queue and attachment budgets are gateway scoped`() async throws { let directory = try makeDatabaseDirectory() defer { try? FileManager.default.removeItem(at: directory) } let databases = try OpenClawClientDatabases(directoryURL: directory) let storeA = databases.store(gatewayID: "gw-a") let storeB = databases.store(gatewayID: "gw-b") for index in 0..