diff --git a/backend/ai-core/internal/background/thinker.go b/backend/ai-core/internal/background/thinker.go index 027f78c..d8ba2d6 100644 --- a/backend/ai-core/internal/background/thinker.go +++ b/backend/ai-core/internal/background/thinker.go @@ -292,27 +292,31 @@ func timePeriod(now time.Time) (string, string) { // SetMessagePusher 设置主动消息推送回调 // SetScheduleLoader sets the dynamic schedule loader for interval calculation. func (t *Thinker) SetScheduleLoader(loader *ScheduleLoader) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.scheduleLoader = loader } func (t *Thinker) SetMessagePusher(pusher func(string, string, string)) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.messagePusher = pusher } // SetPlatformMessagePusher sets the callback for pushing proactive messages to platform adapters (OBv11, etc.). func (t *Thinker) SetPlatformMessagePusher(pusher func(ProactiveTarget, string)) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.platformMessagePusher = pusher } // SetBotUID sets the bot's own platform UID (e.g., OBv11 account). func (t *Thinker) SetBotUID(platform, uid string) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() if t.botUIDs == nil { t.botUIDs = make(map[string]string) @@ -324,7 +328,8 @@ func (t *Thinker) SetBotUID(platform, uid string) { // AddOrUpdatePlatformChannel adds or updates a platform channel with resolved display name. func (t *Thinker) AddOrUpdatePlatformChannel(platform, channelType, channelID, channelName, adapterID, adapterName string) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() for i, ch := range t.platformChannels { @@ -354,7 +359,8 @@ func (t *Thinker) AddOrUpdatePlatformChannel(platform, channelType, channelID, c // SetEmotionTracker sets the emotion tracker. func (t *Thinker) SetEmotionTracker(et *persona.EmotionTracker) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.emotionTracker = et } @@ -362,7 +368,8 @@ func (t *Thinker) SetEmotionTracker(et *persona.EmotionTracker) { // UpdatePresence updates the user online status. // Called by the ai-core presence endpoint when gateway detects connect/disconnect. func (t *Thinker) UpdatePresence(online bool, sessionID string) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() wasOffline := !t.isUserOnline() t.setUserOnline(online) @@ -390,7 +397,8 @@ func (t *Thinker) UpdatePresence(online bool, sessionID string) { // TriggerReminderMessage pushes a reminder message generated by LLM to the user. // Called by the internal reminder-trigger endpoint when a reminder fires. func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() pusher := t.messagePusher pushSessionID := sessionID @@ -410,7 +418,8 @@ func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) { // PushPlatformMessage pushes a message to a platform channel via the platform pusher. func (t *Thinker) PushPlatformMessage(target ProactiveTarget, message string) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() pusher := t.platformMessagePusher @@ -424,7 +433,8 @@ func (t *Thinker) PushPlatformMessage(target ProactiveTarget, message string) { // IsUserRecentlyActive returns true if the user has been active within the given duration. func (t *Thinker) IsUserRecentlyActive(d time.Duration) bool { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() return time.Since(t.lastUserTime()) < d } @@ -557,7 +567,8 @@ func (t *Thinker) restoreContext() { } } if !latest.IsZero() && time.Since(latest) < 24*time.Hour { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.lastUserMessage = latest t.setLastUser(latest) @@ -672,7 +683,8 @@ func (t *Thinker) TriggerPostChatThink() { return } - t.muLock() + if !t.muLock() { + return defer t.muUnlock() canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap @@ -742,7 +754,8 @@ func (t *Thinker) resetSilenceTimer() { return case <-t.silenceTimer.C: // 再次检查:用户是否真的沉默了足够久 - t.muLock() + if !t.muLock() { + return defer t.muUnlock() silenceDuration := time.Since(t.lastUserTime()) canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap @@ -860,7 +873,8 @@ func (t *Thinker) performPlatformObservation() { } observationContent := fmt.Sprintf("[平台观察 %s]\n%s", time.Now().In(t.timeLocation).Format("15:04"), result.Summary) - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.pendingThoughts = append(t.pendingThoughts, &PendingThought{ Content: observationContent, @@ -893,7 +907,8 @@ func (t *Thinker) lightThinkLoop() { log.Println("[后台思考] 轻量思考已停止") return case <-time.After(t.lightThinkInterval): - t.muLock() + if !t.muLock() { + return defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) sinceLastThink := time.Since(t.lastThinkTimeAtomic()) @@ -955,7 +970,8 @@ func (t *Thinker) performLightThink() { // Check if deep think should be woken. if strings.Contains(content, "【需要深思】") { log.Println("[轻量思考] 检测到【需要深思】,唤醒深度思考...") - t.muLock() + if !t.muLock() { + return defer t.muUnlock() canDeep := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap @@ -971,7 +987,8 @@ func (t *Thinker) performLightThink() { topic := strings.TrimSpace(content[idx+len("【话题发起】"):]) if topic != "" { log.Printf("[轻量思考] 话题发起: %s", topic) - t.muLock() + if !t.muLock() { + return defer t.muUnlock() pusher := t.messagePusher sessionID := t.activeSessionID @@ -1049,7 +1066,8 @@ func (t *Thinker) periodicThinkLoop() { // GetPendingThoughts 获取并消费所有待处理的后台思考 func (t *Thinker) GetPendingThoughts() []*PendingThought { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() if len(t.pendingThoughts) == 0 { @@ -1067,7 +1085,8 @@ func (t *Thinker) GetPendingThoughts() []*PendingThought { // HasPendingThoughts 检查是否有待处理的思考 func (t *Thinker) HasPendingThoughts() bool { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() return len(t.pendingThoughts) > 0 } @@ -1079,7 +1098,8 @@ func (t *Thinker) HasPendingThoughts() bool { // 防御性速率限制:即使调用方未检查 minThinkGap,performThink 自身也会 // 强制执行最小间隔,防止并发调用或 bug 导致 LLM 配额被快速消耗。 func (t *Thinker) performThink(triggerReason string) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() gapSinceLast := time.Since(t.lastThinkTimeAtomic()) minGap := t.minThinkGap @@ -1100,14 +1120,16 @@ func (t *Thinker) performThink(triggerReason string) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.thinkCancel = cancel // 0. 让步于前台——如果用户最近有活动(非post_chat),跳过本次思考。 if triggerReason != "post_chat" { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) @@ -1121,7 +1143,8 @@ func (t *Thinker) performThink(triggerReason string) { // 0. 让步于前台——如果用户最近有活动(非post_chat),跳过本次思考。 if triggerReason != "post_chat" { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) @@ -1141,7 +1164,8 @@ func (t *Thinker) performThink(triggerReason string) { // 2. 获取当前活跃会话的对话历史(优先活跃会话,回退到管理员主会话) var convHistory []model.LLMMessage if t.convStore != nil { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() sessionID := t.activeSessionID if sessionID == "" { @@ -1259,7 +1283,8 @@ func (t *Thinker) performThink(triggerReason string) { // 4.5 获取最近平台观察(定期触发和对话后触发时注入) var platformObservation string if triggerReason == "periodic" || triggerReason == "post_chat" { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() for i := len(t.pendingThoughts) - 1; i >= 0; i-- { if strings.HasPrefix(t.pendingThoughts[i].Content, "[平台观察") { @@ -1410,7 +1435,8 @@ func (t *Thinker) performThink(triggerReason string) { if len(parts) == 2 { adapterName, groupID := parts[0], parts[1] // Find the channel to get the correct platform type. - t.muLock() + if !t.muLock() { + return defer t.muUnlock() var platform string for _, ch := range t.platformChannels { @@ -1664,7 +1690,8 @@ func (t *Thinker) buildThinkingUserPrompt( case "post_chat": sb.WriteString("刚有人和你聊完天。你想自然地在心里回味一下刚才的对话……\n") case "silence": - t.muLock() + if !t.muLock() { + return defer t.muUnlock() silenceDuration := time.Since(t.lastUserTime()) @@ -1764,7 +1791,8 @@ func (t *Thinker) buildThinkingUserPrompt( } // OBv11 platform identity and available channels for proactive messaging. - t.muLock() + if !t.muLock() { + return defer t.muUnlock() qqChannels := t.platformChannels botUIDs := t.botUIDs @@ -1913,7 +1941,8 @@ func (t *Thinker) buildOpenAITools() []llm.OpenAITool { // storeThought 存储思考结果到待推送队列,并异步持久化到 memory-service func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCount int) { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() t.pendingThoughts = append(t.pendingThoughts, &PendingThought{ Content: content, @@ -2089,7 +2118,8 @@ func (t *Thinker) extractProactiveMessage(content string) (string, *ProactiveTar channelID = "private_" + m[2] } adapterName := platform // fallback to format key - t.muLock() + if !t.muLock() { + return defer t.muUnlock() for _, ch := range t.platformChannels { if ch.Platform == platform && ch.ChannelType == chatType && ch.ChannelID == channelID { @@ -2681,7 +2711,8 @@ func (t *Thinker) setLastProactive(ts time.Time) { t.lastProactiveNs.Store(ts.Un func (t *Thinker) DeadlockDetected(timeout time.Duration) bool { done := make(chan struct{}) go func() { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() close(done) @@ -2707,7 +2738,8 @@ func (t *Thinker) StartDeadlockMonitor() { for range ticker.C { done := make(chan struct{}) go func() { - t.muLock() + if !t.muLock() { + return defer t.muUnlock() close(done)