dilim/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-13 17:07:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(android): compact chat persistence fixtures (#118456)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-02 21:51:33 -07:00
committed by GitHub
parent e32fda15a8
commit d2ff210679
4 changed files with 369 additions and 689 deletions
Download Patch File Download Diff File Expand all files Collapse all files

211
apps/android/app/src/test/java/ai/openclaw/app/chat/ChatControllerOutboxTest.kt
View File

@@ -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

214
apps/android/app/src/test/java/ai/openclaw/app/chat/ClientDatabasesTest.kt
View File

@@ -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<String>,
registeredGatewayIds: Set<String> = setOf("gateway-a"),
): AndroidClientDatabases =
AndroidClientDatabases.start(
RuntimeEnvironment.getApplication(),
@@ -380,6 +324,30 @@ class ClientDatabasesTest {
registeredGatewayIds = registeredGatewayIds,
)
private suspend fun <T> withDatabases(
names: DatabaseNames,
registeredGatewayIds: Set<String> = setOf("gateway-a"),
block: suspend (AndroidClientDatabases) -> T,
): T {
val databases = open(names, registeredGatewayIds)
return try {
block(databases)
} finally {
databases.close()
}
}
private suspend fun <T> withCleanDatabases(
names: DatabaseNames,
registeredGatewayIds: Set<String> = 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<Any?>("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<Any?>(id, "gateway-test", sessionKey, text, "off", createdAtMs, status, retryCount, lastError),
fixture.run {
arrayOf<Any?>(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",
)
}

417
apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatCommandOutboxTest.kt
View File

@@ -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<OutboxAttachmentPayload> = 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<OutboxAttachmentPayload> = 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<String> = emptySet(),
activeTranscriptEntryIds: Set<String> = 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<ChatOutboxItem>(), 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)
}

216
apps/android/app/src/test/java/ai/openclaw/app/chat/RoomChatTranscriptCacheTest.kt
View File

@@ -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<ChatMessage>,
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<ChatMessage> = store.loadTranscript(gatewayId, agentId, sessionKey)
private suspend fun saveSessions(
sessions: List<ChatSessionEntry>,
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<ChatSessionEntry> = 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<ChatMessage>(), store.loadTranscript("gateway-a", "main", "session-55"))
assertEquals(listOf("kept"), loadTranscript(sessionKey = "session-10").map { it.content.single().text })
assertEquals(emptyList<ChatMessage>(), 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<ChatMessage>(), 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<ChatMessage>(), 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<ChatMessage>(), store.loadTranscript("gateway-b", "main", "main"))
assertEquals(emptyList<ChatSessionEntry>(), store.loadSessions("gateway-b", "main"))
store.saveTranscript(gatewayId = "gateway-b", agentId = "main", sessionKey = "main", messages = listOf(message("gateway b text")))
assertEquals(emptyList<ChatMessage>(), loadTranscript(gatewayId = "gateway-b"))
assertEquals(emptyList<ChatSessionEntry>(), 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<ChatMessage>(), store.loadTranscript("", "main", "main"))
assertEquals(emptyList<ChatSessionEntry>(), store.loadSessions("", "main"))
assertEquals(emptyList<ChatMessage>(), loadTranscript(gatewayId = ""))
assertEquals(emptyList<ChatSessionEntry>(), loadSessions(gatewayId = ""))
// Nothing was written under a fallback scope either.
assertEquals(emptyList<ChatMessage>(), store.loadTranscript("gateway-a", "main", "main"))
assertEquals(emptyList<ChatSessionEntry>(), store.loadSessions("gateway-a", "main"))
assertEquals(emptyList<ChatMessage>(), loadTranscript())
assertEquals(emptyList<ChatSessionEntry>(), 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 })
}
}