495 lines
17 KiB
Go
495 lines
17 KiB
Go
// backthink 包实现 M8 自主后台思考服务
|
||
// （DevDocs/design/12-后台思考与主动.md §0-§3）。
|
||
//
|
||
// 它在两类触发下运行无人值守的思考轮：
|
||
//
|
||
// - 周期性心跳（config.Background.IntervalSec），请求注入的运行时
|
||
// 反思，并允许调用工具（IoT、日历、提醒），
|
||
// 但默认不得发出 chat 风格的消息。
|
||
// - 进程内总线上的异步事件（iot.device_update、
|
||
// reminder.trigger、system.event），它们唤醒一轮思考。
|
||
//
|
||
// 每轮产出一段结论，其结构化尾段决定她
|
||
// 到底想不想说话（design/20 §2 P2），随后咨询 Decide——
|
||
// 作为生产决策路径恢复的重要度级联——来判断此刻是否
|
||
// 合适（事件重要度 / 时段 / 用户状态，如心率、
|
||
// 睡眠、工作、忙碌）。critical 安全事件穿透工作/忙碌
|
||
// 与安静时段（裁决 07.4.2），而心跳轮永远不会被长期
|
||
// 静默的活动间隔拦下（裁决 07.4.3）。当值得打扰时，
|
||
// 推送先走活跃客户端，再降级到 IM 桥。
|
||
// DevDocs/design/11 §6 的主动推送频率门只约束
|
||
// 周期性心跳；事件驱动的推送绕过它（主人要求："该发就发"），
|
||
// 但受同一事件去重约束（见 event_dedup.go）。每轮都经
|
||
// Store 持久化到 SQLite，好让 M9/M10 监控面板能渲染后台思考页
|
||
// （design/12 §4）。
|
||
//
|
||
// 本包并发安全：一把互斥锁守护生命周期状态，
|
||
// 注入的 store/pusher/thinker 按契约要求并发安全，
|
||
// 且内部发布路径绝不会等待慢消费者。
|
||
package backthink
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"sync"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
)
|
||
|
||
// 本包返回的哨兵错误。用 errors.Is 判定；它们被直接
|
||
// 返回或用 %w 包装，因此底层原因保持完整。
|
||
var (
|
||
// 用 nil 数据库构建 Backthink 或 Store 时返回 ErrNilDB。
|
||
ErrNilDB = errors.New("backthink: nil database handle")
|
||
// New 未收到 Thinker 时返回 ErrNoThinker。
|
||
ErrNoThinker = errors.New("backthink: nil thinker")
|
||
// LoopThinker 包装了 nil loop 时由它返回 ErrNoLoop。
|
||
ErrNoLoop = errors.New("backthink: nil loop")
|
||
)
|
||
|
||
// Thinker 在注入的运行时上跑一轮无人值守的思考。它是
|
||
// 服务与 loop/agent 之间的接缝：生产接线交回一个基于
|
||
// *loop.Loop 的 LoopThinker，而测试注入 mock。实现可以
|
||
// 调用工具（IoT、日历、提醒），且必须并发安全。
|
||
type Thinker interface {
|
||
// Think 为给定的会话/触发调用一轮自主思考，
|
||
// 并返回结论文本。
|
||
Think(ctx context.Context, sessionID, trigger string) (string, error)
|
||
}
|
||
|
||
// Deps 是注入给 Backthink 的依赖集合。只有 Thinker 与 DB 是
|
||
// 必需的；其余优雅降级（Bus 为 nil 则关闭事件触发，State 为
|
||
// nil 则给出中性用户状态，Pusher 为 nil 则由 Active/IM 构建）。
|
||
type Deps struct {
|
||
// Bus 是事件触发源。nil 则关闭事件驱动思考。
|
||
Bus bus.Bus
|
||
// Thinker 跑一轮无人值守的思考。必需。
|
||
Thinker Thinker
|
||
// State 提供用户当前状况，用于打扰决策。
|
||
// nil 则给出中性状态。
|
||
State StateProvider
|
||
// Recent 提供主会话最近的消息，好让活动门
|
||
// 感知主人是否刚聊过（最近对话冷却与
|
||
// 静默窗口抑制）。nil 则关闭活动感知。
|
||
Recent recentProvider
|
||
// Pusher 投递主动消息。为 nil 时由 Active（活跃客户端槽）
|
||
// 与 IM（IM 桥槽）加上 Gate 构建。
|
||
Pusher Pusher
|
||
// Active 是活跃 Web/原生客户端槽，推送首选。
|
||
Active ActiveClient
|
||
// IM 是 IM 桥槽，在没有活跃客户端时用作降级。
|
||
IM IMAdapter
|
||
// Exporter 把主动消息镜像到外部推送出口（主人的自托管
|
||
// 统一推送服务，其次是个推）。可选：nil 则
|
||
// 关闭该钩子。它绝不阻塞或破坏 Active/IM 投递。
|
||
Exporter PushExporter
|
||
// Gate 是主动推送限流器。nil 则由 Config 构建默认值。
|
||
Gate PushGate
|
||
// DB 是思考存储所用的 SQLite 句柄。必需。
|
||
DB *sql.DB
|
||
// SessionID 限定一次思考的作用域。为空则用 defaultSessionID。
|
||
SessionID string
|
||
// Now 是决策/记录时间戳的墙钟来源。nil 则用
|
||
// time.Now，从而让依赖时间的行为在测试中可注入。
|
||
Now func() time.Time
|
||
// Log 是基础日志器，来源标记为 "services/backthink"。
|
||
Log *slog.Logger
|
||
}
|
||
|
||
// Backthink 是自主后台思考服务。用 New 构建，用 Start 启动
|
||
// 心跳与订阅，用 Stop 停止它们。
|
||
type Backthink struct {
|
||
cfg Config
|
||
deps Deps
|
||
log *slog.Logger
|
||
store *Store
|
||
pusher Pusher
|
||
gate PushGate
|
||
now func() time.Time
|
||
sessionID string
|
||
|
||
mu sync.Mutex
|
||
running bool
|
||
cancel context.CancelFunc
|
||
subs []func()
|
||
wg sync.WaitGroup
|
||
|
||
// eventMu 守护同一事件去重器（event_dedup.go）。它与
|
||
// 生命周期锁分开，因此去重决策绝不会阻塞槽位换入。
|
||
eventMu sync.Mutex
|
||
lastEventKey string
|
||
}
|
||
|
||
// New 由 cfg 与 deps 构建 Backthink，套用默认值并校验
|
||
// 必需依赖。它先行创建 SQLite 思考存储（经 Store）
|
||
// 并解出推送路径（注入的 Pusher，或由
|
||
// Active/IM 与频率 Gate 构建的默认值）。DB 为 nil 时返回 ErrNilDB，
|
||
// Thinker 为 nil 时返回 ErrNoThinker。
|
||
func New(cfg Config, deps Deps) (*Backthink, error) {
|
||
if deps.Thinker == nil {
|
||
return nil, ErrNoThinker
|
||
}
|
||
if deps.DB == nil {
|
||
return nil, ErrNilDB
|
||
}
|
||
cfg = normalize(cfg)
|
||
|
||
log := logger.WithSource(baseLog(deps.Log), "services/backthink")
|
||
store, err := NewStore(context.Background(), deps.DB, log)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("backthink: new: %w", err)
|
||
}
|
||
|
||
// 提前解出频率门，好让槽位设置器能与它一起重建推送器。
|
||
gate := deps.Gate
|
||
if gate == nil {
|
||
gate = newRateGate(cfg.PushLimit, cfg.PushWindow, cfg.PushCooldown, cfg.KindWindow, log)
|
||
}
|
||
pusher := deps.Pusher
|
||
if pusher == nil {
|
||
pusher = newPusher(deps.Active, deps.IM, gate, log).withExporter(deps.Exporter)
|
||
}
|
||
|
||
now := deps.Now
|
||
if now == nil {
|
||
now = time.Now
|
||
}
|
||
sessionID := deps.SessionID
|
||
if sessionID == "" {
|
||
sessionID = defaultSessionID
|
||
}
|
||
|
||
return &Backthink{
|
||
cfg: cfg,
|
||
deps: deps,
|
||
log: log,
|
||
store: store,
|
||
pusher: pusher,
|
||
gate: gate,
|
||
now: now,
|
||
sessionID: sessionID,
|
||
}, nil
|
||
}
|
||
|
||
// Start 启动后台工作：心跳定时器，以及（配置了 Bus 时）
|
||
// 事件订阅。它是幂等的，调用两次安全；
|
||
// 第二次调用是空操作。传入的 ctx（可为 nil）是所有
|
||
// 内部工作的父级：取消它会像 Stop 一样停止服务。
|
||
func (b *Backthink) Start(ctx context.Context) {
|
||
if ctx == nil {
|
||
ctx = context.Background()
|
||
}
|
||
b.mu.Lock()
|
||
if b.running {
|
||
b.mu.Unlock()
|
||
return
|
||
}
|
||
b.running = true
|
||
cctx, cancel := context.WithCancel(ctx)
|
||
b.cancel = cancel
|
||
b.wg.Add(1)
|
||
go b.tickLoop(cctx)
|
||
b.mu.Unlock()
|
||
|
||
if b.deps.Bus != nil {
|
||
b.subscribe(cctx)
|
||
}
|
||
|
||
b.log.Info("backthink started",
|
||
"interval", b.cfg.Interval.String(),
|
||
"quiet_hours", quietHoursLabel(b.cfg.QuietStartHour, b.cfg.QuietEndHour),
|
||
"event_triggers", b.deps.Bus != nil,
|
||
"session_id", b.sessionID)
|
||
}
|
||
|
||
// Stop 停止心跳、退订所有总线订阅，然后等待
|
||
// 进行中的思考轮（受 ThinkTimeout 约束）排空。它是
|
||
// 幂等的；第二次调用立即返回 nil。若在排空完成前 ctx 已被
|
||
// 取消，则返回包装后的 context 错误，而不是泄漏
|
||
// goroutine。
|
||
func (b *Backthink) Stop(ctx context.Context) error {
|
||
if ctx == nil {
|
||
ctx = context.Background()
|
||
}
|
||
b.mu.Lock()
|
||
if !b.running {
|
||
b.mu.Unlock()
|
||
return nil
|
||
}
|
||
b.running = false
|
||
cancel := b.cancel
|
||
subs := b.subs
|
||
b.subs = nil
|
||
b.mu.Unlock()
|
||
|
||
cancel()
|
||
for _, unsub := range subs {
|
||
unsub()
|
||
}
|
||
|
||
done := make(chan struct{})
|
||
go func() {
|
||
b.wg.Wait()
|
||
close(done)
|
||
}()
|
||
select {
|
||
case <-done:
|
||
b.log.Info("backthink stopped")
|
||
return nil
|
||
case <-ctx.Done():
|
||
return fmt.Errorf("backthink: stop: %w", ctx.Err())
|
||
}
|
||
}
|
||
|
||
// List 返回 M9/M10 监控面板用的最近思考记录，
|
||
// 最新的在前。非正的 limit 选用存储默认值。
|
||
func (b *Backthink) List(ctx context.Context, limit int) ([]Record, error) {
|
||
return b.store.List(ctx, limit)
|
||
}
|
||
|
||
// triggers 返回本服务为事件驱动思考所监听的总线主题。
|
||
// 它是一个方法（而非包级变量），因此不会出现包级全局
|
||
// 切片，符合编码约束。
|
||
func (b *Backthink) triggers() []string {
|
||
return []string{iot.DeviceUpdateSubject, SubjectReminderTrigger, SubjectSystemEvent}
|
||
}
|
||
|
||
// subscribe 为每个触发主题注册一个处理器。每个订阅都
|
||
// 绑定到 ctx，因此取消它会移除注册；退订函数
|
||
// 被保留下来，好让 Stop 也能显式拆除它们。
|
||
func (b *Backthink) subscribe(ctx context.Context) {
|
||
for _, subj := range b.triggers() {
|
||
unsub, err := b.deps.Bus.Subscribe(ctx, subj, b.onEvent)
|
||
if err != nil {
|
||
b.log.Error("backthink subscribe failed", "subject", subj, "error", err)
|
||
continue
|
||
}
|
||
b.mu.Lock()
|
||
b.subs = append(b.subs, unsub)
|
||
b.mu.Unlock()
|
||
}
|
||
}
|
||
|
||
// onEvent 是事件驱动思考的总线处理器。它跑在总线自己的
|
||
// worker goroutine 上，所以不得阻塞在另一个订阅者上。它先
|
||
// 套用最小打扰门（design/12 §6 策略一）：低价值的系统
|
||
// 事件被降级为仅登记一条记录，而不唤醒一整轮
|
||
// 思考，因此平台/系统噪声无法淹没思考器。
|
||
func (b *Backthink) onEvent(ctx context.Context, e bus.Event) error {
|
||
tr := triggerForEvent(e)
|
||
if !b.shouldThink(tr) {
|
||
b.log.Debug("backthink system event downgraded, no thinking",
|
||
"source", tr.Source, "summary", tr.Summary)
|
||
b.record(Record{
|
||
Trigger: tr.Source,
|
||
At: b.now(),
|
||
Conclusion: "system event: low value, not worth waking a thinking pass",
|
||
ShouldPush: false,
|
||
PushResult: PushResult{Route: RouteNone, Detail: "downgraded"},
|
||
})
|
||
return nil
|
||
}
|
||
b.think(ctx, tr)
|
||
return nil
|
||
}
|
||
|
||
// shouldThink 为事件驱动思考把关（design/12 §6 策略一）。在公开
|
||
// 平台上，泛化的系统事件价值低，所以只有高重要度
|
||
// （与主人相关）的系统事件才唤醒一轮。IoT 设备更新与提醒
|
||
// 触发与主人相关 / 价值高，从不受门控，因此这些有产出的
|
||
// 触发保持完全可用。
|
||
func (b *Backthink) shouldThink(tr Trigger) bool {
|
||
if tr.Source == SubjectSystemEvent {
|
||
return tr.Import >= ImportanceHigh
|
||
}
|
||
return true
|
||
}
|
||
|
||
// tickLoop 是心跳 goroutine。按裁决 07.4.4（design/12 §1），它在进程
|
||
// 启动时立即思考一次——产出一句拟人化的启动问候——
|
||
// 随后稳定为每 tick 一轮思考；ctx 被取消时
|
||
// 退出。可接受的副作用是每次进程重启都会唤醒
|
||
// 一轮即时思考。
|
||
func (b *Backthink) tickLoop(ctx context.Context) {
|
||
defer b.wg.Done()
|
||
// 启动即思考：不等满一个心跳周期。已取消的上下文直接退出，不落一条失败的
|
||
// 启动记录。
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
default:
|
||
}
|
||
b.think(ctx, startupTrigger())
|
||
ticker := time.NewTicker(b.cfg.Interval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
b.think(ctx, heartbeatTrigger())
|
||
}
|
||
}
|
||
}
|
||
|
||
// think 跑一轮无人值守的思考：调用运行时、读取用户
|
||
// 状态、判断是否打扰、值得时推送并持久化记录。
|
||
// 只要注入的 Thinker、StateProvider 与 Pusher
|
||
// 并发安全，它就是并发安全的——按契约它们都是。
|
||
func (b *Backthink) think(parent context.Context, tr Trigger) {
|
||
at := b.now()
|
||
cctx, cancel := context.WithTimeout(parent, b.cfg.ThinkTimeout)
|
||
defer cancel()
|
||
|
||
b.log.Debug("backthink trigger", "source", tr.Source, "summary", tr.Summary)
|
||
|
||
conclusion, err := b.deps.Thinker.Think(cctx, b.sessionID, tr.Summary)
|
||
if err != nil {
|
||
b.log.Error("backthink think failed", "source", tr.Source, "error", err)
|
||
b.record(Record{
|
||
Trigger: tr.Source,
|
||
At: at,
|
||
Conclusion: fmt.Sprintf("thinking failed: %v", err),
|
||
ShouldPush: false,
|
||
PushResult: PushResult{Route: RouteNone, Detail: err.Error()},
|
||
})
|
||
return
|
||
}
|
||
|
||
// P2（design/20 §2）：她自己的结论决定她想不想说、
|
||
// 以及发到哪个渠道。parseProactive 稳健地恢复该决策，
|
||
// 并失败安全（格式错乱/含糊 → 不推送）。
|
||
message, channelHint, push := parseProactive(conclusion)
|
||
roundImport := b.escalatedImport(tr, conclusion)
|
||
|
||
state := b.currentState(cctx)
|
||
lastActivity, hasActivity := b.lastUserActivity(cctx)
|
||
// 裁决 07.4.2：importance 级联恢复为生产决策路径——她的结论回答「想不想说」，
|
||
// Decide 只判「此刻能不能说」：critical 安全事件穿透工作/忙碌/安静时段，安静
|
||
// 时段只挡常规推送；心跳类触发（含启动轮）在长期静默段不拦（裁决 07.4.3）。
|
||
// 频率门（PushGate）仍在推送器内部单独生效。
|
||
decision := b.Decide(DecisionInput{
|
||
Trigger: tr.Source,
|
||
Import: roundImport,
|
||
At: at,
|
||
UserState: state,
|
||
LastActivity: lastActivity,
|
||
HasActivity: hasActivity,
|
||
ConclusionPush: push,
|
||
Heartbeat: tr.Kind == TriggerHeartbeat,
|
||
})
|
||
shouldPush := decision.ShouldPush
|
||
|
||
recordConclusion := message
|
||
if recordConclusion == "" {
|
||
recordConclusion = conclusion
|
||
}
|
||
|
||
pushRes := PushResult{Route: RouteNone, Detail: decision.Reason}
|
||
if shouldPush {
|
||
// 同一事件去重：重复的相同事件在首次推送后被抑制
|
||
// （防刷），而真正的新事件仍会发出。心跳
|
||
// 从不去重（它们会重置该链），并保持受门控管辖。
|
||
if b.eventRedundant(tr) {
|
||
pushRes = PushResult{Route: RouteNone, Detail: "duplicate event push suppressed"}
|
||
} else {
|
||
pushRes = b.currentPusher().Push(cctx, PushRequest{
|
||
Message: b.buildProactiveMessage(message, channelHint),
|
||
ChannelHint: channelHint,
|
||
TriggerKind: tr.Kind,
|
||
})
|
||
}
|
||
}
|
||
|
||
b.record(Record{
|
||
Trigger: tr.Source,
|
||
At: at,
|
||
Conclusion: recordConclusion,
|
||
ShouldPush: shouldPush,
|
||
PushResult: pushRes,
|
||
})
|
||
|
||
if shouldPush {
|
||
b.log.Info("backthink proactive push", "source", tr.Source, "route", pushRes.Route, "pushed", pushRes.Pushed, "bypass", pushRes.Bypassed, "reason", pushRes.Detail)
|
||
} else {
|
||
b.log.Debug("backthink did not interrupt", "source", tr.Source, "reason", pushRes.Detail)
|
||
}
|
||
}
|
||
|
||
// escalatedImport 解出本轮打扰决策所用的重要度：触发自带的重要度，叠上她自己结论里
|
||
// 声明的紧急度（结构化标记 [紧急:...]，见 parseProactiveUrgency）。这让她在主动思考轮
|
||
// 里读完外围插件数据、调过工具后，能声明「我认为这条紧急、要突破安静时段发出去」
|
||
// （design/12 §6 安全域豁免 + design/20 §2 P2）。
|
||
//
|
||
// 上抬会记一条日志，使这次穿透可追溯（谁、从哪档抬到哪档）；声明只能上抬不能下压，
|
||
// 因此安全域事件既得的档位不会被一句 routine 抹掉。她自报不是无限通行证：频率门
|
||
// （PushGate）仍独立在推送器内部生效，心跳类轮次的推送照旧受限频约束。
|
||
func (b *Backthink) escalatedImport(tr Trigger, conclusion string) Importance {
|
||
declared := parseProactiveUrgency(conclusion)
|
||
escalated := escalateImport(tr.Import, declared)
|
||
if escalated != tr.Import {
|
||
b.log.Info("backthink declared urgency escalated round",
|
||
"source", tr.Source, "declared", declared.String(),
|
||
"from", tr.Import.String(), "to", escalated.String())
|
||
}
|
||
return escalated
|
||
}
|
||
|
||
// currentState 读取用户当前状况用于打扰决策，
|
||
// 提供者报错时退回中性状态（并记一条 warn 日志）。
|
||
func (b *Backthink) currentState(ctx context.Context) UserState {
|
||
if b.deps.State == nil {
|
||
return UserState{}
|
||
}
|
||
s, err := b.deps.State.Current(ctx)
|
||
if err != nil {
|
||
b.log.Warn("backthink state provider failed", "error", err)
|
||
return UserState{}
|
||
}
|
||
return s
|
||
}
|
||
|
||
// lastUserActivity 报告主人最近一条消息的墙钟时间
|
||
// （在主会话里），以及是否找到了任何用户消息。它是活动
|
||
// 门的活动信号（design/12 §6）：提供者为 nil、读取失败或
|
||
// 尾部为空都得到 has=false，因此活动门会落到当前
|
||
// 行为而不是靠猜。最近的一条用户消息是最近尾部里
|
||
// 最新的 RoleUser 条目（Recent 返回的是最旧在前）。
|
||
func (b *Backthink) lastUserActivity(ctx context.Context) (time.Time, bool) {
|
||
if b.deps.Recent == nil {
|
||
return time.Time{}, false
|
||
}
|
||
msgs, err := b.deps.Recent.Recent(ctx, history.MainSessionID, recentContextCount)
|
||
if err != nil {
|
||
return time.Time{}, false
|
||
}
|
||
for i := len(msgs) - 1; i >= 0; i-- {
|
||
if msgs[i].Role == history.RoleUser {
|
||
return time.UnixMilli(msgs[i].CreatedAt), true
|
||
}
|
||
}
|
||
return time.Time{}, false
|
||
}
|
||
|
||
// record 持久化一轮思考。它使用后台 context，因此即便
|
||
// 该次思考自身的 context 已被取消，审计行仍会写入。
|
||
func (b *Backthink) record(r Record) {
|
||
if _, err := b.store.Append(context.Background(), r); err != nil {
|
||
b.log.Error("backthink record failed", "error", err)
|
||
}
|
||
// 把持久化的这一轮暴露到 WS 流上，好让前端的 backthink
|
||
// 消费者（或其默认兜底）无需轮询存储就能观测到它。
|
||
if b.deps.Bus != nil {
|
||
if err := b.deps.Bus.Publish(context.Background(), "ws.backthink", r); err != nil {
|
||
b.log.Warn("backthink bus publish failed", "error", err)
|
||
}
|
||
}
|
||
}
|