AskaEth
94da6bcd6d
- 追加 TriggerSelfWake（不动既有编号）+ SelfWakeRequest（走既有 urgency 声明通道，不新造第二条） - escalateImport 折算收成唯一 foldDeclared（事件轮与自排轮共用、只上抬取较高档） - DecisionInput.Heartbeat bool → Kind TriggerKind；心跳仍免检、**自排不享心跳豁免**（要求声明 ≥ warning） - push 与档位分离：push.go 零改动即自动正确（eventDriven 对 self_wake 为 false ⇒ 照旧过 PushGate） - 4 个变异自检（折算空操作/删自排分支/常量传心跳/非事件一律豁免） ⚠️ 本片只是判定侧：WakeNow/worker/闸属 C1-2，工具面与提示词属 C1-3 ⇒ **不得写『C 已完成』** ⚠️ 遗留 G-A（建议登记进 design/12 C.4）：TriggerKind 零值＝TriggerHeartbeat ⇒ 漏填＝最宽豁免（fail-open）； 本片已在字段注释与单测里显式化，但规格侧应补一句或引入 TriggerUnknown
470 lines
18 KiB
Go
470 lines
18 KiB
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/iot"
|
||
"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
|
||
// 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_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,
|
||
"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"},
|
||
// 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)
|
||
}
|
||
}
|
||
}
|