diff --git a/backend/ai-core/internal/background/thinker.go b/backend/ai-core/internal/background/thinker.go index 62df366..50d0e26 100644 --- a/backend/ai-core/internal/background/thinker.go +++ b/backend/ai-core/internal/background/thinker.go @@ -363,6 +363,7 @@ func (t *Thinker) SetEmotionTracker(et *persona.EmotionTracker) { // Called by the ai-core presence endpoint when gateway detects connect/disconnect. func (t *Thinker) UpdatePresence(online bool, sessionID string) { t.muLock() + defer t.muUnlock() wasOffline := !t.isUserOnline() t.setUserOnline(online) t.lastOnlineChange = time.Now() @@ -370,7 +371,7 @@ func (t *Thinker) UpdatePresence(online bool, sessionID string) { t.userSessionID = sessionID t.activeSessionID = sessionID } - t.muUnlock() + if online && wasOffline { log.Printf("[后台思考] 用户上线 (session=%s),触发重连思考", sessionID) @@ -390,6 +391,7 @@ func (t *Thinker) UpdatePresence(online bool, sessionID string) { // Called by the internal reminder-trigger endpoint when a reminder fires. func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) { t.muLock() + defer t.muUnlock() pusher := t.messagePusher pushSessionID := sessionID if pushSessionID == "" { @@ -398,7 +400,7 @@ func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) { if pushSessionID == "" { pushSessionID = t.adminSessionID } - t.muUnlock() + if pusher != nil && message != "" { log.Printf("[提醒推送] 推送LLM提醒: user=%s session=%s msg=%s", userID, pushSessionID, message) @@ -409,8 +411,9 @@ 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() + defer t.muUnlock() pusher := t.platformMessagePusher - t.muUnlock() + if pusher != nil { log.Printf("[平台推送] target=%s/%s group=%s msg=%s", target.Platform, target.ChatType, target.GroupID, message) pusher(target, message) @@ -555,9 +558,10 @@ func (t *Thinker) restoreContext() { } if !latest.IsZero() && time.Since(latest) < 24*time.Hour { t.muLock() + defer t.muUnlock() t.lastUserMessage = latest t.setLastUser(latest) - t.muUnlock() + log.Printf("[后台思考] 上下文已恢复: 最近活动 %v 前", time.Since(latest).Round(time.Second)) } else if !latest.IsZero() { log.Printf("[后台思考] 上下文恢复跳过: 最近消息 %v 前(>24h),使用默认值", time.Since(latest).Round(time.Second)) @@ -669,8 +673,9 @@ func (t *Thinker) TriggerPostChatThink() { } t.muLock() + defer t.muUnlock() canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap - t.muUnlock() + if !canThink { log.Printf("[后台思考] 距上次思考仅 %v,跳过 (最小间隔=%v)", time.Since(t.lastThinkTimeAtomic()), t.minThinkGap) @@ -738,9 +743,10 @@ func (t *Thinker) resetSilenceTimer() { case <-t.silenceTimer.C: // 再次检查:用户是否真的沉默了足够久 t.muLock() + defer t.muUnlock() silenceDuration := time.Since(t.lastUserTime()) canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap - t.muUnlock() + if silenceDuration < t.silenceTimeout { log.Printf("[后台思考] 静默检测触发但用户已活动,跳过 (实际静默=%v)", silenceDuration) @@ -855,6 +861,7 @@ func (t *Thinker) performPlatformObservation() { observationContent := fmt.Sprintf("[平台观察 %s]\n%s", time.Now().In(t.timeLocation).Format("15:04"), result.Summary) t.muLock() + defer t.muUnlock() t.pendingThoughts = append(t.pendingThoughts, &PendingThought{ Content: observationContent, CreatedAt: time.Now(), @@ -863,7 +870,7 @@ func (t *Thinker) performPlatformObservation() { if len(t.pendingThoughts) > 10 { t.pendingThoughts = t.pendingThoughts[len(t.pendingThoughts)-10:] } - t.muUnlock() + log.Printf("[后台思考] 平台观察摘要已生成 (长度=%d, 需要关注=%v)", len(result.Summary), result.NeedsAttention) } @@ -887,9 +894,10 @@ func (t *Thinker) lightThinkLoop() { return case <-time.After(t.lightThinkInterval): t.muLock() + defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) sinceLastThink := time.Since(t.lastThinkTimeAtomic()) - t.muUnlock() + // Skip if user was active recently (last 30s). if sinceLastUser < 30*time.Second { @@ -948,8 +956,9 @@ func (t *Thinker) performLightThink() { if strings.Contains(content, "【需要深思】") { log.Println("[轻量思考] 检测到【需要深思】,唤醒深度思考...") t.muLock() + defer t.muUnlock() canDeep := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap - t.muUnlock() + if canDeep { t.performThink("light_wake") } else { @@ -963,13 +972,14 @@ func (t *Thinker) performLightThink() { if topic != "" { log.Printf("[轻量思考] 话题发起: %s", topic) t.muLock() + defer t.muUnlock() pusher := t.messagePusher sessionID := t.activeSessionID if sessionID == "" { sessionID = t.adminSessionID } canPush := time.Since(t.lastProactiveTime()) >= t.proactiveMsgMinGap - t.muUnlock() + if pusher != nil && canPush { go pusher(t.adminUserID, sessionID, topic) if t.convStore != nil && sessionID != "" { @@ -1070,13 +1080,14 @@ func (t *Thinker) HasPendingThoughts() bool { // 强制执行最小间隔,防止并发调用或 bug 导致 LLM 配额被快速消耗。 func (t *Thinker) performThink(triggerReason string) { t.muLock() + defer t.muUnlock() gapSinceLast := time.Since(t.lastThinkTimeAtomic()) minGap := t.minThinkGap if minGap <= 0 { minGap = 5 * time.Second // 默认最小间隔 5 秒 } if gapSinceLast < minGap { - t.muUnlock() + log.Printf("[后台思考] 距上次思考仅 %v,跳过 (最小间隔=%v, 触发原因=%s)", gapSinceLast.Round(time.Second), minGap, triggerReason) return } @@ -1090,14 +1101,16 @@ func (t *Thinker) performThink(triggerReason string) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() t.muLock() + defer t.muUnlock() t.thinkCancel = cancel - t.muUnlock() + // 0. 让步于前台——如果用户最近有活动(非post_chat),跳过本次思考。 if triggerReason != "post_chat" { t.muLock() + defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) - t.muUnlock() + if sinceLastUser < 10*time.Second { log.Printf("[后台思考] 用户 %v 前有活动,让步于前台回复,跳过思考", sinceLastUser.Round(time.Second)) return @@ -1109,8 +1122,9 @@ func (t *Thinker) performThink(triggerReason string) { // 0. 让步于前台——如果用户最近有活动(非post_chat),跳过本次思考。 if triggerReason != "post_chat" { t.muLock() + defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) - t.muUnlock() + if sinceLastUser < 10*time.Second { log.Printf("[后台思考] 用户 %v 前有活动,让步于前台回复,跳过思考", sinceLastUser.Round(time.Second)) return @@ -1128,11 +1142,12 @@ func (t *Thinker) performThink(triggerReason string) { var convHistory []model.LLMMessage if t.convStore != nil { t.muLock() + defer t.muUnlock() sessionID := t.activeSessionID if sessionID == "" { sessionID = t.adminSessionID } - t.muUnlock() + if sessionID != "" { convHistory = t.convStore.GetHistory(sessionID, 30) if len(convHistory) > 0 { @@ -1245,13 +1260,14 @@ func (t *Thinker) performThink(triggerReason string) { var platformObservation string if triggerReason == "periodic" || triggerReason == "post_chat" { t.muLock() + defer t.muUnlock() for i := len(t.pendingThoughts) - 1; i >= 0; i-- { if strings.HasPrefix(t.pendingThoughts[i].Content, "[平台观察") { platformObservation = t.pendingThoughts[i].Content break } } - t.muUnlock() + } // 5. 构建思考提示词(根据触发原因调整) @@ -1395,6 +1411,7 @@ func (t *Thinker) performThink(triggerReason string) { adapterName, groupID := parts[0], parts[1] // Find the channel to get the correct platform type. t.muLock() + defer t.muUnlock() var platform string for _, ch := range t.platformChannels { if ch.AdapterName == adapterName && ch.ChannelID == groupID && ch.ChannelType == "group" { @@ -1402,7 +1419,7 @@ func (t *Thinker) performThink(triggerReason string) { break } } - t.muUnlock() + if platform == "" { platform = "obv11" // fallback } @@ -1648,8 +1665,9 @@ func (t *Thinker) buildThinkingUserPrompt( sb.WriteString("刚有人和你聊完天。你想自然地在心里回味一下刚才的对话……\n") case "silence": t.muLock() + defer t.muUnlock() silenceDuration := time.Since(t.lastUserTime()) - t.muUnlock() + sb.WriteString(fmt.Sprintf("已经大约 %s 没有说话了。你有点想知道大家在做什么……\n", formatDurationHuman(silenceDuration))) default: @@ -1747,11 +1765,12 @@ func (t *Thinker) buildThinkingUserPrompt( // OBv11 platform identity and available channels for proactive messaging. t.muLock() + defer t.muUnlock() qqChannels := t.platformChannels botUIDs := t.botUIDs activeSID := t.activeSessionID lastMsgTime := t.lastUserTime() - t.muUnlock() + if len(qqChannels) > 0 { sb.WriteString("\n\n【你的平台身份与可用频道】\n") @@ -2066,6 +2085,7 @@ func (t *Thinker) extractProactiveMessage(content string) (string, *ProactiveTar } adapterName := platform // fallback to format key t.muLock() + defer t.muUnlock() for _, ch := range t.platformChannels { if ch.Platform == platform && ch.ChannelType == chatType && ch.ChannelID == channelID { if ch.AdapterName != "" { @@ -2074,7 +2094,7 @@ func (t *Thinker) extractProactiveMessage(content string) (string, *ProactiveTar break } } - t.muUnlock() + target = &ProactiveTarget{ Platform: adapterName, @@ -2657,7 +2677,8 @@ func (t *Thinker) DeadlockDetected(timeout time.Duration) bool { done := make(chan struct{}) go func() { t.muLock() - t.muUnlock() + defer t.muUnlock() + close(done) }() select { @@ -2682,7 +2703,8 @@ func (t *Thinker) StartDeadlockMonitor() { done := make(chan struct{}) go func() { t.muLock() - t.muUnlock() + defer t.muUnlock() + close(done) }() select {