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