From 8cde0162e6633d3e3679bc2bcdf1924a9f825a8e Mon Sep 17 00:00:00 2001 From: Aska Date: Mon, 7 Sep 2026 12:18:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=85=AC=E5=85=B1=E5=B9=B3=E5=8F=B0?= =?UTF-8?q?=E6=9C=80=E5=B0=8F=E5=8C=96=E6=89=93=E6=89=B0(design/12=20?= =?UTF-8?q?=C2=A76)=E2=80=94=E2=80=945=E6=9D=A1=E7=AD=96=E7=95=A5:=20?= =?UTF-8?q?=E5=88=86=E7=BA=A7+=E8=BD=BB=E6=80=9D=E8=80=83+=E9=99=90?= =?UTF-8?q?=E6=B5=81+=E5=AE=A1=E6=9F=A5=E5=89=94=E5=99=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 降噪A(策略一/二/五): - channelbridge/light.go: 非owner分级门——小事件当系统消息登记(不进loop)、非owner轻响应(她判断回/不回,规则优先mention/关键词+轻LLM兜底fail-closed静默) - backthink 触发降噪: SubjectSystemEvent 低价值不唤醒thinking(仅高价值/owner才think), iot/reminder不受影响 - quiet hours: 安静时段非owner无信号静默,被@才回(非硬性静默); config QUIET_START/END_HOUR 单源共用 限流B(策略三/四): - adapter/flowgate.go: per-adapter 滑窗限流器+总上限+审查剔噪闸门(Admit两连:速率+上限/审查), owner私聊绕过 - channelbridge/flowgate.go: onMessage入站最前挂闸, 超限/被剔不占worker槽; 审查模型NewLLMReviewer复用model.Adapter(仅>80%阈值触发+跨分钟节流,失败回退FIFO,绝不判身份) - config: ChannelAccount 加 GroupMsgPerMin(默认6)/MsgPerMin/MaxMessages(默认50), CHANNELS_JSON每账号, 向后兼容 - 被剔消息走OnRejected审计不静默丢 main wire接线: 加modelAdapter字段+Review注入(reviewFor nil-safe,无LLM回退FIFO) 全仓 go build/vet/test 绿(53包)+交叉编译, 零新依赖, 前端无涉 --- .env.example | 9 + cmd/cyrene/channels.go | 14 +- cmd/cyrene/wire.go | 29 +- internal/channel/adapter/flowgate.go | 309 ++++++++++++++++++ internal/channel/adapter/flowgate_test.go | 253 ++++++++++++++ internal/config/channels_test.go | 33 ++ internal/config/config.go | 20 +- internal/services/backthink/backthink.go | 42 ++- internal/services/backthink/backthink_test.go | 57 ++++ internal/services/backthink/trigger.go | 20 +- .../services/channelbridge/channelbridge.go | 105 +++++- .../channelbridge/channelbridge_test.go | 22 +- internal/services/channelbridge/flowgate.go | 178 ++++++++++ .../services/channelbridge/flowgate_test.go | 212 ++++++++++++ internal/services/channelbridge/light.go | 190 +++++++++++ internal/services/channelbridge/light_test.go | 285 ++++++++++++++++ 16 files changed, 1750 insertions(+), 28 deletions(-) create mode 100644 internal/channel/adapter/flowgate.go create mode 100644 internal/channel/adapter/flowgate_test.go create mode 100644 internal/services/channelbridge/flowgate.go create mode 100644 internal/services/channelbridge/flowgate_test.go create mode 100644 internal/services/channelbridge/light.go create mode 100644 internal/services/channelbridge/light_test.go diff --git a/.env.example b/.env.example index 1986300..1c8cf12 100644 --- a/.env.example +++ b/.env.example @@ -69,6 +69,10 @@ STT_PROVIDER=dashscope ENABLE_BACKGROUND_THINKING=true THINK_INTERVAL_SEC=300 THINK_SILENCE_TIMEOUT_SEC=120 +# 安静时段（本地时钟小时，默认 23:00-07:00；两端相等且非 0 则关闭）。 +# 同时用于 backthink 后台思考与 channelbridge 公共平台安静时段轻思考（design/12 §6 策略五）。 +QUIET_START_HOUR=23 +QUIET_END_HOUR=7 # ---- WebSocket ---- WS_MAX_CONNECTIONS=1000 @@ -105,4 +109,9 @@ ALLOWED_ORIGINS=http://localhost:5173,http://localhost:3000 # 每个 account 字段：id(平台内唯一)、config(平台特有凭据 map)、 # allowGroups/denyGroups(群黑白名单，allowGroups 非空则成为白名单)、 # allowUsers/denyUsers(私聊用户黑白名单)。未配置时按单账号 env 路径处理。 +# 群/总限流与总上限（可选，design/12 §6 策略三/四，公共平台防刷屏/最小化打扰）： +# groupMsgPerMin 群聊每分钟上限（0=用默认6，-1=不限，正整数=该值） +# msgPerMin 非主人消息每分钟总上限（>0 生效，0/-1=不限） +# maxMessages 待处理上限，接近上限时由审查模型剔除低价值消息（0=用默认50，-1=不限，正整数=该值） +# 旧账号不配这些字段默认走「保护性默认/不限」，保持向后兼容。 # CHANNELS_JSON= diff --git a/cmd/cyrene/channels.go b/cmd/cyrene/channels.go index 97a9683..d599f27 100644 --- a/cmd/cyrene/channels.go +++ b/cmd/cyrene/channels.go @@ -110,13 +110,19 @@ type namedAdapter struct { denyGroups []string allowUsers []string denyUsers []string + // flow limits (design/12 §6 策略三/四). They are always surfaced through + // Config() so the bridge can build a per-account ingress gate; the bridge maps + // the zero default to the protective values and a negative value to "off". + groupMsgPerMin int + msgPerMin int + maxMessages int } func (n *namedAdapter) Name() string { return n.name } func (n *namedAdapter) Config() map[string]any { cfg := n.PlatformAdapter.Config() - out := make(map[string]any, len(cfg)+4) + out := make(map[string]any, len(cfg)+7) for k, v := range cfg { out[k] = v } @@ -124,6 +130,9 @@ func (n *namedAdapter) Config() map[string]any { out["denyGroups"] = n.denyGroups out["allowUsers"] = n.allowUsers out["denyUsers"] = n.denyUsers + out["groupMsgPerMin"] = n.groupMsgPerMin + out["msgPerMin"] = n.msgPerMin + out["maxMessages"] = n.maxMessages return out } @@ -182,6 +191,9 @@ func buildAccountAdapter(platform string, acct config.ChannelAccount, log *slog. denyGroups: acct.DenyGroups, allowUsers: acct.AllowUsers, denyUsers: acct.DenyUsers, + groupMsgPerMin: acct.GroupMsgPerMin, + msgPerMin: acct.MsgPerMin, + maxMessages: acct.MaxMessages, }, nil } diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 46cf8fb..9f6d34b 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -76,6 +76,7 @@ type app struct { log *slog.Logger db *sql.DB bus bus.Bus + modelAdapter model.Adapter personas *personareg.Registry caps *registry.Registry dispatcher *tool.Dispatcher @@ -263,6 +264,9 @@ func (a *app) wireRuntime() error { a.log.Warn("llm adapter unavailable", "error", aerr) adapter = nil } + // Keep the model adapter for cross-function use (e.g. the channel-bridge + // review model that prunes low-value messages, design/12 §6 策略四). + a.modelAdapter = adapter router := model.NewRouter(model.DefaultModelForProvider(a.cfg.LLM), model.WithVisionModel(a.cfg.LLM.VisionModel)) @@ -484,11 +488,16 @@ func (a *app) wireChannels() error { } bridge, berr := channelbridge.New(channelbridge.Options{ - Log: logger.WithSource(a.log, "services/channelbridge"), - Loop: a.loop, - Sessions: a.sessions, - Owner: a.owner, - History: a.history, + Log: logger.WithSource(a.log, "services/channelbridge"), + Loop: a.loop, + Sessions: a.sessions, + Owner: a.owner, + History: a.history, + QuietStartHour: a.cfg.Background.QuietStartHour, + QuietEndHour: a.cfg.Background.QuietEndHour, + // Optional review model prunes low-value queued messages near the per-adapter + // cap (design/12 §6 策略四). Nil adapter → the gate falls back to FIFO evict. + Review: reviewFor(a.modelAdapter, logger.WithSource(a.log, "services/channelbridge/review")), }) if berr != nil { return fmt.Errorf("channel bridge: %w", berr) @@ -652,3 +661,13 @@ func resolveSQLitePath(cfg *config.Config, dataDir string) string { } return p } + +// reviewFor builds the channel-bridge review model (design/12 §6 策略四), or nil +// when no model adapter is available so the gate falls back to FIFO eviction — the +// reviewer is never invoked with a nil adapter (which would panic on .Chat). +func reviewFor(llm model.Adapter, log *slog.Logger) adapter.Reviewer { + if llm == nil { + return nil + } + return channelbridge.NewLLMReviewer(llm, log) +} diff --git a/internal/channel/adapter/flowgate.go b/internal/channel/adapter/flowgate.go new file mode 100644 index 0000000..60afee0 --- /dev/null +++ b/internal/channel/adapter/flowgate.go @@ -0,0 +1,309 @@ +package adapter + +import ( + "context" + "sync" + "time" +) + +// Reviewer decides which buffered messages are worth keeping. It receives the +// current pending set (in arrival order) and returns the indices it wants to keep. +// Messages at a returned index survive; all others are pruned and audited. +// +// It is a pure noise/importance filter (design/12 §6 策略四). Implementations must +// never decide identity: who the sender is was already resolved structurally by the +// channel layer. A nil Reviewer makes the gate fall back to a plain FIFO cap. +type Reviewer func(ctx context.Context, msgs []UnifiedMessage) (keep []int, err error) + +// GateVerdict reports how the gate classified one inbound message. +type GateVerdict struct { + // Proceed reports whether the message should be delivered downstream. + Proceed bool + // Reason is a short token describing a drop: "rate_limited", "review_pruned" + // or "capped". It is empty when Proceed is true. + Reason string + // DroppedMsg is the message that failed admission (nil when Proceed is true). + DroppedMsg *UnifiedMessage + // Pruned lists every message the gate dropped during this admission. On the + // review path it is the noise the reviewer rejected; on the cap path it is the + // oldest messages evicted to honor the total cap. Each is recorded to the + // OnPruned audit hook before being dropped, so nothing is lost silently. + Pruned []UnifiedMessage +} + +// FlowGateOptions configures a per-adapter ingress gate. Every limit is +// independent and a zero value disables that limit, so an account that does not +// declare the new fields keeps its previous unlocked behavior (backward +// compatible, design/12 §6 实现要点). +type FlowGateOptions struct { + // GroupMsgPerMin caps group-chat inbound messages per minute (design/12 §6 + // 策略三). Zero disables the group limit. + GroupMsgPerMin int + // MsgPerMin caps total inbound non-owner messages per minute. Zero disables it. + MsgPerMin int + // MaxMessages is the total pending-message cap. When the pending set nears it + // the Reviewer is invoked to prune noise; with no Reviewer the oldest messages + // are evicted FIFO. Zero disables the cap (design/12 §6 策略四). + MaxMessages int + // Window is the sliding window over which the per-minute limits are counted. + // Zero uses defaultRateWindow. + Window time.Duration + // Review is the optional review model. When nil the gate enforces the cap with + // FIFO eviction instead of model judgment. + Review Reviewer + // OnPruned is called once per pruned/dropped message with a short reason token + // ("review_pruned" or "capped") so the caller can record an audit trail. + // It may be nil (then the caller only sees Pruned on the verdict). + OnPruned func(UnifiedMessage, string) + // Now overrides the clock for tests. Nil uses time.Now. + Now func() time.Time +} + +const ( + // defaultRateWindow is the sliding window over which per-minute limits count. + defaultRateWindow = time.Minute + // defaultReviewPct is how full the pending set must be before the review model + // runs. Design/12 §6 策略四 says "maxMessages 的 80%": this is the cost guard + // that keeps review from firing on every message. + defaultReviewPct = 80 + // FlowLimitDefault is the protective default of 6 group messages per minute + // (design/12 §6 策略三). It is applied when an account pins no group rate limit. + FlowLimitDefault = 6 + // FlowCapDefault is the protective default total-cap of 50 pending messages + // (design/12 §6 策略四). + FlowCapDefault = 50 +) + +// FlowGate is a concurrency-safe, per-adapter ingress gate implementing the two +// measures from design/12 §6 策略三/四: +// +// - 策略三: a sliding-window rate limit on group chat (and, optionally, all +// non-owner messages). Over-limit messages are dropped (downgraded) rather +// than blocking a flood of messages from saturating the loop. +// - 策略四: a bounded pending-message set. When it nears its cap the Reviewer is +// invoked to prune low-value messages; with no Reviewer the oldest messages are +// evicted FIFO. Either way every dropped message is audited. +// +// Owner-private messages are not routed through the gate (the caller bypasses it), +// so the owner is never throttled or pruned. Identity is never examined here. +type FlowGate struct { + groupPerMin int + msgPerMin int + maxMessages int + reviewAt int + window time.Duration + review Reviewer + onPruned func(UnifiedMessage, string) + now func() time.Time + + mu sync.Mutex + groupHits []time.Time + totalHits []time.Time + buffer []queuedMsg + lastReview time.Time + seq uint64 +} + +// queuedMsg wraps a buffer entry with a stable sequence so pruning decisions can +// be audited and the fate of the just-admitted message be tracked across index +// shifts (value equality on UnifiedMessage is unsafe because it holds slices). +type queuedMsg struct { + seq uint64 + msg UnifiedMessage +} + +// NewFlowGate builds a gate from opts. Options that are zero disable their limit, +// except Window which defaults to one minute when left zero. +func NewFlowGate(opts FlowGateOptions) *FlowGate { + g := &FlowGate{ + groupPerMin: opts.GroupMsgPerMin, + msgPerMin: opts.MsgPerMin, + maxMessages: opts.MaxMessages, + window: opts.Window, + review: opts.Review, + onPruned: opts.OnPruned, + now: opts.Now, + } + if g.window <= 0 { + g.window = defaultRateWindow + } + if g.maxMessages > 0 { + g.reviewAt = g.maxMessages * defaultReviewPct / 100 + if g.reviewAt < 1 { + g.reviewAt = 1 + } + } + if g.now == nil { + g.now = time.Now + } + return g +} + +// Admit applies the rate limit then the total-cap/review gate to one inbound +// message. It returns a verdict telling the caller whether to proceed; messages +// that the gate drops are returned (and audited) so nothing is lost silently. +func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict { + now := g.now() + g.mu.Lock() + defer g.mu.Unlock() + + // 策略三: sliding-window rate limits. Group chat has its own cap; MsgPerMin is + // an optional total cap across all non-owner messages. Over-limit messages are + // dropped at the door (downgraded) instead of blocking. + if g.groupPerMin > 0 && msg.ChannelType == ChannelGroup { + g.groupHits = g.trimHits(g.groupHits, now) + if len(g.groupHits) >= g.groupPerMin { + return g.drop("rate_limited", msg) + } + g.groupHits = append(g.groupHits, now) + } + if g.msgPerMin > 0 { + g.totalHits = g.trimHits(g.totalHits, now) + if len(g.totalHits) >= g.msgPerMin { + return g.drop("rate_limited", msg) + } + g.totalHits = append(g.totalHits, now) + } + + if g.maxMessages <= 0 { + return GateVerdict{Proceed: true} + } + + // 策略四: add to the pending set, then prune when it nears the cap. + g.seq++ + seq := g.seq + g.buffer = append(g.buffer, queuedMsg{seq: seq, msg: msg}) + + var dropped []queuedMsg + if g.review != nil && len(g.buffer) >= g.reviewAt && !g.reviewRecently(now) { + keep, err := g.review(ctx, queueMessages(g.buffer)) + if err == nil { + g.buffer, dropped = g.applyReview(g.buffer, keep) + g.lastReview = now + g.auditDropped(dropped, "review_pruned") + } + } + if len(g.buffer) > g.maxMessages { + n := len(g.buffer) - g.maxMessages + var evicted []queuedMsg + g.buffer, evicted = g.dropFront(g.buffer, n) + dropped = append(dropped, evicted...) + g.auditDropped(evicted, "capped") + } + + if !g.containsSeq(seq) { + reason := "capped" + if containsSeq(dropped, seq) { + reason = "review_pruned" + } + v := GateVerdict{Proceed: false, Reason: reason, Pruned: queueMessages(dropped)} + c := msg + v.DroppedMsg = &c + return v + } + return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)} +} + +// drop builds a refused verdict from a single message that never entered the +// buffer (rate-limited at the door), so the audit hook still sees it. +func (g *FlowGate) drop(reason string, msg UnifiedMessage) GateVerdict { + c := msg + v := GateVerdict{Proceed: false, Reason: reason, DroppedMsg: &c, Pruned: []UnifiedMessage{msg}} + g.auditMsg(msg, reason) + return v +} + +// reviewRecently throttles review so it does not run on every message: it fires at +// most once per window, which is the cost guard for the review model. +func (g *FlowGate) reviewRecently(now time.Time) bool { + if g.lastReview.IsZero() { + return false + } + return now.Sub(g.lastReview) < g.window +} + +// trimHits drops event timestamps older than the sliding window. +func (g *FlowGate) trimHits(hits []time.Time, now time.Time) []time.Time { + cutoff := now.Add(-g.window) + keep := 0 + for _, t := range hits { + if t.After(cutoff) { + hits[keep] = t + keep++ + } + } + return hits[:keep] +} + +// applyReview keeps only the entries at the returned indices (validated, de-duped, +// ascending) and returns the kept slice plus the dropped entries in original order. +func (g *FlowGate) applyReview(buffer []queuedMsg, keep []int) (kept, dropped []queuedMsg) { + seen := make(map[int]bool, len(keep)) + for _, i := range keep { + if i >= 0 && i < len(buffer) { + seen[i] = true + } + } + out := make([]queuedMsg, 0, len(buffer)) + drop := make([]queuedMsg, 0) + for i, entry := range buffer { + if seen[i] { + out = append(out, entry) + } else { + drop = append(drop, entry) + } + } + return out, drop +} + +// dropFront removes n entries from the front (oldest) and returns the rest plus the +// dropped ones. +func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedMsg) { + if n >= len(buffer) { + return nil, append([]queuedMsg(nil), buffer...) + } + return buffer[n:], append([]queuedMsg(nil), buffer[:n]...) +} + +// containsSeq reports whether an entry with the given sequence is still buffered. +func (g *FlowGate) containsSeq(seq uint64) bool { + for _, e := range g.buffer { + if e.seq == seq { + return true + } + } + return false +} + +// containsSeq reports whether seq appears among the dropped entries. +func containsSeq(dropped []queuedMsg, seq uint64) bool { + for _, e := range dropped { + if e.seq == seq { + return true + } + } + return false +} + +// auditDropped records every dropped entry to the OnPruned audit hook. +func (g *FlowGate) auditDropped(dropped []queuedMsg, reason string) { + for _, e := range dropped { + g.auditMsg(e.msg, reason) + } +} + +// auditMsg records one dropped message to the OnPruned audit hook. +func (g *FlowGate) auditMsg(msg UnifiedMessage, reason string) { + if g.onPruned != nil { + g.onPruned(msg, reason) + } +} + +// queueMessages extracts the payload messages of queued entries in order. +func queueMessages(queue []queuedMsg) []UnifiedMessage { + out := make([]UnifiedMessage, 0, len(queue)) + for _, e := range queue { + out = append(out, e.msg) + } + return out +} diff --git a/internal/channel/adapter/flowgate_test.go b/internal/channel/adapter/flowgate_test.go new file mode 100644 index 0000000..9b83e4d --- /dev/null +++ b/internal/channel/adapter/flowgate_test.go @@ -0,0 +1,253 @@ +package adapter + +import ( + "context" + "testing" + "time" +) + +// fixedClock returns a Now func pinned to a base time so tests can advance time by +// controlling the returned value. +func fixedClock(base time.Time) (func() time.Time, *time.Time) { + cur := base + return func() time.Time { return cur }, &cur +} + +func groupMsg(i int) UnifiedMessage { + return UnifiedMessage{ + Platform: "kook", + ChannelID: "-100", + ChannelType: ChannelGroup, + GroupID: "-100", + SenderID: "u" + itoa(i), + Content: "msg " + itoa(i), + ContentType: ContentText, + } +} + +func directMsg(i int) UnifiedMessage { + return UnifiedMessage{ + Platform: "telegram", + ChannelID: "100", + ChannelType: ChannelDirect, + SenderID: "u" + itoa(i), + Content: "dm " + itoa(i), + ContentType: ContentText, + } +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [20]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} + +func TestFlowGate_AllZeroAllowsEverything(t *testing.T) { + t.Parallel() + g := NewFlowGate(FlowGateOptions{}) + for i := 0; i < 20; i++ { + if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed { + t.Fatalf("message %d dropped by zero-option gate: %+v", i, v) + } + } +} + +func TestFlowGate_GroupRateLimitDowngradesOverLimit(t *testing.T) { + t.Parallel() + now, clock := fixedClock(time.Unix(1000, 0)) + g := NewFlowGate(FlowGateOptions{ + GroupMsgPerMin: 6, + Window: time.Minute, + Now: now, + }) + + // Six messages within the window are admitted; the seventh is downgraded. + for i := 0; i < 6; i++ { + if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed { + t.Fatalf("message %d should be admitted within limit", i) + } + } + v := g.Admit(context.Background(), groupMsg(6)) + if v.Proceed { + t.Fatal("7th group message in the window must be rate limited") + } + if v.Reason != "rate_limited" { + t.Fatalf("reason = %q, want rate_limited", v.Reason) + } + if v.DroppedMsg == nil || v.DroppedMsg.Content != "msg 6" { + t.Fatalf("dropped msg = %+v, want msg 6", v.DroppedMsg) + } + + // Advancing past the window frees capacity again. + *clock = clock.Add(61 * time.Second) + if v := g.Admit(context.Background(), groupMsg(7)); !v.Proceed { + t.Fatalf("message after window roll should be admitted, got %+v", v) + } +} + +func TestFlowGate_GroupRateLimitDoesNotTouchDirect(t *testing.T) { + t.Parallel() + g := NewFlowGate(FlowGateOptions{GroupMsgPerMin: 1, Window: time.Minute}) + // Direct/private messages bypass the group rate limit. + for i := 0; i < 5; i++ { + if v := g.Admit(context.Background(), directMsg(i)); !v.Proceed { + t.Fatalf("direct message %d wrongly rate limited: %+v", i, v) + } + } +} + +func TestFlowGate_MsgPerMinCapsAllNonOwner(t *testing.T) { + t.Parallel() + g := NewFlowGate(FlowGateOptions{MsgPerMin: 2, Window: time.Minute}) + if v := g.Admit(context.Background(), directMsg(0)); !v.Proceed { + t.Fatal("first dm should pass") + } + if v := g.Admit(context.Background(), directMsg(1)); !v.Proceed { + t.Fatal("second dm should pass") + } + if v := g.Admit(context.Background(), directMsg(2)); v.Proceed { + t.Fatal("third dm should exceed the total per-minute cap") + } +} + +func TestFlowGate_ReviewPrunesWhenNearCap(t *testing.T) { + t.Parallel() + now, _ := fixedClock(time.Unix(1000, 0)) + var audited []string + g := NewFlowGate(FlowGateOptions{ + MaxMessages: 10, + Window: time.Minute, + Now: now, + Review: func(_ context.Context, msgs []UnifiedMessage) ([]int, error) { + // Keep only the first two messages; drop the rest as noise. + return []int{0, 1}, nil + }, + OnPruned: func(_ UnifiedMessage, reason string) { + audited = append(audited, reason) + }, + }) + + // Seven messages stay below the 80% threshold (reviewAt = 8), no review fires. + for i := 0; i < 7; i++ { + if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed { + t.Fatalf("message %d should be admitted before threshold: %+v", i, v) + } + } + if len(audited) != 0 { + t.Fatalf("review fired below threshold: %v", audited) + } + + // The 8th message crosses the 80% threshold → review runs and drops msg2..7, + // including the just-admitted message 7. + v := g.Admit(context.Background(), groupMsg(7)) + if v.Proceed { + t.Fatal("just-admitted message 7 must be pruned by the reviewer") + } + if v.Reason != "review_pruned" { + t.Fatalf("reason = %q, want review_pruned", v.Reason) + } + if len(v.Pruned) != 6 { + t.Fatalf("pruned = %d, want 6 (msg2..msg7)", len(v.Pruned)) + } + if len(audited) != 6 { + t.Fatalf("audit count = %d, want 6", len(audited)) + } + for _, r := range audited { + if r != "review_pruned" { + t.Fatalf("audit reason = %q, want review_pruned", r) + } + } +} + +func TestFlowGate_ReviewThrottledWithinWindow(t *testing.T) { + t.Parallel() + now, clock := fixedClock(time.Unix(1000, 0)) + reviews := 0 + g := NewFlowGate(FlowGateOptions{ + MaxMessages: 3, + Window: time.Minute, + Now: now, + Review: func(_ context.Context, msgs []UnifiedMessage) ([]int, error) { + reviews++ + return []int{0}, nil + }, + }) + + // First threshold crossing triggers a review. + for i := 0; i < 3; i++ { + g.Admit(context.Background(), groupMsg(i)) + } + if reviews != 1 { + t.Fatalf("reviews after crossing threshold = %d, want 1", reviews) + } + + // Within the same window the gate must not re-review even when it refills. + for i := 3; i < 6; i++ { + g.Admit(context.Background(), groupMsg(i)) + } + if reviews != 1 { + t.Fatalf("reviews within window = %d, want 1 (throttled)", reviews) + } + + // A fresh window allows a review again. + *clock = clock.Add(2 * time.Minute) + for i := 6; i < 9; i++ { + g.Admit(context.Background(), groupMsg(i)) + } + if reviews != 2 { + t.Fatalf("reviews after window roll = %d, want 2", reviews) + } +} + +func TestFlowGate_FifoCapWhenNoReview(t *testing.T) { + t.Parallel() + var audited []string + g := NewFlowGate(FlowGateOptions{ + MaxMessages: 3, + OnPruned: func(_ UnifiedMessage, reason string) { + audited = append(audited, reason) + }, + }) + + for i := 0; i < 3; i++ { + if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed { + t.Fatalf("message %d should be admitted at cap: %+v", i, v) + } + } + // The 4th evicts the oldest (msg0) to honor the cap; the new message is admitted. + v := g.Admit(context.Background(), groupMsg(3)) + if !v.Proceed { + t.Fatalf("new message should be admitted after FIFO eviction, got %+v", v) + } + if len(v.Pruned) != 1 || v.Pruned[0].Content != "msg 0" { + t.Fatalf("pruned = %+v, want [msg 0]", v.Pruned) + } + if len(audited) != 1 || audited[0] != "capped" { + t.Fatalf("audit = %v, want [capped]", audited) + } +} + +func TestFlowGate_Defaults(t *testing.T) { + t.Parallel() + if FlowLimitDefault != 6 { + t.Fatalf("FlowLimitDefault = %d, want 6", FlowLimitDefault) + } + if FlowCapDefault != 50 { + t.Fatalf("FlowCapDefault = %d, want 50", FlowCapDefault) + } + g := NewFlowGate(FlowGateOptions{}) + if g.window != time.Minute { + t.Fatalf("default window = %v, want 1m", g.window) + } + if g.reviewAt != 0 || g.maxMessages != 0 { + t.Fatalf("zero-option gate should disable both cap and review, got reviewAt=%d max=%d", g.reviewAt, g.maxMessages) + } +} diff --git a/internal/config/channels_test.go b/internal/config/channels_test.go index 31535fd..142cba4 100644 --- a/internal/config/channels_test.go +++ b/internal/config/channels_test.go @@ -53,6 +53,39 @@ func TestParseChannelAccounts_MultiAccountWithAllowDeny(t *testing.T) { } } +func TestParseChannelAccounts_FlowLimitFields(t *testing.T) { + t.Parallel() + raw := `{ + "telegram": [ + {"id":"conf0","config":{"bot_token":"t0"},"groupMsgPerMin":10,"maxMessages":80,"msgPerMin":30} + ] + }` + got, err := ParseChannelAccounts(raw) + if err != nil { + t.Fatalf("ParseChannelAccounts: %v", err) + } + a := got["telegram"][0] + if a.GroupMsgPerMin != 10 || a.MsgPerMin != 30 || a.MaxMessages != 80 { + t.Fatalf("flow fields = group:%d msg:%d max:%d, want 10/30/80", + a.GroupMsgPerMin, a.MsgPerMin, a.MaxMessages) + } +} + +func TestParseChannelAccounts_NoFlowFieldsDefaultsToZero(t *testing.T) { + t.Parallel() + raw := `{"telegram":[{"id":"conf0","config":{"bot_token":"t0"}}]}` + got, err := ParseChannelAccounts(raw) + if err != nil { + t.Fatalf("ParseChannelAccounts: %v", err) + } + a := got["telegram"][0] + // Absent fields stay zero; the bridge maps zero to the protective defaults + // (6 group/min, 50 max), so old accounts are backward compatible. + if a.GroupMsgPerMin != 0 || a.MsgPerMin != 0 || a.MaxMessages != 0 { + t.Fatalf("absent flow fields should be zero, got %+v", a) + } +} + func TestParseChannelAccounts_Malformed(t *testing.T) { t.Parallel() if _, err := ParseChannelAccounts(`{`); err == nil { diff --git a/internal/config/config.go b/internal/config/config.go index 262e6d4..c2b1062 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -104,11 +104,17 @@ type VoiceConfig struct { DashscopeAPIKey string `env:"DASHSCOPE_API_KEY"` } -// BackgroundConfig holds the autonomous thinking settings. +// BackgroundConfig holds the autonomous thinking settings. QuietStartHour and +// QuietEndHour bound the do-not-disturb window in local clock hours; they are the +// single source shared by the backthink service and the channel bridge's +// minimal-disturbance quiet-hours gate (design/12 §6 策略五). Equal non-zero hours +// disable the window. type BackgroundConfig struct { Enabled bool `env:"ENABLE_BACKGROUND_THINKING" default:"true"` IntervalSec int `env:"THINK_INTERVAL_SEC" default:"300"` SilenceTimeout int `env:"THINK_SILENCE_TIMEOUT_SEC" default:"120"` + QuietStartHour int `env:"QUIET_START_HOUR" default:"23"` + QuietEndHour int `env:"QUIET_END_HOUR" default:"7"` } // WSHubConfig holds the WebSocket hub limits. @@ -193,6 +199,18 @@ type ChannelAccount struct { // DenyUsers is a black list of user uids: a direct message from a user in this // list is dropped at ingress. DenyUsers []string `json:"denyUsers,omitempty"` + // GroupMsgPerMin caps group-chat inbound messages per minute for this account + // (design/12 §6 策略三). Zero follows the protective default (6); a negative + // value disables the group rate limit; a positive value pins it. + GroupMsgPerMin int `json:"groupMsgPerMin,omitempty"` + // MsgPerMin caps total inbound non-owner messages per minute for this account. + // Zero or negative disables it. + MsgPerMin int `json:"msgPerMin,omitempty"` + // MaxMessages is the total pending/active message cap for this account + // (design/12 §6 策略四). Zero follows the protective default (50); a negative + // value disables the cap; a positive value pins it. When the pending set nears + // this cap the review model prunes low-value messages. + MaxMessages int `json:"maxMessages,omitempty"` } // Load reads configuration, merging an optional .env file (development only) diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index 9ac9fb9..4ec8227 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -104,10 +104,14 @@ type Config struct { } // ConfigFromBackground maps a config.BackgroundConfig (the source of the -// heartbeat interval) into a backthink.Config, leaving the push/decide knobs at -// their defaults for normalize to fill in. +// heartbeat interval and the shared quiet-hours window) into a backthink.Config, +// leaving the push/decide knobs at their defaults for normalize to fill in. func ConfigFromBackground(bg config.BackgroundConfig) Config { - return Config{IntervalSec: bg.IntervalSec} + return Config{ + IntervalSec: bg.IntervalSec, + QuietStartHour: bg.QuietStartHour, + QuietEndHour: bg.QuietEndHour, + } } // Deps is the injected dependency set for a Backthink. Only Thinker and DB are @@ -308,12 +312,40 @@ func (b *Backthink) subscribe(ctx context.Context) { } // onEvent is the bus handler for event-triggered thinking. It runs on the bus's -// own worker goroutine, so it must not block on another subscriber. +// own worker goroutine, so it must not block on another subscriber. It first +// applies the minimal-disturbance gate (design/12 §6 策略一): a low-value system +// event is downgraded to a registration-only record rather than waking a full +// thinking pass, so platform/system noise cannot flood the thinker. func (b *Backthink) onEvent(ctx context.Context, e bus.Event) error { - b.think(ctx, triggerForEvent(e)) + 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 gates event-triggered thinking (design/12 §6 策略一). On a public +// platform a generic system event is low value, so only a high-importance +// (owner-relevant) system event wakes a pass. IoT device updates and reminder +// triggers are owner-relevant / high value and are never gated, so the productive +// triggers stay fully functional. +func (b *Backthink) shouldThink(tr Trigger) bool { + if tr.Source == SubjectSystemEvent { + return tr.Import >= ImportanceHigh + } + return true +} + // tickLoop is the heartbeat goroutine. It fires one thinking pass per tick and // exits when ctx is cancelled. func (b *Backthink) tickLoop(ctx context.Context) { diff --git a/internal/services/backthink/backthink_test.go b/internal/services/backthink/backthink_test.go index 337f970..5a9c422 100644 --- a/internal/services/backthink/backthink_test.go +++ b/internal/services/backthink/backthink_test.go @@ -136,6 +136,63 @@ func TestBackthink_EventTrigger_NoPush(t *testing.T) { } } +func TestBackthink_SystemEventLowValueDowngraded(t *testing.T) { + t.Parallel() + // design/12 §6 策略一: a low-value system event on a public platform does not + // wake a full thinking pass — it is downgraded to a registration-only record. + thinker := &mockThinker{text: "should not be called"} + bs := bus.New() + b, _ := newBackthink(t, Config{}, Deps{ + Bus: bs, + Thinker: thinker, + Now: func() time.Time { return dayAt() }, + }) + + b.Start(context.Background()) + defer func() { _ = b.Stop(context.Background()) }() + + if err := bs.Publish(context.Background(), SubjectSystemEvent, SystemEvent{Summary: "member joined"}); err != nil { + t.Fatalf("Publish: %v", err) + } + + eventually(t, 2*time.Second, func() bool { + recs, _ := b.List(context.Background(), 10) + return len(recs) >= 1 && recs[0].Trigger == SubjectSystemEvent + }) + if got := thinker.count(); got != 0 { + t.Fatalf("thinker calls = %d, want 0 (low-value system event downgraded)", got) + } + recs, _ := b.List(context.Background(), 10) + if recs[0].ShouldPush { + t.Fatalf("ShouldPush = true for a downgraded system event") + } + if recs[0].PushResult.Route != RouteNone { + t.Fatalf("PushResult.Route = %q, want %q", recs[0].PushResult.Route, RouteNone) + } +} + +func TestBackthink_SystemEventHighValueThinks(t *testing.T) { + t.Parallel() + // A high-importance (owner-relevant) system event still wakes a thinking pass — + // the denoise gate only downgrades low-value noise (design/12 §6 策略一). + thinker := &mockThinker{text: "important notice"} + bs := bus.New() + b, _ := newBackthink(t, Config{}, Deps{ + Bus: bs, + Thinker: thinker, + Now: func() time.Time { return dayAt() }, + }) + + b.Start(context.Background()) + defer func() { _ = b.Stop(context.Background()) }() + + if err := bs.Publish(context.Background(), SubjectSystemEvent, SystemEvent{Summary: "owner tagged", Import: ImportanceHigh}); err != nil { + t.Fatalf("Publish: %v", err) + } + + eventually(t, 2*time.Second, func() bool { return thinker.count() >= 1 }) +} + func TestBackthink_List_NewestFirst(t *testing.T) { t.Parallel() thinker := &mockThinker{text: "nothing to report"} diff --git a/internal/services/backthink/trigger.go b/internal/services/backthink/trigger.go index 23a13c5..2eb6828 100644 --- a/internal/services/backthink/trigger.go +++ b/internal/services/backthink/trigger.go @@ -17,6 +17,17 @@ const ( SubjectSystemEvent = "system.event" ) +// SystemEvent is the optional typed payload a caller may publish to +// SubjectSystemEvent to convey a system notice. Import defaults to Normal. Under +// design/12 §6 策略一, a low-value system notice (a join/leave ping, generic +// platform noise) is downgraded to a registration-only record and does not wake a +// full thinking pass; only a high-importance (owner-relevant) system event thinks, +// so a busy public platform cannot trigger a flood of expensive reflections. +type SystemEvent struct { + Summary string + Import Importance +} + // Trigger describes what woke a thinking pass: the source subsystem, a // human-readable summary and the event's estimated importance. type Trigger struct { @@ -48,7 +59,14 @@ func triggerForEvent(e bus.Event) Trigger { case SubjectReminderTrigger: return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceHigh} case SubjectSystemEvent: - return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal} + tr := Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal} + if sy, ok := e.Payload.(SystemEvent); ok { + if sy.Summary != "" { + tr.Summary = sy.Summary + } + tr.Import = sy.Import + } + return tr default: return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal} } diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 4b56be1..5a0b16c 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -90,7 +90,30 @@ type Options struct { // allow/deny black-or-white lists (design/16 §4.6 §6), so the composition root // can route it into an audit trail. It is invoked with the rejected message and // a short reason token ("group_denied", "user_not_allowed", ...). Nil = log only. + // + // It is also wired as the flow gate's OnPruned audit hook (design/12 §6 策略三/四), + // so a message dropped by the rate limiter or the review prune reaches the same + // audit trail with reason "rate_limited", "review_pruned" or "capped". OnRejected func(adapter.UnifiedMessage, string) + // Review is the optional review model (design/12 §6 策略四) that prunes low-value + // queued messages. Nil falls back to FIFO eviction at the total cap. Use + // NewLLMReviewer to wrap a model.Adapter into this signature. + Review adapter.Reviewer + // Light is the optional light-response judge for non-owner group/stranger + // messages (design/12 §6 策略二/五). When nil the bridge uses its built-in + // rule-based judge and fails-closed to silence, so no chat message is sent to an + // LLM just to decide whether to bother the owner. It is only consulted when the + // rule judge is inconclusive. + Light LightJudge + // QuietStartHour and QuietEndHour bound the do-not-disturb window (local clock + // hours), shared with the backthink service so 策略五 reuses the same definition. + // Equal non-zero hours disable the window; both zero pick the backthink defaults + // (23-07). This is the price of keeping one source of truth across services. + QuietStartHour int + QuietEndHour int + // Now is the wall-clock source for quiet-hour judgement. Nil uses time.Now, + // which makes time-dependent behaviour injectable in tests. + Now func() time.Time } // Bridge owns the channel adapter lifecycle and the inbound message pipeline. It @@ -103,12 +126,24 @@ type Bridge struct { history history.HistoryStore onPersisted func(history.Message) onRejected func(adapter.UnifiedMessage, string) + review adapter.Reviewer + + light LightJudge + quietStartHour int + quietEndHour int + now func() time.Time mu sync.RWMutex adapters map[string]adapter.PlatformAdapter connected map[string]bool closed bool + // gateMu guards the per-adapter flow gates (design/12 §6 策略三/四). They are + // built lazily on the first inbound message per account and are independent of + // the adapter/connected lifecycle, so the same gate survives reconnects. + gateMu sync.Mutex + gates map[string]*adapter.FlowGate + ctx context.Context cancel context.CancelFunc wg sync.WaitGroup @@ -129,18 +164,34 @@ func New(opts Options) (*Bridge, error) { if log == nil { log = logger.New(logger.Options{Out: nil}) } + // Quiet hours default to the same window the backthink service uses, so the two + // services never disagree about the do-not-disturb window (design/12 §6 策略五). + qStart, qEnd := opts.QuietStartHour, opts.QuietEndHour + if qStart == 0 && qEnd == 0 { + qStart, qEnd = defaultQuietStartHour, defaultQuietEndHour + } + now := opts.Now + if now == nil { + now = time.Now + } return &Bridge{ - log: logger.WithSource(log, "services/channelbridge"), - run: opts.Loop, - sessions: opts.Sessions, - owner: opts.Owner, - history: opts.History, - onPersisted: opts.OnPersisted, - onRejected: opts.OnRejected, - adapters: make(map[string]adapter.PlatformAdapter), - connected: make(map[string]bool), - cancel: func() {}, - sem: make(chan struct{}, maxConcurrent), + log: logger.WithSource(log, "services/channelbridge"), + run: opts.Loop, + sessions: opts.Sessions, + owner: opts.Owner, + history: opts.History, + onPersisted: opts.OnPersisted, + onRejected: opts.OnRejected, + review: opts.Review, + light: opts.Light, + quietStartHour: qStart, + quietEndHour: qEnd, + now: now, + adapters: make(map[string]adapter.PlatformAdapter), + connected: make(map[string]bool), + gates: make(map[string]*adapter.FlowGate), + cancel: func() {}, + sem: make(chan struct{}, maxConcurrent), }, nil } @@ -341,6 +392,23 @@ func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a ad return } + // design/12 §6 策略三/四: per-adapter minimal-disturbance gate. Owner-private + // messages bypass it entirely (the owner is never throttled or pruned); every + // other message passes a sliding-window rate limit (策略三) and the bounded + // pending set whose saturation triggers the review model to prune noise (策略四). + // Dropped messages are audited through the gate's OnPruned hook (wired to + // OnRejected) and never reach the pipeline, so a flood cannot saturate the loop. + actor := session.ResolveActor(msg, b.owner) + if !actor.IsOwner() { + verdict := b.gateFor(a).Admit(ctx, msg) + if !verdict.Proceed { + b.log.Warn("inbound message dropped by flow gate", + "platform", msg.Platform, "channel", msg.ChannelID, + "reason", verdict.Reason, "dropped", len(verdict.Pruned)) + return + } + } + select { case b.sem <- struct{}{}: case <-ctx.Done(): @@ -415,6 +483,21 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a "session", sess.SessionID, "actor_role", sess.ActorRole, "env", sess.Env(), "mem_writable", sess.MemWritable) + // design/12 §6 策略一/二/五: minimal-disturbance gate for non-owner messages. A + // non-owner group/stranger message that the cheap light judgement deems not + // worth a response is registered to its isolated session log and silenced — it + // never drives the full agent loop. Owner messages are never gated here, so + // owner-private chat and owner group @/high-value replies are unchanged. + if !actor.IsOwner() { + if ok, reason := b.shouldRespond(ctx, msg, actor); !ok { + b.registerSilenced(ctx, msg, actor, sess) + b.log.Debug("non-owner message silenced (minimal disturbance)", + "platform", msg.Platform, "channel", msg.ChannelID, + "session", sess.SessionID, "reason", reason) + return + } + } + // design/16 §4.4 + design/02 §3: an owner private chat is routed INTO the single // main web session (shared across platforms) and persisted to chat_messages with // the channel platform, so the Web client mirrors it; group chats and strangers diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index b9b61f7..e590c8b 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -84,12 +84,17 @@ func (s *stubSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage return s.sess, nil } +// directMessage is an owner-private direct message: the sender is tagged as the +// owner (structural, not from text), so it exercises the owner-private path. The +// single-account stub session in newTestBridge reports Owner=true to match, since +// the production SessionManager derives Owner from the resolved actor. func directMessage() adapter.UnifiedMessage { return adapter.UnifiedMessage{ Platform: "telegram", ChannelID: "100", ChannelType: adapter.ChannelDirect, SenderID: "100", + Sender: adapter.Actor{Role: adapter.RoleOwner}, Content: "hello", ContentType: adapter.ContentText, } @@ -180,6 +185,8 @@ func TestProcessGroupMessageKeepsGroupTarget(t *testing.T) { t.Parallel() b, chat, _, ad := newTestBridge(t) + // A non-owner group message that is addressed (mentioned) gets a response under + // design/12 §6 策略二, so it still exercises group reply targeting. msg := adapter.UnifiedMessage{ Platform: "telegram", ChannelID: "-20", @@ -187,6 +194,8 @@ func TestProcessGroupMessageKeepsGroupTarget(t *testing.T) { GroupID: "-20", GroupName: "Gang", SenderID: "9", + Sender: adapter.Actor{Role: adapter.RoleMember}, + Mentions: []adapter.Mention{{UserID: "9"}}, Content: "hello all", ContentType: adapter.ContentText, } @@ -351,10 +360,13 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) { t.Fatalf("New: %v", err) } + // Addressed (mentioned) group message → responds (design/12 §6 策略二), still + // isolated to the group session. msg := adapter.UnifiedMessage{ Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup, - GroupID: "-20", GroupName: "Gang", SenderID: "9", Content: "hello all", - ContentType: adapter.ContentText, + GroupID: "-20", GroupName: "Gang", SenderID: "9", + Sender: adapter.Actor{Role: adapter.RoleMember}, Mentions: []adapter.Mention{{UserID: "9"}}, + Content: "hello all", ContentType: adapter.ContentText, } b.process(context.Background(), msg) @@ -463,11 +475,13 @@ func TestGroupMessageThirdPartySenderNameAndBoundary(t *testing.T) { t.Fatalf("New: %v", err) } + // Addressed (mentioned) group message → responds (design/12 §6 策略二), so the + // third-party boundary and the [第三方 · 张三] prefix are injected. msg := adapter.UnifiedMessage{ Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup, GroupID: "-20", GroupName: "Gang", SenderID: "9", SenderName: "张三", - Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "hello all", - ContentType: adapter.ContentText, + Sender: adapter.Actor{Role: adapter.RoleMember}, Mentions: []adapter.Mention{{UserID: "9"}}, + Content: "hello all", ContentType: adapter.ContentText, } b.process(context.Background(), msg) diff --git a/internal/services/channelbridge/flowgate.go b/internal/services/channelbridge/flowgate.go new file mode 100644 index 0000000..5432343 --- /dev/null +++ b/internal/services/channelbridge/flowgate.go @@ -0,0 +1,178 @@ +package channelbridge + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "sort" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// gateFor returns the per-account flow gate for an adapter, building it lazily +// from the adapter's Config snapshot on first use. Gates are keyed by the adapter +// name so every account (platform:confId) gets an independent rate limit and +// pending-cap (design/12 §6 策略三/四), and they survive adapter reconnects. +func (b *Bridge) gateFor(a adapter.PlatformAdapter) *adapter.FlowGate { + name := a.Name() + b.gateMu.Lock() + defer b.gateMu.Unlock() + if g, ok := b.gates[name]; ok { + return g + } + g := adapter.NewFlowGate(b.gateOptions(a)) + b.gates[name] = g + return g +} + +// gateOptions resolves the per-account flow-limits out of the adapter's Config +// snapshot into a FlowGate. An absent key (single-account env adapters) leaves the +// limit off, so existing channels run unchanged; a present key with value 0 follows +// the protective default (6 group/min, 50 cap) and a negative value disables it — +// the backward-compatible contract for CHANNELS_JSON accounts (design/12 §6 实现要点). +func (b *Bridge) gateOptions(a adapter.PlatformAdapter) adapter.FlowGateOptions { + opts := adapter.FlowGateOptions{ + Review: b.review, + OnPruned: b.onRejected, + Now: b.now, + } + cfg := a.Config() + if grp, ok := cfgInt(cfg, "groupMsgPerMin"); ok { + opts.GroupMsgPerMin = flowOrDefault(grp, adapter.FlowLimitDefault) + } + if tot, ok := cfgInt(cfg, "msgPerMin"); ok && tot > 0 { + opts.MsgPerMin = tot + } + if max, ok := cfgInt(cfg, "maxMessages"); ok { + opts.MaxMessages = flowOrDefault(max, adapter.FlowCapDefault) + } + return opts +} + +// flowOrDefault maps a configured limit value to an effective limit: a negative +// value disables the limit (0), zero follows the protective default, and a positive +// value pins it. This is the "old account without the field keeps prior behavior or +// gets the safe default" contract. +func flowOrDefault(v, def int) int { + switch { + case v < 0: + return 0 + case v == 0: + return def + default: + return v + } +} + +// cfgInt reads an integer field from an adapter config map, returning (0,false) +// when the key is absent or of an unexpected type (so an absent field stays "off"). +func cfgInt(m map[string]any, key string) (int, bool) { + if m == nil { + return 0, false + } + switch v := m[key].(type) { + case int: + return v, true + case int64: + return int(v), true + case float64: + return int(v), true + default: + return 0, false + } +} + +const ( + // reviewMaxTokens bounds the review model output so a noisy queue listing never + // produces a giant completion (design/12 §6 策略四 cost guard). + reviewMaxTokens = 256 + // reviewSnippetLen is how many bytes of each message's content are shown to the + // reviewer, enough to judge importance without leaking the whole queue. + reviewSnippetLen = 80 + // reviewPrompt instructs the review model to pick which queued messages matter. + // It only ever decides noise/importance — never who the sender is (project red + // line: identity comes from actor metadata, not the LLM). + reviewPrompt = "你是群聊消息去噪助手。下面列出若干进入队列的消息（每条带编号）。" + + "请判断哪些消息没有保留价值（灌水、无关、低信息量），只返回你认为【有必要保留】的编号 JSON 数组。" + + "只判断信息价值，不判断发送者身份；不要编造编号；只输出 JSON 数组。" +) + +// NewLLMReviewer wraps a model.Adapter into an adapter.Reviewer (design/12 §6 策略四). +// It asks the LLM, via a light single-turn prompt, which of the queued messages are +// worth keeping and routes the answer back to the gate. The gate only invokes it +// when a pending set nears its cap, so the LLM is never called per-message. +func NewLLMReviewer(llm model.Adapter, log *slog.Logger) adapter.Reviewer { + if log == nil { + log = slog.Default() + } + return func(ctx context.Context, msgs []adapter.UnifiedMessage) ([]int, error) { + resp, err := llm.Chat(ctx, model.Request{ + Messages: []model.Message{ + model.TextMessage(model.RoleSystem, reviewPrompt), + model.TextMessage(model.RoleUser, buildReviewContent(msgs)), + }, + Purpose: "noise_review", + MaxTokens: reviewMaxTokens, + }, model.Options{}) + if err != nil { + return nil, fmt.Errorf("flowgate: review model: %w", err) + } + keep, err := parseKeepIndices(resp.Content, len(msgs)) + if err != nil { + log.Warn("flow gate review output unparsable, falling back to FIFO", + "error", err) + return nil, err + } + return keep, nil + } +} + +// buildReviewContent renders the queued messages as a numbered, truncated listing so +// the reviewer can judge importance cheaply. No identity or LLM judgement is made +// here — only a compact, index-labelled text summary. +func buildReviewContent(msgs []adapter.UnifiedMessage) string { + var b strings.Builder + for i, m := range msgs { + name := senderDisplayName(m, m.Sender) + if name == "" { + name = "(未标注)" + } + fmt.Fprintf(&b, "[%d] %s: %s\n", i, name, truncateText(m.Content, reviewSnippetLen)) + } + return b.String() +} + +// truncateText shortens s to at most n bytes (by rune count) for display. +func truncateText(s string, n int) string { + if n <= 0 { + return "" + } + runes := []rune(s) + if len(runes) <= n { + return s + } + return string(runes[:n]) + "…" +} + +// parseKeepIndices decodes a JSON array of indices from the review model and keeps +// only in-range, de-duplicated values in ascending order. Out-of-range or duplicate +// indices are dropped so a hallucinated number can never evict a real message. +func parseKeepIndices(s string, bound int) ([]int, error) { + var idx []int + if err := json.Unmarshal([]byte(s), &idx); err != nil { + return nil, err + } + seen := make(map[int]bool, len(idx)) + out := make([]int, 0, len(idx)) + for _, i := range idx { + if i >= 0 && i < bound && !seen[i] { + seen[i] = true + out = append(out, i) + } + } + sort.Ints(out) + return out, nil +} diff --git a/internal/services/channelbridge/flowgate_test.go b/internal/services/channelbridge/flowgate_test.go new file mode 100644 index 0000000..7a500c8 --- /dev/null +++ b/internal/services/channelbridge/flowgate_test.go @@ -0,0 +1,212 @@ +package channelbridge + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [20]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} + +func groupSender(i int) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "kook", + ChannelID: "-100", + ChannelType: adapter.ChannelGroup, + GroupID: "-100", + SenderID: "u" + itoa(i), + Content: "spam " + itoa(i), + ContentType: adapter.ContentText, + } +} + +func groupOwnerMsg(i int) adapter.UnifiedMessage { + m := groupSender(i) + m.Sender = adapter.Actor{Role: adapter.RoleOwner} + return m +} + +// TestFlowGate_RateLimitDropsNonOwnerAndAudits verifies the gate is wired into the +// ingress (onMessage) and that a rate-limited non-owner message is dropped with the +// reason surfaced through the OnRejected audit hook. The dropped path is synchronous +// (no worker goroutine is spawned), so the assertion is deterministic. +func TestFlowGate_RateLimitDropsNonOwnerAndAudits(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "hi"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}} + var rejected []string + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + ad := &fakeAdapter{name: "kook", cfg: map[string]any{"bot_token": "x", "msgPerMin": 1}} + b.Add(ad) + + // First message is admitted (async worker), only auditing the second, which the + // total per-minute cap refuses. + b.onMessage(context.Background(), groupSender(0), ad) + b.onMessage(context.Background(), groupSender(1), ad) + + if len(rejected) != 1 || rejected[0] != "rate_limited" { + t.Fatalf("rejected = %v, want [rate_limited]", rejected) + } +} + +// TestFlowGate_OwnerPrivateBypassesGate verifies owner messages bypass the flow gate +// entirely: even with an aggressive rate limit, an owner message is never dropped as +// rate_limited / review_pruned / capped. +func TestFlowGate_OwnerPrivateBypassesGate(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "hi"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + var rejected []string + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + ad := &fakeAdapter{name: "kook", cfg: map[string]any{"bot_token": "x", "groupMsgPerMin": 1, "maxMessages": 1}} + b.Add(ad) + + // Three owner messages: if the gate applied, the 2nd and 3rd would be rate + // limited / capped and reach OnRejected. Owner bypass leaves the audit empty. + for i := 0; i < 3; i++ { + b.onMessage(context.Background(), groupOwnerMsg(i), ad) + } + if len(rejected) != 0 { + t.Fatalf("owner messages must bypass the gate, got rejected %v", rejected) + } +} + +// TestFlowGate_GateOptionsDefaults pins the config-to-limit resolution that keeps +// old CHANNELS_JSON accounts backward compatible: absent key → off, 0 → protective +// default, negative → off, positive → pinned. +func TestFlowGate_GateOptionsDefaults(t *testing.T) { + t.Parallel() + + b, err := New(Options{Loop: &stubChatter{}, Sessions: &stubSessions{}}) + if err != nil { + t.Fatalf("New: %v", err) + } + + cases := []struct { + name string + cfg map[string]any + group, max int + }{ + {"absent keys off", map[string]any{"bot_token": "x"}, 0, 0}, + {"group 0 defaults 6", map[string]any{"bot_token": "x", "groupMsgPerMin": 0}, 6, 0}, + {"group -1 off", map[string]any{"bot_token": "x", "groupMsgPerMin": -1}, 0, 0}, + {"group 10 pinned", map[string]any{"bot_token": "x", "groupMsgPerMin": 10}, 10, 0}, + {"max 0 defaults 50", map[string]any{"bot_token": "x", "maxMessages": 0}, 0, 50}, + {"max -1 off", map[string]any{"bot_token": "x", "maxMessages": -1}, 0, 0}, + {"max 80 pinned", map[string]any{"bot_token": "x", "maxMessages": 80}, 0, 80}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + opts := b.gateOptions(&fakeAdapter{name: "kook", cfg: c.cfg}) + if opts.GroupMsgPerMin != c.group { + t.Fatalf("GroupMsgPerMin = %d, want %d", opts.GroupMsgPerMin, c.group) + } + if opts.MaxMessages != c.max { + t.Fatalf("MaxMessages = %d, want %d", opts.MaxMessages, c.max) + } + }) + } +} + +// mockLLM is a model.Adapter stub returning a fixed completion. +type mockLLM struct { + content string + err error +} + +func (m *mockLLM) Name() string { return "mock-llm" } +func (m *mockLLM) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + if m.err != nil { + return model.Response{}, m.err + } + return model.Response{Content: m.content, Message: model.TextMessage(model.RoleAssistant, m.content)}, nil +} +func (m *mockLLM) ChatStream(_ context.Context, _ model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return model.Response{}, nil +} + +// TestNewLLMReviewer verifies the review model is wrapped into a Reviewer that asks +// the LLM which queued messages to keep and filters the returned indices. +func TestNewLLMReviewer(t *testing.T) { + t.Parallel() + + llm := &mockLLM{content: `[0,1,99,0]`} + review := NewLLMReviewer(llm, nil) + + msgs := []adapter.UnifiedMessage{ + {SenderID: "a", Content: "x", ContentType: adapter.ContentText}, + {SenderID: "b", Content: "y", ContentType: adapter.ContentText}, + {SenderID: "c", Content: "z", ContentType: adapter.ContentText}, + } + keep, err := review(context.Background(), msgs) + if err != nil { + t.Fatalf("review: %v", err) + } + // Out-of-range index 99 and the duplicate 0 are dropped; keep = [0,1]. + if len(keep) != 2 || keep[0] != 0 || keep[1] != 1 { + t.Fatalf("keep = %v, want [0 1]", keep) + } + + if llm.err != nil { + t.Fatal("unexpected mock error") + } +} + +// TestNewLLMReviewerError verifies a review-model failure is surfaced to the gate, +// which then falls back to FIFO eviction rather than fabricating a success. +func TestNewLLMReviewerError(t *testing.T) { + t.Parallel() + + review := NewLLMReviewer(&mockLLM{err: context.DeadlineExceeded}, nil) + if _, err := review(context.Background(), []adapter.UnifiedMessage{ + {SenderID: "a", Content: "x", ContentType: adapter.ContentText}, + }); err == nil { + t.Fatal("expected error from failed review model") + } +} + +func TestParseKeepIndices(t *testing.T) { + t.Parallel() + + got, err := parseKeepIndices(`[2,0,0,5]`, 3) + if err != nil { + t.Fatalf("parse: %v", err) + } + if len(got) != 2 || got[0] != 0 || got[1] != 2 { + t.Fatalf("keep = %v, want [0 2] (out-of-range/dup dropped)", got) + } + if _, err := parseKeepIndices(`not json`, 3); err == nil { + t.Fatal("expected error for non-JSON") + } +} diff --git a/internal/services/channelbridge/light.go b/internal/services/channelbridge/light.go new file mode 100644 index 0000000..3f49e0e --- /dev/null +++ b/internal/services/channelbridge/light.go @@ -0,0 +1,190 @@ +package channelbridge + +import ( + "context" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// This file holds the channel bridge's minimal-disturbance inbound classification +// for public platforms (DevDocs/design/12 §6 策略一/二/五). Non-owner group/stranger +// messages are judged cheaply before they reach the full agent loop, so a busy +// public group does not fling every message into an expensive LLM turn. The logic +// lives here to keep channelbridge.go under the module size bound; it never reads +// message content to decide identity — identity is already resolved structurally by +// the session/actor layer (a project red line). + +// Quiet-hour defaults, mirroring the backthink service so the bridge and the +// background thinker share one do-not-disturb window (design/12 §6 策略五). +const ( + defaultQuietStartHour = 23 + defaultQuietEndHour = 7 +) + +// lightVerdict is the outcome of a rule-based light judgement before the optional +// injected LightJudge is consulted (design/12 §6 策略二/五). +type lightVerdict int + +const ( + // verdictRespond means the rule signals the message is worth a response. + verdictRespond lightVerdict = iota + // verdictSilent means the rule signals the message should be registered but not + // answered (e.g. during quiet hours with no strong signal). + verdictSilent + // verdictInconclusive means the rules cannot tell; defer to the LightJudge (or + // fail-closed to silence when none is wired). + verdictInconclusive +) + +// LightInput is the set of signals a LightJudge receives to decide response +// worthiness. Identity is already resolved structurally by the session/actor +// layer, so the judge weighs only 值不值得回 — it never decides who the sender is. +type LightInput struct { + Msg adapter.UnifiedMessage + Actor adapter.Actor + SenderName string + InQuietHours bool + QuietStart int + QuietEnd int +} + +// LightDecision is a LightJudge's verdict. +type LightDecision struct { + Respond bool + Reason string +} + +// LightJudge is the optional cost-controlled light-response judge (design/12 §6 +// 策略二/五). The bridge consults it only when the built-in rule judge is +// inconclusive, so a chat message is not repeatedly sent to an expensive LLM. A +// nil LightJudge (the default) keeps the bridge rule-only and fail-closed toward +// silence (不打扰为第一原则). The production wiring does not supply one yet, so the +// seam is ready but not token-consuming until it is deliberately wired. +type LightJudge interface { + Judge(ctx context.Context, in LightInput) (LightDecision, error) +} + +// highValueKeywords is the small term list the rule judge matches to decide a +// message is worth a response even when the sender was not directly addressed +// (设计/12 §6 策略二: 关键词重要度). It is a cheap substring match — never an LLM call, +// and never used to infer identity. +var highValueKeywords = []string{ + "紧急", "救命", "故障", "坏掉", "坏了", "报错", "出错", "崩溃", + "求助", "帮帮我", "help", "urgent", "error", "bug", +} + +// isLowValueEvent reports whether a non-owner message is platform/system noise +// that must be registered but never drives the full agent loop (设计/12 §6 策略一). +// It is only consulted for non-owner messages: a system/synthetic sender is the +// canonical low-value signal. Messages with no meaningful chat text are dropped at +// the bridge head anyway, so this primarily gates system-actor notifications. +func isLowValueEvent(actor adapter.Actor) bool { + return actor.IsSystem() +} + +// hasHighValueKeyword reports whether text carries a term worth responding to even +// when the sender was not directly addressed. Kept tiny so the rule path stays O(1) +// per message. +func hasHighValueKeyword(text string) bool { + lower := strings.ToLower(text) + for _, k := range highValueKeywords { + if strings.Contains(lower, k) { + return true + } + } + return false +} + +// ruleLightJudge decides response-worthiness from cheap rule signals only +// (设计/12 §6 策略二/五). Priority: a direct mention, a mention of the owner's display +// name, a high-value keyword, then a quiet-hours bias toward silence; otherwise the +// verdict is inconclusive (defer to the injected LightJudge, else silence). It +// reads only already-resolved actor metadata and message content — never infers +// identity. +func (b *Bridge) ruleLightJudge(msg adapter.UnifiedMessage, actor adapter.Actor, inQuiet bool) lightVerdict { + if len(msg.Mentions) > 0 { + return verdictRespond + } + if b.owner != nil && b.owner.DisplayName != "" && + strings.Contains(msg.Content, b.owner.DisplayName) { + return verdictRespond + } + if hasHighValueKeyword(msg.Content) { + return verdictRespond + } + if inQuiet { + return verdictSilent + } + return verdictInconclusive +} + +// shouldRespond implements the minimal-disturbance classification for a non-owner +// inbound message (设计/12 §6 策略一/二/五). It returns whether the message should +// drive the full agent loop and reply, or be registered-and-silenced. Owner +// messages never reach here (processFrom gates on !actor.IsOwner), so owner-private +// chat and owner group @/high-value replies are untouched. +func (b *Bridge) shouldRespond(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor) (bool, string) { + // 策略一: platform/system noise → register only, never the full loop. + if isLowValueEvent(actor) { + b.log.Debug("non-owner low-value system event, registered only", + "platform", msg.Platform, "channel", msg.ChannelID, "actor_role", actor.Role) + return false, "low-value system event; registered only" + } + + // 策略二/五: rule-based light judgement, quiet-hours aware. + inQuiet := b.inQuietHours(b.now()) + switch b.ruleLightJudge(msg, actor, inQuiet) { + case verdictRespond: + return true, "rule signal: worth responding" + case verdictSilent: + b.log.Debug("non-owner message silenced (quiet/no signal), registered only", + "platform", msg.Platform, "channel", msg.ChannelID, "quiet", inQuiet, "actor_role", actor.Role) + return false, "rule signal: not worth responding" + } + + // Inconclusive → consult the optional injected light judge. Fail-closed to + // silence on error or absence (不打扰为第一原则). + if b.light != nil { + d, err := b.light.Judge(ctx, LightInput{ + Msg: msg, Actor: actor, SenderName: senderDisplayName(msg, actor), + InQuietHours: inQuiet, QuietStart: b.quietStartHour, QuietEnd: b.quietEndHour, + }) + if err == nil { + if d.Respond { + return true, "light judge: " + d.Reason + } + return false, "light judge: " + d.Reason + } + b.log.Warn("light judge failed; silencing", "error", err) + } + b.log.Debug("non-owner message has no rule signal, silenced", + "platform", msg.Platform, "channel", msg.ChannelID, "actor_role", actor.Role) + return false, "no rule signal; silenced (fail-closed)" +} + +// registerSilenced persists a silenced non-owner inbound turn into its isolated +// session's platform log so the adapter log page still renders the non-owner +// stream (design/16 §4.6 §4), then stops — no loop, no reply. It is the 登记 side +// of the minimal-disturbance path. +func (b *Bridge) registerSilenced(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session) { + b.persistIsolated(ctx, history.RoleUser, msg.Content, deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor)) + b.log.Debug("non-owner message silenced, registered only", + "platform", msg.Platform, "channel", msg.ChannelID, "session", sess.SessionID) +} + +// inQuietHours reports whether at falls inside the configured do-not-disturb +// window (shared semantics with backthink: equal non-zero hours disable it, and a +// window that wraps midnight is handled). It is a pure method on Bridge so tests +// drive it deterministically via the Now source. +func (b *Bridge) inQuietHours(at time.Time) bool { + h := at.Hour() + start, end := b.quietStartHour, b.quietEndHour + if start <= end { + return h >= start && h < end + } + return h >= start || h < end +} diff --git a/internal/services/channelbridge/light_test.go b/internal/services/channelbridge/light_test.go new file mode 100644 index 0000000..70ccfaf --- /dev/null +++ b/internal/services/channelbridge/light_test.go @@ -0,0 +1,285 @@ +package channelbridge + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// stubLightJudge is a configurable LightJudge recording how often it was consulted. +type stubLightJudge struct { + resp LightDecision + err error + calls int +} + +func (s *stubLightJudge) Judge(context.Context, LightInput) (LightDecision, error) { + s.calls++ + return s.resp, s.err +} + +// nonOwnerGroupMsg returns a third-party group message from a member sender, with +// the given mentions. Content is left untouched so tests control the signal. +func nonOwnerGroupMsg(content string, mentions ...adapter.Mention) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup, + GroupID: "-20", GroupName: "Gang", SenderID: "9", SenderName: "张三", + Sender: adapter.Actor{Role: adapter.RoleMember}, Mentions: mentions, + Content: content, ContentType: adapter.ContentText, + } +} + +// nonOwnerSession is the restricted (group / stranger) session binding used by the +// non-owner tests: read-only memory, no owner. +func nonOwnerSession() session.Session { + return session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false} +} + +func TestNonOwnerSilencedRegisteredOnly(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "不该回"}} + sess := &stubSessions{sess: nonOwnerSession()} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // A non-owner message with no @/keyword/name signal is silenced (设计/12 §6 策略二): + // it is registered to the isolated session log but never drives the loop. + b.process(context.Background(), nonOwnerGroupMsg("hello all")) + + if chat.sessionID != "" || len(chat.messages) != 0 { + t.Fatalf("loop ran for a no-signal non-owner message: session=%q messages=%+v", chat.sessionID, chat.messages) + } + recs := hist.records() + if len(recs) != 1 { + t.Fatalf("records = %d, want 1 (inbound registered, no assistant reply)", len(recs)) + } + if recs[0].SessionID != "sess-g" || recs[0].Role != history.RoleUser || recs[0].Content != "hello all" { + t.Fatalf("registered record = %+v, want isolated sess-g user 'hello all'", recs[0]) + } +} + +func TestNonOwnerMentionResponds(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "收到"}} + sess := &stubSessions{sess: nonOwnerSession()} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // An addressed (mentioned) non-owner message is worth a response (策略二), so it + // drives the full loop against the isolated group session. + b.process(context.Background(), nonOwnerGroupMsg("hello", adapter.Mention{UserID: "bot"})) + + if chat.sessionID != "sess-g" { + t.Fatalf("loop session = %q, want isolated sess-g", chat.sessionID) + } + // Restricted group context → the third-party boundary system message plus the + // user message (design/16 §4.6 §1), mirroring TestGroupMessageThirdPartySenderNameAndBoundary. + if len(chat.messages) != 2 || chat.messages[1].Role != model.RoleUser { + t.Fatalf("loop messages = %+v, want boundary + user", chat.messages) + } + if chat.messages[1].Content[0].Text != "[第三方 · 张三] hello" { + t.Fatalf("loop user text = %q, want [第三方 · 张三] hello", chat.messages[1].Content[0].Text) + } +} + +func TestNonOwnerHighValueKeywordResponds(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "我看看"}} + sess := &stubSessions{sess: nonOwnerSession()} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // A high-value keyword (求助) is worth a response even without an @ (策略二). + b.process(context.Background(), nonOwnerGroupMsg("有人在吗，紧急求助")) + if chat.sessionID != "sess-g" { + t.Fatalf("loop session = %q, want isolated sess-g for a high-value keyword", chat.sessionID) + } +} + +func TestNonOwnerQuietHoursSilenced(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "不该回"}} + sess := &stubSessions{sess: nonOwnerSession()} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + QuietStartHour: 23, QuietEndHour: 7, + Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // 策略五: during quiet hours a no-signal non-owner message is silenced — it is + // registered but not answered. + b.process(context.Background(), nonOwnerGroupMsg("hello all")) + + if chat.sessionID != "" || len(chat.messages) != 0 { + t.Fatalf("loop ran during quiet hours for a no-signal message: session=%q messages=%+v", chat.sessionID, chat.messages) + } + if len(hist.records()) != 1 { + t.Fatalf("records = %d, want 1 (registered, no reply)", len(hist.records())) + } +} + +func TestNonOwnerQuietHoursMentionResponds(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "在的"}} + sess := &stubSessions{sess: nonOwnerSession()} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + QuietStartHour: 23, QuietEndHour: 7, + Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // 策略五 is not a hard silence: a directly-addressed message during quiet hours + // still gets a response. + b.process(context.Background(), nonOwnerGroupMsg("在吗", adapter.Mention{UserID: "bot"})) + if chat.sessionID != "sess-g" { + t.Fatalf("loop session = %q, want sess-g (mention should break quiet-hours silence)", chat.sessionID) + } +} + +func TestOwnerUnaffectedInQuietHours(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "主人好"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + ad := &fakeAdapter{name: "telegram"} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: &captureHistory{}, + QuietStartHour: 23, QuietEndHour: 7, + Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + b.Add(ad) + + // An owner message is never gated, even during quiet hours (策略五 only ever + // governs non-owner traffic): owner-private chat still routes to the main web + // session and replies. + b.process(context.Background(), directMessage()) + + if chat.sessionID != history.MainSessionID { + t.Fatalf("owner loop session = %q, want %q", chat.sessionID, history.MainSessionID) + } + if len(ad.sends) != 1 || ad.sends[0].Content != "主人好" { + t.Fatalf("owner reply = %+v, want '主人好'", ad.sends) + } +} + +func TestLightJudgeConsultedWhenInconclusive(t *testing.T) { + t.Parallel() + + // A no-signal daytime message is inconclusive to the rules, so the injected + // LightJudge gets a chance to decide (设计/12 §6 策略二: 规则不确定才轻 LLM). + judge := &stubLightJudge{resp: LightDecision{Respond: true, Reason: "值得回"}} + chat := &stubChatter{res: &loop.Result{Text: "嗯"}} + sess := &stubSessions{sess: nonOwnerSession()} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + Light: judge, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + b.process(context.Background(), nonOwnerGroupMsg("今天天气不错")) + if judge.calls != 1 { + t.Fatalf("light judge calls = %d, want 1 (inconclusive rules escalated)", judge.calls) + } + if chat.sessionID != "sess-g" { + t.Fatalf("loop session = %q, want sess-g (judge said respond)", chat.sessionID) + } +} + +func TestLightJudgeSilenceIsRespected(t *testing.T) { + t.Parallel() + + judge := &stubLightJudge{resp: LightDecision{Respond: false, Reason: "不值得"}} + chat := &stubChatter{res: &loop.Result{Text: "不该回"}} + sess := &stubSessions{sess: nonOwnerSession()} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + Light: judge, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + b.process(context.Background(), nonOwnerGroupMsg("今天天气不错")) + if judge.calls != 1 { + t.Fatalf("light judge calls = %d, want 1", judge.calls) + } + if chat.sessionID != "" { + t.Fatalf("loop ran after the judge said silence: session=%q", chat.sessionID) + } + if len(hist.records()) != 1 { + t.Fatalf("records = %d, want 1 (registered, no reply)", len(hist.records())) + } +} + +func TestLightJudgeErrorFailsClosedToSilence(t *testing.T) { + t.Parallel() + + judge := &stubLightJudge{err: context.DeadlineExceeded} + chat := &stubChatter{res: &loop.Result{Text: "不该回"}} + sess := &stubSessions{sess: nonOwnerSession()} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + Light: judge, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // A failing light judge is fail-closed toward silence (不打扰为第一原则). + b.process(context.Background(), nonOwnerGroupMsg("今天天气不错")) + if judge.calls != 1 { + t.Fatalf("light judge calls = %d, want 1", judge.calls) + } + if chat.sessionID != "" || len(chat.messages) != 0 { + t.Fatalf("loop ran after judge error: session=%q messages=%+v", chat.sessionID, chat.messages) + } + if len(hist.records()) != 1 { + t.Fatalf("records = %d, want 1 (registered, no reply)", len(hist.records())) + } +}