Cyrene_v2/internal/services/backthink/backthink.go

// backthink 包实现 M8 自主后台思考服务
// （DevDocs/design/12-后台思考与主动.md §0-§3）。
//
// 它在两类触发下运行无人值守的思考轮：
//
//   - 周期性心跳（config.Background.IntervalSec），请求注入的运行时
//     反思，并允许调用工具（与主人对话**同一套**工具面：同一个
//     loop / dispatcher / 能力注册表，见 thinker.go 的 LoopThinker），
//     但默认不得发出 chat 风格的消息。心跳周期**按当前时刻选**：
//     落在安静窗口内降频到 config.Background.QuietIntervalSec，
//     窗外回落 IntervalSec（design/12 §N A1/A3，见 heartbeat.go）。
//     降频**只作用于周期性心跳**，事件驱动通路照旧立刻触发（§N A2）。
//   - 进程内总线上的异步事件（iot.device_update、
//     reminder.trigger、system.event），它们唤醒一轮思考。
//
// 每轮产出一段结论，其结构化尾段决定她
// 到底想不想说话（design/20 §2 P2），随后咨询 Decide——
// 作为生产决策路径恢复的重要度级联——来判断此刻是否
// 合适（事件重要度 / 时段 / 用户状态，如心率、
// 睡眠、工作、忙碌）。用户状态在**思考之前**读一次，同一份快照同时喂给
// 思考轮（经 WithRoundState 渲染成上下文块）与这里的决策（X1 缺口 d）。
// 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/logger"
)

// 本包返回的哨兵错误。用 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。实现可以
// 调用**与主人对话同一套**已注册工具（工具面 = 运行时的能力注册表，
// 不是本包枚举出来的清单），且必须并发安全。
type Thinker interface {
	// Think 为给定的会话/触发调用一轮自主思考，
	// 并返回结论文本。
	//
	// 本轮的用户状态快照经 ctx 挂载（WithRoundState / RoundStateFrom）：
	// 实现**可选**读取它并把状态写进自己的上下文（LoopThinker 就是这么做的），
	// 不读的实现（旧实现/测试替身）行为逐字不变。
	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
	// Audit 是自排审计出口（design/12 §N C.5）。它实现 Auditor 这个三字段窄接口，
	// 生产接线注入网关（后端组装晚于 backthink，故也提供 SetAuditSink 晚绑定）。
	// nil 则审计静默跳过：审计是留痕，不是门。
	Audit Auditor
	// 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

	// selfWake 是她自排一轮思考的成本闸与待跑槽（selfwake.go，design/12 §N C）。
	// 它在 New 里按**已 normalize** 的配置构建（G3），Start 时由一个 worker 消费。
	selfWake *selfWakeGate
}

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

	now := deps.Now
	if now == nil {
		now = time.Now
	}
	sessionID := deps.SessionID
	if sessionID == "" {
		sessionID = defaultSessionID
	}

	b := &Backthink{
		cfg:       cfg,
		deps:      deps,
		log:       log,
		store:     store,
		pusher:    pusher,
		gate:      gate,
		now:       now,
		sessionID: sessionID,
		// 闸只认**已 normalize** 的旋钮：未回落默认的 ≤0 会被滑窗读成「无规则」
		// （恒放行且不记录），那就把「≤0 回落默认」静默变成「无上限」（G3）。
		selfWake: newSelfWakeGate(cfg.SelfWakeLimit, cfg.SelfWakeWindow),
	}
	// 未注入自定义推送器时，默认推送器在结构体构造**之后**组装：它要挂上
	// b.auditEvent（读取当前审计出口的回调），而该回调需要 b 本身。
	if pusher == nil {
		b.rebuildPusher()
	}
	return b, 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)
	// 自排 worker（design/12 §N C.1）：与心跳并列的第二个常驻 goroutine。
	// Stop 的 cancel()+wg.Wait() 自动收走它，不引入新的生命周期状态。
	b.wg.Add(1)
	go b.selfWakeLoop(cctx)
	b.mu.Unlock()

	if b.deps.Bus != nil {
		b.subscribe(cctx)
	}

	b.log.Info("backthink started",
		"interval", b.cfg.Interval.String(),
		"quiet_interval", b.cfg.QuietInterval.String(),
		"quiet_interval_enabled", b.quietIntervalEnabled(),
		"quiet_hours", quietHoursLabel(b.cfg.QuietStartHour, b.cfg.QuietEndHour),
		"event_triggers", b.deps.Bus != nil,
		// 自排的成本闸口径（design/12 §N C.5 的观测要求）：与 quiet_interval 同族，
		// 缺了它主人无法从日志确认「她一小时最多能叫醒自己几次」。
		"self_wake_limit", b.cfg.SelfWakeLimit,
		"self_wake_window", b.cfg.SelfWakeWindow.String(),
		"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 返回本服务为事件驱动思考所监听的总线主题。它是
// `EventSubjects`（trigger_domain.go 的唯一真源）的转述 —— 订阅表与 admin
// 的 `?trigger=` 筛选白名单共用同一份，新增主题时两边同时生效。
func (b *Backthink) triggers() []string {
	return EventSubjects()
}

// 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"},
			// Urgency 留空（未知）：这一行**不是一轮思考**（结论即写明未唤醒思考），
			// 没有「该轮生效的档位」可落；宁可如实为未知，也不拿事件的档位冒充——
			// 与 P3+P4 的 fail-safe 同口径。（降级只发生在 shouldThink 判到
			// Import < High 的系统事件上，故它的档位恒为 routine，面板若要显示，
			// 由既有下限推导即可。）
		})
		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
}

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

	// X1（缺口 d）：用户状态在**思考之前**读一次，随后同时交给思考轮（经 ctx）与打扰决策。
	// 此前它只在思考返回后读取、只喂 Decide，于是她写结论时不知道主人在睡/在忙，语气与时机
	// 判断和真正生效的打扰门脱节；现在两侧是**同一份**快照（读取失败时 Known=false，
	// 思考轮不注入状态块、决策按中性状态走，行为与接线前一致）。
	state, stateKnown := b.currentState(cctx)
	thinkCtx := WithRoundState(cctx, RoundState{State: state, Known: stateKnown})
	conclusion, err := b.deps.Thinker.Think(thinkCtx, 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()},
			// Urgency 留空（未知）：这一轮没跑完，**没有任何档位生效**；
			// 拿触发的档位冒充「该轮档位」就是给一行失败记录安它没有的事实。
		})
		return
	}

	// P2（design/20 §2）：她自己的结论决定她想不想说、
	// 以及发到哪个渠道。parseProactive 稳健地恢复该决策，
	// 并失败安全（格式错乱/含糊 → 不推送）。
	message, channelHint, push := parseProactive(conclusion)
	roundImport := b.escalatedImport(tr, conclusion)

	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,
		Kind:           tr.Kind,
	})
	shouldPush := decision.ShouldPush

	recordConclusion := message
	if recordConclusion == "" {
		recordConclusion = conclusion
	}

	pushRes := b.deliver(cctx, tr, message, channelHint, shouldPush, decision.Reason)

	b.record(Record{
		Trigger:    tr.Source,
		At:         at,
		Conclusion: recordConclusion,
		ShouldPush: shouldPush,
		PushResult: pushRes,
		// X2a：把**该轮实际生效**的档位落进存档（面板此前只能给「下限」）。
		// 取数点就是上面决策真正吃进去的 roundImport —— 她自报的紧急度经
		// escalatedImport 上抬后已折叠在内，因此记下的是生效值；她自报得不比
		// 事件自带档位低时，生效值就是事件那一档（声明只能上抬，见 escalateImport）。
		Urgency: urgencyForImportance(roundImport).String(),
	})

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

// deliver 投递一轮结论（已由 Decide 判完能不能说）：值得打断时先过同一事件去重，
// 再交给推送器。它不做任何决策，只把「决定 + 结论」变成一条 PushResult —— 抽出来是
// 为了让 think 保持在项目规范的可读长度内（函数 ≤80 行）。
//
// 同一事件去重：重复的相同事件在首次推送后被抑制（防刷），而真正的新事件仍会发出。
// 心跳从不去重（它们会重置该链），并保持受门控管辖。
func (b *Backthink) deliver(ctx context.Context, tr Trigger, message, channelHint string, shouldPush bool, reason string) PushResult {
	if !shouldPush {
		return PushResult{Route: RouteNone, Detail: reason}
	}
	if b.eventRedundant(tr) {
		return PushResult{Route: RouteNone, Detail: "duplicate event push suppressed"}
	}
	return b.currentPusher().Push(ctx, PushRequest{
		Message:     b.buildProactiveMessage(message, channelHint),
		ChannelHint: channelHint,
		TriggerKind: tr.Kind,
	})
}

// 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 与 lastUserActivity 在本包的状态面（state.go）里：
// 前者现在同时回答「读到了没」（X1 缺口 d），后者是活动门的活动信号。

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