From d2ff2106792b254dc1449eafcd2ba49ec009f9a1 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 2 Aug 2026 21:51:33 -0700 Subject: [PATCH] test(android): compact chat persistence fixtures (#118456) --- .../app/chat/ChatControllerOutboxTest.kt | 211 ++------- .../openclaw/app/chat/ClientDatabasesTest.kt | 214 +++------ .../app/chat/RoomChatCommandOutboxTest.kt | 417 +++++++----------- .../app/chat/RoomChatTranscriptCacheTest.kt | 216 ++++----- 4 files changed, 369 insertions(+), 689 deletions(-) diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerOutboxTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerOutboxTest.kt index 7b206d633137..f34839521a36 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerOutboxTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerOutboxTest.kt @@ -502,6 +502,21 @@ class ChatControllerOutboxTest { OutboxScenario(this, capacity).block() } + private fun ChatController.singleOutboxStatus(): ChatOutboxStatus = + outboxItems.value + .single() + .status + + private fun FakeCommandOutbox.singleStatus(): ChatOutboxStatus = + rows.values + .single() + .status + + private fun FakeCommandOutbox.statusFor(text: String): ChatOutboxStatus = + rows.values + .first { it.text == text } + .status + @Test fun enqueueWhileOfflineShowsQueuedRowAndSurvivesControllerRecreation() = outboxTest { @@ -566,12 +581,7 @@ class ChatControllerOutboxTest { runCurrent() assertTrue(gateway.sentMessages.isEmpty()) - assertEquals( - ChatOutboxStatus.Queued, - chat.outboxItems.value - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Queued, chat.singleOutboxStatus()) gateway.settingsPatchGate?.complete(Unit) advanceUntilIdle() @@ -598,12 +608,7 @@ class ChatControllerOutboxTest { runCurrent() assertTrue(gateway.sentMessages.isEmpty()) - assertEquals( - ChatOutboxStatus.Queued, - chat.outboxItems.value - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Queued, chat.singleOutboxStatus()) gateway.settingsPatchGate?.complete(Unit) advanceUntilIdle() @@ -682,18 +687,8 @@ class ChatControllerOutboxTest { assertEquals(listOf("accepted"), gateway.sentMessages) assertFalse(chat.healthOk.value) - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .first { it.text == "accepted" } - .status, - ) - assertEquals( - ChatOutboxStatus.Queued, - outbox.rows.values - .first { it.text == "younger" } - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.statusFor("accepted")) + assertEquals(ChatOutboxStatus.Queued, outbox.statusFor("younger")) outbox.acceptedStatusUpdateFailure = null chat.handleGatewayEvent("health", null) @@ -708,12 +703,7 @@ class ChatControllerOutboxTest { val parked = outbox.rows.values.first { it.text == "accepted" } assertEquals(ChatOutboxStatus.Failed, parked.status) assertEquals(OUTBOX_DELIVERY_UNCONFIRMED_ERROR, parked.lastError) - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .first { it.text == "younger" } - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.statusFor("younger")) } @Test @@ -902,18 +892,8 @@ class ChatControllerOutboxTest { assertEquals(listOf("ambiguous"), gateway.sentMessages) assertFalse(chat.healthOk.value) - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .first { it.text == "ambiguous" } - .status, - ) - assertEquals( - ChatOutboxStatus.Queued, - outbox.rows.values - .first { it.text == "younger" } - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.statusFor("ambiguous")) + assertEquals(ChatOutboxStatus.Queued, outbox.statusFor("younger")) outbox.failedStatusUpdateFailure = null gateway.sendResponse = { key -> """{"runId":"$key","status":"started"}""" } @@ -930,12 +910,7 @@ class ChatControllerOutboxTest { restarted.handleGatewayEvent("health", null) advanceUntilIdle() assertEquals(listOf("ambiguous", "younger"), gateway.sentMessages) - assertEquals( - ChatOutboxStatus.Failed, - restarted.outboxItems.value - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Failed, restarted.singleOutboxStatus()) } @Test @@ -984,18 +959,8 @@ class ChatControllerOutboxTest { assertTrue(gateway.sentMessages.isEmpty()) assertFalse(chat.healthOk.value) - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .first { it.text == "older" } - .status, - ) - assertEquals( - ChatOutboxStatus.Queued, - outbox.rows.values - .first { it.text == "younger" } - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.statusFor("older")) + assertEquals(ChatOutboxStatus.Queued, outbox.statusFor("younger")) outbox.queuedStatusUpdateFailure = null gateway.sendFailureBeforeDispatch = null @@ -1039,12 +1004,7 @@ class ChatControllerOutboxTest { advanceUntilIdle() assertEquals(1, gateway.sentMessages.size) - assertEquals( - ChatOutboxStatus.Failed, - restarted.outboxItems.value - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Failed, restarted.singleOutboxStatus()) restarted.retryOutboxCommand(ambiguous.id) advanceUntilIdle() @@ -1068,12 +1028,7 @@ class ChatControllerOutboxTest { advanceUntilIdle() assertTrue(gateway.sentMessages.isEmpty()) - assertEquals( - ChatOutboxStatus.Failed, - chat.outboxItems.value - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Failed, chat.singleOutboxStatus()) chat.retryOutboxCommand("migrated-ambiguous") advanceUntilIdle() @@ -1213,12 +1168,7 @@ class ChatControllerOutboxTest { restarted.handleGatewayEvent("health", null) advanceUntilIdle() assertEquals(1, gateway.sentMessages.size) - assertEquals( - ChatOutboxStatus.Failed, - restarted.outboxItems.value - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Failed, restarted.singleOutboxStatus()) restarted.retryOutboxCommand(ambiguous.id) advanceUntilIdle() @@ -1486,12 +1436,7 @@ class ChatControllerOutboxTest { advanceUntilIdle() assertEquals(listOf("interrupted send"), gateway.sentMessages) - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.singleStatus()) processJob.cancel() gateway.sendFailureAfterDispatch = null @@ -1702,12 +1647,7 @@ class ChatControllerOutboxTest { // The row keeps its 'sending' claim; the next process surfaces it as delivery-unconfirmed // instead of silently replaying a possibly delivered dispatch. - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.singleStatus()) gateway.sendFailureAfterDispatch = null gateway.echoDeliveredSendsInHistory = false val restarted = controller() @@ -1734,12 +1674,7 @@ class ChatControllerOutboxTest { gateway.online = true first.handleGatewayEvent("health", null) runCurrent() - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) processJob.cancel() // The next process proves the turn against canonical history and retires the row @@ -1766,12 +1701,7 @@ class ChatControllerOutboxTest { gateway.online = true first.handleGatewayEvent("health", null) runCurrent() - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) processJob.cancel() // The gateway lost the turn (crash between ACK and transcript write): an idle history @@ -2042,22 +1972,12 @@ class ChatControllerOutboxTest { chat.switchSession("agent:other:main") gateway.sendGate?.complete(Unit) assertTrue(send.await()) - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) advanceTimeBy(120_001) runCurrent() - assertEquals( - ChatOutboxStatus.Failed, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Failed, outbox.singleStatus()) chat.switchSession("agent:main:main") assertEquals(0, chat.pendingRunCount.value) assertTrue(chat.messages.value.none { message -> message.content.any { it.text == "hidden accepted turn" } }) @@ -2086,12 +2006,7 @@ class ChatControllerOutboxTest { advanceTimeBy(120_001) runCurrent() - assertEquals( - ChatOutboxStatus.Failed, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Failed, outbox.singleStatus()) chat.switchSession("agent:main:main") assertEquals(0, chat.pendingRunCount.value) } @@ -2392,12 +2307,7 @@ class ChatControllerOutboxTest { send.join() assertEquals(true, accepted) - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.singleStatus()) assertTrue(gateway.sentMessages.isEmpty()) } @@ -2485,12 +2395,7 @@ class ChatControllerOutboxTest { assertTrue(accepted.await()) assertTrue(chat.healthOk.value) - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) assertEquals(2, sendRequestCount) assertEquals(1, chat.pendingRunCount.value) } @@ -2510,12 +2415,7 @@ class ChatControllerOutboxTest { val accepted = chat.sendMessageAwaitAcceptance(message = "slow turn", thinkingLevel = "off", attachments = emptyList()) advanceTimeBy(1_000) assertTrue(accepted) - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) // A follow-up send must see the accepted head as live-owned: it dispatches directly, // and the reconciliation sweep must not park the head while its run is in flight. @@ -2541,12 +2441,7 @@ class ChatControllerOutboxTest { chat.handleGatewayEvent("health", null) advanceTimeBy(5_000) assertEquals(listOf("queued turn"), gateway.sentMessages) - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) // The run completes under the acknowledged id and its turn becomes visible in // canonical history. The adopted send must resolve with the live run: without the @@ -2697,12 +2592,7 @@ class ChatControllerOutboxTest { gateway.echoDeliveredSendsInHistory = false chat.sendMessageAwaitAcceptance(message = "never confirmed", thinkingLevel = "off", attachments = emptyList()) runCurrent() - assertEquals( - ChatOutboxStatus.Accepted, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Accepted, outbox.singleStatus()) // Run ownership expires without proof; the row surfaces for manual review. advanceUntilIdle() @@ -2729,12 +2619,7 @@ class ChatControllerOutboxTest { chat.sendMessageAwaitAcceptance(message = "survives caller death", thinkingLevel = "off", attachments = emptyList()) } runCurrent() - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.singleStatus()) callerJob.cancel() gate.complete(Unit) advanceUntilIdle() @@ -2758,12 +2643,7 @@ class ChatControllerOutboxTest { val accepted = chat.sendMessageAwaitAcceptance(message = "owned by flush", thinkingLevel = "off", attachments = emptyList()) advanceUntilIdle() assertTrue(accepted) - assertEquals( - ChatOutboxStatus.Queued, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Queued, outbox.singleStatus()) assertTrue(gateway.sentMessages.isEmpty()) assertFalse(chat.healthOk.value) @@ -2788,12 +2668,7 @@ class ChatControllerOutboxTest { outbox.acceptedStatusUpdateFailure = IllegalStateException("storage unavailable") chat.sendMessageAwaitAcceptance(message = "stranded claim", thinkingLevel = "off", attachments = emptyList()) runCurrent() - assertEquals( - ChatOutboxStatus.Sending, - outbox.rows.values - .single() - .status, - ) + assertEquals(ChatOutboxStatus.Sending, outbox.singleStatus()) assertFalse(chat.healthOk.value) // The re-armed recovery sweep parks the row on the next health transition, so the diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/ClientDatabasesTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/ClientDatabasesTest.kt index 0e4fcbda348e..256c7939a66a 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/ClientDatabasesTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/ClientDatabasesTest.kt @@ -18,8 +18,7 @@ class ClientDatabasesTest { fun deferredOutboxPersistsAtomicMutationDemotion() = runTest { val names = databaseNames() - val databases = open(names, registeredGatewayIds = setOf("gateway-a")) - try { + withCleanDatabases(names) { databases -> val outbox = databases.commandOutbox() val scope = ChatOutboxScope("main", "main") val lease = requireNotNull(outbox.beginSessionMutation("gateway-a", scope, nowMs = 1_000)) @@ -31,9 +30,6 @@ class ClientDatabasesTest { val persisted = requireNotNull(outbox.branchState("gateway-a", scope)) assertTrue(persisted.needsReconciliation) assertNull(persisted.switchPendingSinceMs) - } finally { - databases.close() - delete(names) } } @@ -44,8 +40,7 @@ class ClientDatabasesTest { val context = RuntimeEnvironment.getApplication() createV2Fixture(context.getDatabasePath(names.legacy).path) - val databases = open(names, registeredGatewayIds = setOf("gateway-test")) - try { + withCleanDatabases(names, setOf("gateway-test")) { databases -> assertEquals( 2, databases @@ -95,9 +90,6 @@ class ClientDatabasesTest { assertFalse(context.getDatabasePath(names.legacy).exists()) assertTrue(context.getDatabasePath(names.cache).exists()) assertTrue(context.getDatabasePath(names.state).exists()) - } finally { - databases.close() - delete(names) } } @@ -110,27 +102,20 @@ class ClientDatabasesTest { val bytes = ByteArray((OUTBOX_ATTACHMENT_CHUNK_BYTES * 9) + 77) { (it % 127).toByte() } addV8AttachmentFixture(names.legacy, bytes) - val first = open(names, registeredGatewayIds = setOf("gateway-test")) - try { + withDatabases(names, setOf("gateway-test")) { first -> val loaded = first.commandOutbox().loadAttachments("media-command") assertEquals(1, loaded.size) assertTrue(bytes.contentEquals(loaded.single().bytes)) assertTrue(first.commandOutbox().wasAdmitted("media-command")) - } finally { - first.close() } // A fresh open reads only client-state.db. The completion marker prevents a stale legacy // file from being imported twice if deletion was interrupted. - val reopened = open(names, registeredGatewayIds = setOf("gateway-test")) - try { + withCleanDatabases(names, setOf("gateway-test")) { reopened -> val loaded = reopened.commandOutbox().loadAttachments("media-command") assertEquals(1, loaded.size) assertTrue(bytes.contentEquals(loaded.single().bytes)) assertTrue(reopened.commandOutbox().wasAdmitted("media-command")) - } finally { - reopened.close() - delete(names) } } @@ -139,39 +124,23 @@ class ClientDatabasesTest { runTest { val names = databaseNames() val context = RuntimeEnvironment.getApplication() - val first = open(names, registeredGatewayIds = setOf("gateway-a")) - try { + withDatabases(names) { first -> first.transcriptCache().saveTranscript( gatewayId = "gateway-a", agentId = "main", sessionKey = "main", messages = listOf(cachedMessage("cache me")), ) - assertTrue( - first.commandOutbox().enqueue( - gatewayId = "gateway-a", - sessionKey = "main", - text = "preserve me", - thinkingLevel = "off", - nowMs = 1, - ownerAgentId = "main", - ) is ChatOutboxEnqueueResult.Queued, - ) - } finally { - first.close() + first.enqueue("gateway-a", "preserve me") } SQLiteDatabase.openDatabase(context.getDatabasePath(names.cache).path, null, SQLiteDatabase.OPEN_READWRITE).use { it.version = 99 } - val reopened = open(names, registeredGatewayIds = setOf("gateway-a")) - try { + withCleanDatabases(names) { reopened -> assertTrue(reopened.transcriptCache().loadTranscript("gateway-a", "main", "main").isEmpty()) assertEquals(listOf("preserve me"), reopened.commandOutbox().load("gateway-a").map { it.text }) - } finally { - reopened.close() - delete(names) } } @@ -180,11 +149,8 @@ class ClientDatabasesTest { runTest { val names = databaseNames() val context = RuntimeEnvironment.getApplication() - val first = open(names, registeredGatewayIds = setOf("gateway-a")) - try { + withDatabases(names) { first -> seedGateway(first, "gateway-a", "preserve") - } finally { - first.close() } val statePath = context.getDatabasePath(names.state).path @@ -192,7 +158,7 @@ class ClientDatabasesTest { it.version = 99 } - val failedOpen = open(names, registeredGatewayIds = setOf("gateway-a")) + val failedOpen = open(names) val failure = runCatching { failedOpen.clientStateDatabase() } failedOpen.close() assertTrue(failure.isFailure) @@ -207,24 +173,17 @@ class ClientDatabasesTest { fun absentGatewayCommitsStagedRemovalAcrossBothDatabasesAndKeepsOtherGateway() = runTest { val names = databaseNames() - val first = open(names, registeredGatewayIds = setOf("gateway-a", "gateway-b")) - try { + withDatabases(names, setOf("gateway-a", "gateway-b")) { first -> seedGateway(first, "gateway-a", "remove") seedGateway(first, "gateway-b", "keep") first.stageGatewayRemoval("gateway-a") - } finally { - first.close() } - val reopened = open(names, registeredGatewayIds = setOf("gateway-b")) - try { + withCleanDatabases(names, setOf("gateway-b")) { reopened -> assertTrue(reopened.transcriptCache().loadTranscript("gateway-a", "main", "main").isEmpty()) assertTrue(reopened.commandOutbox().load("gateway-a").isEmpty()) assertEquals(listOf("keep"), reopened.transcriptCache().loadTranscript("gateway-b", "main", "main").map { it.content.single().text }) assertEquals(listOf("keep"), reopened.commandOutbox().load("gateway-b").map { it.text }) - } finally { - reopened.close() - delete(names) } } @@ -232,33 +191,20 @@ class ClientDatabasesTest { fun cachePendingRemovalNeverDeletesNewDurableRowsOnResume() = runTest { val names = databaseNames() - val first = open(names, registeredGatewayIds = setOf("gateway-a", "gateway-b")) - try { + withDatabases(names, setOf("gateway-a", "gateway-b")) { first -> seedGateway(first, "gateway-a", "remove") seedGateway(first, "gateway-b", "keep") // Force only the disposable half to fail after the durable state transaction commits. first.gatewayCacheDatabase().close() first.commitGatewayRemoval("gateway-a") assertTrue(first.commandOutbox().load("gateway-a").isEmpty()) - assertTrue( - first.commandOutbox().enqueue( - gatewayId = "gateway-a", - sessionKey = "main", - text = "new after purge", - thinkingLevel = "off", - nowMs = 2, - ownerAgentId = "main", - ) is ChatOutboxEnqueueResult.Queued, - ) + first.enqueue("gateway-a", "new after purge", nowMs = 2) // A retry may stage again before restart; it must not downgrade cache-pending into a // cancelable marker that could strand the old derived rows. first.stageGatewayRemoval("gateway-a") - } finally { - first.close() } - val reopened = open(names, registeredGatewayIds = setOf("gateway-a", "gateway-b")) - try { + withCleanDatabases(names, setOf("gateway-a", "gateway-b")) { reopened -> assertTrue(reopened.transcriptCache().loadTranscript("gateway-a", "main", "main").isEmpty()) assertEquals(listOf("new after purge"), reopened.commandOutbox().load("gateway-a").map { it.text }) assertEquals(listOf("keep"), reopened.transcriptCache().loadTranscript("gateway-b", "main", "main").map { it.content.single().text }) @@ -270,9 +216,6 @@ class ClientDatabasesTest { .gatewayRemovals() .isEmpty(), ) - } finally { - reopened.close() - delete(names) } } @@ -280,21 +223,14 @@ class ClientDatabasesTest { fun stillRegisteredGatewayCancelsCancelableStagedRemoval() = runTest { val names = databaseNames() - val first = open(names, registeredGatewayIds = setOf("gateway-a")) - try { + withDatabases(names) { first -> seedGateway(first, "gateway-a", "keep") first.stageGatewayRemoval("gateway-a") - } finally { - first.close() } - val reopened = open(names, registeredGatewayIds = setOf("gateway-a")) - try { + withCleanDatabases(names) { reopened -> assertEquals(listOf("keep"), reopened.transcriptCache().loadTranscript("gateway-a", "main", "main").map { it.content.single().text }) assertEquals(listOf("keep"), reopened.commandOutbox().load("gateway-a").map { it.text }) - } finally { - reopened.close() - delete(names) } } @@ -309,13 +245,21 @@ class ClientDatabasesTest { sessionKey = "main", messages = listOf(cachedMessage(text)), ) + databases.enqueue(gatewayId, text) + } + + private suspend fun AndroidClientDatabases.enqueue( + gatewayId: String, + text: String, + nowMs: Long = 1, + ) { assertTrue( - databases.commandOutbox().enqueue( + commandOutbox().enqueue( gatewayId = gatewayId, sessionKey = "main", text = text, thinkingLevel = "off", - nowMs = 1, + nowMs = nowMs, ownerAgentId = "main", ) is ChatOutboxEnqueueResult.Queued, ) @@ -370,7 +314,7 @@ class ClientDatabasesTest { private fun open( names: DatabaseNames, - registeredGatewayIds: Set, + registeredGatewayIds: Set = setOf("gateway-a"), ): AndroidClientDatabases = AndroidClientDatabases.start( RuntimeEnvironment.getApplication(), @@ -380,6 +324,30 @@ class ClientDatabasesTest { registeredGatewayIds = registeredGatewayIds, ) + private suspend fun withDatabases( + names: DatabaseNames, + registeredGatewayIds: Set = setOf("gateway-a"), + block: suspend (AndroidClientDatabases) -> T, + ): T { + val databases = open(names, registeredGatewayIds) + return try { + block(databases) + } finally { + databases.close() + } + } + + private suspend fun withCleanDatabases( + names: DatabaseNames, + registeredGatewayIds: Set = setOf("gateway-a"), + block: suspend (AndroidClientDatabases) -> T, + ): T = + try { + withDatabases(names, registeredGatewayIds, block) + } finally { + delete(names) + } + private fun databaseNames(): DatabaseNames { val id = UUID.randomUUID().toString() return DatabaseNames( @@ -433,76 +401,40 @@ class ClientDatabasesTest { "VALUES (?, ?, ?, ?, ?, ?, ?)", arrayOf("gateway-test", "main", 0, "assistant", "[\"legacy transcript\"]", 10L, null), ) - insertOutbox(database, id = "pristine", status = "queued", retryCount = 0, lastError = null, createdAtMs = now) - insertOutbox( - database, - id = "legacy-queued-error", - status = "queued", - retryCount = 0, - lastError = "socket closed after send", - createdAtMs = now + 1, - ) - insertOutbox( - database, - id = "interrupted-send", - status = "sending", - retryCount = 1, - lastError = null, - createdAtMs = now + 2, - ) - insertOutbox( - database, - id = "already-failed", - status = "failed", - retryCount = 3, - lastError = "original failure", - createdAtMs = now + 3, - ) - insertOutbox( - database, - id = "legacy-command", - status = "queued", - retryCount = 0, - lastError = null, - createdAtMs = now + 4, - text = "/clear", - ) - insertOutbox( - database, - id = "accepted", - status = "accepted", - retryCount = 0, - lastError = null, - createdAtMs = now + 5, - ) - insertOutbox( - database, - id = "explicit-owner", - status = "queued", - retryCount = 0, - lastError = null, - createdAtMs = now + 6, - sessionKey = "agent:ops:side", - ) + listOf( + LegacyOutboxFixture("pristine", "queued"), + LegacyOutboxFixture("legacy-queued-error", "queued", lastError = "socket closed after send"), + LegacyOutboxFixture("interrupted-send", "sending", retryCount = 1), + LegacyOutboxFixture("already-failed", "failed", retryCount = 3, lastError = "original failure"), + LegacyOutboxFixture("legacy-command", "queued", text = "/clear"), + LegacyOutboxFixture("accepted", "accepted"), + LegacyOutboxFixture("explicit-owner", "queued", sessionKey = "agent:ops:side"), + ).forEachIndexed { index, fixture -> insertOutbox(database, fixture, now + index) } database.version = 2 } } private fun insertOutbox( database: SQLiteDatabase, - id: String, - status: String, - retryCount: Int, - lastError: String?, + fixture: LegacyOutboxFixture, createdAtMs: Long, - text: String = id, - sessionKey: String = "main", ) { database.execSQL( "INSERT INTO outbox_commands " + "(id, gatewayId, sessionKey, text, thinkingLevel, createdAtMs, status, retryCount, lastError) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - arrayOf(id, "gateway-test", sessionKey, text, "off", createdAtMs, status, retryCount, lastError), + fixture.run { + arrayOf(id, "gateway-test", sessionKey, text, "off", createdAtMs, status, retryCount, lastError) + }, ) } + + private data class LegacyOutboxFixture( + val id: String, + val status: String, + val retryCount: Int = 0, + val lastError: String? = null, + val text: String = id, + val sessionKey: String = "main", + ) } diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatCommandOutboxTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatCommandOutboxTest.kt index 6367e59d2249..1e589b8eb4ae 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatCommandOutboxTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatCommandOutboxTest.kt @@ -26,6 +26,29 @@ class RoomChatCommandOutboxTest { database.close() } + private suspend fun ChatCommandOutbox.enqueueResult( + text: String, + nowMs: Long, + gatewayId: String = "gateway-a", + sessionKey: String = "main", + thinkingLevel: String = "off", + ownerAgentId: String = "main", + idempotencyKey: String? = null, + attachments: List = emptyList(), + gatedEpoch: Long? = null, + ): ChatOutboxEnqueueResult = + enqueue( + gatewayId = gatewayId, + sessionKey = sessionKey, + text = text, + thinkingLevel = thinkingLevel, + nowMs = nowMs, + ownerAgentId = ownerAgentId, + idempotencyKey = idempotencyKey, + attachments = attachments, + gatedEpoch = gatedEpoch, + ) + private suspend fun ChatCommandOutbox.enqueueQueued( text: String, nowMs: Long, @@ -33,17 +56,79 @@ class RoomChatCommandOutboxTest { sessionKey: String = "main", thinkingLevel: String = "off", ownerAgentId: String = "main", - ): ChatOutboxItem { - val result = - enqueue( + idempotencyKey: String? = null, + attachments: List = emptyList(), + gatedEpoch: Long? = null, + ): ChatOutboxItem = + ( + enqueueResult( + text = text, + nowMs = nowMs, gatewayId = gatewayId, sessionKey = sessionKey, - text = text, thinkingLevel = thinkingLevel, - nowMs = nowMs, ownerAgentId = ownerAgentId, - ) - return (result as ChatOutboxEnqueueResult.Queued).item + idempotencyKey = idempotencyKey, + attachments = attachments, + gatedEpoch = gatedEpoch, + ) as ChatOutboxEnqueueResult.Queued + ).item + + private suspend fun ChatCommandOutbox.requeueCurrent( + item: ChatOutboxItem, + nowMs: Long, + replacementId: String, + ): Int = + requeueForRetryIfCurrent( + gatewayId = "gateway-a", + id = item.id, + expectedAttemptVersion = item.attemptVersion, + expectedRetryCount = item.retryCount, + expectedLastError = item.lastError, + nowMs = nowMs, + gatedEpoch = null, + ownerAgentId = "main", + replacementId = replacementId, + ) + + private suspend fun ChatCommandOutbox.reconcile( + scope: ChatOutboxScope, + previousState: ChatOutboxBranchState, + activeLeafEntryId: String? = null, + branchLeafEntryIds: Set = emptySet(), + activeTranscriptEntryIds: Set = emptySet(), + ): Boolean = + reconcileBranchScope( + gatewayId = "gateway-a", + scope = scope, + previousState = previousState, + activeLeafEntryId = activeLeafEntryId, + branchLeafEntryIds = branchLeafEntryIds, + activeTranscriptEntryIds = activeTranscriptEntryIds, + lastError = OUTBOX_BRANCH_CHANGED_ERROR, + ) + + private suspend fun insertLegacyCommand( + id: String, + status: ChatOutboxStatus, + retryCount: Int, + lastError: String?, + ) { + database.outboxDao().insert( + OutboxCommandEntity( + id = id, + gatewayId = "gateway-a", + sessionKey = "main", + text = "legacy", + thinkingLevel = "off", + createdAtMs = 10, + status = status.dbValue, + retryCount = retryCount, + lastError = lastError, + gatedEpoch = null, + ownerAgentId = "main", + ), + ) } @Test @@ -69,17 +154,14 @@ class RoomChatCommandOutboxTest { fun callerSuppliedIdempotencyKeyCanReconcileComposerAdmissionAfterRestart() = runTest { val result = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "agent:main:device", + store.enqueueQueued( text = "send once", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", + sessionKey = "agent:main:device", idempotencyKey = "composer-command-a", - ) as ChatOutboxEnqueueResult.Queued + ) - assertEquals("composer-command-a", result.item.id) + assertEquals("composer-command-a", result.id) assertTrue(store.wasAdmitted("composer-command-a")) store.delete("composer-command-a") assertTrue(store.wasAdmitted("composer-command-a")) @@ -91,13 +173,10 @@ class RoomChatCommandOutboxTest { runTest { repeat(OUTBOX_ADMISSION_RECEIPTS_PER_ROUTING_OWNER + 2) { index -> val id = "composer-command-$index" - store.enqueue( - gatewayId = "gateway-a", + store.enqueueQueued( sessionKey = "agent:main:device-$index", text = "message $index", - thinkingLevel = "off", nowMs = index.toLong(), - ownerAgentId = "main", idempotencyKey = id, ) store.delete(id) @@ -114,24 +193,18 @@ class RoomChatCommandOutboxTest { fun activeAdmissionReceiptSurvivesFallbackPruningUntilCommandRetires() = runTest { val protectedId = "active-checkpoint" - store.enqueue( - gatewayId = "gateway-a", + store.enqueueQueued( sessionKey = "agent:main:protected", text = "still pending", - thinkingLevel = "off", nowMs = 0, - ownerAgentId = "main", idempotencyKey = protectedId, ) repeat(OUTBOX_ADMISSION_RECEIPTS_PER_ROUTING_OWNER + 2) { index -> val id = "retired-command-$index" - store.enqueue( - gatewayId = "gateway-a", + store.enqueueQueued( sessionKey = "agent:main:device-$index", text = "message $index", - thinkingLevel = "off", nowMs = index.toLong() + 1, - ownerAgentId = "main", idempotencyKey = id, ) store.delete(id) @@ -140,13 +213,10 @@ class RoomChatCommandOutboxTest { store.delete(protectedId) assertTrue(store.wasAdmitted(protectedId)) val nextId = "next-retired-command" - store.enqueue( - gatewayId = "gateway-a", + store.enqueueQueued( sessionKey = "agent:main:next", text = "advance the recovery window", - thinkingLevel = "off", nowMs = 100, - ownerAgentId = "main", idempotencyKey = nextId, ) store.delete(nextId) @@ -160,15 +230,7 @@ class RoomChatCommandOutboxTest { store.enqueueQueued("m$index", nowMs = index.toLong()) } - val refused = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", - text = "overflow", - thinkingLevel = "off", - nowMs = 999, - ownerAgentId = "main", - ) + val refused = store.enqueueResult(text = "overflow", nowMs = 999) assertEquals(ChatOutboxEnqueueResult.QueueFull, refused) assertEquals(OUTBOX_MAX_QUEUED, store.load("gateway-a").size) @@ -227,21 +289,7 @@ class RoomChatCommandOutboxTest { @Test fun restartRecoveryCreatesAmbiguityStateForRowsWithoutDeliveryMetadata() = runTest { - database.outboxDao().insert( - OutboxCommandEntity( - id = "legacy-sending", - gatewayId = "gateway-a", - sessionKey = "main", - text = "legacy", - thinkingLevel = "off", - createdAtMs = 10, - status = ChatOutboxStatus.Sending.dbValue, - retryCount = 0, - lastError = null, - gatedEpoch = null, - ownerAgentId = "main", - ), - ) + insertLegacyCommand("legacy-sending", ChatOutboxStatus.Sending, retryCount = 0, lastError = null) store.failSendingAfterRestart() @@ -253,37 +301,18 @@ class RoomChatCommandOutboxTest { @Test fun legacyAmbiguousFailureBackfillsFreshRetryIdentityEvidence() = runTest { - database.outboxDao().insert( - OutboxCommandEntity( - id = "legacy-ambiguous", - gatewayId = "gateway-a", - sessionKey = "main", - text = "legacy", - thinkingLevel = "off", - createdAtMs = 10, - status = ChatOutboxStatus.Failed.dbValue, - retryCount = 1, - lastError = OUTBOX_DELIVERY_UNCONFIRMED_ERROR, - gatedEpoch = null, - ownerAgentId = "main", - ), + insertLegacyCommand( + "legacy-ambiguous", + ChatOutboxStatus.Failed, + retryCount = 1, + lastError = OUTBOX_DELIVERY_UNCONFIRMED_ERROR, ) val legacy = store.load("gateway-a").single() assertTrue(legacy.hadUnacknowledgedSend) store.confirmBranchChange("gateway-a", ChatOutboxScope("main", "main"), "leaf-new", OUTBOX_BRANCH_CHANGED_ERROR) val parked = store.load("gateway-a").single() - store.requeueForRetryIfCurrent( - gatewayId = "gateway-a", - id = parked.id, - expectedAttemptVersion = parked.attemptVersion, - expectedRetryCount = parked.retryCount, - expectedLastError = parked.lastError, - nowMs = 20, - gatedEpoch = null, - ownerAgentId = "main", - replacementId = "legacy-fresh-id", - ) + store.requeueCurrent(parked, nowMs = 20, replacementId = "legacy-fresh-id") assertEquals("legacy-fresh-id", store.load("gateway-a").single().id) } @@ -355,14 +384,7 @@ class RoomChatCommandOutboxTest { runTest { assertEquals( ChatOutboxEnqueueResult.Unavailable, - store.enqueue( - gatewayId = " ", - sessionKey = "main", - text = "hi", - thinkingLevel = "off", - nowMs = 1, - ownerAgentId = "main", - ), + store.enqueueResult(text = "hi", nowMs = 1, gatewayId = " "), ) assertEquals(emptyList(), store.load(" ")) @@ -398,17 +420,7 @@ class RoomChatCommandOutboxTest { assertEquals( 1, - store.requeueForRetryIfCurrent( - gatewayId = "gateway-a", - id = parkedAccepted.id, - expectedAttemptVersion = parkedAccepted.attemptVersion, - expectedRetryCount = parkedAccepted.retryCount, - expectedLastError = parkedAccepted.lastError, - nowMs = 20, - gatedEpoch = null, - ownerAgentId = "main", - replacementId = "fresh-client-id", - ), + store.requeueCurrent(parkedAccepted, nowMs = 20, replacementId = "fresh-client-id"), ) val retriedAccepted = store.load("gateway-a").single() assertEquals("fresh-client-id", retriedAccepted.id) @@ -418,17 +430,7 @@ class RoomChatCommandOutboxTest { val queued = store.enqueueQueued("never dispatched", nowMs = 30) store.confirmBranchChange("gateway-a", scope, "leaf-newer", OUTBOX_BRANCH_CHANGED_ERROR) val parkedQueued = store.load("gateway-a").single() - store.requeueForRetryIfCurrent( - gatewayId = "gateway-a", - id = parkedQueued.id, - expectedAttemptVersion = parkedQueued.attemptVersion, - expectedRetryCount = parkedQueued.retryCount, - expectedLastError = parkedQueued.lastError, - nowMs = 40, - gatedEpoch = null, - ownerAgentId = "main", - replacementId = "unused-replacement", - ) + store.requeueCurrent(parkedQueued, nowMs = 40, replacementId = "unused-replacement") val retriedQueued = store.load("gateway-a").single() assertEquals(queued.id, retriedQueued.id) assertEquals(2, retriedQueued.attemptVersion) @@ -439,32 +441,18 @@ class RoomChatCommandOutboxTest { runTest { val bytes = ByteArray(OUTBOX_ATTACHMENT_CHUNK_BYTES + 17) { (it % 251).toByte() } val queued = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "attachment retry", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf(payload(bytes, fileName = "proof.jpg")), - ) as ChatOutboxEnqueueResult.Queued - store.updateStatusIfAttempt(queued.item.id, 1, ChatOutboxStatus.Accepted, 0, null) + ) + store.updateStatusIfAttempt(queued.id, 1, ChatOutboxStatus.Accepted, 0, null) store.confirmBranchChange("gateway-a", ChatOutboxScope("main", "main"), "leaf-new", OUTBOX_BRANCH_CHANGED_ERROR) val parked = store.load("gateway-a").single() assertEquals( 1, - store.requeueForRetryIfCurrent( - gatewayId = "gateway-a", - id = parked.id, - expectedAttemptVersion = parked.attemptVersion, - expectedRetryCount = parked.retryCount, - expectedLastError = parked.lastError, - nowMs = 20, - gatedEpoch = null, - ownerAgentId = "main", - replacementId = "fresh-attachment-id", - ), + store.requeueCurrent(parked, nowMs = 20, replacementId = "fresh-attachment-id"), ) val loaded = store.loadAttachments("fresh-attachment-id").single() @@ -572,17 +560,7 @@ class RoomChatCommandOutboxTest { val staleLease = requireNotNull(store.beginSessionMutation("gateway-a", scope, nowMs = 1_000)) assertTrue(store.demoteSessionMutationToReconciliation("gateway-a", scope, lease = null)) val reconciliationState = requireNotNull(store.branchState("gateway-a", scope)) - assertTrue( - store.reconcileBranchScope( - gatewayId = "gateway-a", - scope = scope, - previousState = reconciliationState, - activeLeafEntryId = null, - branchLeafEntryIds = emptySet(), - activeTranscriptEntryIds = emptySet(), - lastError = OUTBOX_BRANCH_CHANGED_ERROR, - ), - ) + assertTrue(store.reconcile(scope, reconciliationState)) val currentLease = requireNotNull(store.beginSessionMutation("gateway-a", scope, nowMs = 2_000)) assertFalse( @@ -606,14 +584,12 @@ class RoomChatCommandOutboxTest { val advanceState = requireNotNull(store.branchState("gateway-a", advancingScope)) val advancingRow = store.enqueueQueued("stay active", nowMs = 10, sessionKey = "advance") assertTrue( - store.reconcileBranchScope( - gatewayId = "gateway-a", - scope = advancingScope, - previousState = advanceState, + store.reconcile( + advancingScope, + advanceState, activeLeafEntryId = "leaf-new", branchLeafEntryIds = setOf("leaf-new"), activeTranscriptEntryIds = setOf("leaf-old", "leaf-new"), - lastError = OUTBOX_BRANCH_CHANGED_ERROR, ), ) assertEquals(ChatOutboxStatus.Queued, store.load("gateway-a").single { it.id == advancingRow.id }.status) @@ -624,14 +600,12 @@ class RoomChatCommandOutboxTest { val switchState = requireNotNull(store.branchState("gateway-a", switchedScope)) val switchedRow = store.enqueueQueued("park me", nowMs = 20, sessionKey = "switched") assertTrue( - store.reconcileBranchScope( - gatewayId = "gateway-a", - scope = switchedScope, - previousState = switchState, + store.reconcile( + switchedScope, + switchState, activeLeafEntryId = "leaf-b", branchLeafEntryIds = setOf("leaf-a", "leaf-b"), activeTranscriptEntryIds = setOf("leaf-b"), - lastError = OUTBOX_BRANCH_CHANGED_ERROR, ), ) assertEquals(ChatOutboxStatus.Failed, store.load("gateway-a").single { it.id == switchedRow.id }.status) @@ -645,17 +619,7 @@ class RoomChatCommandOutboxTest { val mainState = requireNotNull(store.branchState("gateway-a", mainScope)) val opsState = requireNotNull(store.branchState("gateway-a", opsScope)) - assertTrue( - store.reconcileBranchScope( - "gateway-a", - mainScope, - mainState, - activeLeafEntryId = null, - branchLeafEntryIds = emptySet(), - activeTranscriptEntryIds = emptySet(), - lastError = OUTBOX_BRANCH_CHANGED_ERROR, - ), - ) + assertTrue(store.reconcile(mainScope, mainState)) assertTrue(store.confirmBranchChange("gateway-a", mainScope, "main-leaf", OUTBOX_BRANCH_CHANGED_ERROR)) assertEquals(1, store.branchState("gateway-a", mainScope)?.epoch) assertEquals(0, store.branchState("gateway-a", opsScope)?.epoch) @@ -670,14 +634,12 @@ class RoomChatCommandOutboxTest { val admitted = store.enqueueQueued("after snapshot", nowMs = 10) assertTrue( - store.reconcileBranchScope( - gatewayId = "gateway-a", - scope = scope, - previousState = emptyRoot, + store.reconcile( + scope, + emptyRoot, activeLeafEntryId = "leaf-current", branchLeafEntryIds = setOf("leaf-current"), activeTranscriptEntryIds = setOf("leaf-current"), - lastError = OUTBOX_BRANCH_CHANGED_ERROR, ), ) @@ -717,21 +679,14 @@ class RoomChatCommandOutboxTest { @Test fun deleteForSessionRemovesOnlyThatSessionsRows() = runTest { - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "for main", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", idempotencyKey = "main-admission", ) store.enqueueQueued("for other", nowMs = 20, sessionKey = "agent:other:main") - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "other owner", - thinkingLevel = "off", nowMs = 30, ownerAgentId = "other", idempotencyKey = "other-owner-admission", @@ -759,26 +714,22 @@ class RoomChatCommandOutboxTest { val big = ByteArray(OUTBOX_ATTACHMENT_CHUNK_BYTES + 1234) { (it % 251).toByte() } val small = byteArrayOf(5, 4, 3) val queued = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "with media", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf( payload(big, fileName = "big.jpg"), payload(small, fileName = "note.m4a", type = "audio", mimeType = "audio/mp4", durationMs = 900L), ), - ) as ChatOutboxEnqueueResult.Queued + ) val loadedItem = store.load("gateway-a").single() assertEquals(listOf("big.jpg", "note.m4a"), loadedItem.attachments.map { it.fileName }) assertEquals(listOf(big.size.toLong(), small.size.toLong()), loadedItem.attachments.map { it.byteLength }) assertEquals(900L, loadedItem.attachments[1].durationMs) - val loaded = store.loadAttachments(queued.item.id) + val loaded = store.loadAttachments(queued.id) assertTrue(big.contentEquals(loaded[0].bytes)) assertTrue(small.contentEquals(loaded[1].bytes)) } @@ -788,13 +739,9 @@ class RoomChatCommandOutboxTest { runTest { val oversized = ByteArray((OUTBOX_MAX_COMMAND_ATTACHMENT_BYTES + 1).toInt()) val refused = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueResult( text = "too big", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf(payload(oversized)), ) assertEquals(ChatOutboxEnqueueResult.AttachmentsTooLarge, refused) @@ -806,23 +753,15 @@ class RoomChatCommandOutboxTest { runTest { val aboveDefaultCap = ByteArray((OUTBOX_MAX_COMMAND_ATTACHMENT_BYTES + 1L).toInt()) val video = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueResult( text = "video", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf(payload(aboveDefaultCap, type = "video", mimeType = "video/mp4")), ) val document = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueResult( text = "document", - thinkingLevel = "off", nowMs = 11, - ownerAgentId = "main", attachments = listOf(payload(aboveDefaultCap, type = "file", mimeType = "application/pdf")), ) @@ -836,13 +775,9 @@ class RoomChatCommandOutboxTest { runTest { val document = ByteArray(5 * 1024 * 1024) val refused = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueResult( text = "mixed", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf( payload(document, fileName = "one.pdf", type = "file", mimeType = "application/pdf"), @@ -862,13 +797,9 @@ class RoomChatCommandOutboxTest { var index = 0 while (true) { val result = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueResult( text = "bulk $index", - thinkingLevel = "off", nowMs = index.toLong(), - ownerAgentId = "main", attachments = listOf(payload(chunk)), ) if (result !is ChatOutboxEnqueueResult.Queued) { @@ -883,13 +814,9 @@ class RoomChatCommandOutboxTest { // Deleting a queued row releases its bytes, so admission recovers. store.delete(stored.first()) val retried = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueResult( text = "fits again", - thinkingLevel = "off", nowMs = 999, - ownerAgentId = "main", attachments = listOf(payload(chunk)), ) assertTrue(retried is ChatOutboxEnqueueResult.Queued) @@ -898,18 +825,7 @@ class RoomChatCommandOutboxTest { @Test fun conditionalDeleteNeverRemovesAClaimedRow() = runTest { - val first = - ( - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", - text = "delete queued", - thinkingLevel = "off", - nowMs = 1, - ownerAgentId = "main", - idempotencyKey = "rollback-receipt", - ) as ChatOutboxEnqueueResult.Queued - ).item + val first = store.enqueueQueued(text = "delete queued", nowMs = 1, idempotencyKey = "rollback-receipt") assertTrue(store.wasAdmitted("rollback-receipt")) assertTrue(store.deleteIfQueued(first.id)) assertTrue(store.load("gateway-a").isEmpty()) @@ -926,55 +842,45 @@ class RoomChatCommandOutboxTest { runTest { val bytes = byteArrayOf(1, 2, 3) val queued = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "confirmed", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf(payload(bytes)), - ) as ChatOutboxEnqueueResult.Queued - store.updateStatus(queued.item.id, ChatOutboxStatus.Accepted, retryCount = 0, lastError = null) + ) + store.updateStatus(queued.id, ChatOutboxStatus.Accepted, retryCount = 0, lastError = null) val keep = store.enqueueQueued("kept", nowMs = 20) - assertEquals(1, store.confirmDelivered(setOf(queued.item.id, "missing-row"))) + assertEquals(1, store.confirmDelivered(setOf(queued.id, "missing-row"))) assertEquals(listOf(keep.id), store.load("gateway-a").map { it.id }) - assertTrue(store.loadAttachments(queued.item.id).isEmpty()) + assertTrue(store.loadAttachments(queued.id).isEmpty()) } @Test fun clearGatewayAndSessionDeleteAlsoDropAttachmentBytes() = runTest { val a = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "a", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", attachments = listOf(payload(byteArrayOf(1))), - ) as ChatOutboxEnqueueResult.Queued + ) val b = - store.enqueue( - gatewayId = "gateway-b", + store.enqueueQueued( sessionKey = "other", text = "b", - thinkingLevel = "off", nowMs = 20, - ownerAgentId = "main", + gatewayId = "gateway-b", attachments = listOf(payload(byteArrayOf(2))), - ) as ChatOutboxEnqueueResult.Queued + ) store.deleteForSession("gateway-b", "other", "main") store.clearGateway("gateway-a") assertTrue(store.load("gateway-a").isEmpty()) assertTrue(store.load("gateway-b").isEmpty()) - assertTrue(store.loadAttachments(a.item.id).isEmpty()) - assertTrue(store.loadAttachments(b.item.id).isEmpty()) + assertTrue(store.loadAttachments(a.id).isEmpty()) + assertTrue(store.loadAttachments(b.id).isEmpty()) } @Test @@ -989,22 +895,19 @@ class RoomChatCommandOutboxTest { fun retryAndExactSessionDeletionCanonicalizeOwnerAgentIds() = runTest { val queued = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "mixed owner", - thinkingLevel = "off", nowMs = 10, ownerAgentId = "Main", - ) as ChatOutboxEnqueueResult.Queued + ) assertEquals("main", store.load("gateway-a").single().ownerAgentId) - store.updateStatus(queued.item.id, ChatOutboxStatus.Failed, retryCount = 1, lastError = "retry") + store.updateStatus(queued.id, ChatOutboxStatus.Failed, retryCount = 1, lastError = "retry") assertEquals( 1, store.requeueForRetry( gatewayId = "gateway-a", - id = queued.item.id, + id = queued.id, nowMs = 20, gatedEpoch = null, ownerAgentId = "MAIN", @@ -1020,19 +923,15 @@ class RoomChatCommandOutboxTest { fun gatedEpochSurvivesPersistenceAndRetryRestamping() = runTest { val queued = - store.enqueue( - gatewayId = "gateway-a", - sessionKey = "main", + store.enqueueQueued( text = "/clear", - thinkingLevel = "off", nowMs = 10, - ownerAgentId = "main", gatedEpoch = 7L, - ) as ChatOutboxEnqueueResult.Queued + ) assertEquals(7L, store.load("gateway-a").single().gatedEpoch) - store.updateStatus(queued.item.id, ChatOutboxStatus.Failed, retryCount = 0, lastError = OUTBOX_CONNECTION_CHANGED_ERROR) - assertEquals(1, store.requeueForRetry(gatewayId = "gateway-a", id = queued.item.id, nowMs = 20, gatedEpoch = 9L)) + store.updateStatus(queued.id, ChatOutboxStatus.Failed, retryCount = 0, lastError = OUTBOX_CONNECTION_CHANGED_ERROR) + assertEquals(1, store.requeueForRetry(gatewayId = "gateway-a", id = queued.id, nowMs = 20, gatedEpoch = 9L)) assertEquals(9L, store.load("gateway-a").single().gatedEpoch) } diff --git a/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatTranscriptCacheTest.kt b/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatTranscriptCacheTest.kt index 28371d884732..53291f6c8a63 100644 --- a/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatTranscriptCacheTest.kt +++ b/apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatTranscriptCacheTest.kt @@ -16,14 +16,13 @@ class RoomChatTranscriptCacheTest { Room .inMemoryDatabaseBuilder(RuntimeEnvironment.getApplication(), GatewayCacheDatabase::class.java) .build() + private val store = RoomChatTranscriptCache(database = database) @After fun tearDown() { database.close() } - private fun cache(): RoomChatTranscriptCache = RoomChatTranscriptCache(database = database) - private fun message( text: String, role: String = "user", @@ -39,10 +38,34 @@ class RoomChatTranscriptCacheTest { idempotencyKey = idempotencyKey, ) + private suspend fun saveTranscript( + messages: List, + gatewayId: String = "gateway-a", + agentId: String = "main", + sessionKey: String = "main", + ) = store.saveTranscript(gatewayId, agentId, sessionKey, messages) + + private suspend fun loadTranscript( + gatewayId: String = "gateway-a", + agentId: String = "main", + sessionKey: String = "main", + ): List = store.loadTranscript(gatewayId, agentId, sessionKey) + + private suspend fun saveSessions( + sessions: List, + gatewayId: String = "gateway-a", + agentId: String = "main", + retainedSessionKey: String? = null, + ) = store.saveSessions(gatewayId, agentId, sessions, retainedSessionKey) + + private suspend fun loadSessions( + gatewayId: String = "gateway-a", + agentId: String = "main", + ): List = store.loadSessions(gatewayId, agentId) + @Test fun transcriptRoundTripKeepsTextAndManagedReferencesWithoutBinaryParts() = runTest { - val store = cache() val imagePart = ChatMessageContent(type = "image", mimeType = "image/png", fileName = "a.png", base64 = "AAAA") val managedImage = ChatMessageContent( @@ -52,10 +75,7 @@ class RoomChatTranscriptCacheTest { url = "/api/chat/media/outgoing/main/11111111-1111-4111-8111-111111111111/full", alt = "Managed image", ) - store.saveTranscript( - gatewayId = "gateway-a", - agentId = "main", - sessionKey = "main", + saveTranscript( messages = listOf( message("hello", role = "user", timestampMs = 10, idempotencyKey = "run-1:user", extraParts = listOf(imagePart)), @@ -66,7 +86,7 @@ class RoomChatTranscriptCacheTest { ), ) - val loaded = store.loadTranscript("gateway-a", "main", "main") + val loaded = loadTranscript() assertEquals(listOf("hello", null, "world"), loaded.map { it.content.single().text }) assertTrue(loaded.all { message -> message.content.all { part -> part.base64 == null } }) @@ -79,7 +99,6 @@ class RoomChatTranscriptCacheTest { @Test fun transcriptRoundTripKeepsManagedAudioAndVideoMetadata() = runTest { - val store = cache() val audio = ChatMessageContent( type = "audio", @@ -99,10 +118,7 @@ class RoomChatTranscriptCacheTest { width = 1920, height = 1080, ) - store.saveTranscript( - gatewayId = "gateway-a", - agentId = "main", - sessionKey = "main", + saveTranscript( messages = listOf( ChatMessage(id = "audio", role = "assistant", content = listOf(audio), timestampMs = 10), @@ -110,7 +126,7 @@ class RoomChatTranscriptCacheTest { ), ) - val loaded = store.loadTranscript("gateway-a", "main", "main") + val loaded = loadTranscript() assertEquals(listOf(audio, video), loaded.map { it.content.single() }) } @@ -133,7 +149,7 @@ class RoomChatTranscriptCacheTest { ), ) - val loaded = cache().loadTranscript("gateway-a", "main", "main").single() + val loaded = loadTranscript().single() assertEquals(listOf("legacy one", "legacy two"), loaded.content.map { it.text }) } @@ -141,7 +157,6 @@ class RoomChatTranscriptCacheTest { @Test fun lastDefaultOwnerIsGatewayScopedAndClearedWithItsCache() = runTest { - val store = cache() store.saveLastDefaultAgentId("gateway-a", "agent-a") store.saveLastDefaultAgentId("gateway-b", "agent-b") @@ -157,11 +172,7 @@ class RoomChatTranscriptCacheTest { @Test fun transcriptRoundTripDropsInternalRoleRows() = runTest { - val store = cache() - store.saveTranscript( - gatewayId = "gateway-a", - agentId = "main", - sessionKey = "main", + saveTranscript( messages = listOf( message("hello", role = "user"), @@ -171,7 +182,7 @@ class RoomChatTranscriptCacheTest { ), ) - val loaded = store.loadTranscript("gateway-a", "main", "main") + val loaded = loadTranscript() assertEquals(listOf("hello", "visible plugin notice", "reply"), loaded.map { it.content.single().text }) assertEquals(listOf("user", "custom", "assistant"), loaded.map { it.role }) @@ -180,15 +191,11 @@ class RoomChatTranscriptCacheTest { @Test fun transcriptWriteKeepsOnlyNewestBoundedMessages() = runTest { - val store = cache() - store.saveTranscript( - gatewayId = "gateway-a", - agentId = "main", - sessionKey = "main", + saveTranscript( messages = (0 until MAX_CACHED_MESSAGES_PER_SESSION + 50).map { index -> message("m$index", timestampMs = index.toLong()) }, ) - val loadedTexts = store.loadTranscript("gateway-a", "main", "main").map { it.content.single().text } + val loadedTexts = loadTranscript().map { it.content.single().text } assertEquals(MAX_CACHED_MESSAGES_PER_SESSION, loadedTexts.size) assertEquals("m50", loadedTexts.first()) @@ -198,35 +205,29 @@ class RoomChatTranscriptCacheTest { @Test fun sessionWriteEvictsBeyondBoundAndDropsOrphanedTranscripts() = runTest { - val store = cache() - store.saveTranscript(gatewayId = "gateway-a", agentId = "main", sessionKey = "session-10", messages = listOf(message("kept"))) - store.saveTranscript(gatewayId = "gateway-a", agentId = "main", sessionKey = "session-55", messages = listOf(message("evicted"))) + saveTranscript(sessionKey = "session-10", messages = listOf(message("kept"))) + saveTranscript(sessionKey = "session-55", messages = listOf(message("evicted"))) - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = (0 until MAX_CACHED_SESSIONS + 10).map { index -> ChatSessionEntry(key = "session-$index", updatedAtMs = 1000L - index, displayName = "Session $index") }, ) - val sessions = store.loadSessions("gateway-a", "main") + val sessions = loadSessions() assertEquals(MAX_CACHED_SESSIONS, sessions.size) assertEquals("session-0", sessions.first().key) assertEquals("session-${MAX_CACHED_SESSIONS - 1}", sessions.last().key) assertEquals("Session 0", sessions.first().displayName) - assertEquals(listOf("kept"), store.loadTranscript("gateway-a", "main", "session-10").map { it.content.single().text }) - assertEquals(emptyList(), store.loadTranscript("gateway-a", "main", "session-55")) + assertEquals(listOf("kept"), loadTranscript(sessionKey = "session-10").map { it.content.single().text }) + assertEquals(emptyList(), loadTranscript(sessionKey = "session-55")) } @Test fun sessionRoundTripKeepsRunMetadata() = runTest { - val store = cache() - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = listOf( ChatSessionEntry( @@ -241,7 +242,7 @@ class RoomChatTranscriptCacheTest { ), ) - val loaded = store.loadSessions("gateway-a", "main").single() + val loaded = loadSessions().single() assertEquals("done", loaded.status) assertEquals(1_000L, loaded.startedAt) @@ -254,20 +255,17 @@ class RoomChatTranscriptCacheTest { @Test fun transcriptForSessionOutsideFullCachedListSurvivesEviction() = runTest { - val store = cache() - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = (0 until MAX_CACHED_SESSIONS).map { index -> ChatSessionEntry(key = "session-$index", updatedAtMs = 1000L - index) }, ) - store.saveTranscript(gatewayId = "gateway-a", agentId = "main", sessionKey = "deep-session", messages = listOf(message("deep text"))) + saveTranscript(sessionKey = "deep-session", messages = listOf(message("deep text"))) - assertEquals(listOf("deep text"), store.loadTranscript("gateway-a", "main", "deep-session").map { it.content.single().text }) - val sessionKeys = store.loadSessions("gateway-a", "main").map { it.key } + assertEquals(listOf("deep text"), loadTranscript(sessionKey = "deep-session").map { it.content.single().text }) + val sessionKeys = loadSessions().map { it.key } assertEquals(MAX_CACHED_SESSIONS, sessionKeys.size) assertTrue(sessionKeys.contains("deep-session")) } @@ -275,26 +273,22 @@ class RoomChatTranscriptCacheTest { @Test fun sessionCacheIsBoundedAcrossEveryAgentInOneGateway() = runTest { - val store = cache() repeat(MAX_CACHED_SESSIONS + 1) { index -> - store.saveTranscript( - gatewayId = "gateway-a", + saveTranscript( agentId = "agent-$index", - sessionKey = "main", messages = listOf(message("message-$index")), ) } val cachedSessionCount = (0..MAX_CACHED_SESSIONS).sumOf { index -> - store.loadSessions("gateway-a", "agent-$index").size + loadSessions(agentId = "agent-$index").size } assertEquals(MAX_CACHED_SESSIONS, cachedSessionCount) - assertTrue(store.loadTranscript("gateway-a", "agent-0", "main").isEmpty()) + assertTrue(loadTranscript(agentId = "agent-0").isEmpty()) assertEquals( listOf("message-$MAX_CACHED_SESSIONS"), - store - .loadTranscript("gateway-a", "agent-$MAX_CACHED_SESSIONS", "main") + loadTranscript(agentId = "agent-$MAX_CACHED_SESSIONS") .map { it.content.single().text }, ) } @@ -302,143 +296,123 @@ class RoomChatTranscriptCacheTest { @Test fun activeDeepTranscriptSurvivesSessionListRefresh() = runTest { - val store = cache() val listedSessions = (0 until MAX_CACHED_SESSIONS).map { index -> ChatSessionEntry(key = "session-$index", updatedAtMs = 1000L - index) } - store.saveSessions(gatewayId = "gateway-a", agentId = "main", sessions = listedSessions) - store.saveTranscript( - gatewayId = "gateway-a", - agentId = "main", + saveSessions(sessions = listedSessions) + saveTranscript( sessionKey = "deep-session", messages = listOf(message("deep text")), ) - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = listedSessions, retainedSessionKey = "deep-session", ) - assertEquals(MAX_CACHED_SESSIONS, store.loadSessions("gateway-a", "main").size) - assertTrue(store.loadSessions("gateway-a", "main").any { it.key == "deep-session" }) + assertEquals(MAX_CACHED_SESSIONS, loadSessions().size) + assertTrue(loadSessions().any { it.key == "deep-session" }) assertEquals( listOf("deep text"), - store.loadTranscript("gateway-a", "main", "deep-session").map { it.content.single().text }, + loadTranscript(sessionKey = "deep-session").map { it.content.single().text }, ) } @Test fun completeSessionListRefreshDropsMissingDeepTranscript() = runTest { - val store = cache() - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = listOf(ChatSessionEntry(key = "deep-session", updatedAtMs = 1)), ) - store.saveTranscript( - gatewayId = "gateway-a", - agentId = "main", + saveTranscript( sessionKey = "deep-session", messages = listOf(message("deleted remotely")), ) - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = listOf(ChatSessionEntry(key = "main", updatedAtMs = 2)), ) - assertEquals(listOf("main"), store.loadSessions("gateway-a", "main").map { it.key }) - assertTrue(store.loadTranscript("gateway-a", "main", "deep-session").isEmpty()) + assertEquals(listOf("main"), loadSessions().map { it.key }) + assertTrue(loadTranscript(sessionKey = "deep-session").isEmpty()) } @Test fun deleteSessionRemovesSessionRowAndTranscript() = runTest { - val store = cache() - store.saveSessions( - gatewayId = "gateway-a", - agentId = "main", + saveSessions( sessions = listOf( ChatSessionEntry(key = "main", updatedAtMs = 1), ChatSessionEntry(key = "other", updatedAtMs = 2), ), ) - store.saveTranscript(gatewayId = "gateway-a", agentId = "main", sessionKey = "main", messages = listOf(message("delete me"))) - store.saveTranscript(gatewayId = "gateway-a", agentId = "other", sessionKey = "main", messages = listOf(message("delete me too"))) - store.saveTranscript(gatewayId = "gateway-a", agentId = "main", sessionKey = "other", messages = listOf(message("keep me"))) + saveTranscript(messages = listOf(message("delete me"))) + saveTranscript(agentId = "other", messages = listOf(message("delete me too"))) + saveTranscript(sessionKey = "other", messages = listOf(message("keep me"))) store.deleteSession("gateway-a", "main", "main") - assertEquals(emptyList(), store.loadTranscript("gateway-a", "main", "main")) - assertEquals(listOf("delete me too"), store.loadTranscript("gateway-a", "other", "main").map { it.content.single().text }) - assertEquals(listOf("other"), store.loadSessions("gateway-a", "main").map { it.key }) - assertEquals(listOf("keep me"), store.loadTranscript("gateway-a", "main", "other").map { it.content.single().text }) + assertEquals(emptyList(), loadTranscript()) + assertEquals(listOf("delete me too"), loadTranscript(agentId = "other").map { it.content.single().text }) + assertEquals(listOf("other"), loadSessions().map { it.key }) + assertEquals(listOf("keep me"), loadTranscript(sessionKey = "other").map { it.content.single().text }) } @Test fun transcriptsAreScopedToGatewayIdentity() = runTest { - val store = cache() - store.saveTranscript(gatewayId = "gateway-a", agentId = "main", sessionKey = "main", messages = listOf(message("gateway a text"))) - store.saveSessions("gateway-a", "main", listOf(ChatSessionEntry(key = "main", updatedAtMs = 1))) + saveTranscript(messages = listOf(message("gateway a text"))) + saveSessions(listOf(ChatSessionEntry(key = "main", updatedAtMs = 1))) - assertEquals(emptyList(), store.loadTranscript("gateway-b", "main", "main")) - assertEquals(emptyList(), store.loadSessions("gateway-b", "main")) - store.saveTranscript(gatewayId = "gateway-b", agentId = "main", sessionKey = "main", messages = listOf(message("gateway b text"))) + assertEquals(emptyList(), loadTranscript(gatewayId = "gateway-b")) + assertEquals(emptyList(), loadSessions(gatewayId = "gateway-b")) + saveTranscript(gatewayId = "gateway-b", messages = listOf(message("gateway b text"))) - assertEquals(listOf("gateway a text"), store.loadTranscript("gateway-a", "main", "main").map { it.content.single().text }) - assertEquals(listOf("main"), store.loadSessions("gateway-a", "main").map { it.key }) + assertEquals(listOf("gateway a text"), loadTranscript().map { it.content.single().text }) + assertEquals(listOf("main"), loadSessions().map { it.key }) } @Test fun blankGatewayIdentityDisablesReadsAndWrites() = runTest { - val store = cache() - store.saveTranscript(gatewayId = "", agentId = "main", sessionKey = "main", messages = listOf(message("must not persist"))) - store.saveSessions("", "main", listOf(ChatSessionEntry(key = "main", updatedAtMs = 1))) + saveTranscript(gatewayId = "", messages = listOf(message("must not persist"))) + saveSessions(listOf(ChatSessionEntry(key = "main", updatedAtMs = 1)), gatewayId = "") - assertEquals(emptyList(), store.loadTranscript("", "main", "main")) - assertEquals(emptyList(), store.loadSessions("", "main")) + assertEquals(emptyList(), loadTranscript(gatewayId = "")) + assertEquals(emptyList(), loadSessions(gatewayId = "")) // Nothing was written under a fallback scope either. - assertEquals(emptyList(), store.loadTranscript("gateway-a", "main", "main")) - assertEquals(emptyList(), store.loadSessions("gateway-a", "main")) + assertEquals(emptyList(), loadTranscript()) + assertEquals(emptyList(), loadSessions()) } @Test fun transcriptsAreScopedToAgentOwnership() = runTest { - val store = cache() - store.saveTranscript("gateway-a", "agent-a", "custom", listOf(message("agent a text"))) - store.saveTranscript("gateway-a", "agent-b", "custom", listOf(message("agent b text"))) - store.saveSessions( - "gateway-a", - "agent-a", - listOf(ChatSessionEntry(key = "agent-a-session", updatedAtMs = 1)), + saveTranscript(agentId = "agent-a", sessionKey = "custom", messages = listOf(message("agent a text"))) + saveTranscript(agentId = "agent-b", sessionKey = "custom", messages = listOf(message("agent b text"))) + saveSessions( + agentId = "agent-a", + sessions = listOf(ChatSessionEntry(key = "agent-a-session", updatedAtMs = 1)), retainedSessionKey = "custom", ) - store.saveSessions( - "gateway-a", - "agent-b", - listOf(ChatSessionEntry(key = "agent-b-session", updatedAtMs = 2)), + saveSessions( + agentId = "agent-b", + sessions = listOf(ChatSessionEntry(key = "agent-b-session", updatedAtMs = 2)), retainedSessionKey = "custom", ) assertEquals( listOf("agent a text"), - store.loadTranscript("gateway-a", "agent-a", "custom").map { it.content.single().text }, + loadTranscript(agentId = "agent-a", sessionKey = "custom").map { it.content.single().text }, ) assertEquals( listOf("agent b text"), - store.loadTranscript("gateway-a", "agent-b", "custom").map { it.content.single().text }, + loadTranscript(agentId = "agent-b", sessionKey = "custom").map { it.content.single().text }, ) - assertEquals(listOf("agent-a-session", "custom"), store.loadSessions("gateway-a", "agent-a").map { it.key }) - assertEquals(listOf("agent-b-session", "custom"), store.loadSessions("gateway-a", "agent-b").map { it.key }) + assertEquals(listOf("agent-a-session", "custom"), loadSessions(agentId = "agent-a").map { it.key }) + assertEquals(listOf("agent-b-session", "custom"), loadSessions(agentId = "agent-b").map { it.key }) } }