profile: rebase concurrent conversation updates

src/main/kotlin/profile/ConversationProfileReducer.kt

@@ -1,6 +1,8 @@
 package top.jie65535.mirai.profile
 
 object ConversationProfileReducer {
+    private const val MISSING_ITEM_REFERENCE = "P2147483647"
+
     fun reduce(
         profiles: Map<Long, UserProfileSnapshot>,
         batch: ConversationProfileBatch,
@@ -33,4 +35,55 @@ object ConversationProfileReducer {
             )
         }
     }
+
+    fun reduceRebased(
+        expectedProfiles: Map<Long, UserProfileSnapshot>,
+        latestProfiles: Map<Long, UserProfileSnapshot>,
+        batch: ConversationProfileBatch,
+        eligibleUserIds: Set<Long>,
+        response: ConversationProfileModelResponse,
+        model: String,
+        promptVersion: String,
+        summaryMaxLength: Int,
+    ): List<ProfileReduction> {
+        val rebasedResponse = response.copy(
+            users = response.users.map { userResponse ->
+                val userId = batch.aliasToUserId[userResponse.userAlias]
+                val expected = userId?.let(expectedProfiles::get)
+                val latest = userId?.let(latestProfiles::get)
+                if (userId == null || userId !in eligibleUserIds || expected == null || latest == null) {
+                    userResponse
+                } else {
+                    userResponse.copy(
+                        operations = userResponse.operations.map { operation ->
+                            operation.rebaseItemReference(expected, latest)
+                        }
+                    )
+                }
+            }
+        )
+        return reduce(
+            profiles = latestProfiles,
+            batch = batch,
+            eligibleUserIds = eligibleUserIds,
+            response = rebasedResponse,
+            model = model,
+            promptVersion = promptVersion,
+            summaryMaxLength = summaryMaxLength,
+        )
+    }
+
+    private fun ProfileModelOperation.rebaseItemReference(
+        expected: UserProfileSnapshot,
+        latest: UserProfileSnapshot,
+    ): ProfileModelOperation {
+        if (action == ProfileOperationAction.ADD) return this
+        val expectedItem = ProfileItemReferences.resolve(expected, itemRef)
+            ?: return copy(itemRef = MISSING_ITEM_REFERENCE)
+        val latestReference = ProfileItemReferences.entries(latest)
+            .firstOrNull { entry -> entry.item.id == expectedItem.id }
+            ?.reference
+            ?: MISSING_ITEM_REFERENCE
+        return copy(itemRef = latestReference)
+    }
 }

src/main/kotlin/profile/UserProfileAnalysisService.kt

@@ -14,8 +14,6 @@ import java.security.MessageDigest
 import java.util.concurrent.ConcurrentHashMap
 
 object UserProfileAnalysisService {
-    private const val MAX_CONVERSATION_CONFLICT_RETRIES = 3
-
     private val runningUsers = ConcurrentHashMap.newKeySet<Long>()
     private val runningGroups = ConcurrentHashMap.newKeySet<Long>()
     private val runningCompactions = ConcurrentHashMap.newKeySet<Long>()
@@ -433,75 +431,70 @@ object UserProfileAnalysisService {
             .filterValues { it >= minAuthoredTextChars.coerceAtLeast(1) }
             .keys
         if (eligibleUserIds.isEmpty()) return null
-        var profiles = userLocks.withUserLocks(eligibleUserIds) {
+        val profiles = userLocks.withUserLocks(eligibleUserIds) {
             withContext(Dispatchers.IO) {
                 if (UserProfileStore.isConversationProcessed(batch.inputHash)) null
                 else loadConversationProfiles(eligibleUserIds)
             }
         } ?: return null
-        var conflictRetries = 0
-        var totalUsage = ProfileTokenUsage()
-
-        while (true) {
-            val (result, reductions) = analyzeConversationWithRetry(
-                model = model,
-                profiles = profiles,
-                batch = batch,
-                eligibleUserIds = eligibleUserIds,
-                retryMax = retryMax,
-                summaryMaxLength = summaryMaxLength,
-                onRetryFailure = onRetryFailure,
-            )
-            totalUsage += result.usage
-            val commitOutcome = userLocks.withUserLocks(eligibleUserIds) {
-                withContext(Dispatchers.IO) {
-                    if (UserProfileStore.isConversationProcessed(batch.inputHash)) {
-                        ConversationCommitOutcome.AlreadyProcessed
+        val (result, reductions) = analyzeConversationWithRetry(
+            model = model,
+            profiles = profiles,
+            batch = batch,
+            eligibleUserIds = eligibleUserIds,
+            retryMax = retryMax,
+            summaryMaxLength = summaryMaxLength,
+            onRetryFailure = onRetryFailure,
+        )
+        val totalUsage = result.usage
+        val commitOutcome = userLocks.withUserLocks(eligibleUserIds) {
+            withContext(Dispatchers.IO) {
+                if (UserProfileStore.isConversationProcessed(batch.inputHash)) {
+                    ConversationCommitOutcome.AlreadyProcessed
+                } else {
+                    val latestProfiles = loadConversationProfiles(eligibleUserIds)
+                    val committedReductions = if (hasProfileVersionConflict(profiles, latestProfiles)) {
+                        ConversationProfileReducer.reduceRebased(
+                            expectedProfiles = profiles,
+                            latestProfiles = latestProfiles,
+                            batch = batch,
+                            eligibleUserIds = eligibleUserIds,
+                            response = result.response,
+                            model = model.modelName,
+                            promptVersion = ProfilePromptStore.PROMPT_VERSION,
+                            summaryMaxLength = summaryMaxLength.coerceAtLeast(100),
+                        )
                     } else {
-                        val latestProfiles = loadConversationProfiles(eligibleUserIds)
-                        if (hasProfileVersionConflict(profiles, latestProfiles)) {
-                            ConversationCommitOutcome.Conflict(latestProfiles)
-                        } else {
-                            UserProfileStore.commitConversation(
-                                reductions = reductions.map { reduction ->
-                                    reduction to batch.forUser(reduction.profile.userId)
-                                },
-                                usage = totalUsage,
-                            )
-                            ConversationCommitOutcome.Committed
-                        }
+                        reductions
                     }
-                }
-            }
-
-            when (commitOutcome) {
-                ConversationCommitOutcome.AlreadyProcessed -> return null
-                ConversationCommitOutcome.Committed -> {
-                    onCommittedOperations(
-                        "source=CONVERSATION bot=${batch.botId} group=${batch.groupId} " +
-                                "batch=[${batch.startTime},${batch.endTime})",
-                        reductions,
-                    )
-                    return ConversationProfileAnalysisReport(
-                        analyzedUsers = eligibleUserIds.size,
-                        processedMessages = batch.messages.size,
-                        appliedOperations = reductions.sumOf { it.operations.size },
-                        skippedOperations = reductions.sumOf { it.skippedOperations.size },
+                    UserProfileStore.commitConversation(
+                        reductions = committedReductions.map { reduction ->
+                            reduction to batch.forUser(reduction.profile.userId)
+                        },
                         usage = totalUsage,
                     )
-                }
-
-                is ConversationCommitOutcome.Conflict -> {
-                    conflictRetries++
-                    if (conflictRetries > MAX_CONVERSATION_CONFLICT_RETRIES) {
-                        throw IllegalStateException(
-                            "群 ${batch.groupId} 会话画像提交连续冲突 $conflictRetries 次，未提交结果"
-                        )
-                    }
-                    profiles = commitOutcome.latestProfiles
+                    ConversationCommitOutcome.Committed(committedReductions)
                 }
             }
         }
+
+        val committedReductions = when (commitOutcome) {
+            ConversationCommitOutcome.AlreadyProcessed -> return null
+            is ConversationCommitOutcome.Committed -> commitOutcome.reductions
+        }
+
+        onCommittedOperations(
+            "source=CONVERSATION bot=${batch.botId} group=${batch.groupId} " +
+                    "batch=[${batch.startTime},${batch.endTime})",
+            committedReductions,
+        )
+        return ConversationProfileAnalysisReport(
+            analyzedUsers = eligibleUserIds.size,
+            processedMessages = batch.messages.size,
+            appliedOperations = committedReductions.sumOf { it.operations.size },
+            skippedOperations = committedReductions.sumOf { it.skippedOperations.size },
+            usage = totalUsage,
+        )
     }
 
     private fun loadConversationProfiles(userIds: Set<Long>): Map<Long, UserProfileSnapshot> =
@@ -755,11 +748,10 @@ object UserProfileAnalysisService {
     }
 
     private sealed class ConversationCommitOutcome {
-        object Committed : ConversationCommitOutcome()
-        object AlreadyProcessed : ConversationCommitOutcome()
-        data class Conflict(
-            val latestProfiles: Map<Long, UserProfileSnapshot>,
+        data class Committed(
+            val reductions: List<ProfileReduction>,
         ) : ConversationCommitOutcome()
+        object AlreadyProcessed : ConversationCommitOutcome()
     }
 }
 

src/test/kotlin/profile/ConversationProfileReducerTest.kt

@@ -144,6 +144,78 @@ class ConversationProfileReducerTest {
         assertFalse(prompt.contains(oldItem.id))
     }
 
+    @Test
+    fun rebasesItemReferenceByStableIdWhenConcurrentUpdateChangesItemOrder() {
+        val target = item("target", "原始目标条目", 100)
+        val expected = emptyProfile(USER_A).copy(version = 1, items = listOf(target))
+        val inserted = item("inserted", "并发新增条目", 50)
+        val latest = expected.copy(version = 2, items = listOf(inserted, target))
+
+        val reduction = ConversationProfileReducer.reduceRebased(
+            expectedProfiles = mapOf(USER_A to expected, USER_B to emptyProfile(USER_B)),
+            latestProfiles = mapOf(USER_A to latest, USER_B to emptyProfile(USER_B)),
+            batch = batch(),
+            eligibleUserIds = USERS,
+            response = ConversationProfileModelResponse(
+                users = listOf(
+                    ConversationProfileUserResponse(
+                        userAlias = "U1",
+                        operations = listOf(
+                            ProfileModelOperation(
+                                action = ProfileOperationAction.UPDATE,
+                                itemRef = "P1",
+                                content = "更新后的目标条目",
+                                evidenceRefs = listOf(1),
+                            )
+                        ),
+                    )
+                )
+            ),
+            model = "test-model",
+            promptVersion = "test-prompt",
+            summaryMaxLength = 500,
+        ).single { it.profile.userId == USER_A }
+
+        assertEquals("并发新增条目", reduction.profile.items.single { it.id == "inserted" }.content)
+        assertEquals("更新后的目标条目", reduction.profile.items.single { it.id == "target" }.content)
+        assertEquals(1, reduction.operations.size)
+    }
+
+    @Test
+    fun skipsRebasedOperationWhenConcurrentUpdateRemovedTargetItem() {
+        val target = item("target", "稍后被删除的条目", 100)
+        val expected = emptyProfile(USER_A).copy(version = 1, items = listOf(target))
+        val latest = expected.copy(version = 2, items = emptyList())
+
+        val reduction = ConversationProfileReducer.reduceRebased(
+            expectedProfiles = mapOf(USER_A to expected, USER_B to emptyProfile(USER_B)),
+            latestProfiles = mapOf(USER_A to latest, USER_B to emptyProfile(USER_B)),
+            batch = batch(),
+            eligibleUserIds = USERS,
+            response = ConversationProfileModelResponse(
+                users = listOf(
+                    ConversationProfileUserResponse(
+                        userAlias = "U1",
+                        operations = listOf(
+                            ProfileModelOperation(
+                                action = ProfileOperationAction.DELETE,
+                                itemRef = "P1",
+                                evidenceRefs = listOf(1),
+                            )
+                        ),
+                    )
+                )
+            ),
+            model = "test-model",
+            promptVersion = "test-prompt",
+            summaryMaxLength = 500,
+        ).single { it.profile.userId == USER_A }
+
+        assertTrue(reduction.operations.isEmpty())
+        assertTrue(reduction.skippedOperations.single().contains("不存在"))
+        assertTrue(reduction.profile.items.isEmpty())
+    }
+
     @Test
     fun acceptsMoreThanFourOperationsForOneUser() {
         val operations = (1..5).map { index ->
@@ -297,6 +369,15 @@ class ConversationProfileReducerTest {
         snapshotEndTime = 0,
     )
 
+    private fun item(id: String, content: String, firstSeenAt: Int) = UserProfileItem(
+        id = id,
+        category = ProfileCategory.NOTABLE_FACT,
+        content = content,
+        confidence = ProfileConfidence.MEDIUM,
+        firstSeenAt = firstSeenAt,
+        lastConfirmedAt = firstSeenAt,
+    )
+
     private fun message(ref: Int, fromId: Long, text: String) = ProfilePromptMessage(
         record = ChatMessageRecord(
             botId = BOT,

src/test/kotlin/profile/UserProfileAnalysisServiceTest.kt

@@ -149,12 +149,11 @@ class UserProfileAnalysisServiceTest {
     }
 
     @Test
-    fun sharedUserRequestsRunConcurrentlyAndConflictReloadsLatestProfile() = withProfileStore {
+    fun sharedUserRequestsRunConcurrentlyAndConflictRebasesWithoutAnotherModelCall() = withProfileStore {
         val firstEntered = CompletableDeferred<Unit>()
         val releaseFirst = CompletableDeferred<Unit>()
         val secondFirstEntered = CompletableDeferred<Unit>()
         val releaseSecondFirst = CompletableDeferred<Unit>()
-        val secondRetried = CompletableDeferred<Unit>()
         val secondCalls = AtomicInteger()
         val firstModel = InspectingConversationProfileModel { profiles ->
             assertEquals(0, profiles.getValue(USER_A).version)
@@ -163,20 +162,10 @@ class UserProfileAnalysisServiceTest {
             result(responseFor("U1", "第一群归纳的信息", evidenceRef = 1))
         }
         val secondModel = InspectingConversationProfileModel { profiles ->
-            when (secondCalls.incrementAndGet()) {
-                1 -> {
-                    assertEquals(0, profiles.getValue(USER_A).version)
-                    secondFirstEntered.complete(Unit)
-                    releaseSecondFirst.await()
-                }
-
-                2 -> {
-                    assertEquals(1, profiles.getValue(USER_A).version)
-                    secondRetried.complete(Unit)
-                }
-
-                else -> error("共享用户冲突只应触发一次重算")
-            }
+            secondCalls.incrementAndGet()
+            assertEquals(0, profiles.getValue(USER_A).version)
+            secondFirstEntered.complete(Unit)
+            releaseSecondFirst.await()
             result(responseFor("U1", "第二群归纳的信息", evidenceRef = 1))
         }
 
@@ -188,11 +177,10 @@ class UserProfileAnalysisServiceTest {
             releaseFirst.complete(Unit)
             assertNotNull(first.await())
             releaseSecondFirst.complete(Unit)
-            withTimeout(1_000) { secondRetried.await() }
             assertNotNull(second.await())
         }
 
-        assertEquals(2, secondCalls.get())
+        assertEquals(1, secondCalls.get())
         val profile = assertNotNull(UserProfileStore.load(USER_A))
         assertEquals(2, profile.version)
         assertEquals(2, profile.items.size)