Operit 中断回合统计修复实战：将取消收尾从 Room 重载迁移到运行态消息所有权

AI Agent人工智能大模型AI 应用工具调用本地部署MCP ClientsAgent 记忆【免费下载链接】OperitThe most powerful AI agent and AI chat software on Android/Operit是一款Android上能力最为强大、发展最久的AI Agent项目地址https://gitcode.com/gh_mirrors/op/Operit点击查看免费下载导读本文围绕 Operit 项目中 issue-741「中断回合统计」修复展开剖析用户中断 AI 输出后「部分内容、Token、耗时、完成时间丢失」这一缺陷的根因以及官方采用的两层修正方案其一由MessageProcessingDelegate的每会话运行态ChatRuntime持有当前流式 AI 消息取消收尾直接持久化同一对象不再依赖数据库实体推断运行态身份其二协调 Web 端删除会话与丢弃式取消的执行顺序避免部分消息写入已删除会话。读完本文你将掌握 Operit 聊天运行时中断收尾的完整调用链、回合编号互斥机制与对应的单元测试验证方式。一、问题背景取消收尾为何丢失中断统计在 Operit 的聊天消息处理层中用户中断模型输出是一个高频交互动作。修复前旧实现的取消收尾流程是消息处理层在取消模型输出之前已经从当前服务读取了本轮 Token 与耗时快照取消任务结束后收尾逻辑重新从 Room 数据库加载聊天记录通过只存在于内存中的contentStream字段查找「当前流式 AI 消息」对该消息回写部分回复内容、Token、耗时与完成时间。问题在于第 3 步contentStream是一个瞬态字段。查看 ChatMessage.kt 可以看到该字段被Transient注解标记Transient var contentStream: StreamString? null // 修改为StreamString类型与EnhancedAIService.sendMessage返回类型匹配Transient意味着该字段不属于数据库实体、不参与 Room 持久化。从 Room 重载后的所有消息contentStream恒为null因此「按contentStream非空定位流式消息」的查找永远无法命中——于是中断时已生成的部分回复、输入/输出 Token、等待耗时、输出耗时以及completedAt全部不会写回用户界面上表现为「中断即丢失」。这正是 index.md 中描述的原始状况取消前快照已读取收尾却因身份识别失败而无法落盘。二、修正总览运行态消息所有权官方修正的核心思想是把「当前可持久化的流式 AI 消息」的所有权从数据库模型迁移到每个会话的运行态对象彻底切断收尾逻辑对 Room 重载结果的依赖。方案共五步见 01-bind-cancellation-to-runtime-message.md在ChatRuntime中保存当前流式 AI 消息及 Waifu 分段列表取消任务前读取该对象与统计快照任务停止后将两者一起交给收尾逻辑收尾逻辑只从数据库读取匹配的用户消息用于回写回合统计不再用数据库对象识别流式 AI 消息通过回合编号与取消互斥保证旧取消请求不会清理新回合正常完成与取消完成后均清理运行态消息引用。从源码结构看MessageProcessingDelegate为每个会话维护了一个ChatRuntime实例ConcurrentHashMapString, ChatRuntime其中关键字段定义在 MessageProcessingDelegate.ktprivate data class ActiveStreamingTurn( val message: ChatMessage, val segmentedMessages: MutableListChatMessage? null, ) private data class ChatRuntime( var sendJob: Job? null, var responseStream: SharedStreamString? null, // 取消收尾必须持有运行态对象Room 不保存 contentStream重载后无法识别当前流消息。 var activeStreamingTurn: ActiveStreamingTurn? null, var streamCollectionJob: Job? null, var stateCollectionJob: Job? null, var currentTurnOptions: ChatTurnOptions ChatTurnOptions(), var requestSentAt: Long 0L, var requestStartElapsed: Long 0L, var firstResponseElapsed: Long? null, val turnSequence: AtomicLong AtomicLong(0L), Volatile var activeTurnId: Long 0L, val cancellationMutex: Mutex Mutex(), Volatile var cancellationInProgress: Boolean false, val isLoading: MutableStateFlowBoolean MutableStateFlow(false) )其中ActiveStreamingTurn同时持有单条流式 AI 消息message与Waifu 分段消息列表segmentedMessages这是 Waifu分段长回复模式下多个已形成分段能被统一统计的前提。注释也直接点明了设计动机Room 不保存contentStream重载后无法识别当前流消息。三、取消收尾完整调用链3.1 取消入口普通取消 vs 丢弃式取消MessageProcessingDelegate提供两个取消入口见 MessageProcessingDelegate.ktfun cancelMessage(chatId: String) { val expectedTurnId runtimeFor(chatId).activeTurnId coroutineScope.launch(Dispatchers.IO) { cancelMessageInternal( chatId chatId, keepPartialResponse true, expectedTurnId expectedTurnId, ) } } suspend fun cancelMessageForDestructiveMutation(chatId: String) { cancelMessageInternal(chatId, keepPartialResponse false) }cancelMessage(chatId)用户点击「停止」触发的普通取消keepPartialResponse true会持久化部分回复cancelMessageForDestructiveMutation(chatId)删除会话等破坏性操作前的丢弃式取消keepPartialResponse false不持久化部分回复。3.2 取消互斥与回合编号守卫cancelMessageInternal的骨架MessageProcessingDelegate.kt体现了两个并发安全设计val chatRuntime runtimeFor(chatId) chatRuntime.cancellationMutex.withLock { val turnId expectedTurnId ?: chatRuntime.activeTurnId if (!chatRuntime.isLoading.value || chatRuntime.activeTurnId ! turnId) { returnwithLock } chatRuntime.cancellationInProgress true ... }回合编号守卫turnSequence.incrementAndGet()在每次sendUserMessage开始时产生新turnId并写入activeTurnId。取消时把发起时刻的activeTurnId作为expectedTurnId传入进入临界区后若发现当前activeTurnId已变化即用户已发起新回合则直接返回保证旧取消请求永远不会清理新回合取消互斥整个取消过程持有cancellationMutex同一会话同一时刻只有一个取消操作在执行避免重复取消导致运行态被多次清理。3.3 快照捕获取消前的 Token 与耗时在任务真正停止之前代码通过readCurrentTurnCancellationSnapshot(chatId)捕获统计快照MessageProcessingDelegate.kt。快照类型为internal data class TurnCancellationSnapshot( val inputTokens: Long, val outputTokens: Long, val cachedInputTokens: Long, val sentAt: Long, val outputDurationMs: Long, val waitDurationMs: Long, )各字段来源inputTokens/outputTokens/cachedInputTokens来自service.captureCurrentTurnTokenSnapshot()即模型服务当前回合的实时计数含缓存命中 tokensentAtruntime.requestSentAt本轮请求发送时间戳waitDurationMs(firstResponseElapsed - requestStartElapsed)即「等待首包耗时」outputDurationMs(messageTimingNow() - firstResponseElapsed)即「首包到取消时刻的输出耗时」。快照读取失败时仅记录AppLogger.w并返回null不会阻断取消流程。3.4 任务停止与收尾持久化取消在cancellationMutex内依次完成清除工具调用计数、调用AIMessageManager.cancelOperation(chatId)、逐个cancel()并join()发送/状态收集/流收集协程随后若activeTurn ! null即keepPartialResponse true且存在活跃流调用detachStreamingAiMessage完成收尾持久化MessageProcessingDelegate.kt解析最终内容resolveFinalContent从contentStream的 replayCache 或事件载体拼接已生成的全部文本写回streamingMessage.content构造完成消息completeInterruptedMessage将快照各统计字段复制到消息并写入completedAt同时把contentStream置空MessageProcessingDelegate.kt回写用户消息统计从getRuntimeChatHistory(chatId)中按sender user且sentAt匹配找到本轮用户消息用withTurnMetrics写入相同的 Token 与耗时——这就是「中断前已写入的用户消息获得相同的回合统计」Waifu 分段处理若activeTurn.segmentedMessages非空则不再落单条消息而是把每一条已持久化分段copy上相同的统计字段与completedAt后逐一写回保证 Waifu 模式下已形成的分段消息获得相同统计持久化若currentTurnOptions.persistTurn为真调用saveCurrentChat()落盘。3.5 运行态清理finally块中只有当chatRuntime.activeTurnId turnId仍是当前回合时才执行清理将sendJob、stateCollectionJob、streamCollectionJob、responseStream、activeStreamingTurn全部置空重置时间戳字段与isLoading刷新全局加载状态。这样无论是正常完成还是取消完成运行态消息引用都会被回收不会残留影响下一回合。ChatRuntime的cancellationInProgress标志在finally中先复位供上层判断取消是否仍在进行。四、协调破坏性删除Web 端删除会话的执行顺序第二个独立缺陷见 02-coordinate-destructive-delete.md旧实现中Web 接口发现目标会话仍在输出时会异步请求取消并立即删除数据库会话。由于中断收尾需要写入部分消息这两个操作可能交错造成消息写入一个已删除的会话产生悬挂数据。修正后的执行顺序为Web 删除接口先等待总结与消息处理的丢弃式取消完全结束再执行会话删除。丢弃式取消keepPartialResponse false不持久化部分回复因此中断统计的普通保存路径完全不会参与破坏性删除。代码层面的落地有两处ChatServiceCore初始化时注册了破坏性变更前钩子ChatServiceCore.ktchatHistoryDelegate.setBeforeDestructiveHistoryMutation { chatId - messageCoordinationDelegate.cancelSummaryForDestructiveMutation(chatId) messageProcessingDelegate.cancelMessageForDestructiveMutation(chatId) }Web HTTP 桥的handleDeleteChat在删除前先挂起等待取消完成WebChatHttpBridge.ktif (core.activeStreamingChatIds.value.contains(chatId)) { runBlocking { core.cancelMessageForDestructiveMutation(chatId) } } val deleted runBlocking { chatHistoryManager.deleteChatHistory(chatId) }由于cancelMessageForDestructiveMutation是suspend函数且内部使用job.join()等任务停止其协程runBlocking会阻塞到丢弃式取消完整结束后才继续执行deleteChatHistory从执行顺序上根除了「消息写入已删除会话」的竞态。五、验收与测试验证修复的验收标准01-bind-cancellation-to-runtime-message.md共四条中断消息写入部分内容、completedAt、Token 和耗时中断前已写入的用户消息获得相同的回合统计Waifu 模式下已持久化的分段消息获得相同统计流式消息身份不依赖 Room 中不存在的字段即不再依赖contentStream。仓库为此新增了 MessageProcessingDelegateTest.kt其中completeInterruptedMessage_appliesTurnSnapshotAndCompletesPartialContent用例直接构造一个带contentStream emptyStream()的 AI 消息与TurnCancellationSnapshot断言completeInterruptedMessage返回的消息满足content被替换为部分响应文本partial responsecontentStream被置为nullinputTokens、outputTokens、cachedInputTokens与快照一致120 / 34 / 56sentAt、outputDurationMs、waitDurationMs、completedAt与快照一致30 / 4000 / 500 / 5000。该测试从纯函数层面锁定了「快照应用 部分内容完成 流引用清理」的核心契约。按仓库本地工作约束index.md 验证记录注明「本次未执行构建或测试命令」读者可在自己的构建环境中通过./gradlew :app:testDebugUnitTest --tests com.ai.assistance.operit.services.core.MessageProcessingDelegateTest运行该用例。六、设计要点小结关注点旧实现修正后流式 AI 消息身份取消收尾从 Room 重载后按contentStream非空查找ChatRuntime.activeStreamingTurn运行态持有统计快照来源取消前读取收尾时丢失TurnCancellationSnapshot与消息引用一起传递用户消息统计回写无因 AI 消息未命中按sentAt匹配用户消息并withTurnMetrics回写Waifu 分段无segmentedMessages逐条复制统计与completedAt并发安全无回合编号activeTurnIdcancellationMutex互斥Web 删除会话异步取消与删除交错runBlocking等待丢弃式取消结束后再删除运行态清理无finally中按回合校验后清理全部引用整体上这一修复把「谁拥有当前流式消息」从持久化模型的身份推断改为运行态对象的显式持有配合回合编号互斥与破坏性操作顺序协调同时保障了普通取消的统计保留、Waifu 分段统计一致性与删除会话的数据完整性。若要进一步深入可继续阅读 preserve_interrupted_ai_output_20260814网络失败消息定稿的同类收尾问题以及 chat_runtime_foreground_service_plan.mdChatRuntime运行态的服务化演进两者与本文共用同一套运行态消息生命周期模型。赞分享AI Agent人工智能大模型AI 应用工具调用本地部署MCP ClientsAgent 记忆【免费下载链接】OperitThe most powerful AI agent and AI chat software on Android/Operit是一款Android上能力最为强大、发展最久的AI Agent项目地址https://gitcode.com/gh_mirrors/op/Operit点击查看免费下载相关推荐Operit 中断回合统计修复将取消快照与运行态流式消息绑定保留部分回复与 Token/耗时数据Operit 中断回合统计修复将取消快照与运行态流式消息绑定保留部分回复与 Token/耗时数据 中断回合统计是 AI 聊天应用中一个极易被忽略、却直接影响AI Agent人工智能大模型AI 应用工具调用本地部署MCP ClientsAgent 记忆GUI 自动化Operit 会话破坏性删除与中断收尾的协调丢弃式取消如何阻止消息写入已删会话Operit 会话破坏性删除与中断收尾的协调丢弃式取消如何阻止消息写入已删会话 导读 本文围绕 Operit 中中断回合统计修复issue 741的第AI Agent人工智能大模型AI 应用工具调用本地部署MCP ClientsAgent 记忆GUI 自动化Operit 超大聊天消息读取修复基于 Room 事务分块读取规避 Android CursorWindow 溢出Operit 超大聊天消息读取修复基于 Room 事务分块读取规避 Android CursorWindow 溢出 本指南聚焦 OperitAndroidAI Agent人工智能大模型AI 应用工具调用本地部署MCP ClientsAgent 记忆GUI 自动化上一篇MOOTDXPython通达信数据接口的完整指南下一篇推荐开源项目NASA每日天文图片APOD微服务创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考