conversation: preserve triggers and await follow-ups

src/main/kotlin/conversation/ConversationEngine.kt

@@ -8,7 +8,6 @@ import com.aallam.openai.api.chat.ToolCall
 import com.aallam.openai.api.chat.ToolChoice
 import com.aallam.openai.api.core.Usage
 import com.aallam.openai.api.model.ModelId
-import io.ktor.util.collections.ConcurrentSet
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
@@ -19,6 +18,7 @@ import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.launch
 import net.mamoe.mirai.event.events.GroupMessageEvent
 import net.mamoe.mirai.event.events.MessageEvent
+import net.mamoe.mirai.message.data.source
 import top.jie65535.mirai.JChatGPT
 import top.jie65535.mirai.config.PluginConfig
 import top.jie65535.mirai.data.TokenUsageStore
@@ -51,11 +51,10 @@ import top.jie65535.mirai.tools.WebSearch
 import top.jie65535.mirai.util.RetryBackoff
 import java.time.OffsetDateTime
 import java.time.format.DateTimeFormatter
-import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
 internal object ConversationEngine {
-    private val activeRequests = ConcurrentSet<Long>()
+    private val runtimeState = ConversationRuntimeState<MessageEvent>()
     private val dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy年MM月dd E HH:mm:ss")
     private val thinkRegex = Regex("<think>[\\s\\S]*?</think>")
     private val tools: List<BaseAgent> = listOf(
@@ -84,16 +83,53 @@ internal object ConversationEngine {
     )
 
     fun clear() {
-        activeRequests.clear()
+        runtimeState.clear()
+    }
+
+    fun isExpectedUser(event: MessageEvent): Boolean = runtimeState.isExpectedUser(
+        key = event.toConversationKey(),
+        userId = event.sender.id,
+        nowEpochSecond = currentEpochSecond(),
+    )
+
+    suspend fun resumeObserved(event: MessageEvent): Boolean {
+        val started = runtimeState.beginObserved(
+            key = event.toConversationKey(),
+            userId = event.sender.id,
+            nowEpochSecond = currentEpochSecond(),
+        ) ?: return false
+        runConversation(event, started.running, started.resumedWait)
+        return true
     }
 
     suspend fun start(event: MessageEvent) {
-        val subjectId = event.subject.id
-        if (!activeRequests.add(subjectId)) {
-            JChatGPT.logger.warning("The current Contact is busy!")
-            return
-        }
+        when (val result = runtimeState.beginExplicit(event.toConversationKey(), event)) {
+            is ConversationRuntimeState.BeginResult.Queued -> {
+                if (result.newlyQueued) {
+                    JChatGPT.logger.info(
+                        "当前会话忙碌，已暂存用户 ${event.senderName}(${event.sender.id}) 的二次触发"
+                    )
+                } else {
+                    JChatGPT.logger.info(
+                        "当前会话已有待处理触发，用户 ${event.senderName}(${event.sender.id}) 的消息将通过增量历史合并"
+                    )
+                }
+            }
 
+            is ConversationRuntimeState.BeginResult.Started -> {
+                runConversation(event, result.running, result.resumedWait)
+            }
+        }
+    }
+
+    private suspend fun runConversation(
+        initialEvent: MessageEvent,
+        running: ConversationRuntimeState.Running<MessageEvent>,
+        resumedWait: FollowUpWaitDirective?,
+    ) {
+        val subjectId = initialEvent.subject.id
+        var currentEvent = initialEvent
+        var indexesReleased = false
         try {
             val cache = ConversationContext.cache(subjectId)
             val reuseCache = PluginConfig.enableContextCache && cache != null &&
@@ -112,16 +148,25 @@ internal object ConversationEngine {
             } else mutableListOf()
 
             if (history.isEmpty() || cache == null) {
-                val prompt = ConversationContext.getSystemPrompt(event)
+                val prompt = ConversationContext.getSystemPrompt(currentEvent)
                 if (PluginConfig.logPrompt) JChatGPT.logger.info("Prompt: $prompt")
                 history += ChatMessage(ChatRole.System, prompt)
-                val historyText = ConversationContext.getHistory(event)
+                val historyText = ConversationContext.getHistory(currentEvent)
                 JChatGPT.logger.info("注入聊天记录：\n$historyText")
                 history += ChatMessage.User(historyText)
             } else {
-                val newMessages = ConversationContext.getAfterHistory(cache.lastActivityAt, event)
+                val newMessages = ConversationContext.getAfterHistory(cache.lastActivityAt, currentEvent)
                 JChatGPT.logger.info("补充聊天记录：\n$newMessages")
-                history += ChatMessage.User("## 以下是上次对话结束至今的新消息\n\n$newMessages")
+                history += ChatMessage.User(
+                    if (resumedWait == null) {
+                        "## 以下是上次对话结束至今的新消息\n\n$newMessages"
+                    } else {
+                        buildObservationResumePrompt(resumedWait, newMessages)
+                    }
+                )
+            }
+            if (resumedWait != null && !reuseCache) {
+                history += ChatMessage.User(buildObservationResumePrompt(resumedWait, null))
             }
 
             val endpoints = LargeLanguageModels.orderedChatEndpoints()
@@ -134,6 +179,7 @@ internal object ConversationEngine {
             var consecutiveFailures = 0
             do {
                 val endpoint = endpoints[endpointIndex]
+                val roundEvent = currentEvent
                 var streamingOk = false
                 try {
                     val startedAt = OffsetDateTime.now().toEpochSecond().toInt()
@@ -161,7 +207,7 @@ internal object ConversationEngine {
                             if (index >= responseToolCalls.size) {
                                 responseToolCalls.lastOrNull()?.let { toolCall ->
                                     toolCallTasks += JChatGPT.async {
-                                        toolCall.toResultMessage(event)
+                                        toolCall.toResultMessage(roundEvent)
                                     }
                                 }
                                 val id = toolCallChunk.id
@@ -196,22 +242,48 @@ internal object ConversationEngine {
                         toolCalls = responseToolCalls.ifEmpty { null },
                         reasoningContent = if (responseToolCalls.isNotEmpty()) reasoningContent?.toString() else null,
                     )
-                    recordUsage(event, lastTokenUsage, lastCacheUsage)
+                    recordUsage(roundEvent, lastTokenUsage, lastCacheUsage)
                     completedRounds++
 
                     if (responseToolCalls.size > toolCallTasks.size) {
-                        val finalToolResult = responseToolCalls.last().toResultMessage(event)
+                        val finalToolResult = responseToolCalls.last().toResultMessage(roundEvent)
                         if (toolCallTasks.isNotEmpty()) history += toolCallTasks.awaitAll()
                         history += finalToolResult
-                        done = responseToolCalls.any { it.function.name == "endConversation" }
-                    } else {
-                        done = true
                     }
 
-                    if (!done) {
+                    val endCalls = responseToolCalls.filter {
+                        it.function.name == END_CONVERSATION_TOOL_NAME
+                    }
+                    if (endCalls.size > 1) {
+                        JChatGPT.logger.warning("模型在同一轮调用了多次 endConversation，将采用第一次调用的参数")
+                    }
+                    val endCall = endCalls.firstOrNull()
+                    val endArguments = endCall?.let { call ->
+                        runCatching { call.function.argumentsAsJsonOrNull() }
+                            .onFailure {
+                                JChatGPT.logger.warning("无法解析 endConversation 参数，将按普通结束处理", it)
+                            }
+                            .getOrNull()
+                    }
+                    val waitDirective = parseFollowUpWait(endArguments)
+                    if (endArguments?.containsKey(FOLLOW_UP_WAIT_ARGUMENT) == true && waitDirective == null) {
+                        JChatGPT.logger.warning("endConversation.waitForFollowUp 参数无效，将按普通结束处理")
+                    }
+
+                    val requestedEnd = responseToolCalls.isEmpty() || endCall != null
+                    val canContinue = completedRounds < maxRounds
+                    if (!requestedEnd && canContinue) {
+                        val pendingEvent = runtimeState.takePending(running)
+                        if (pendingEvent != null) currentEvent = pendingEvent
                         history += ChatMessage.User(
-                            buildContinuationPrompt(maxRounds - completedRounds, startedAt, event)
+                            buildContinuationPrompt(
+                                remainingRounds = maxRounds - completedRounds,
+                                startedAt = startedAt,
+                                event = currentEvent,
+                                pendingTrigger = pendingEvent != null,
+                            )
                         )
+                        done = false
                     } else {
                         if (PluginConfig.enableContextCache) {
                             ConversationContext.saveCache(
@@ -220,8 +292,48 @@ internal object ConversationEngine {
                             )
                             JChatGPT.logger.debug("已保存对话上下文到缓存")
                         }
-                        if (event is GroupMessageEvent) {
-                            ProfileAutoMaintenance.recordCompletedConversation(event, startedAt)
+
+                        when (val finish = runtimeState.finish(
+                            running = running,
+                            waitDirective = waitDirective.takeIf { requestedEnd },
+                            nowEpochSecond = currentEpochSecond(),
+                            allowPendingContinuation = canContinue,
+                            onFinished = {
+                                ConversationContext.releaseActiveIndexes(subjectId)
+                                indexesReleased = true
+                            },
+                        )) {
+                            is ConversationRuntimeState.FinishResult.Continue -> {
+                                currentEvent = finish.event
+                                history += ChatMessage.User(
+                                    buildContinuationPrompt(
+                                        remainingRounds = maxRounds - completedRounds,
+                                        startedAt = startedAt,
+                                        event = currentEvent,
+                                        pendingTrigger = true,
+                                    )
+                                )
+                                done = false
+                            }
+
+                            is ConversationRuntimeState.FinishResult.Observing -> {
+                                scheduleObservationTimeout(finish.observation)
+                                (currentEvent as? GroupMessageEvent)?.let {
+                                    ProfileAutoMaintenance.recordCompletedConversation(it, startedAt)
+                                }
+                                JChatGPT.logger.info(
+                                    "会话已结束，等待用户 ${finish.observation.directive.fromUserIds.joinToString()} " +
+                                            "在 ${finish.observation.directive.timeoutSeconds} 秒内发言"
+                                )
+                                done = true
+                            }
+
+                            ConversationRuntimeState.FinishResult.Ended -> {
+                                (currentEvent as? GroupMessageEvent)?.let {
+                                    ProfileAutoMaintenance.recordCompletedConversation(it, startedAt)
+                                }
+                                done = true
+                            }
                         }
                     }
                 } catch (cause: Exception) {
@@ -264,16 +376,55 @@ internal object ConversationEngine {
             throw cause
         } catch (cause: Throwable) {
             JChatGPT.logger.warning(cause)
-            event.subject.sendMessage("很抱歉，发生异常，请稍后重试")
+            currentEvent.subject.sendMessage("很抱歉，发生异常，请稍后重试")
         } finally {
-            ConversationContext.releaseActiveIndexes(subjectId)
-            JChatGPT.launch {
-                delay(500.milliseconds)
-                activeRequests.remove(subjectId)
+            if (!indexesReleased) {
+                runtimeState.abort(running) {
+                    ConversationContext.releaseActiveIndexes(subjectId)
+                    indexesReleased = true
+                }
             }
         }
     }
 
+    private fun scheduleObservationTimeout(observation: ConversationRuntimeState.Observation) {
+        val job = JChatGPT.launch {
+            val remainingSeconds = observation.expiresAtEpochSecond - currentEpochSecond()
+            if (remainingSeconds > 0) delay(remainingSeconds.seconds)
+            if (runtimeState.expire(observation)) {
+                JChatGPT.logger.debug(
+                    "等待用户 ${observation.directive.fromUserIds.joinToString()} 的观察窗口已超时"
+                )
+            }
+        }
+        runtimeState.attachTimeoutJob(observation, job)
+    }
+
+    private fun MessageEvent.toConversationKey(): ConversationKey = ConversationKey(
+        botId = bot.id,
+        kind = message.source.kind,
+        subjectId = subject.id,
+    )
+
+    private fun currentEpochSecond(): Long = OffsetDateTime.now().toEpochSecond()
+
+    private fun buildObservationResumePrompt(
+        directive: FollowUpWaitDirective,
+        newMessages: String?,
+    ): String = buildString {
+        appendLine("## 观察状态恢复")
+        appendLine("你此前结束发言后，选择等待指定用户在当前会话中的下一条消息。")
+        append("等待用户：").appendLine(directive.fromUserIds.joinToString())
+        append("等待条件：").appendLine(directive.condition)
+        appendLine("被观察状态唤醒不代表必须回复。请判断新消息是否满足等待条件、是否承接当前话题。")
+        appendLine("如果无关，不要发送任何内容，直接调用 endConversation。")
+        if (newMessages != null) {
+            appendLine()
+            appendLine("## 等待后出现的新消息")
+            append(newMessages)
+        }
+    }
+
     private fun chatCompletions(
         history: List<ChatMessage>,
         endpoint: LargeLanguageModels.ChatEndpoint,
@@ -336,13 +487,21 @@ internal object ConversationEngine {
         return truncated
     }
 
-    private fun buildContinuationPrompt(remainingRounds: Int, startedAt: Int, event: MessageEvent): String = buildString {
+    private fun buildContinuationPrompt(
+        remainingRounds: Int,
+        startedAt: Int,
+        event: MessageEvent,
+        pendingTrigger: Boolean,
+    ): String = buildString {
         appendLine("## 系统提示")
         append("本次运行最多还剩").append(remainingRounds).appendLine("轮。")
         appendLine("如果要多次发言，可以一次性调用多次发言工具。")
         appendLine("如果没有什么要做的，可以提前结束。")
+        if (pendingTrigger) appendLine("运行期间收到了新的显式触发，请优先处理水位后的新消息。")
         appendLine("当前时间：${dateTimeFormatter.format(OffsetDateTime.now())}")
-        val messages = ConversationContext.getAfterHistory(startedAt, event)
+        val messages = ConversationContext.getAfterHistory(startedAt, event).ifEmpty {
+            if (pendingTrigger && !JChatGPT.includeHistory) ConversationContext.getHistory(event) else ""
+        }
         if (messages.isNotEmpty()) append("## 以下是上次运行至今的新消息\n\n$messages")
     }