feat(backthink): 事件驱动主动消息绕过PushGate——IoT等事件该发就发(heartbeat仍受60min节流)

- 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

@@ -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)
 	}

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)
+	}
+}

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
+}

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

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)
+	}
+}

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}
 	}
 }