fix: 回退muLock为简单版本,保留deadlock监控

This commit is contained in:
2026-06-28 16:57:36 +08:00
parent 3b8b35069a
commit 32160d9453
+64 -32
View File
@@ -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 {
// 防御性速率限制:即使调用方未检查 minThinkGapperformThink 自身也会
// 强制执行最小间隔,防止并发调用或 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)