Cyrene_v2/internal/services/backthink/backthink.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)

	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:         tr.Import,
		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)
	}
}

// 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)
		}
	}
}