fix(backthink): 主动推送降频(15min心跳+60min≤1次+推后30min冷却+刚聊过10min冷避) + 情绪约束(分类限频/活跃感知) + 剥离'我决定:push'内部标记
This commit is contained in:
1 parent
03d7050e1e
commit
12031bd13c
14 files changed
+662
-149
No files matched your search
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
@@ -82,11 +83,15 @@ type IMAdapter interface {
|
||||
}
|
||||
|
||||
// PushGate decides whether a proactive push may proceed right now, applying the
|
||||
// rate limits of design/11 §6. It is a seam over channel/session's
|
||||
// ProactivePolicy/RateLimiter so the service can be tested with a mock.
|
||||
// rate limits of design/11 §6 plus a per-flavor limit and a post-push cooldown. It
|
||||
// is a seam over channel/session's ProactivePolicy/RateLimiter so the service can
|
||||
// be tested with a mock.
|
||||
type PushGate interface {
|
||||
// Allow reports whether a proactive push may be sent now.
|
||||
Allow(ctx context.Context) bool
|
||||
// Allow reports whether a proactive push of the given kind may be sent now.
|
||||
Allow(ctx context.Context, kind PushKind) bool
|
||||
// RecordPush notes a successful delivery so the cooldown/per-kind windows
|
||||
// start from a real successful push, not a merely-attempted one.
|
||||
RecordPush(kind PushKind)
|
||||
}
|
||||
|
||||
// PushRequest carries one proactive push attempt: the outbound message plus the
|
||||
@@ -135,8 +140,9 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return PushResult{Pushed: false, Route: RouteNone, Err: err.Error()}
|
||||
}
|
||||
if p.gate != nil && !p.gate.Allow(ctx) {
|
||||
p.log.Warn("backthink push rate limited")
|
||||
kind := classifyPush(req.Message.Content)
|
||||
if p.gate != nil && !p.gate.Allow(ctx, kind) {
|
||||
p.log.Warn("backthink push rate limited", "kind", kind.String())
|
||||
return PushResult{Pushed: false, Route: RouteRateLimited, Detail: "proactive push rate limited"}
|
||||
}
|
||||
|
||||
@@ -165,6 +171,11 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult {
|
||||
res = p.pushDefault(ctx, msg)
|
||||
}
|
||||
}
|
||||
// Only a real delivery advances the cooldown/per-kind windows, so an
|
||||
// attempted-but-failed push does not suppress a later genuine one.
|
||||
if res.Pushed && p.gate != nil {
|
||||
p.gate.RecordPush(kind)
|
||||
}
|
||||
return annotateHint(res, req.ChannelHint)
|
||||
}
|
||||
|
||||
@@ -249,31 +260,74 @@ func annotateHint(res PushResult, hint string) PushResult {
|
||||
}
|
||||
|
||||
// rateGate is the production PushGate: a sliding-window counter over
|
||||
// session.RateLimiter (design/11 §6 anti-spam rule).
|
||||
// session.RateLimiter (design/11 §6 anti-spam rule) combined with a post-push
|
||||
// cooldown and a per-flavor (问候/提醒/分享/关心) window so the owner is never
|
||||
// spammed with the same style (design/12 §6 策略二).
|
||||
type rateGate struct {
|
||||
log *slog.Logger
|
||||
limiter *session.RateLimiter
|
||||
log *slog.Logger
|
||||
limiter *session.RateLimiter
|
||||
cooldown time.Duration
|
||||
kindWindow time.Duration
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
lastPushTime time.Time
|
||||
lastKind map[PushKind]time.Time
|
||||
}
|
||||
|
||||
// newRateGate builds a gate that allows at most limit pushes within window. A
|
||||
// non-positive limit or window yields an always-allow gate (no rule configured).
|
||||
func newRateGate(limit int, window time.Duration, log *slog.Logger) *rateGate {
|
||||
return &rateGate{log: log, limiter: session.NewRateLimiter(limit, window)}
|
||||
// newRateGate builds a gate that allows at most limit pushes within window, keeps
|
||||
// pushes cooldown apart after a success, and limits each PushKind to one push per
|
||||
// kindWindow. A non-positive limit or window yields an always-allow limiter (no
|
||||
// rate rule); a non-positive cooldown or kindWindow disables that specific guard.
|
||||
func newRateGate(limit int, window, cooldown, kindWindow time.Duration, log *slog.Logger) *rateGate {
|
||||
return &rateGate{
|
||||
log: log,
|
||||
limiter: session.NewRateLimiter(limit, window),
|
||||
cooldown: cooldown,
|
||||
kindWindow: kindWindow,
|
||||
now: time.Now,
|
||||
lastKind: make(map[PushKind]time.Time),
|
||||
}
|
||||
}
|
||||
|
||||
// Allow implements PushGate. A canceled context denies (fail closed).
|
||||
func (g *rateGate) Allow(ctx context.Context) bool {
|
||||
// Allow implements PushGate. A canceled context denies (fail closed). It refuses
|
||||
// when the last successful push is inside the cooldown, or when the same flavor
|
||||
// was delivered within kindWindow, before consulting the sliding-window limiter.
|
||||
func (g *rateGate) Allow(ctx context.Context, kind PushKind) bool {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return false
|
||||
}
|
||||
now := g.now()
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if !g.lastPushTime.IsZero() && now.Sub(g.lastPushTime) < g.cooldown {
|
||||
return false
|
||||
}
|
||||
if lt, ok := g.lastKind[kind]; ok && now.Sub(lt) < g.kindWindow {
|
||||
return false
|
||||
}
|
||||
return g.limiter.Allow()
|
||||
}
|
||||
|
||||
// RecordPush implements PushGate: it stamps the last successful push time and the
|
||||
// per-flavor last time so the cooldown and kind window start from a real delivery.
|
||||
func (g *rateGate) RecordPush(kind PushKind) {
|
||||
now := g.now()
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
g.lastPushTime = now
|
||||
g.lastKind[kind] = now
|
||||
}
|
||||
|
||||
// buildProactiveMessage assembles the normalized outbound message carrying a
|
||||
// think conclusion's clean body. It tags the platform with the requested channel
|
||||
// when the conclusion names a concrete platform, otherwise the default backthink
|
||||
// platform. The concrete active-client and IM adapters apply their own
|
||||
// routing/targeting to the owner (red line: owner only).
|
||||
//
|
||||
// The body is run through stripInternalMarkers as a final guard so a structured
|
||||
// decision/channel marker that survived parsing can never reach the user-facing
|
||||
// message (only our logs/records may carry it).
|
||||
func (b *Backthink) buildProactiveMessage(message, channelHint string) adapter.UnifiedMessage {
|
||||
platform := defaultProactivePlatform
|
||||
if h := normalizeChannelHint(channelHint); isSpecificChannel(h) {
|
||||
@@ -282,7 +336,7 @@ func (b *Backthink) buildProactiveMessage(message, channelHint string) adapter.U
|
||||
return adapter.UnifiedMessage{
|
||||
Platform: platform,
|
||||
ChannelType: adapter.ChannelDirect,
|
||||
Content: message,
|
||||
Content: stripInternalMarkers(message),
|
||||
ContentType: adapter.ContentText,
|
||||
Sender: adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
|
||||
Timestamp: b.now(),
|
||||
|
||||
Reference in new issue
Block a user