feat(backthink): 主动推送 sink 打通——hub.SendProactive + bridge.SendProactive + wire 接线
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

backthink 决策+路由(P2)就绪但无投递 sink(ActiveClient/IMAdapter 无实现)。补齐:
- gateway/proactive.go: hub.SendProactive 向 owner/admin 连接广播 proactive_message WS事件,
  内容脱敏(去RawData/media/token),返回接收数,count=0→IM兜底;Server.SendProactive 委托
- channelbridge/proactive.go: Bridge.SendProactive 解析 owner 账户(msg.Platform→匹配白名单账号,
  否则首个已配置主人频道),ChannelType=Direct+ChannelID/SenderID=owner id,走 adapter.SendProactive
- backthink/sinks.go: SetActiveSink/SetIMSink 晚注入(New后重建pusher,currentPusher 并发安全)
- wire.go: 在 gw/bridge 建好后 bt.SetActiveSink(gw)/SetIMSink(bridge)
- 投递链: think→parseProactive→ShouldPush→currentPusher.Push→gate→active(active_client)/IM(im_bridge)
  →record(route+送达数)
- 测试: sinks/gateway proactive/channelbridge proactive; 后端 build/vet/test 全绿
- 注: -race 本机 TSan 不可用(环境); 多平台IM按msg.Platform匹配 owner 账号(单sink诚实);
  weixin 24h窗口按 ErrProactiveWindow 记录为 im_bridge 错误不静默
This commit is contained in:
Aska committed 2026-09-07 19:43:48 +08:00
1 parent 2283383b9e
commit 6a2b8b9f12
9 files changed
+615 -6

No files matched your search

+8 -5
View File
@@ -157,6 +157,7 @@ type Backthink struct {
log *slog.Logger
store *Store
pusher Pusher
gate PushGate
now func() time.Time
sessionID string
@@ -187,12 +188,13 @@ func New(cfg Config, deps Deps) (*Backthink, error) {
return nil, fmt.Errorf("backthink: new: %w", err)
}
// Resolve the gate up front so the sink setters can rebuild the pusher with it.
gate := deps.Gate
if gate == nil {
gate = newRateGate(cfg.MaxPushWindow, cfg.PushWindow, log)
}
pusher := deps.Pusher
if pusher == nil {
gate := deps.Gate
if gate == nil {
gate = newRateGate(cfg.MaxPushWindow, cfg.PushWindow, log)
}
pusher = newPusher(deps.Active, deps.IM, gate, log)
}
@@ -211,6 +213,7 @@ func New(cfg Config, deps Deps) (*Backthink, error) {
log: log,
store: store,
pusher: pusher,
gate: gate,
now: now,
sessionID: sessionID,
}, nil
@@ -407,7 +410,7 @@ func (b *Backthink) think(parent context.Context, tr Trigger) {
pushRes := PushResult{Route: RouteNone, Detail: b.interruptReason(push, at, state)}
if shouldPush {
pushRes = b.pusher.Push(cctx, PushRequest{
pushRes = b.currentPusher().Push(cctx, PushRequest{
Message: b.buildProactiveMessage(message, channelHint),
ChannelHint: channelHint,
})
+40
View File
@@ -0,0 +1,40 @@
package backthink
// The production composition root builds the gateway hub and the channel bridge
// *after* backthink.New (the hub needs the loop, the bridge needs the loop and
// owner profile, both of which are assembled later), so the ActiveClient and
// IMAdapter sinks are unavailable at construction time. These setters inject them
// once the gateway / bridge exist and rebuild the pusher, keeping the resolved
// rate gate so a later push routes correctly.
// SetActiveSink wires the production active web/native client sink (the gateway
// hub) and rebuilds the pusher. Call it after the gateway exists, before Start.
func (b *Backthink) SetActiveSink(active ActiveClient) {
b.mu.Lock()
defer b.mu.Unlock()
b.deps.Active = active
b.rebuildPusherLocked()
}
// SetIMSink wires the production IM bridge sink (the channel bridge) and rebuilds
// the pusher. Call it after the channel bridge exists, before Start.
func (b *Backthink) SetIMSink(im IMAdapter) {
b.mu.Lock()
defer b.mu.Unlock()
b.deps.IM = im
b.rebuildPusherLocked()
}
// rebuildPusherLocked replaces the pusher from the current sinks and gate. The
// caller must hold b.mu.
func (b *Backthink) rebuildPusherLocked() {
b.pusher = newPusher(b.deps.Active, b.deps.IM, b.gate, b.log)
}
// currentPusher returns the active pusher under the lifecycle lock, so a sink
// swap never races with a concurrent thinking pass that is about to push.
func (b *Backthink) currentPusher() Pusher {
b.mu.Lock()
defer b.mu.Unlock()
return b.pusher
}
+102
View File
@@ -0,0 +1,102 @@
package backthink
import (
"context"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)
// TestBackthink_SetActiveSink_RebuildsPusher verifies the sink setters rebuild the
// pusher so a later push routes to the injected active client instead of the
// default nil-sink RouteNone pusher built at New.
func TestBackthink_SetActiveSink_RebuildsPusher(t *testing.T) {
db := newTestDB(t)
bt, err := New(Config{}, Deps{
Thinker: &mockThinker{text: "[我决定:push]\n[渠道:web]\nhi"},
State: &mockState{},
Gate: &mockGate{allow: true},
DB: db,
})
if err != nil {
t.Fatalf("new: %v", err)
}
active := &mockActive{n: 2}
im := &mockIMAdapter{}
bt.SetActiveSink(active)
bt.SetIMSink(im)
res := bt.pusher.Push(context.Background(), PushRequest{
Message: adapter.UnifiedMessage{Content: "hi"},
ChannelHint: "web",
})
if !res.Pushed || res.Route != RouteActiveClient {
t.Fatalf("push = %+v, want active_client pushed", res)
}
if active.calls != 1 {
t.Fatalf("active calls = %d, want 1", active.calls)
}
if im.count() != 0 {
t.Fatalf("im calls = %d, want 0 (active preferred)", im.count())
}
}
// TestBackthink_SetIMSink_Fallback verifies a wired IM sink receives the push when
// no active client is online (so the owner is still reached).
func TestBackthink_SetIMSink_Fallback(t *testing.T) {
db := newTestDB(t)
bt, err := New(Config{}, Deps{
Thinker: &mockThinker{text: "[我决定:push]\nhi"},
State: &mockState{},
Gate: &mockGate{allow: true},
DB: db,
})
if err != nil {
t.Fatalf("new: %v", err)
}
active := &mockActive{n: 0} // no active client online
im := &mockIMAdapter{}
bt.SetActiveSink(active)
bt.SetIMSink(im)
res := bt.pusher.Push(context.Background(), PushRequest{
Message: adapter.UnifiedMessage{Content: "hi"},
})
if !res.Pushed || res.Route != RouteIMBridge {
t.Fatalf("push = %+v, want im_bridge pushed", res)
}
if im.count() != 1 {
t.Fatalf("im calls = %d, want 1", im.count())
}
}
// TestBackthink_SinkSetters_ConcurrentSafe verifies the setter and the read of the
// current pusher do not race under the lifecycle lock, so an in-flight pass never
// touches a half-rebuilt pusher.
func TestBackthink_SinkSetters_ConcurrentSafe(t *testing.T) {
db := newTestDB(t)
bt, err := New(Config{}, Deps{
Thinker: &mockThinker{},
State: &mockState{},
Gate: &mockGate{allow: true},
DB: db,
})
if err != nil {
t.Fatalf("new: %v", err)
}
done := make(chan struct{})
go func() {
defer close(done)
for i := 0; i < 200; i++ {
bt.SetActiveSink(&mockActive{n: 1})
bt.SetIMSink(&mockIMAdapter{})
}
}()
for i := 0; i < 200; i++ {
_ = bt.currentPusher().Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{Content: "x"}})
}
<-done
}