AskaEth/Cyrene
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: 全链路优化 — 死锁修复、MD3主题、上下文持久化、群聊自然化、打字状态、知识库

Browse Source 
**死锁根因修复**
- periodicThinkLoop:1015 orphaned lock → 删除（字段已原子化）
- RecordUserMessage 隔离为 recordMu
- atomic.Int64 替换 lastUserMessage/lastThinkTime 等

**MD3 / Android 17 主题**
- 毛玻璃卡片 (backdrop-filter)
- MD3 色彩令牌 (pink primary #f472b6)
- icons.js 独立矢量图标库 + 运行时 emoji 替换
- 无边框卡片、圆角按钮、阴影层次

**上下文持久化**
- AddMessage → saveToDB 异步写 PostgreSQL
- LoadFromDB 恢复 (admin-session-main + 懒加载)
- LLMMessage.Timestamp 字段

**群聊与适配器**
- group_ambient 模式: 非@消息让 LLM 自己判断是否插话
- 戳一戳动作消息总是回复
- NapCat 打字状态 (set_input_status, 最小3秒显示)
- HTTP API 配置 (http_url/http_token)

**知识库 & 防编造**
- knowledge.CanHandle 对 chat 意图也触发
- 关键词预筛选避免无关 embedding 调用
- persona + synthesizer 三重诚实规则
- 工具结果持久化到会话历史

**平台桥接器**
- detached:true Go进程独立存活
- ethend 重启自动接管已运行服务
- stop() 接管模式 taskkill/F/ PID
- Windows netstat 替代 fuser 获取 PID
- 重复适配器种子逻辑修复
- 失败转发日志 Direction: error

**崩溃诊断**
- crashlog 包 (Recover + WrapHTTP + LLMCall)
- /api/v1/debug/goroutines 端点
- thinker 操作日志 + 30s stats
- 日志写入 logs/ 目录持久化

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
AskaEth
2026-06-28 11:29:29 +08:00
parent 0d6970a2d3
commit fd44b15d81
23 changed files with 1447 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ai-core/internal/background/thinker.go
+167 -102
View File
@@ -7,9 +7,11 @@ import (
"log"
"os"
"regexp"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
ctxbuild "git.yeij.top/AskaEth/Cyrene/ai-core/internal/context"
@@ -124,8 +126,11 @@ func ParsePlatformChannels(raw string) []PlatformChannel {
//
// 主动消息：思考中如有【主动消息】标记，会通过 messagePusher 回调推送给在线用户（带频率限制）。
type Thinker struct {
mu sync.Mutex
wg sync.WaitGroup
mu sync.Mutex
recordMu sync.Mutex
muLockedBy string // debug: file:line of last Lock()
muLockedAt time.Time // debug: when Lock() was acquired // RecordUserMessage专用，隔离于mu死锁
wg sync.WaitGroup
stopCh chan struct{}
enabled bool
@@ -198,11 +203,11 @@ type Thinker struct {
pendingThoughts []*PendingThought
lastUserMessage time.Time
lastThinkTime time.Time
lastProactiveMsgTime time.Time
lastProactiveMsgTime time.Time // deprecated, use lastProactiveTime()
thinkCancel context.CancelFunc // 取消当前思考，让步于前台
// 思考计数器（用于周期性记忆维护，每 N 次思考触发一次）
thinkCount int
thinkCountAtomic atomic.Int64
// Phase 1 Step 4: 思考链 + 自主工具安全策略
chain *ThinkChain
@@ -218,7 +223,10 @@ type Thinker struct {
scheduleLoader *ScheduleLoader
// Phase 2: 在线状态追踪
userOnline bool
userOnlineAtomic atomic.Bool
lastUserMsgNs atomic.Int64 // UnixNano (replaces lastUserMessage)
lastThinkNs atomic.Int64 // UnixNano (replaces lastThinkTime)
lastProactiveNs atomic.Int64 // UnixNano (replaces lastProactiveMsgTime)
lastOnlineChange time.Time
userSessionID string // 当前活跃的 session ID (用于重连)
@@ -284,28 +292,28 @@ func timePeriod(now time.Time) (string, string) {
// SetMessagePusher 设置主动消息推送回调
// SetScheduleLoader sets the dynamic schedule loader for interval calculation.
func (t *Thinker) SetScheduleLoader(loader *ScheduleLoader) {
t.mu.Lock()
defer t.mu.Unlock()
t.muLock()
defer t.muUnlock()
t.scheduleLoader = loader
}
func (t *Thinker) SetMessagePusher(pusher func(string, string, string)) {
t.mu.Lock()
defer t.mu.Unlock()
t.muLock()
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.mu.Lock()
defer t.mu.Unlock()
t.muLock()
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.mu.Lock()
defer t.mu.Unlock()
t.muLock()
defer t.muUnlock()
if t.botUIDs == nil {
t.botUIDs = make(map[string]string)
}
@@ -316,8 +324,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.mu.Lock()
defer t.mu.Unlock()
t.muLock()
defer t.muUnlock()
for i, ch := range t.platformChannels {
if ch.Platform == platform && ch.ChannelType == channelType && ch.ChannelID == channelID {
@@ -346,23 +354,23 @@ func (t *Thinker) AddOrUpdatePlatformChannel(platform, channelType, channelID, c
// SetEmotionTracker sets the emotion tracker.
func (t *Thinker) SetEmotionTracker(et *persona.EmotionTracker) {
t.mu.Lock()
defer t.mu.Unlock()
t.muLock()
defer t.muUnlock()
t.emotionTracker = et
}
// 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.mu.Lock()
wasOffline := !t.userOnline
t.userOnline = online
t.muLock()
wasOffline := !t.isUserOnline()
t.setUserOnline(online)
t.lastOnlineChange = time.Now()
if sessionID != "" {
t.userSessionID = sessionID
t.activeSessionID = sessionID
}
t.mu.Unlock()
t.muUnlock()
if online && wasOffline {
log.Printf("[后台思考] 用户上线 (session=%s)，触发重连思考", sessionID)
@@ -381,7 +389,7 @@ 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.mu.Lock()
t.muLock()
pusher := t.messagePusher
pushSessionID := sessionID
if pushSessionID == "" {
@@ -390,7 +398,7 @@ func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) {
if pushSessionID == "" {
pushSessionID = t.adminSessionID
}
t.mu.Unlock()
t.muUnlock()
if pusher != nil && message != "" {
log.Printf("[提醒推送] 推送LLM提醒: user=%s session=%s msg=%s", userID, pushSessionID, message)
@@ -400,9 +408,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.mu.Lock()
t.muLock()
pusher := t.platformMessagePusher
t.mu.Unlock()
t.muUnlock()
if pusher != nil {
log.Printf("[平台推送] target=%s/%s group=%s msg=%s", target.Platform, target.ChatType, target.GroupID, message)
pusher(target, message)
@@ -413,9 +421,9 @@ 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.mu.Lock()
defer t.mu.Unlock()
return time.Since(t.lastUserMessage) < d
t.muLock()
defer t.muUnlock()
return time.Since(t.lastUserTime()) < d
}
@@ -532,7 +540,7 @@ func (t *Thinker) restoreContext() {
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
defer func() { t.mu.Lock(); t.thinkCancel = nil; t.mu.Unlock() }()
defer func() { t.muLock(); t.thinkCancel = nil; t.muUnlock() }()
memories, err := t.memClient.Query(ctx, model.MemoryQuery{
UserID: t.adminUserID,
@@ -551,9 +559,10 @@ func (t *Thinker) restoreContext() {
}
// Only use if reasonably recent (within 24h). Old memories don't represent user activity.
if !latest.IsZero() && time.Since(latest) < 24*time.Hour {
t.mu.Lock()
t.muLock()
t.lastUserMessage = latest
t.mu.Unlock()
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))
@@ -573,6 +582,10 @@ func (t *Thinker) Start() {
// 恢复上下文：从持久化存储查询最近活动时间，避免重启后"失忆"
t.restoreContext()
// 同步原子字段（restoreContext 写入 lastUserMessage 后需要同步）
t.setLastUser(t.lastUserMessage)
t.setLastThink(t.lastThinkTime)
t.setLastProactive(t.lastProactiveMsgTime)
// 初始化静默检测定时器（但不启动，等第一次用户消息后启动）
if t.silenceTimeout > 0 {
@@ -628,14 +641,16 @@ func (t *Thinker) Stop() {
// 2. 记录当前活跃的前端会话 ID（用于对话上下文检索和主动消息推送）
// 3. 重置静默检测的一次性定时器（如果启用）
func (t *Thinker) RecordUserMessage(sessionID string) {
t.mu.Lock()
t.lastUserMessage = time.Now()
now := time.Now()
t.recordMu.Lock()
t.lastUserMessage = now
t.setLastUser(now)
if sessionID != "" {
t.activeSessionID = sessionID
}
// 用户主动发消息时重置主动消息推送冷却——活跃对话中应允许昔涟回复
t.lastProactiveMsgTime = time.Time{}
t.mu.Unlock()
t.setLastProactive(time.Time{})
t.recordMu.Unlock()
if t.thinkCancel != nil {
t.thinkCancel()
t.thinkCancel = nil
@@ -656,12 +671,12 @@ func (t *Thinker) TriggerPostChatThink() {
return
}
t.mu.Lock()
canThink := time.Since(t.lastThinkTime) >= t.minThinkGap
t.mu.Unlock()
t.muLock()
canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap
t.muUnlock()
if !canThink {
log.Printf("[后台思考] 距上次思考仅 %v，跳过 (最小间隔=%v)", time.Since(t.lastThinkTime), t.minThinkGap)
log.Printf("[后台思考] 距上次思考仅 %v，跳过 (最小间隔=%v)", time.Since(t.lastThinkTimeAtomic()), t.minThinkGap)
return
}
@@ -725,10 +740,10 @@ func (t *Thinker) resetSilenceTimer() {
return
case <-t.silenceTimer.C:
// 再次检查：用户是否真的沉默了足够久
t.mu.Lock()
silenceDuration := time.Since(t.lastUserMessage)
canThink := time.Since(t.lastThinkTime) >= t.minThinkGap
t.mu.Unlock()
t.muLock()
silenceDuration := time.Since(t.lastUserTime())
canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap
t.muUnlock()
if silenceDuration < t.silenceTimeout {
log.Printf("[后台思考] 静默检测触发但用户已活动，跳过 (实际静默=%v)", silenceDuration)
@@ -842,7 +857,7 @@ func (t *Thinker) performPlatformObservation() {
}
observationContent := fmt.Sprintf("[平台观察 %s]\n%s", time.Now().In(t.timeLocation).Format("15:04"), result.Summary)
t.mu.Lock()
t.muLock()
t.pendingThoughts = append(t.pendingThoughts, &PendingThought{
Content: observationContent,
CreatedAt: time.Now(),
@@ -851,7 +866,7 @@ func (t *Thinker) performPlatformObservation() {
if len(t.pendingThoughts) > 10 {
t.pendingThoughts = t.pendingThoughts[len(t.pendingThoughts)-10:]
}
t.mu.Unlock()
t.muUnlock()
log.Printf("[后台思考] 平台观察摘要已生成 (长度=%d, 需要关注=%v)", len(result.Summary), result.NeedsAttention)
}
@@ -874,10 +889,10 @@ func (t *Thinker) lightThinkLoop() {
log.Println("[后台思考] 轻量思考已停止")
return
case <-time.After(t.lightThinkInterval):
t.mu.Lock()
sinceLastUser := time.Since(t.lastUserMessage)
sinceLastThink := time.Since(t.lastThinkTime)
t.mu.Unlock()
t.muLock()
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 {
@@ -895,6 +910,8 @@ func (t *Thinker) lightThinkLoop() {
// performLightThink runs a single lightweight thinking cycle using the fast model.
func (t *Thinker) performLightThink() {
log.Printf("[debug] lightThink ENTER")
defer func() { log.Printf("[debug] lightThink EXIT") }()
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
@@ -933,9 +950,9 @@ func (t *Thinker) performLightThink() {
// Check if deep think should be woken.
if strings.Contains(content, "【需要深思】") {
log.Println("[轻量思考] 检测到【需要深思】，唤醒深度思考...")
t.mu.Lock()
canDeep := time.Since(t.lastThinkTime) >= t.minThinkGap
t.mu.Unlock()
t.muLock()
canDeep := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap
t.muUnlock()
if canDeep {
t.performThink("light_wake")
} else {
@@ -948,14 +965,14 @@ func (t *Thinker) performLightThink() {
topic := strings.TrimSpace(content[idx+len("【话题发起】"):])
if topic != "" {
log.Printf("[轻量思考] 话题发起: %s", topic)
t.mu.Lock()
t.muLock()
pusher := t.messagePusher
sessionID := t.activeSessionID
if sessionID == "" {
sessionID = t.adminSessionID
}
canPush := time.Since(t.lastProactiveMsgTime) >= t.proactiveMsgMinGap
t.mu.Unlock()
canPush := time.Since(t.lastProactiveTime()) >= t.proactiveMsgMinGap
t.muUnlock()
if pusher != nil && canPush {
go pusher(t.adminUserID, sessionID, topic)
if t.convStore != nil && sessionID != "" {
@@ -995,15 +1012,10 @@ func (t *Thinker) periodicThinkLoop() {
log.Println("[后台思考] 周期性思考已停止")
return
case <-time.After(interval):
t.mu.Lock()
sinceLastThink := time.Since(t.lastThinkTime)
sinceLastUser := time.Since(t.lastUserMessage)
t.mu.Unlock()
// 离线时降低思考频率（可配置，默认 10 分钟）
t.mu.Lock()
isOffline := !t.userOnline
t.mu.Unlock()
// (atomic fields, no lock needed — all reads are lock-free)
sinceLastThink := time.Since(t.lastThinkTimeAtomic())
sinceLastUser := time.Since(t.lastUserTime())
isOffline := !t.isUserOnline()
offlineMinGap := t.offlineThinkGap
// 跳过条件：用户最近在活动（30s 内有消息），说明正在对话中
@@ -1022,7 +1034,7 @@ func (t *Thinker) periodicThinkLoop() {
continue
}
log.Printf("[后台思考] 周期性触发 (间隔=%v, 上次思考=%v前, 上次用户消息=%v前)", interval, sinceLastThink.Round(time.Second), sinceLastUser.Round(time.Second))
log.Printf("[debug] periodicThink TRIGGER interval=%v lastThink=%v lastUser=%v", interval, sinceLastThink.Round(time.Second), sinceLastUser.Round(time.Second))
t.performThink("periodic")
}
}
@@ -1030,8 +1042,8 @@ func (t *Thinker) periodicThinkLoop() {
// GetPendingThoughts 获取并消费所有待处理的后台思考
func (t *Thinker) GetPendingThoughts() []*PendingThought {
t.mu.Lock()
defer t.mu.Unlock()
t.muLock()
defer t.muUnlock()
if len(t.pendingThoughts) == 0 {
return nil
@@ -1048,8 +1060,8 @@ func (t *Thinker) GetPendingThoughts() []*PendingThought {
// HasPendingThoughts 检查是否有待处理的思考
func (t *Thinker) HasPendingThoughts() bool {
t.mu.Lock()
defer t.mu.Unlock()
t.muLock()
defer t.muUnlock()
return len(t.pendingThoughts) > 0
}
@@ -1060,34 +1072,35 @@ func (t *Thinker) HasPendingThoughts() bool {
// 防御性速率限制：即使调用方未检查 minThinkGap，performThink 自身也会
// 强制执行最小间隔，防止并发调用或 bug 导致 LLM 配额被快速消耗。
func (t *Thinker) performThink(triggerReason string) {
t.mu.Lock()
gapSinceLast := time.Since(t.lastThinkTime)
t.muLock()
gapSinceLast := time.Since(t.lastThinkTimeAtomic())
minGap := t.minThinkGap
if minGap <= 0 {
minGap = 5 * time.Second // 默认最小间隔 5 秒
}
if gapSinceLast < minGap {
t.mu.Unlock()
t.muUnlock()
log.Printf("[后台思考] 距上次思考仅 %v，跳过 (最小间隔=%v, 触发原因=%s)", gapSinceLast.Round(time.Second), minGap, triggerReason)
return
}
t.lastThinkTime = time.Now()
t.thinkCount++
currentCount := t.thinkCount
t.mu.Unlock()
t.setLastThink(t.lastThinkTime)
t.incThinkCount()
currentCount := t.thinkCount()
t.muUnlock()
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()
t.mu.Lock()
t.muLock()
t.thinkCancel = cancel
t.mu.Unlock()
t.muUnlock()
// 0. 让步于前台——如果用户最近有活动（非post_chat），跳过本次思考。
if triggerReason != "post_chat" {
t.mu.Lock()
sinceLastUser := time.Since(t.lastUserMessage)
t.mu.Unlock()
t.muLock()
sinceLastUser := time.Since(t.lastUserTime())
t.muUnlock()
if sinceLastUser < 10*time.Second {
log.Printf("[后台思考] 用户 %v 前有活动，让步于前台回复，跳过思考", sinceLastUser.Round(time.Second))
return
@@ -1098,9 +1111,9 @@ func (t *Thinker) performThink(triggerReason string) {
// 0. 让步于前台——如果用户最近有活动（非post_chat），跳过本次思考。
if triggerReason != "post_chat" {
t.mu.Lock()
sinceLastUser := time.Since(t.lastUserMessage)
t.mu.Unlock()
t.muLock()
sinceLastUser := time.Since(t.lastUserTime())
t.muUnlock()
if sinceLastUser < 10*time.Second {
log.Printf("[后台思考] 用户 %v 前有活动，让步于前台回复，跳过思考", sinceLastUser.Round(time.Second))
return
@@ -1117,12 +1130,12 @@ func (t *Thinker) performThink(triggerReason string) {
// 2. 获取当前活跃会话的对话历史（优先活跃会话，回退到管理员主会话）
var convHistory []model.LLMMessage
if t.convStore != nil {
t.mu.Lock()
t.muLock()
sessionID := t.activeSessionID
if sessionID == "" {
sessionID = t.adminSessionID
}
t.mu.Unlock()
t.muUnlock()
if sessionID != "" {
convHistory = t.convStore.GetHistory(sessionID, 30)
if len(convHistory) > 0 {
@@ -1234,14 +1247,14 @@ func (t *Thinker) performThink(triggerReason string) {
// 4.5 获取最近平台观察（定期触发和对话后触发时注入）
var platformObservation string
if triggerReason == "periodic" || triggerReason == "post_chat" {
t.mu.Lock()
t.muLock()
for i := len(t.pendingThoughts) - 1; i >= 0; i-- {
if strings.HasPrefix(t.pendingThoughts[i].Content, "[平台观察") {
platformObservation = t.pendingThoughts[i].Content
break
}
}
t.mu.Unlock()
t.muUnlock()
}
// 5. 构建思考提示词（根据触发原因调整）
@@ -1384,7 +1397,7 @@ 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.mu.Lock()
t.muLock()
var platform string
for _, ch := range t.platformChannels {
if ch.AdapterName == adapterName && ch.ChannelID == groupID && ch.ChannelType == "group" {
@@ -1392,7 +1405,7 @@ func (t *Thinker) performThink(triggerReason string) {
break
}
}
t.mu.Unlock()
t.muUnlock()
if platform == "" {
platform = "obv11" // fallback
}
@@ -1466,7 +1479,7 @@ func (t *Thinker) performThink(triggerReason string) {
log.Printf("[后台思考] 完成 (触发原因=%s, 轮数=%d, 内容长度=%d, 工具调用=%d次)", triggerReason, len(allContents), len(finalContent), totalToolCalls)
// 9. 记忆维护：机械合并(每10次) + LLM整理(每次)
t.maybeMaintainMemories(currentCount)
t.maybeMaintainMemories(int(currentCount))
t.performMemoryConsolidation(ctx)
}
@@ -1637,9 +1650,9 @@ func (t *Thinker) buildThinkingUserPrompt(
case "post_chat":
sb.WriteString("刚有人和你聊完天。你想自然地在心里回味一下刚才的对话……\n")
case "silence":
t.mu.Lock()
silenceDuration := time.Since(t.lastUserMessage)
t.mu.Unlock()
t.muLock()
silenceDuration := time.Since(t.lastUserTime())
t.muUnlock()
sb.WriteString(fmt.Sprintf("已经大约 %s 没有说话了。你有点想知道大家在做什么……\n",
formatDurationHuman(silenceDuration)))
default:
@@ -1736,12 +1749,12 @@ func (t *Thinker) buildThinkingUserPrompt(
}
// OBv11 platform identity and available channels for proactive messaging.
t.mu.Lock()
t.muLock()
qqChannels := t.platformChannels
botUIDs := t.botUIDs
activeSID := t.activeSessionID
lastMsgTime := t.lastUserMessage
t.mu.Unlock()
lastMsgTime := t.lastUserTime()
t.muUnlock()
if len(qqChannels) > 0 {
sb.WriteString("\n\n【你的平台身份与可用频道】\n")
@@ -1884,7 +1897,7 @@ func (t *Thinker) buildOpenAITools() []llm.OpenAITool {
// storeThought 存储思考结果到待推送队列，并异步持久化到 memory-service
func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCount int) {
t.mu.Lock()
t.muLock()
t.pendingThoughts = append(t.pendingThoughts, &PendingThought{
Content: content,
CreatedAt: time.Now(),
@@ -1915,21 +1928,23 @@ func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCou
canPush = false
}
if canPush && t.proactiveGuard != nil {
decision := t.proactiveGuard.Evaluate(time.Now(), t.lastProactiveMsgTime, urgency, "active")
decision := t.proactiveGuard.Evaluate(time.Now(), t.lastProactiveTime(), urgency, "active")
logDecision(decision)
if !decision.ShouldSend {
canPush = false
} else {
t.lastProactiveMsgTime = time.Now()
t.setLastProactive(t.lastProactiveMsgTime)
t.proactiveGuard.RecordSend(time.Now())
}
} else if canPush {
gapSinceLast := time.Since(t.lastProactiveMsgTime)
gapSinceLast := time.Since(t.lastProactiveTime())
if gapSinceLast < 30*time.Minute {
log.Printf("[后台思考] 主动消息距上次仅 %v，跳过推送", gapSinceLast.Round(time.Second))
canPush = false
} else {
t.lastProactiveMsgTime = time.Now()
t.setLastProactive(t.lastProactiveMsgTime)
}
}
}
@@ -1939,7 +1954,7 @@ func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCou
copy := *proactiveTarget
targetCopy = &copy
}
t.mu.Unlock()
t.muUnlock()
log.Printf("[后台思考] 思考已存储 (当前累积 %d 条待推送思考)", len(t.pendingThoughts))
@@ -2037,7 +2052,7 @@ func (t *Thinker) extractProactiveMessage(content string) (string, *ProactiveTar
channelID = "private_" + m[2]
}
adapterName := platform // fallback to format key
t.mu.Lock()
t.muLock()
for _, ch := range t.platformChannels {
if ch.Platform == platform && ch.ChannelType == chatType && ch.ChannelID == channelID {
if ch.AdapterName != "" {
@@ -2046,7 +2061,7 @@ func (t *Thinker) extractProactiveMessage(content string) (string, *ProactiveTar
break
}
}
t.mu.Unlock()
t.muUnlock()
target = &ProactiveTarget{
Platform: adapterName,
@@ -2594,3 +2609,53 @@ func getEnvDuration(key string, fallbackSec int) time.Duration {
}
return time.Duration(sec) * time.Second
}
// muLock acquires t.mu with caller tracking.
func (t *Thinker) muLock() {
t.mu.Lock()
_, file, line, _ := runtime.Caller(1)
if idx := strings.LastIndex(file, "Cyrene/"); idx >= 0 {
file = file[idx+len("Cyrene/"):]
}
t.muLockedBy = fmt.Sprintf("%s:%d", file, line)
t.muLockedAt = time.Now()
}
// muUnlock releases t.mu and clears the tracking.
func (t *Thinker) muUnlock() {
t.muLockedBy = ""
t.mu.Unlock()
}
// ========== lock-free atomic accessors ==========
func (t *Thinker) lastUserTime() time.Time { return time.Unix(0, t.lastUserMsgNs.Load()) }
func (t *Thinker) setLastUser(ts time.Time) { t.lastUserMsgNs.Store(ts.UnixNano()) }
func (t *Thinker) lastThinkTimeAtomic() time.Time { return time.Unix(0, t.lastThinkNs.Load()) }
func (t *Thinker) setLastThink(ts time.Time) { t.lastThinkNs.Store(ts.UnixNano()) }
func (t *Thinker) thinkCount() int64 { return t.thinkCountAtomic.Add(0) }
func (t *Thinker) incThinkCount() int64 { return t.thinkCountAtomic.Add(1) }
func (t *Thinker) isUserOnline() bool { return t.userOnlineAtomic.Load() }
func (t *Thinker) setUserOnline(v bool) { t.userOnlineAtomic.Store(v) }
func (t *Thinker) lastProactiveTime() time.Time { return time.Unix(0, t.lastProactiveNs.Load()) }
func (t *Thinker) setLastProactive(ts time.Time) { t.lastProactiveNs.Store(ts.UnixNano()) }
// DeadlockDetected tries to acquire t.mu with a timeout. Returns true if the lock appears orphaned.
func (t *Thinker) DeadlockDetected(timeout time.Duration) bool {
done := make(chan struct{})
go func() {
t.muLock()
t.muUnlock()
close(done)
}()
select {
case <-done:
return false
case <-time.After(timeout):
return true
}
}
// LockHolder returns info about who currently holds t.mu (for debugging).
func (t *Thinker) LockHolder() (string, time.Time) {
return t.muLockedBy, t.muLockedAt
}