From d09b25071de652471941b1f9534fe3fc562b5249 Mon Sep 17 00:00:00 2001 From: Aska Date: Tue, 8 Sep 2026 13:46:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(backthink):=20=E4=BA=8B=E4=BB=B6=E9=A9=B1?= =?UTF-8?q?=E5=8A=A8=E4=B8=BB=E5=8A=A8=E6=B6=88=E6=81=AF=E7=BB=95=E8=BF=87?= =?UTF-8?q?PushGate=E2=80=94=E2=80=94IoT=E7=AD=89=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E8=AF=A5=E5=8F=91=E5=B0=B1=E5=8F=91(heartbeat=E4=BB=8D?= =?UTF-8?q?=E5=8F=9760min=E8=8A=82=E6=B5=81)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - trigger.go: 新增 TriggerKind(TriggerHeartbeat/TriggerEvent) + Trigger.Kind 标记触发来源 - push.go: PushRequest.TriggerKind 分流——event 跳过 gate.Allow/RecordPush(该发就发)，heartbeat 仍受 60min≤1/冷却/分类限频；PushResult.Bypassed 标记 + detail=event_bypass 区分 - backthink.go: think() 传 TriggerKind + 过同事件去重门 - event_dedup.go(新): 同事件(Source+Summary)去重防刷，无 60min 节流；不同事件/心跳后重置链 - 测试: event 连续 3 次全推 bypass / heartbeat 第2次 rate_limited / event 不消耗心跳限 / 同事件去重 - build/vet/test 全绿 + windows/darwin 交叉编译通过 --- internal/services/backthink/backthink.go | 33 ++++++--- internal/services/backthink/backthink_test.go | 64 ++++++++++++++++ internal/services/backthink/event_dedup.go | 31 ++++++++ internal/services/backthink/push.go | 43 +++++++++-- internal/services/backthink/push_test.go | 73 +++++++++++++++++++ internal/services/backthink/trigger.go | 44 +++++++++-- 6 files changed, 266 insertions(+), 22 deletions(-) create mode 100644 internal/services/backthink/event_dedup.go diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index 82bc2d9..cc8b543 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -12,10 +12,12 @@ // Each pass produces a conclusion, then consults Decide to judge whether the user // should be interrupted at all (event importance / time-of-day / user state such // as heart rate, sleep, work, busy / boredom). When it is worth interrupting the -// push goes to the active client first and falls back to the IM bridge, bounded -// by the proactive-push rate limit of DevDocs/design/11 §6. Every pass is -// persisted to SQLite via Store so the M9/M10 monitor panel can render the -// background-thinking page (design/12 §4). +// push goes to the active client first and falls back to the IM bridge. The +// proactive-push rate gate of DevDocs/design/11 §6 governs the periodic heartbeat +// only; an event-driven push bypasses it (owner mandate: "该发就发") subject to +// same-event dedup (see event_dedup.go). Every pass is persisted to SQLite via +// Store so the M9/M10 monitor panel can render the background-thinking page +// (design/12 §4). // // The package is concurrency-safe: a mutex guards the lifecycle state, the // injected store/pusher/thinker are required to be concurrency-safe by contract, @@ -110,6 +112,11 @@ type Backthink struct { cancel context.CancelFunc subs []func() wg sync.WaitGroup + + // eventMu guards the same-event dedup guard (event_dedup.go). It is separate + // from the lifecycle lock so a dedup decision never blocks a sink swap. + eventMu sync.Mutex + lastEventKey string } // New builds a Backthink from cfg and deps, applying defaults and validating the @@ -356,10 +363,18 @@ func (b *Backthink) think(parent context.Context, tr Trigger) { pushRes := PushResult{Route: RouteNone, Detail: b.interruptReason(push, at, state, tr.Import, lastActivity, hasActivity)} if shouldPush { - pushRes = b.currentPusher().Push(cctx, PushRequest{ - Message: b.buildProactiveMessage(message, channelHint), - ChannelHint: channelHint, - }) + // Same-event dedup: a repeating identical event is suppressed after the + // first push (防刷), while a genuinely new event still goes out. Heartbeats + // are never deduped (they reset the chain) and stay gate-governed. + if b.eventRedundant(tr) { + pushRes = PushResult{Route: RouteNone, Detail: "duplicate event push suppressed"} + } else { + pushRes = b.currentPusher().Push(cctx, PushRequest{ + Message: b.buildProactiveMessage(message, channelHint), + ChannelHint: channelHint, + TriggerKind: tr.Kind, + }) + } } b.record(Record{ @@ -371,7 +386,7 @@ func (b *Backthink) think(parent context.Context, tr Trigger) { }) if shouldPush { - b.log.Info("backthink proactive push", "source", tr.Source, "route", pushRes.Route, "pushed", pushRes.Pushed, "reason", pushRes.Detail) + 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) } diff --git a/internal/services/backthink/backthink_test.go b/internal/services/backthink/backthink_test.go index b9d137f..fbf0b8a 100644 --- a/internal/services/backthink/backthink_test.go +++ b/internal/services/backthink/backthink_test.go @@ -3,6 +3,7 @@ package backthink import ( "context" "errors" + "strings" "testing" "time" @@ -308,3 +309,66 @@ func TestBackthink_Stop_UnsubscribesBus(t *testing.T) { t.Fatalf("thinker calls grew after stop (bus not unsubscribed): %d -> %d", before, got) } } + +// TestBackthink_EventPush_DedupSameEvent verifies the same-event dedup guard: a +// repeating identical event (same source + summary) is suppressed after the first +// push (防刷), while a genuinely different event still goes out. The duplicate is +// recorded as suppressed (route none), not delivered. +func TestBackthink_EventPush_DedupSameEvent(t *testing.T) { + t.Parallel() + thinker := &mockThinker{text: "washing finished, clothes ready"} + pusher := &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}} + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: thinker, + State: &mockState{state: UserState{}}, + Pusher: pusher, + Now: func() time.Time { return dayAt() }, + }) + + ev := triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: iot.DeviceUpdate{ + Device: iot.Device{ID: "washer", Type: "sensor"}, Action: "finished", + }}) + + b.think(context.Background(), ev) // first: pushes + b.think(context.Background(), ev) // second: same event -> suppressed + b.think(context.Background(), ev) // third: same event -> suppressed + + if got := pusher.count(); got != 1 { + t.Fatalf("pusher calls = %d, want 1 (repeating same event deduped)", got) + } + recs, err := b.List(context.Background(), 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) < 3 { + t.Fatalf("len(recs) = %d, want at least 3 (one per think)", len(recs)) + } + // Newest first: the two most recent records are the suppressed duplicates. + for i := 0; i < 2; i++ { + if recs[i].PushResult.Pushed || recs[i].PushResult.Route != RouteNone { + t.Fatalf("recs[%d] = %+v, want suppressed duplicate", i, recs[i].PushResult) + } + if !strings.Contains(recs[i].PushResult.Detail, "duplicate") { + t.Fatalf("recs[%d].Detail = %q, want it to mention duplicate", i, recs[i].PushResult.Detail) + } + } + // The third-most-recent record is the first (delivered) push. + if !recs[2].PushResult.Pushed || recs[2].PushResult.Route != RouteActiveClient { + t.Fatalf("recs[2] = %+v, want the first delivered push", recs[2].PushResult) + } + + // A genuinely different event resets the chain and goes out. + ev2 := triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: iot.DeviceUpdate{ + Device: iot.Device{ID: "dryer", Type: "sensor"}, Action: "finished", + }}) + b.think(context.Background(), ev2) + if got := pusher.count(); got != 2 { + t.Fatalf("pusher calls after a different event = %d, want 2", got) + } + // After the different event, the original identical event is no longer a + // consecutive duplicate, so it pushes again — the guard only dedups consecutively. + b.think(context.Background(), ev) + if got := pusher.count(); got != 3 { + t.Fatalf("pusher calls after an intervening different event = %d, want 3", got) + } +} diff --git a/internal/services/backthink/event_dedup.go b/internal/services/backthink/event_dedup.go new file mode 100644 index 0000000..21d76a7 --- /dev/null +++ b/internal/services/backthink/event_dedup.go @@ -0,0 +1,31 @@ +package backthink + +// eventRedundant reports whether tr is a duplicate of the immediately preceding +// event-driven push (same source + summary), so a repeating device/reminder event +// does not spam the owner (防刷). It is the one guard that still applies to an +// event-driven push while the PushGate itself is bypassed for events (owner +// mandate: "该发就发" — a genuine event goes out, only a true duplicate is held). +// +// It atomically records tr as the latest event key when the push proceeds, and +// clears the key on any non-event push (heartbeat) so a later repeat of a prior +// event is not spuriously suppressed by an intervening reflection. +// +// The caller runs it before handing the push to the Pusher; a true duplicate is +// recorded as suppressed (RouteNone) rather than delivered. +func (b *Backthink) eventRedundant(tr Trigger) bool { + if tr.Kind != TriggerEvent { + // A non-event push (heartbeat) breaks the same-event chain. + b.eventMu.Lock() + b.lastEventKey = "" + b.eventMu.Unlock() + return false + } + key := tr.Source + "\x00" + tr.Summary + b.eventMu.Lock() + defer b.eventMu.Unlock() + if b.lastEventKey == key { + return true + } + b.lastEventKey = key + return false +} diff --git a/internal/services/backthink/push.go b/internal/services/backthink/push.go index bdf68e4..d3b5398 100644 --- a/internal/services/backthink/push.go +++ b/internal/services/backthink/push.go @@ -50,6 +50,10 @@ type PushResult struct { Pushed bool // Route is the delivery route taken. Route Route + // Bypassed reports whether this push was event-driven and skipped the PushGate + // (a real event, "该发就发"). A heartbeat push is gate-governed and never + // bypassed. + Bypassed bool // Detail is a short human-readable note (e.g. client count, decision reason). Detail string // Err is the underlying error text when delivery failed, or that of the @@ -60,9 +64,9 @@ type PushResult struct { // String returns a concise summary for logs. func (p PushResult) String() string { if p.Err != "" { - return fmt.Sprintf("pushed=%v route=%s detail=%q err=%q", p.Pushed, p.Route, p.Detail, p.Err) + return fmt.Sprintf("pushed=%v route=%s bypass=%v detail=%q err=%q", p.Pushed, p.Route, p.Bypassed, p.Detail, p.Err) } - return fmt.Sprintf("pushed=%v route=%s detail=%q", p.Pushed, p.Route, p.Detail) + return fmt.Sprintf("pushed=%v route=%s bypass=%v detail=%q", p.Pushed, p.Route, p.Bypassed, p.Detail) } // ActiveClient delivers a proactive message to every currently active web/native @@ -106,6 +110,9 @@ type PushRequest struct { Message adapter.UnifiedMessage // ChannelHint is the requested target channel, "" when she did not choose one. ChannelHint string + // TriggerKind records what woke this pass. An event-driven push bypasses the + // PushGate (owner mandate: "该发就发"); a heartbeat stays gate-governed. + TriggerKind TriggerKind } // Pusher delivers a proactive message to ONE target channel. It is bounded by the @@ -148,7 +155,12 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult { return PushResult{Pushed: false, Route: RouteNone, Err: err.Error()} } kind := classifyPush(req.Message.Content) - if p.gate != nil && !p.gate.Allow(ctx, kind) { + // A real event-driven push bypasses the PushGate: it should go out as-is ("该发就发"), + // independent of the heartbeat throttle (60min≤1 / cooldown / per-flavor window). + // Only the periodic heartbeat stays gate-governed, so the owner is never spammed + // by an empty reflection. Same-event dedup is applied upstream in think(). + eventDriven := req.TriggerKind == TriggerEvent + if p.gate != nil && !eventDriven && !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"} } @@ -174,12 +186,18 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult { res = p.pushActive(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 { + // Only a real delivery advances the cooldown/per-kind windows (and only for a + // gate-governed push), so an attempted-but-failed push does not suppress a later + // genuine one, and an event push never contributes to the heartbeat throttle. + if res.Pushed && p.gate != nil && !eventDriven { p.gate.RecordPush(kind) } - return annotateHint(res, req.ChannelHint) + res = annotateHint(res, req.ChannelHint) + if eventDriven { + res.Bypassed = true + res = annotateBypass(res) + } + return res } // pushActive delivers only to the active web/native client. @@ -246,6 +264,17 @@ func annotateHint(res PushResult, hint string) PushResult { return res } +// annotateBypass tags an event-driven push result so the record/panel shows it +// skipped the PushGate, distinct from a heartbeat push that was rate-limited. +func annotateBypass(res PushResult) PushResult { + if res.Detail == "" { + res.Detail = "event_bypass" + } else { + res.Detail = res.Detail + ", event_bypass" + } + return res +} + // rateGate is the production PushGate: a sliding-window counter over // session.RateLimiter (design/11 §6 anti-spam rule) combined with a post-push // cooldown and a per-flavor (问候/提醒/分享/关心) window so the owner is never diff --git a/internal/services/backthink/push_test.go b/internal/services/backthink/push_test.go index 20fa7d8..7f46605 100644 --- a/internal/services/backthink/push_test.go +++ b/internal/services/backthink/push_test.go @@ -237,3 +237,76 @@ func TestRateGate_KindWindowLimitsSameFlavor(t *testing.T) { t.Fatal("same flavor after the kind window should be allowed") } } + +// TestPusher_EventBypassesGate verifies an event-driven push skips the PushGate: +// with a limit of 1 push per window, three consecutive event pushes are ALL +// delivered (owner mandate "该发就发"), each marked Bypassed and each advancing +// no cooldown/kind window. +func TestPusher_EventBypassesGate(t *testing.T) { + t.Parallel() + active := &mockActive{n: 1} + p := newPusher(active, nil, newRateGate(1, time.Minute, 0, 0, discardLog()), discardLog()) + req := PushRequest{Message: adapter.UnifiedMessage{Content: "洗衣机洗好了"}, TriggerKind: TriggerEvent} + + first := p.Push(context.Background(), req) + second := p.Push(context.Background(), req) + third := p.Push(context.Background(), req) + + if !first.Pushed || !first.Bypassed { + t.Fatalf("first event push = %+v, want delivered + Bypassed", first) + } + if !second.Pushed || !second.Bypassed { + t.Fatalf("second event push = %+v, want delivered + Bypassed (gate bypassed)", second) + } + if !third.Pushed || !third.Bypassed { + t.Fatalf("third event push = %+v, want delivered + Bypassed (gate bypassed)", third) + } + if active.calls != 3 { + t.Fatalf("active calls = %d, want 3 (events are not rate-capped)", active.calls) + } +} + +// TestPusher_HeartbeatStillRateLimited verifies a heartbeat-driven push stays +// gate-governed: with a limit of 1 push per window, the first passes and the +// second is refused as rate_limited, never Bypassed. +func TestPusher_HeartbeatStillRateLimited(t *testing.T) { + t.Parallel() + active := &mockActive{n: 1} + p := newPusher(active, nil, newRateGate(1, time.Minute, 0, 0, discardLog()), discardLog()) + req := PushRequest{Message: adapter.UnifiedMessage{Content: "想跟你说声晚安"}, TriggerKind: TriggerHeartbeat} + + first := p.Push(context.Background(), req) + second := p.Push(context.Background(), req) + + if !first.Pushed || first.Bypassed { + t.Fatalf("first heartbeat push = %+v, want delivered non-bypassed", first) + } + if second.Pushed || second.Route != RouteRateLimited || second.Bypassed { + t.Fatalf("second heartbeat push = %+v, want rate_limited non-bypassed", second) + } + if active.calls != 1 { + t.Fatalf("active calls = %d, want 1 (heartbeat still gated)", active.calls) + } +} + +// TestPusher_EventBypass_DoesNotAdvanceGate verifies an event-driven push does not +// consume the heartbeat limit: after an event push, a following heartbeat push +// still gets its own full allowance. +func TestPusher_EventBypass_DoesNotAdvanceGate(t *testing.T) { + t.Parallel() + active := &mockActive{n: 1} + p := newPusher(active, nil, newRateGate(1, time.Minute, 0, 0, discardLog()), discardLog()) + + ev := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{Content: "洗衣机洗好了"}, TriggerKind: TriggerEvent}) + hb := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{Content: "想跟你说声晚安"}, TriggerKind: TriggerHeartbeat}) + + if !ev.Pushed || !ev.Bypassed { + t.Fatalf("event push = %+v, want delivered + Bypassed", ev) + } + if !hb.Pushed || hb.Bypassed { + t.Fatalf("heartbeat push after an event = %+v, want delivered non-bypassed (event did not consume the gate)", hb) + } + if active.calls != 2 { + t.Fatalf("active calls = %d, want 2", active.calls) + } +} diff --git a/internal/services/backthink/trigger.go b/internal/services/backthink/trigger.go index 2eb6828..b62390c 100644 --- a/internal/services/backthink/trigger.go +++ b/internal/services/backthink/trigger.go @@ -7,6 +7,33 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" ) +// TriggerKind classifies what woke a thinking pass, so the push path can apply the +// right policy: a heartbeat is gate-governed (60min≤1 / cooldown / per-flavor), an +// event is a real signal the owner wants to hear about and bypasses the rate gate. +type TriggerKind int + +const ( + // TriggerHeartbeat is the periodic autonomous reflection. Its proactive push is + // throttled by the PushGate so the owner is never spammed by empty reflections. + TriggerHeartbeat TriggerKind = iota + // TriggerEvent is a real event (IoT device change, reminder, owner-relevant + // system event). Its push bypasses the PushGate — "该发就发" — subject only to + // same-event dedup. + TriggerEvent +) + +// String returns a stable label for logs and records. +func (k TriggerKind) String() string { + switch k { + case TriggerEvent: + return "event" + case TriggerHeartbeat: + return "heartbeat" + default: + return fmt.Sprintf("triggerkind(%d)", int(k)) + } +} + // Bus subjects the service subscribes to for event-triggered thinking. They are // the contract defined by design/12 §1: an IoT state change, a reminder firing, // or a generic system event. @@ -34,17 +61,21 @@ type Trigger struct { Source string Summary string Import Importance + // Kind records whether this pass was the periodic heartbeat or a real event, so + // the push path can bypass the rate gate for a genuine event (owner mandate). + Kind TriggerKind } // heartbeatTrigger is the periodic, low-importance trigger. Low importance means // the pass rarely interrupts the user, honouring "默认不发送 chat 类型消息". func heartbeatTrigger() Trigger { - return Trigger{Source: "heartbeat", Summary: "periodic autonomous reflection", Import: ImportanceLow} + return Trigger{Source: "heartbeat", Summary: "periodic autonomous reflection", Import: ImportanceLow, Kind: TriggerHeartbeat} } // triggerForEvent maps a bus event to a Trigger. An IoT device change is treated // as normal importance, a reminder firing as high importance, and a generic -// system event as normal importance. +// system event as normal importance. Every event is tagged TriggerEvent so its +// proactive push bypasses the heartbeat PushGate. func triggerForEvent(e bus.Event) Trigger { switch e.Subject { case iot.DeviceUpdateSubject: @@ -53,13 +84,14 @@ func triggerForEvent(e bus.Event) Trigger { Source: e.Subject, Summary: fmt.Sprintf("device %s (%s) %s", d.Device.ID, d.Device.Type, d.Action), Import: ImportanceNormal, + Kind: TriggerEvent, } } - return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal} + return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal, Kind: TriggerEvent} case SubjectReminderTrigger: - return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceHigh} + return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceHigh, Kind: TriggerEvent} case SubjectSystemEvent: - tr := Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal} + tr := Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal, Kind: TriggerEvent} if sy, ok := e.Payload.(SystemEvent); ok { if sy.Summary != "" { tr.Summary = sy.Summary @@ -68,6 +100,6 @@ func triggerForEvent(e bus.Event) Trigger { } return tr default: - return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal} + return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal, Kind: TriggerEvent} } }