feat(backthink): 后台自主思考（heartbeat+事件触发 + 打扰决策 fail-closed + 主动推送活跃→IM 兜底 + 限制 + 思考记录）

internal/services/backthink/backthink.go

@@ -0,0 +1,438 @@
+// Package backthink implements the M8 autonomous background-think service
+// (DevDocs/design/12-后台思考与主动.md §0-§3).
+//
+// It runs unattended thinking passes on two triggers:
+//
+//   - A periodic heartbeat (config.Background.IntervalSec) that asks the injected
+//     runtime to reflect and is allowed to call tools (IoT, calendar, reminders)
+//     but must not emit chat-style messages by default.
+//   - Asynchronous events on the in-process bus (iot.device_update,
+//     reminder.trigger, system.event) that wake a thinking pass.
+//
+// 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).
+//
+// The package is concurrency-safe: a mutex guards the lifecycle state, the
+// injected store/pusher/thinker are required to be concurrency-safe by contract,
+// and the internal publish path never waits on a slow consumer.
+package backthink
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"sync"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
+)
+
+// Defaults applied by normalize when a Config field is left at its zero value.
+// They are compile-time constants so no magic number appears in the run path.
+const (
+	// defaultIntervalSec is the heartbeat period in seconds when unused.
+	defaultIntervalSec = 300
+	// defaultQuietStartHour and defaultQuietEndHour bound the do-not-disturb
+	// window in local clock hours (00:00-07:00 vs 23:00-07:00: the window may
+	// wrap midnight; two equal non-zero hours disable it).
+	defaultQuietStartHour = 23
+	defaultQuietEndHour   = 7
+	// defaultMaxPushWindow caps proactive pushes within defaultPushWindow to
+	// honor the anti-spam rule of design/11 §6 (QQ wind control, iLink window).
+	defaultMaxPushWindow = 3
+	defaultPushWindow    = 10 * time.Minute
+	// defaultThinkTimeout bounds a single unattended thought so it can never
+	// wedge a worker goroutine.
+	defaultThinkTimeout = 30 * time.Second
+	// defaultSessionID scopes a think when the caller supplies none. The
+	// background pass owns no conversation, so a stable synthetic id is used.
+	defaultSessionID = "sess-backthink"
+)
+
+// Sentinel errors returned by this package. Use errors.Is to test for them; they
+// are returned directly or wrapped with %w so the underlying cause stays intact.
+var (
+	// ErrNilDB is returned when Backthink or Store is built with a nil database.
+	ErrNilDB = errors.New("backthink: nil database handle")
+	// ErrNoThinker is returned when New is given no Thinker.
+	ErrNoThinker = errors.New("backthink: nil thinker")
+	// ErrNoLoop is returned by LoopThinker when it wraps a nil loop.
+	ErrNoLoop = errors.New("backthink: nil loop")
+)
+
+// Thinker runs one unattended thinking pass over the injected runtime. It is the
+// seam between the service and the loop/agent: the production wiring hands back a
+// LoopThinker over a *loop.Loop, while tests inject a mock. Implementations may
+// call tools (IoT, calendar, reminders) and must be concurrency-safe.
+type Thinker interface {
+	// Think invokes one autonomous thinking pass for the given session/trigger
+	// and returns the concluded text.
+	Think(ctx context.Context, sessionID, trigger string) (string, error)
+}
+
+// Config carries the autonomous-think tuning knobs. Zero values pick safe
+// defaults via normalize, so a partially filled Config is still usable.
+type Config struct {
+	// Interval is the heartbeat period. When zero it is derived from
+	// IntervalSec (or its default).
+	Interval time.Duration
+	// IntervalSec is the heartbeat period in seconds, mapped from
+	// config.BackgroundConfig.IntervalSec. Ignored when Interval is set.
+	IntervalSec int
+	// QuietStartHour and QuietEndHour bound the do-not-disturb window in local
+	// clock hours. Equal non-zero hours disable the window.
+	QuietStartHour int
+	QuietEndHour   int
+	// MaxPushWindow bounds proactive pushes within PushWindow. Zero uses the
+	// default. Non-positive PushWindow or MaxPushWindow disables the limit.
+	MaxPushWindow int
+	// PushWindow is the sliding window over which MaxPushWindow is enforced.
+	PushWindow time.Duration
+	// ThinkTimeout bounds a single thinking pass. Zero uses the default.
+	ThinkTimeout time.Duration
+}
+
+// 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.
+func ConfigFromBackground(bg config.BackgroundConfig) Config {
+	return Config{IntervalSec: bg.IntervalSec}
+}
+
+// Deps is the injected dependency set for a Backthink. Only Thinker and DB are
+// required; the rest degrade gracefully (a nil Bus disables event triggers, a nil
+// State yields a neutral user state, and a nil Pusher is built from Active/IM).
+type Deps struct {
+	// Bus is the event-trigger source. Nil disables event-driven thinking.
+	Bus bus.Bus
+	// Thinker runs one unattended thinking pass. Required.
+	Thinker Thinker
+	// State supplies the user's current condition for the interrupt decision.
+	// Nil yields a neutral state.
+	State StateProvider
+	// Pusher delivers a proactive message. When nil it is built from Active (the
+	// active-client sink) and IM (the IM bridge sink) plus Gate.
+	Pusher Pusher
+	// Active is the active web/native client sink, preferred for pushes.
+	Active ActiveClient
+	// IM is the IM bridge sink, used as the fallback when no active client is up.
+	IM IMAdapter
+	// Gate is the proactive-push rate limiter. Nil builds a default from Config.
+	Gate PushGate
+	// DB is the SQLite handle for the think store. Required.
+	DB *sql.DB
+	// SessionID scopes a think. Empty uses defaultSessionID.
+	SessionID string
+	// Now is the wall-clock source for decision/record timestamps. Nil uses
+	// time.Now, which makes time-dependent behaviour injectable in tests.
+	Now func() time.Time
+	// Log is the base logger, tagged with source "services/backthink".
+	Log *slog.Logger
+}
+
+// Backthink is the autonomous background-think service. Build it with New, start
+// the heartbeat and subscriptions with Start, and stop them with Stop.
+type Backthink struct {
+	cfg       Config
+	deps      Deps
+	log       *slog.Logger
+	store     *Store
+	pusher    Pusher
+	now       func() time.Time
+	sessionID string
+
+	mu      sync.Mutex
+	running bool
+	cancel  context.CancelFunc
+	subs    []func()
+	wg      sync.WaitGroup
+}
+
+// New builds a Backthink from cfg and deps, applying defaults and validating the
+// required dependencies. It creates the SQLite think store (via Store) up front
+// and resolves the push path (an injected Pusher, or a default built from
+// Active/IM and the rate Gate). It returns ErrNilDB when DB is nil and
+// ErrNoThinker when Thinker is nil.
+func New(cfg Config, deps Deps) (*Backthink, error) {
+	if deps.Thinker == nil {
+		return nil, ErrNoThinker
+	}
+	if deps.DB == nil {
+		return nil, ErrNilDB
+	}
+	cfg = normalize(cfg)
+
+	log := logger.WithSource(baseLog(deps.Log), "services/backthink")
+	store, err := NewStore(context.Background(), deps.DB, log)
+	if err != nil {
+		return nil, fmt.Errorf("backthink: new: %w", err)
+	}
+
+	pusher := deps.Pusher
+	if pusher == nil {
+		gate := deps.Gate
+		if gate == nil {
+			gate = newRateGate(cfg.MaxPushWindow, cfg.PushWindow, log)
+		}
+		pusher = newPusher(deps.Active, deps.IM, gate, log)
+	}
+
+	now := deps.Now
+	if now == nil {
+		now = time.Now
+	}
+	sessionID := deps.SessionID
+	if sessionID == "" {
+		sessionID = defaultSessionID
+	}
+
+	return &Backthink{
+		cfg:       cfg,
+		deps:      deps,
+		log:       log,
+		store:     store,
+		pusher:    pusher,
+		now:       now,
+		sessionID: sessionID,
+	}, nil
+}
+
+// Start begins the background work: the heartbeat ticker and, when a Bus is
+// configured, the event subscriptions. It is idempotent and safe to call twice; a
+// second call is a no-op. The provided ctx (may be nil) is the parent of all
+// internal work: cancelling it stops the service exactly like Stop.
+func (b *Backthink) Start(ctx context.Context) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	b.mu.Lock()
+	if b.running {
+		b.mu.Unlock()
+		return
+	}
+	b.running = true
+	cctx, cancel := context.WithCancel(ctx)
+	b.cancel = cancel
+	b.wg.Add(1)
+	go b.tickLoop(cctx)
+	b.mu.Unlock()
+
+	if b.deps.Bus != nil {
+		b.subscribe(cctx)
+	}
+
+	b.log.Info("backthink started",
+		"interval", b.cfg.Interval.String(),
+		"quiet_hours", quietHoursLabel(b.cfg.QuietStartHour, b.cfg.QuietEndHour),
+		"event_triggers", b.deps.Bus != nil,
+		"session_id", b.sessionID)
+}
+
+// Stop halts the heartbeat and unsubscribes every bus subscription, then waits
+// for the in-flight thinking pass (bounded by ThinkTimeout) to drain. It is
+// idempotent; a second call returns nil immediately. When ctx is cancelled before
+// the drain completes it returns a wrapped context error rather than leaking the
+// goroutine.
+func (b *Backthink) Stop(ctx context.Context) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	b.mu.Lock()
+	if !b.running {
+		b.mu.Unlock()
+		return nil
+	}
+	b.running = false
+	cancel := b.cancel
+	subs := b.subs
+	b.subs = nil
+	b.mu.Unlock()
+
+	cancel()
+	for _, unsub := range subs {
+		unsub()
+	}
+
+	done := make(chan struct{})
+	go func() {
+		b.wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+		b.log.Info("backthink stopped")
+		return nil
+	case <-ctx.Done():
+		return fmt.Errorf("backthink: stop: %w", ctx.Err())
+	}
+}
+
+// List returns the most recent thinking records for the M9/M10 monitor panel,
+// newest first. A non-positive limit selects the store default.
+func (b *Backthink) List(ctx context.Context, limit int) ([]Record, error) {
+	return b.store.List(ctx, limit)
+}
+
+// triggers returns the bus subjects the service watches for event-driven
+// thinking. It is a method (not a package var) so no package-level global slice
+// appears, per the coding constraint.
+func (b *Backthink) triggers() []string {
+	return []string{iot.DeviceUpdateSubject, SubjectReminderTrigger, SubjectSystemEvent}
+}
+
+// subscribe registers one handler per trigger subject. Each subscription is
+// bound to ctx so cancelling it removes the registration; the unregister
+// functions are kept so Stop can also tear them down explicitly.
+func (b *Backthink) subscribe(ctx context.Context) {
+	for _, subj := range b.triggers() {
+		unsub, err := b.deps.Bus.Subscribe(ctx, subj, b.onEvent)
+		if err != nil {
+			b.log.Error("backthink subscribe failed", "subject", subj, "error", err)
+			continue
+		}
+		b.mu.Lock()
+		b.subs = append(b.subs, unsub)
+		b.mu.Unlock()
+	}
+}
+
+// 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.
+func (b *Backthink) onEvent(ctx context.Context, e bus.Event) error {
+	b.think(ctx, triggerForEvent(e))
+	return nil
+}
+
+// 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) {
+	defer b.wg.Done()
+	ticker := time.NewTicker(b.cfg.Interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			b.think(ctx, heartbeatTrigger())
+		}
+	}
+}
+
+// think runs one unattended thinking pass: invoke the runtime, read the user
+// state, decide whether to interrupt, push when warranted and persist the record.
+// It is concurrency-safe provided the injected Thinker, StateProvider and Pusher
+// are concurrency-safe, which they are by contract.
+func (b *Backthink) think(parent context.Context, tr Trigger) {
+	at := b.now()
+	cctx, cancel := context.WithTimeout(parent, b.cfg.ThinkTimeout)
+	defer cancel()
+
+	b.log.Debug("backthink trigger", "source", tr.Source, "summary", tr.Summary)
+
+	conclusion, err := b.deps.Thinker.Think(cctx, b.sessionID, tr.Summary)
+	if err != nil {
+		b.log.Error("backthink think failed", "source", tr.Source, "error", err)
+		b.record(Record{
+			Trigger:    tr.Source,
+			At:         at,
+			Conclusion: fmt.Sprintf("thinking failed: %v", err),
+			ShouldPush: false,
+			PushResult: PushResult{Route: RouteNone, Detail: err.Error()},
+		})
+		return
+	}
+
+	state := b.currentState(cctx)
+	decision := b.Decide(DecisionInput{Trigger: tr.Source, Import: tr.Import, At: at, UserState: state})
+
+	pushRes := PushResult{Route: RouteNone, Detail: decision.Reason}
+	if decision.ShouldPush {
+		pushRes = b.pusher.Push(cctx, b.buildProactiveMessage(conclusion))
+	}
+
+	b.record(Record{
+		Trigger:    tr.Source,
+		At:         at,
+		Conclusion: conclusion,
+		ShouldPush: decision.ShouldPush,
+		PushResult: pushRes,
+	})
+
+	if decision.ShouldPush {
+		b.log.Info("backthink proactive push", "source", tr.Source, "route", pushRes.Route, "pushed", pushRes.Pushed, "reason", decision.Reason)
+	} else {
+		b.log.Debug("backthink did not interrupt", "source", tr.Source, "reason", decision.Reason)
+	}
+}
+
+// currentState reads the user's current condition for the interrupt decision,
+// falling back to a neutral state (and a warn log) when the provider errors.
+func (b *Backthink) currentState(ctx context.Context) UserState {
+	if b.deps.State == nil {
+		return UserState{}
+	}
+	s, err := b.deps.State.Current(ctx)
+	if err != nil {
+		b.log.Warn("backthink state provider failed", "error", err)
+		return UserState{}
+	}
+	return s
+}
+
+// record persists a thinking pass. It uses a background context so an audit row
+// is written even when the think's own context has already been cancelled.
+func (b *Backthink) record(r Record) {
+	if _, err := b.store.Append(context.Background(), r); err != nil {
+		b.log.Error("backthink record failed", "error", err)
+	}
+}
+
+// normalize fills a Config's zero fields with safe defaults.
+func normalize(cfg Config) Config {
+	if cfg.IntervalSec <= 0 {
+		cfg.IntervalSec = defaultIntervalSec
+	}
+	if cfg.Interval <= 0 {
+		cfg.Interval = time.Duration(cfg.IntervalSec) * time.Second
+	}
+	if cfg.QuietStartHour == 0 && cfg.QuietEndHour == 0 {
+		cfg.QuietStartHour = defaultQuietStartHour
+		cfg.QuietEndHour = defaultQuietEndHour
+	}
+	if cfg.MaxPushWindow <= 0 {
+		cfg.MaxPushWindow = defaultMaxPushWindow
+	}
+	if cfg.PushWindow <= 0 {
+		cfg.PushWindow = defaultPushWindow
+	}
+	if cfg.ThinkTimeout <= 0 {
+		cfg.ThinkTimeout = defaultThinkTimeout
+	}
+	return cfg
+}
+
+// quietHoursLabel renders the do-not-disturb window for logs.
+func quietHoursLabel(start, end int) string {
+	return fmt.Sprintf("%02d:00-%02d:00", start, end)
+}
+
+// baseLog returns l when non-nil, otherwise a discard logger, so New always has a
+// safe logger to tag.
+func baseLog(l *slog.Logger) *slog.Logger {
+	if l != nil {
+		return l
+	}
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}

internal/services/backthink/backthink_test.go

@@ -0,0 +1,224 @@
+package backthink
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
+)
+
+// newBackthink builds a Backthink with a fresh in-memory DB and the given deps,
+// returning the zeroed deps filled with the DB handle so tests can inspect deps.
+func newBackthink(t *testing.T, cfg Config, deps Deps) (*Backthink, Deps) {
+	t.Helper()
+	deps.DB = newTestDB(t)
+	b, err := New(cfg, deps)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	t.Cleanup(func() { _ = b.Stop(context.Background()) })
+	return b, deps
+}
+
+func TestNew_Validation(t *testing.T) {
+	t.Parallel()
+	if _, err := New(Config{}, Deps{DB: newTestDB(t)}); !errors.Is(err, ErrNoThinker) {
+		t.Fatalf("missing thinker: err = %v, want ErrNoThinker", err)
+	}
+	if _, err := New(Config{}, Deps{Thinker: &mockThinker{}}); !errors.Is(err, ErrNilDB) {
+		t.Fatalf("missing db: err = %v, want ErrNilDB", err)
+	}
+}
+
+func TestBackthink_Heartbeat_TriggersThink(t *testing.T) {
+	t.Parallel()
+	thinker := &mockThinker{text: "nothing to report"}
+	b, _ := newBackthink(t, Config{Interval: 20 * time.Millisecond}, Deps{Thinker: thinker})
+
+	b.Start(context.Background())
+	defer func() { _ = b.Stop(context.Background()) }()
+
+	eventually(t, 2*time.Second, func() bool { return thinker.count() >= 1 })
+	eventually(t, 2*time.Second, func() bool {
+		recs, _ := b.List(context.Background(), 10)
+		return len(recs) >= 1
+	})
+	recs, err := b.List(context.Background(), 10)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if recs[0].Trigger != "heartbeat" {
+		t.Fatalf("Trigger = %q, want heartbeat", recs[0].Trigger)
+	}
+}
+
+func TestBackthink_EventTrigger_Push(t *testing.T) {
+	t.Parallel()
+	thinker := &mockThinker{text: "washing finished, clothes ready"}
+	pusher := &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}
+	bs := bus.New()
+	b, _ := newBackthink(t, Config{}, Deps{
+		Bus:     bs,
+		Thinker: thinker,
+		State:   &mockState{state: UserState{}},
+		Pusher:  pusher,
+		Now:     func() time.Time { return dayAt() },
+	})
+
+	b.Start(context.Background())
+	defer func() { _ = b.Stop(context.Background()) }()
+
+	if err := bs.Publish(context.Background(), SubjectReminderTrigger, "reminder due"); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+
+	eventually(t, 2*time.Second, func() bool {
+		recs, _ := b.List(context.Background(), 10)
+		return len(recs) >= 1
+	})
+	if got := thinker.count(); got != 1 {
+		t.Fatalf("thinker calls = %d, want 1", got)
+	}
+	if got := pusher.count(); got != 1 {
+		t.Fatalf("pusher calls = %d, want 1 (important + daytime)", got)
+	}
+	recs, _ := b.List(context.Background(), 10)
+	if !recs[0].ShouldPush {
+		t.Fatalf("ShouldPush = false, want true")
+	}
+	if !recs[0].PushResult.Pushed || recs[0].PushResult.Route != RouteActiveClient {
+		t.Fatalf("PushResult = %+v, want pushed active_client", recs[0].PushResult)
+	}
+}
+
+func TestBackthink_EventTrigger_NoPush(t *testing.T) {
+	t.Parallel()
+	thinker := &mockThinker{text: "washing finished, clothes ready"}
+	pusher := &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}
+	bs := bus.New()
+	b, _ := newBackthink(t, Config{}, Deps{
+		Bus:     bs,
+		Thinker: thinker,
+		State:   &mockState{state: UserState{Sleeping: true}},
+		Pusher:  pusher,
+		Now:     func() time.Time { return nightAt() },
+	})
+
+	b.Start(context.Background())
+	defer func() { _ = b.Stop(context.Background()) }()
+
+	if err := bs.Publish(context.Background(), iot.DeviceUpdateSubject, iot.DeviceUpdate{
+		Device: iot.Device{ID: "washer", Type: "sensor"},
+		Action: "finished",
+	}); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+
+	eventually(t, 2*time.Second, func() bool {
+		recs, _ := b.List(context.Background(), 10)
+		return len(recs) >= 1
+	})
+	if got := pusher.count(); got != 0 {
+		t.Fatalf("pusher calls = %d, want 0 (asleep + night)", got)
+	}
+	recs, _ := b.List(context.Background(), 10)
+	if recs[0].ShouldPush {
+		t.Fatalf("ShouldPush = true, want false")
+	}
+	if recs[0].PushResult.Route != RouteNone {
+		t.Fatalf("PushResult.Route = %q, want %q", recs[0].PushResult.Route, RouteNone)
+	}
+	if recs[0].Trigger != "iot.device_update" {
+		t.Fatalf("Trigger = %q, want iot.device_update", recs[0].Trigger)
+	}
+}
+
+func TestBackthink_List_NewestFirst(t *testing.T) {
+	t.Parallel()
+	thinker := &mockThinker{text: "nothing to report"}
+	b, _ := newBackthink(t, Config{Interval: 15 * time.Millisecond}, Deps{Thinker: thinker})
+
+	b.Start(context.Background())
+	defer func() { _ = b.Stop(context.Background()) }()
+
+	eventually(t, 2*time.Second, func() bool {
+		recs, _ := b.List(context.Background(), 10)
+		return len(recs) >= 2
+	})
+	top, err := b.List(context.Background(), 1)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if len(top) != 1 {
+		t.Fatalf("len = %d, want 1", len(top))
+	}
+	all, err := b.List(context.Background(), 10)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	// The single-row listing must be the same newest record as the head of the
+	// full listing, and the full listing must be strictly newest-first.
+	if top[0].ID != all[0].ID {
+		t.Fatalf("List(1)[0].ID = %d, want %d (== newest)", top[0].ID, all[0].ID)
+	}
+	for i := 1; i < len(all); i++ {
+		if all[i].ID >= all[i-1].ID {
+			t.Fatalf("records not newest-first: %d before %d", all[i-1].ID, all[i].ID)
+		}
+	}
+}
+
+func TestBackthink_Stop_NoLeak(t *testing.T) {
+	t.Parallel()
+	thinker := &mockThinker{text: "nothing to report"}
+	b, _ := newBackthink(t, Config{Interval: 10 * time.Millisecond}, Deps{Thinker: thinker})
+
+	b.Start(context.Background())
+	eventually(t, 2*time.Second, func() bool { return thinker.count() >= 1 })
+
+	if err := b.Stop(context.Background()); err != nil {
+		t.Fatalf("Stop: %v", err)
+	}
+	// Idempotent.
+	if err := b.Stop(context.Background()); err != nil {
+		t.Fatalf("second Stop: %v", err)
+	}
+
+	before := thinker.count()
+	time.Sleep(40 * time.Millisecond)
+	if got := thinker.count(); got != before {
+		t.Fatalf("thinker calls grew after stop: %d -> %d", before, got)
+	}
+}
+
+func TestBackthink_Stop_UnsubscribesBus(t *testing.T) {
+	t.Parallel()
+	thinker := &mockThinker{text: "nothing to report"}
+	bs := bus.New()
+	b, _ := newBackthink(t, Config{Interval: 25 * time.Millisecond}, Deps{Bus: bs, Thinker: thinker})
+
+	b.Start(context.Background())
+	// Confirm the subscription is live by triggering one think.
+	if err := bs.Publish(context.Background(), SubjectReminderTrigger, "first"); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+	eventually(t, 2*time.Second, func() bool { return thinker.count() >= 1 })
+
+	if err := b.Stop(context.Background()); err != nil {
+		t.Fatalf("Stop: %v", err)
+	}
+
+	before := thinker.count()
+	if err := bs.Publish(context.Background(), iot.DeviceUpdateSubject, iot.DeviceUpdate{
+		Device: iot.Device{ID: "washer", Type: "sensor"}, Action: "finished",
+	}); err != nil {
+		t.Fatalf("Publish after stop: %v", err)
+	}
+	time.Sleep(40 * time.Millisecond)
+	if got := thinker.count(); got != before {
+		t.Fatalf("thinker calls grew after stop (bus not unsubscribed): %d -> %d", before, got)
+	}
+}

internal/services/backthink/decide.go

@@ -0,0 +1,141 @@
+package backthink
+
+import (
+	"fmt"
+	"time"
+)
+
+// Importance is the estimated severity of an event that woke a thinking pass. It
+// is the primary positive signal in the interrupt decision: a low-importance
+// heartbeat almost never interrupts, while a critical safety event normally does.
+type Importance int
+
+const (
+	// ImportanceLow is a routine trigger (e.g. the periodic heartbeat).
+	ImportanceLow Importance = iota
+	// ImportanceNormal is an ordinary event worth reflecting on (device change).
+	ImportanceNormal
+	// ImportanceHigh is a notable event (a reminder the user set firing).
+	ImportanceHigh
+	// ImportanceCritical is a safety-critical event that may override work/busy.
+	ImportanceCritical
+)
+
+// String returns a stable label for logs and records.
+func (i Importance) String() string {
+	switch i {
+	case ImportanceLow:
+		return "low"
+	case ImportanceNormal:
+		return "normal"
+	case ImportanceHigh:
+		return "high"
+	case ImportanceCritical:
+		return "critical"
+	default:
+		return fmt.Sprintf("importance(%d)", int(i))
+	}
+}
+
+// highBoredomThreshold is the boredom (0..1) above which a normal daytime event
+// is worth interrupting a bored, otherwise idle user. It is a const so no magic
+// number appears in the decision path.
+const highBoredomThreshold = 0.7
+
+// UserState describes the user's current condition as observed from device
+// telemetry (heart rate, sleep) and schedule (work, busy). It is the negative
+// signal in the interrupt decision: a sleeping, working or busy user is left
+// alone. Boredom (0..1) is the soft positive dial that widens interruption.
+type UserState struct {
+	// HeartRateBPM is the current heart rate, 0 when unknown.
+	HeartRateBPM int
+	// Sleeping reports whether the user is currently asleep from sleep tracking.
+	Sleeping bool
+	// Working reports whether the user is in a work/scheduled block.
+	Working bool
+	// Busy reports whether the user is marked busy (calendar / do-not-disturb).
+	Busy bool
+	// Boredom ranges 0..1. Higher boredom means the user is idle and more open
+	// to a gentle interruption.
+	Boredom float64
+}
+
+// String returns a concise human-readable summary for logs/records.
+func (s UserState) String() string {
+	return fmt.Sprintf("bpm=%d sleeping=%v working=%v busy=%v boredom=%.2f",
+		s.HeartRateBPM, s.Sleeping, s.Working, s.Busy, s.Boredom)
+}
+
+// DecisionInput is the set of signals Decide weighs: the event's importance, the
+// current wall-clock time, the user's current state and their boredom.
+type DecisionInput struct {
+	// Trigger is a human-readable source label, recorded for traceability.
+	Trigger string
+	// Import is the event's estimated importance.
+	Import Importance
+	// At is the current wall-clock time, in the local timezone used for the
+	// do-not-disturb window.
+	At time.Time
+	// UserState carries the user's physiological/schedule condition.
+	UserState UserState
+}
+
+// Decision is the outcome of Decide.
+type Decision struct {
+	// ShouldPush reports whether the user should be interrupted.
+	ShouldPush bool
+	// Reason is a short human-readable justification, surfaced on the panel.
+	Reason string
+}
+
+// Decide applies the interrupt rule from design/12 §2 (and design/11 §6) with
+// "不打扰 (do not interrupt) as the first principle" as its bias. It is a pure
+// method on Backthink only so it can read the configured quiet-hours window; it
+// does not touch state.
+//
+// The evaluation order is deliberately fail-closed (prefer not interrupting):
+//
+//  1. A sleeping user is never interrupted.
+//  2. A working or busy user is interrupted only by a critical event.
+//  3. Inside the quiet-hours (deep-night) window only high/critical events
+//     interrupt; low/normal events are deferred.
+//  4. In the daytime, high/critical events always interrupt; a normal event
+//     interrupts only when the user is quite bored; a low event never does.
+func (b *Backthink) Decide(in DecisionInput) Decision {
+	s := in.UserState
+	if s.Sleeping {
+		return Decision{ShouldPush: false, Reason: "user is asleep; not interrupting"}
+	}
+	if (s.Working || s.Busy) && in.Import < ImportanceCritical {
+		return Decision{ShouldPush: false, Reason: "user is working/busy; not interrupting"}
+	}
+	if b.inQuietHours(in.At) {
+		if in.Import >= ImportanceHigh {
+			return Decision{ShouldPush: true, Reason: "quiet hours but event is important"}
+		}
+		return Decision{ShouldPush: false, Reason: "quiet hours; not interrupting"}
+	}
+	switch in.Import {
+	case ImportanceCritical, ImportanceHigh:
+		return Decision{ShouldPush: true, Reason: "high/critical importance during daytime"}
+	case ImportanceNormal:
+		if s.Boredom >= highBoredomThreshold {
+			return Decision{ShouldPush: true, Reason: "normal importance and user is bored"}
+		}
+		return Decision{ShouldPush: false, Reason: "normal importance; not worth interrupting"}
+	default:
+		return Decision{ShouldPush: false, Reason: "low importance; not interrupting"}
+	}
+}
+
+// inQuietHours reports whether at falls inside the configured do-not-disturb
+// window. The window may wrap midnight (e.g. 23:00-07:00); two equal non-zero
+// hours disable it.
+func (b *Backthink) inQuietHours(at time.Time) bool {
+	h := at.Hour()
+	start, end := b.cfg.QuietStartHour, b.cfg.QuietEndHour
+	if start <= end {
+		return h >= start && h < end
+	}
+	return h >= start || h < end
+}

internal/services/backthink/decide_test.go

@@ -0,0 +1,139 @@
+package backthink
+
+import (
+	"testing"
+	"time"
+)
+
+// dayAt returns a mid-afternoon time in UTC (outside the quiet-hours window).
+func dayAt() time.Time { return time.Date(2026, 9, 6, 14, 0, 0, 0, time.UTC) }
+
+// nightAt returns a 02:00 time in UTC (inside the default 23-7 quiet window).
+func nightAt() time.Time { return time.Date(2026, 9, 6, 2, 0, 0, 0, time.UTC) }
+
+// testBackthink builds a Backthink with a normalized empty Config so Decide uses
+// the default quiet-hours window (23:00-07:00).
+func testBackthink() *Backthink {
+	return &Backthink{cfg: normalize(Config{})}
+}
+
+func TestDecide_DoNotDisturb(t *testing.T) {
+	t.Parallel()
+	b := testBackthink()
+	cases := []struct {
+		name string
+		in   DecisionInput
+		want bool
+	}{
+		{
+			name: "凌晨+睡眠_不打扰",
+			in:   DecisionInput{Trigger: "iot.device_update", Import: ImportanceNormal, At: nightAt(), UserState: UserState{Sleeping: true}},
+			want: false,
+		},
+		{
+			name: "睡眠_重要也不打扰",
+			in:   DecisionInput{Trigger: "reminder.trigger", Import: ImportanceHigh, At: dayAt(), UserState: UserState{Sleeping: true}},
+			want: false,
+		},
+		{
+			name: "上班_不打扰",
+			in:   DecisionInput{Trigger: "iot.device_update", Import: ImportanceNormal, At: dayAt(), UserState: UserState{Working: true}},
+			want: false,
+		},
+		{
+			name: "忙碌_普通_不打扰",
+			in:   DecisionInput{Trigger: "iot.device_update", Import: ImportanceNormal, At: dayAt(), UserState: UserState{Busy: true}},
+			want: false,
+		},
+		{
+			name: "安静时段_普通_不打扰",
+			in:   DecisionInput{Trigger: "iot.device_update", Import: ImportanceNormal, At: nightAt(), UserState: UserState{Boredom: 0.9}},
+			want: false,
+		},
+		{
+			name: "白天_普通_不无聊_不打扰",
+			in:   DecisionInput{Trigger: "iot.device_update", Import: ImportanceNormal, At: dayAt(), UserState: UserState{}},
+			want: false,
+		},
+		{
+			name: "白天_低重要_不打扰",
+			in:   DecisionInput{Trigger: "heartbeat", Import: ImportanceLow, At: dayAt(), UserState: UserState{}},
+			want: false,
+		},
+	}
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got := b.Decide(tc.in)
+			if got.ShouldPush != tc.want {
+				t.Fatalf("Decide(%+v).ShouldPush = %v, want %v (reason %q)", tc.in, got.ShouldPush, tc.want, got.Reason)
+			}
+		})
+	}
+}
+
+func TestDecide_ShouldPush(t *testing.T) {
+	t.Parallel()
+	b := testBackthink()
+	cases := []struct {
+		name string
+		in   DecisionInput
+	}{
+		{
+			name: "白天+重要_推",
+			in:   DecisionInput{Trigger: "reminder.trigger", Import: ImportanceHigh, At: dayAt(), UserState: UserState{}},
+		},
+		{
+			name: "白天+危急_推",
+			in:   DecisionInput{Trigger: "system.event", Import: ImportanceCritical, At: dayAt(), UserState: UserState{}},
+		},
+		{
+			name: "白天+普通+无聊_推",
+			in:   DecisionInput{Trigger: "iot.device_update", Import: ImportanceNormal, At: dayAt(), UserState: UserState{Boredom: 0.8}},
+		},
+		{
+			name: "安静时段+重要_推",
+			in:   DecisionInput{Trigger: "reminder.trigger", Import: ImportanceHigh, At: nightAt(), UserState: UserState{}},
+		},
+		{
+			name: "危急_可打断忙碌_推",
+			in:   DecisionInput{Trigger: "system.event", Import: ImportanceCritical, At: dayAt(), UserState: UserState{Working: true}},
+		},
+	}
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got := b.Decide(tc.in)
+			if !got.ShouldPush {
+				t.Fatalf("Decide(%+v).ShouldPush = false, want true (reason %q)", tc.in, got.Reason)
+			}
+		})
+	}
+}
+
+func TestInQuietHours(t *testing.T) {
+	t.Parallel()
+	b := &Backthink{cfg: Config{QuietStartHour: 23, QuietEndHour: 7}}
+	cases := []struct {
+		name string
+		at   time.Time
+		want bool
+	}{
+		{name: "午夜", at: time.Date(2026, 9, 6, 0, 0, 0, 0, time.UTC), want: true},
+		{name: "凌晨两点", at: nightAt(), want: true},
+		{name: "早上七点_偏出", at: time.Date(2026, 9, 6, 7, 0, 0, 0, time.UTC), want: false},
+		{name: "傍晚", at: time.Date(2026, 9, 6, 18, 30, 0, 0, time.UTC), want: false},
+		{name: "深夜十一点", at: time.Date(2026, 9, 6, 23, 0, 0, 0, time.UTC), want: true},
+	}
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			if got := b.inQuietHours(tc.at); got != tc.want {
+				t.Fatalf("inQuietHours(%v) = %v, want %v", tc.at, got, tc.want)
+			}
+		})
+	}
+}

internal/services/backthink/mocks_test.go

@@ -0,0 +1,137 @@
+package backthink
+
+import (
+	"context"
+	"database/sql"
+	"io"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
+)
+
+// mockThinker records every Think invocation and returns a fixed conclusion. It
+// respects ctx cancellation so Stop can drain a hung pass.
+type mockThinker struct {
+	mu    sync.Mutex
+	text  string
+	err   error
+	sleep time.Duration
+	calls int
+	last  string
+}
+
+func (m *mockThinker) Think(ctx context.Context, sessionID, trigger string) (string, error) {
+	if err := ctx.Err(); err != nil {
+		return "", err
+	}
+	if m.sleep > 0 {
+		select {
+		case <-ctx.Done():
+			return "", ctx.Err()
+		case <-time.After(m.sleep):
+		}
+	}
+	m.mu.Lock()
+	m.calls++
+	m.last = trigger
+	m.mu.Unlock()
+	return m.text, m.err
+}
+
+func (m *mockThinker) count() int          { m.mu.Lock(); defer m.mu.Unlock(); return m.calls }
+func (m *mockThinker) lastTrigger() string { m.mu.Lock(); defer m.mu.Unlock(); return m.last }
+
+// mockState returns a fixed UserState.
+type mockState struct{ state UserState }
+
+func (m *mockState) Current(context.Context) (UserState, error) { return m.state, nil }
+
+// mockPusher records Push calls and returns a fixed result.
+type mockPusher struct {
+	mu      sync.Mutex
+	result  PushResult
+	calls   int
+	lastMsg adapter.UnifiedMessage
+}
+
+func (p *mockPusher) Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult {
+	p.mu.Lock()
+	p.calls++
+	p.lastMsg = msg
+	p.mu.Unlock()
+	return p.result
+}
+
+func (p *mockPusher) count() int { p.mu.Lock(); defer p.mu.Unlock(); return p.calls }
+
+// mockActive implements ActiveClient.
+type mockActive struct {
+	mu    sync.Mutex
+	n     int
+	err   error
+	calls int
+}
+
+func (m *mockActive) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) {
+	m.mu.Lock()
+	m.calls++
+	m.mu.Unlock()
+	return m.n, m.err
+}
+
+// mockIMAdapter implements IMAdapter.
+type mockIMAdapter struct {
+	mu    sync.Mutex
+	err   error
+	calls int
+	last  adapter.UnifiedMessage
+}
+
+func (m *mockIMAdapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error {
+	m.mu.Lock()
+	m.calls++
+	m.last = msg
+	m.mu.Unlock()
+	return m.err
+}
+
+func (m *mockIMAdapter) count() int { m.mu.Lock(); defer m.mu.Unlock(); return m.calls }
+
+// mockGate implements PushGate.
+type mockGate struct{ allow bool }
+
+func (g *mockGate) Allow(context.Context) bool { return g.allow }
+
+// newTestDB opens an isolated in-memory SQLite handle for one test and closes it
+// on cleanup.
+func newTestDB(t *testing.T) *sql.DB {
+	t.Helper()
+	h, err := db.Open(context.Background(), ":memory:", nil)
+	if err != nil {
+		t.Fatalf("open test db: %v", err)
+	}
+	t.Cleanup(func() { _ = h.Close() })
+	return h
+}
+
+// discardLog returns a logger that writes nowhere.
+func discardLog() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+// eventually polls cond until it is true or the deadline passes.
+func eventually(t *testing.T, timeout time.Duration, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("condition not met within %v", timeout)
+}

internal/services/backthink/push.go

@@ -0,0 +1,167 @@
+package backthink
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
+)
+
+// Route identifies how a proactive message was actually delivered, for the
+// monitor panel and the push result record.
+type Route string
+
+const (
+	// RouteNone means no push was attempted or no target was available.
+	RouteNone Route = "none"
+	// RouteRateLimited means the push was refused by the rate limiter.
+	RouteRateLimited Route = "rate_limited"
+	// RouteActiveClient means the message went to an active web/native client.
+	RouteActiveClient Route = "active_client"
+	// RouteIMBridge means the message fell back to the IM bridge.
+	RouteIMBridge Route = "im_bridge"
+)
+
+// String returns the route as a plain label.
+func (r Route) String() string { return string(r) }
+
+// PushResult is the outcome of one proactive push attempt, recorded with each
+// thinking pass (design/12 §4).
+type PushResult struct {
+	// Pushed reports whether the message was delivered.
+	Pushed bool
+	// Route is the delivery route taken.
+	Route Route
+	// 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
+	// provider that decided not to push.
+	Err string
+}
+
+// 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 detail=%q", p.Pushed, p.Route, p.Detail)
+}
+
+// ActiveClient delivers a proactive message to every currently active web/native
+// client and reports how many received it. It is the preferred push target. The
+// gateway WebSocket hub implements this in production; tests inject a mock.
+type ActiveClient interface {
+	// SendProactive pushes msg to each active client and returns the number of
+	// clients that accepted it.
+	SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error)
+}
+
+// IMAdapter is the IM bridge fallback: it pushes a proactive message over an IM
+// channel. Every adapter.PlatformAdapter satisfies this subset, so the service
+// depends on the narrow interface rather than the concrete adapter type.
+type IMAdapter interface {
+	// SendProactive pushes msg to the owner's IM channel.
+	SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error
+}
+
+// PushGate decides whether a proactive push may proceed right now, applying the
+// rate limits of design/11 §6. It is a seam over channel/session's
+// ProactivePolicy/RateLimiter so the service can be tested with a mock.
+type PushGate interface {
+	// Allow reports whether a proactive push may be sent now.
+	Allow(ctx context.Context) bool
+}
+
+// Pusher delivers a proactive message under the active-client-first, IM-fallback
+// policy (design/12 §3), bounded by the rate gate.
+type Pusher interface {
+	// Push delivers msg, preferring the active client and falling back to the IM
+	// bridge. It never blocks on a slow target.
+	Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult
+}
+
+// pusher is the concrete Pusher: it consults the gate, then tries the active
+// client and falls back to IM.
+type pusher struct {
+	active ActiveClient
+	im     IMAdapter
+	gate   PushGate
+	log    *slog.Logger
+}
+
+// newPusher builds a Pusher from the optional active-client sink, IM sink and
+// rate gate. Any of them may be nil; a nil target simply makes that route
+// unavailable.
+func newPusher(active ActiveClient, im IMAdapter, gate PushGate, log *slog.Logger) *pusher {
+	return &pusher{active: active, im: im, gate: gate, log: log}
+}
+
+// Push implements Pusher.
+func (p *pusher) Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult {
+	if err := ctx.Err(); err != nil {
+		return PushResult{Pushed: false, Route: RouteNone, Err: err.Error()}
+	}
+	if p.gate != nil && !p.gate.Allow(ctx) {
+		p.log.Warn("backthink push rate limited")
+		return PushResult{Pushed: false, Route: RouteRateLimited, Detail: "proactive push rate limited"}
+	}
+
+	// Active client first.
+	if p.active != nil {
+		n, err := p.active.SendProactive(ctx, msg)
+		if err != nil {
+			p.log.Warn("backthink active client push failed", "error", err)
+		} else if n > 0 {
+			return PushResult{Pushed: true, Route: RouteActiveClient, Detail: fmt.Sprintf("delivered to %d active client(s)", n)}
+		}
+	}
+
+	// IM fallback.
+	if p.im != nil {
+		if err := p.im.SendProactive(ctx, msg); err != nil {
+			p.log.Warn("backthink IM push failed", "error", err)
+			return PushResult{Pushed: false, Route: RouteIMBridge, Err: err.Error()}
+		}
+		return PushResult{Pushed: true, Route: RouteIMBridge, Detail: "delivered via IM bridge"}
+	}
+
+	return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client or IM target"}
+}
+
+// rateGate is the production PushGate: a sliding-window counter over
+// session.RateLimiter (design/11 §6 anti-spam rule).
+type rateGate struct {
+	log     *slog.Logger
+	limiter *session.RateLimiter
+}
+
+// newRateGate builds a gate that allows at most limit pushes within window. A
+// non-positive limit or window yields an always-allow gate (no rule configured).
+func newRateGate(limit int, window time.Duration, log *slog.Logger) *rateGate {
+	return &rateGate{log: log, limiter: session.NewRateLimiter(limit, window)}
+}
+
+// Allow implements PushGate. A canceled context denies (fail closed).
+func (g *rateGate) Allow(ctx context.Context) bool {
+	if err := ctx.Err(); err != nil {
+		return false
+	}
+	return g.limiter.Allow()
+}
+
+// buildProactiveMessage assembles the normalized outbound message carrying a
+// think conclusion. It is intentionally minimal: the concrete active-client and
+// IM adapters apply their own routing/targeting to the owner.
+func (b *Backthink) buildProactiveMessage(conclusion string) adapter.UnifiedMessage {
+	return adapter.UnifiedMessage{
+		Platform:    "cyrene-backthink",
+		ChannelType: adapter.ChannelDirect,
+		Content:     conclusion,
+		ContentType: adapter.ContentText,
+		Sender:      adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
+		Timestamp:   b.now(),
+	}
+}

internal/services/backthink/push_test.go

@@ -0,0 +1,147 @@
+package backthink
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+func TestPusher_ActiveFirst(t *testing.T) {
+	t.Parallel()
+	active := &mockActive{n: 2}
+	im := &mockIMAdapter{}
+	p := newPusher(active, im, &mockGate{allow: true}, discardLog())
+
+	res := p.Push(context.Background(), adapter.UnifiedMessage{})
+
+	if !res.Pushed {
+		t.Fatalf("Pushed = false, want true")
+	}
+	if res.Route != RouteActiveClient {
+		t.Fatalf("Route = %q, want %q", res.Route, RouteActiveClient)
+	}
+	if active.calls != 1 {
+		t.Fatalf("active calls = %d, want 1", active.calls)
+	}
+	if im.count() != 0 {
+		t.Fatalf("im calls = %d, want 0 (active should be preferred)", im.count())
+	}
+}
+
+func TestPusher_IMFallback_NoActive(t *testing.T) {
+	t.Parallel()
+	active := &mockActive{n: 0} // no active client online
+	im := &mockIMAdapter{}
+	p := newPusher(active, im, &mockGate{allow: true}, discardLog())
+
+	res := p.Push(context.Background(), adapter.UnifiedMessage{})
+
+	if !res.Pushed {
+		t.Fatalf("Pushed = false, want true")
+	}
+	if res.Route != RouteIMBridge {
+		t.Fatalf("Route = %q, want %q", res.Route, RouteIMBridge)
+	}
+	if im.count() != 1 {
+		t.Fatalf("im calls = %d, want 1", im.count())
+	}
+}
+
+func TestPusher_IMFallback_ActiveError(t *testing.T) {
+	t.Parallel()
+	active := &mockActive{err: context.DeadlineExceeded}
+	im := &mockIMAdapter{}
+	p := newPusher(active, im, &mockGate{allow: true}, discardLog())
+
+	res := p.Push(context.Background(), adapter.UnifiedMessage{})
+
+	if !res.Pushed {
+		t.Fatalf("Pushed = false, want true")
+	}
+	if res.Route != RouteIMBridge {
+		t.Fatalf("Route = %q, want %q", res.Route, RouteIMBridge)
+	}
+	if im.count() != 1 {
+		t.Fatalf("im calls = %d, want 1", im.count())
+	}
+}
+
+func TestPusher_NoTarget_Skipped(t *testing.T) {
+	t.Parallel()
+	p := newPusher(nil, nil, &mockGate{allow: true}, discardLog())
+
+	res := p.Push(context.Background(), adapter.UnifiedMessage{})
+
+	if res.Pushed {
+		t.Fatalf("Pushed = true, want false")
+	}
+	if res.Route != RouteNone {
+		t.Fatalf("Route = %q, want %q", res.Route, RouteNone)
+	}
+}
+
+func TestPusher_GateDenied(t *testing.T) {
+	t.Parallel()
+	active := &mockActive{n: 1}
+	im := &mockIMAdapter{}
+	p := newPusher(active, im, &mockGate{allow: false}, discardLog())
+
+	res := p.Push(context.Background(), adapter.UnifiedMessage{})
+
+	if res.Pushed {
+		t.Fatalf("Pushed = true, want false")
+	}
+	if res.Route != RouteRateLimited {
+		t.Fatalf("Route = %q, want %q", res.Route, RouteRateLimited)
+	}
+	if active.calls != 0 || im.count() != 0 {
+		t.Fatalf("targets called despite denial: active=%d im=%d", active.calls, im.count())
+	}
+}
+
+func TestPusher_CancelledContext_Denied(t *testing.T) {
+	t.Parallel()
+	active := &mockActive{n: 1}
+	p := newPusher(active, nil, &mockGate{allow: true}, discardLog())
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	res := p.Push(ctx, adapter.UnifiedMessage{})
+
+	if res.Pushed {
+		t.Fatalf("Pushed = true, want false")
+	}
+	if res.Route != RouteNone {
+		t.Fatalf("Route = %q, want %q", res.Route, RouteNone)
+	}
+	if active.calls != 0 {
+		t.Fatalf("active calls = %d, want 0", active.calls)
+	}
+}
+
+// TestPusher_RateLimit exercises the production rateGate over session.RateLimiter:
+// with a limit of 1 push per window, only the first of three attempts passes.
+func TestPusher_RateLimit(t *testing.T) {
+	t.Parallel()
+	active := &mockActive{n: 1}
+	p := newPusher(active, nil, newRateGate(1, time.Minute, discardLog()), discardLog())
+
+	first := p.Push(context.Background(), adapter.UnifiedMessage{})
+	second := p.Push(context.Background(), adapter.UnifiedMessage{})
+	third := p.Push(context.Background(), adapter.UnifiedMessage{})
+
+	if !first.Pushed || first.Route != RouteActiveClient {
+		t.Fatalf("first push = %+v, want active_client pushed", first)
+	}
+	if second.Pushed || second.Route != RouteRateLimited {
+		t.Fatalf("second push = %+v, want rate_limited not pushed", second)
+	}
+	if third.Pushed || third.Route != RouteRateLimited {
+		t.Fatalf("third push = %+v, want rate_limited not pushed", third)
+	}
+	if active.calls != 1 {
+		t.Fatalf("active calls = %d, want 1 (rate limit caps the sink)", active.calls)
+	}
+}

internal/services/backthink/state.go

@@ -0,0 +1,123 @@
+package backthink
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strconv"
+	"strings"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
+)
+
+// StateProvider supplies the user's current condition for the interrupt decision
+// (design/12 §2). Production wires IoT/telemetry and calendar; tests inject a
+// mock. It must be concurrency-safe.
+type StateProvider interface {
+	// Current returns the user's current UserState. A nil-able error means "use
+	// a neutral state"; a hard error is logged by the caller and treated as
+	// neutral.
+	Current(ctx context.Context) (UserState, error)
+}
+
+// IoTStateProvider is a best-effort StateProvider built over an iot.Manager. It
+// scans the managed devices and reads heart-rate and sleep signals from any
+// device whose id/name/type mentions the corresponding keyword (e.g. "heart",
+// "心率", "sleep", "睡眠"). Work/busy come from a device/state flag; when nothing
+// matches it returns a neutral state rather than an error, so the service still
+// runs without full telemetry wiring.
+type IoTStateProvider struct {
+	mgr *iot.Manager
+	log *slog.Logger
+}
+
+// NewIoTStateProvider wraps mgr as a StateProvider.
+func NewIoTStateProvider(mgr *iot.Manager, log *slog.Logger) *IoTStateProvider {
+	return &IoTStateProvider{mgr: mgr, log: logger.WithSource(baseLog(log), "services/backthink")}
+}
+
+// Current implements StateProvider.
+func (p *IoTStateProvider) Current(ctx context.Context) (UserState, error) {
+	if p.mgr == nil {
+		return UserState{}, fmt.Errorf("backthink: state: %w", iot.ErrProviderNotRegistered)
+	}
+	devices, err := p.mgr.List(ctx)
+	if err != nil {
+		return UserState{}, fmt.Errorf("backthink: state: list: %w", err)
+	}
+
+	var state UserState
+	for _, d := range devices {
+		key := strings.ToLower(d.ID + " " + d.Name + " " + d.Type)
+		switch {
+		case strings.Contains(key, "heart") || strings.Contains(key, "心率"):
+			if v, ok := numState(d, "bpm", "heart_rate", "rate"); ok {
+				state.HeartRateBPM = v
+			}
+		case strings.Contains(key, "sleep") || strings.Contains(key, "睡眠"):
+			state.Sleeping = boolState(d, "sleeping", "asleep", "state")
+		}
+		if b, ok := boolStateOK(d, "working"); ok {
+			state.Working = b
+		}
+		if b, ok := boolStateOK(d, "busy"); ok {
+			state.Busy = b
+		}
+	}
+	if p.log != nil {
+		p.log.Debug("backthink iot state sampled", "devices", len(devices), "state", state.String())
+	}
+	return state, nil
+}
+
+// numState reads the first numeric value among keys from a device's state map.
+func numState(d iot.Device, keys ...string) (int, bool) {
+	for _, k := range keys {
+		v, ok := d.State[k]
+		if !ok {
+			continue
+		}
+		switch n := v.(type) {
+		case int:
+			return n, true
+		case int64:
+			return int(n), true
+		case float64:
+			return int(n), true
+		case string:
+			if f, err := strconv.Atoi(n); err == nil {
+				return f, true
+			}
+		}
+	}
+	return 0, false
+}
+
+// boolState reads a bool (or a "sleep"/"asleep" string) among keys.
+func boolState(d iot.Device, keys ...string) bool {
+	b, _ := boolStateOK(d, keys...)
+	return b
+}
+
+// boolStateOK reports whether a bool-typed state flag was found and its value.
+func boolStateOK(d iot.Device, keys ...string) (bool, bool) {
+	for _, k := range keys {
+		v, ok := d.State[k]
+		if !ok {
+			continue
+		}
+		switch n := v.(type) {
+		case bool:
+			return n, true
+		case string:
+			switch strings.ToLower(n) {
+			case "sleep", "asleep", "busy", "working", "yes", "true":
+				return true, true
+			case "awake", "free", "idle", "no", "false":
+				return false, true
+			}
+		}
+	}
+	return false, false
+}

internal/services/backthink/state_test.go

@@ -0,0 +1,85 @@
+package backthink
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
+)
+
+// stateMockProvider is a tiny iot.Provider that returns a heart-rate watch, a
+// sleep band and a working-calendar device, to exercise the state decoding.
+type stateMockProvider struct{}
+
+func (stateMockProvider) Name() string { return "state-mock" }
+
+func (stateMockProvider) List(context.Context) ([]iot.Device, error) {
+	return []iot.Device{
+		{ID: "watch.heart", Name: "心率手表", Type: "sensor", State: map[string]any{"bpm": 62}},
+		{ID: "band.sleep", Name: "睡眠手环", Type: "sensor", State: map[string]any{"sleeping": true}},
+		{ID: "cal.schedule", Name: "日程", State: map[string]any{"working": true}},
+	}, nil
+}
+
+func (stateMockProvider) Control(context.Context, string, string, map[string]any) (*iot.Device, error) {
+	return nil, iot.ErrUnsupported
+}
+
+func (stateMockProvider) Status(context.Context, string) (*iot.Device, error) {
+	return nil, iot.ErrDeviceNotFound
+}
+
+func TestIoTStateProvider_Current(t *testing.T) {
+	t.Parallel()
+	mgr, err := iot.New(config.IoTConfig{Provider: "none"}, nil)
+	if err != nil {
+		t.Fatalf("iot.New: %v", err)
+	}
+	if err := mgr.Register(stateMockProvider{}); err != nil {
+		t.Fatalf("Register: %v", err)
+	}
+
+	p := NewIoTStateProvider(mgr, nil)
+	st, err := p.Current(context.Background())
+	if err != nil {
+		t.Fatalf("Current: %v", err)
+	}
+	if st.HeartRateBPM != 62 {
+		t.Fatalf("HeartRateBPM = %d, want 62", st.HeartRateBPM)
+	}
+	if !st.Sleeping {
+		t.Fatalf("Sleeping = false, want true")
+	}
+	if !st.Working {
+		t.Fatalf("Working = false, want true")
+	}
+}
+
+func TestIoTStateProvider_NilManager(t *testing.T) {
+	t.Parallel()
+	p := NewIoTStateProvider(nil, nil)
+	_, err := p.Current(context.Background())
+	if !errors.Is(err, iot.ErrProviderNotRegistered) {
+		t.Fatalf("err = %v, want ErrProviderNotRegistered", err)
+	}
+}
+
+func TestIoTStateProvider_EmptyNeutral(t *testing.T) {
+	t.Parallel()
+	mgr, err := iot.New(config.IoTConfig{Provider: "mock"}, nil)
+	if err != nil {
+		t.Fatalf("iot.New: %v", err)
+	}
+	p := NewIoTStateProvider(mgr, nil)
+	// The mock provider has no heart-rate/sleep/working device, so it must yield
+	// a zero state instead of an error, keeping the service operable.
+	st, err := p.Current(context.Background())
+	if err != nil {
+		t.Fatalf("Current: %v", err)
+	}
+	if st != (UserState{}) {
+		t.Fatalf("state = %+v, want zero UserState", st)
+	}
+}

internal/services/backthink/store.go

@@ -0,0 +1,181 @@
+package backthink
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
+)
+
+// defaultListLimit is the number of most-recent records returned by List when the
+// caller passes a non-positive limit.
+const defaultListLimit = 50
+
+// DDL for the thinking record store. at/created_at follow the project timestamp
+// convention (RFC3339 with a timezone); trigger and conclusion are the free-text
+// fields the panel renders; the push_* columns carry a decomposed PushResult so
+// List need not parse a blob. The primary key is the project-standard auto id.
+const createRecordsSQL = `
+CREATE TABLE IF NOT EXISTS backthink_records (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    trigger      TEXT NOT NULL,
+    at           TEXT NOT NULL,
+    conclusion   TEXT NOT NULL,
+    should_push  INTEGER NOT NULL,
+    push_pushed  INTEGER NOT NULL DEFAULT 0,
+    push_route   TEXT NOT NULL,
+    push_detail  TEXT NOT NULL DEFAULT '',
+    push_error   TEXT NOT NULL DEFAULT '',
+    created_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+)`
+
+const createRecordsIndexSQL = `
+CREATE INDEX IF NOT EXISTS idx_backthink_records_at ON backthink_records(at)`
+
+const insertRecordSQL = `
+INSERT INTO backthink_records
+    (trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+
+const selectRecordsSQL = `
+SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error
+FROM backthink_records
+ORDER BY id DESC
+LIMIT ?`
+
+// Record is one persisted thinking pass (design/12 §4), returned by Store.List
+// and the service's List. Trigger is the source subsystem label, At the decision
+// timestamp, Conclusion the runtime's output, ShouldPush the interrupt decision
+// and PushResult how (or whether) the push went out.
+type Record struct {
+	ID         int64
+	Trigger    string
+	At         time.Time
+	Conclusion string
+	ShouldPush bool
+	PushResult PushResult
+}
+
+// Store persists thinking records in SQLite. It creates its table and index in
+// NewStore so the module is self-contained (no migration bookkeeping), mirrors
+// how channel/session creates its own table. It is concurrency-safe by virtue of
+// the underlying *sql.DB.
+type Store struct {
+	db  *sql.DB
+	log *slog.Logger
+}
+
+// NewStore builds a Store, ensuring the backthink_records table and index exist.
+// It returns ErrNilDB when db is nil. ctx is used for the DDL.
+func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
+	if db == nil {
+		return nil, ErrNilDB
+	}
+	log = logger.WithSource(baseLog(log), "services/backthink")
+	if _, err := db.ExecContext(ctx, createRecordsSQL); err != nil {
+		return nil, fmt.Errorf("backthink: ensure backthink_records: %w", err)
+	}
+	if _, err := db.ExecContext(ctx, createRecordsIndexSQL); err != nil {
+		return nil, fmt.Errorf("backthink: ensure backthink_records index: %w", err)
+	}
+	return &Store{db: db, log: log}, nil
+}
+
+// Append persists r and returns the new record id.
+func (s *Store) Append(ctx context.Context, r Record) (int64, error) {
+	res, err := s.db.ExecContext(ctx, insertRecordSQL,
+		r.Trigger,
+		r.At.UTC().Format(time.RFC3339Nano),
+		r.Conclusion,
+		boolInt(r.ShouldPush),
+		boolInt(r.PushResult.Pushed),
+		string(r.PushResult.Route),
+		r.PushResult.Detail,
+		r.PushResult.Err,
+	)
+	if err != nil {
+		return 0, fmt.Errorf("backthink: insert record: %w", err)
+	}
+	id, err := res.LastInsertId()
+	if err != nil {
+		return 0, fmt.Errorf("backthink: record id: %w", err)
+	}
+	if s.log != nil {
+		s.log.Debug("backthink record appended", "id", id, "trigger", r.Trigger, "should_push", r.ShouldPush, "route", r.PushResult.Route)
+	}
+	return id, nil
+}
+
+// List returns the most recent records newest-first. A non-positive limit selects
+// defaultListLimit.
+func (s *Store) List(ctx context.Context, limit int) ([]Record, error) {
+	if limit <= 0 {
+		limit = defaultListLimit
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	rows, err := s.db.QueryContext(ctx, selectRecordsSQL, limit)
+	if err != nil {
+		return nil, fmt.Errorf("backthink: list records: %w", err)
+	}
+	defer rows.Close()
+
+	records := make([]Record, 0, limit)
+	for rows.Next() {
+		var (
+			r          Record
+			at         string
+			route      string
+			pushed     int
+			shouldPush int
+		)
+		if err := rows.Scan(
+			&r.ID,
+			&r.Trigger,
+			&at,
+			&r.Conclusion,
+			&shouldPush,
+			&pushed,
+			&route,
+			&r.PushResult.Detail,
+			&r.PushResult.Err,
+		); err != nil {
+			return nil, fmt.Errorf("backthink: scan record: %w", err)
+		}
+		r.ShouldPush = intBool(shouldPush)
+		r.PushResult.Pushed = intBool(pushed)
+		r.PushResult.Route = Route(route)
+		r.At = parseTime(at)
+		records = append(records, r)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("backthink: iterate records: %w", err)
+	}
+	return records, nil
+}
+
+// boolInt maps a bool to the SQLite INTEGER 1/0 representation.
+func boolInt(b bool) int {
+	if b {
+		return 1
+	}
+	return 0
+}
+
+// intBool maps a SQLite INTEGER to a bool.
+func intBool(n int) bool { return n != 0 }
+
+// parseTime parses a stored RFC3339(nano/with-zone) timestamp, falling back to
+// the zero time so a malformed row never breaks a listing.
+func parseTime(s string) time.Time {
+	for _, layout := range []string{time.RFC3339Nano, time.RFC3339} {
+		if t, err := time.Parse(layout, s); err == nil {
+			return t
+		}
+	}
+	return time.Time{}
+}

internal/services/backthink/store_test.go

@@ -0,0 +1,86 @@
+package backthink
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestStore_AppendList(t *testing.T) {
+	t.Parallel()
+	s, err := NewStore(context.Background(), newTestDB(t), discardLog())
+	if err != nil {
+		t.Fatalf("NewStore: %v", err)
+	}
+
+	base := time.Date(2026, 9, 6, 10, 0, 0, 0, time.UTC)
+	for i := 0; i < 3; i++ {
+		id, err := s.Append(context.Background(), Record{
+			Trigger:    "iot.device_update",
+			At:         base.Add(time.Duration(i) * time.Minute),
+			Conclusion: "nothing to report",
+			ShouldPush: i%2 == 0,
+			PushResult: PushResult{Pushed: i == 2, Route: RouteActiveClient, Detail: "ok"},
+		})
+		if err != nil {
+			t.Fatalf("Append %d: %v", i, err)
+		}
+		if id != int64(i+1) {
+			t.Fatalf("Append id = %d, want %d", id, i+1)
+		}
+	}
+
+	records, err := s.List(context.Background(), 0)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if len(records) != 3 {
+		t.Fatalf("len = %d, want 3", len(records))
+	}
+	// Newest first (descending id).
+	if records[0].ID != 3 || records[1].ID != 2 || records[2].ID != 1 {
+		t.Fatalf("order = %d,%d,%d, want 3,2,1", records[0].ID, records[1].ID, records[2].ID)
+	}
+	if !records[0].ShouldPush {
+		t.Fatalf("records[0].ShouldPush = false, want true")
+	}
+	if !records[0].PushResult.Pushed || records[0].PushResult.Route != RouteActiveClient {
+		t.Fatalf("records[0].PushResult = %+v, want pushed active_client", records[0].PushResult)
+	}
+	if !records[0].At.Equal(base.Add(2 * time.Minute)) {
+		t.Fatalf("records[0].At = %v, want %v", records[0].At, base.Add(2*time.Minute))
+	}
+}
+
+func TestStore_List_Limit(t *testing.T) {
+	t.Parallel()
+	s, err := NewStore(context.Background(), newTestDB(t), discardLog())
+	if err != nil {
+		t.Fatalf("NewStore: %v", err)
+	}
+	for i := 0; i < 5; i++ {
+		if _, err := s.Append(context.Background(), Record{Trigger: "heartbeat", Conclusion: "x"}); err != nil {
+			t.Fatalf("Append %d: %v", i, err)
+		}
+	}
+
+	records, err := s.List(context.Background(), 2)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if len(records) != 2 {
+		t.Fatalf("len = %d, want 2", len(records))
+	}
+	if records[0].ID != 5 || records[1].ID != 4 {
+		t.Fatalf("ids = %d,%d, want 5,4", records[0].ID, records[1].ID)
+	}
+}
+
+func TestStore_NewStore_NilDB(t *testing.T) {
+	t.Parallel()
+	_, err := NewStore(context.Background(), nil, discardLog())
+	if !errors.Is(err, ErrNilDB) {
+		t.Fatalf("err = %v, want ErrNilDB", err)
+	}
+}

internal/services/backthink/thinker.go

@@ -0,0 +1,45 @@
+package backthink
+
+import (
+	"context"
+	"fmt"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
+)
+
+// LoopThinker adapts a *loop.Loop into a Thinker. It is the production wiring
+// that lets a background pass drive the full tool loop (memory injection, model
+// steps, tool calls such as IoT/calendar/reminders) without the service knowing
+// the loop's concrete shape.
+type LoopThinker struct {
+	loop *loop.Loop
+}
+
+// NewLoopThinker wraps l as a Thinker.
+func NewLoopThinker(l *loop.Loop) *LoopThinker { return &LoopThinker{loop: l} }
+
+// Think runs one unattended think through the wrapped loop and returns the
+// conclusion text. It returns ErrNoLoop when no loop was injected.
+func (t *LoopThinker) Think(ctx context.Context, sessionID, trigger string) (string, error) {
+	if t.loop == nil {
+		return "", ErrNoLoop
+	}
+	messages := []model.Message{model.TextMessage(model.RoleUser, backgroundPrompt(trigger))}
+	res, err := t.loop.Run(ctx, sessionID, messages)
+	if err != nil {
+		return "", fmt.Errorf("backthink: think: %w", err)
+	}
+	return res.Text, nil
+}
+
+// backgroundPrompt is the instruction handed to the runtime for one unattended
+// thought. It explicitly avoids asking for a chat-style message to the user.
+func backgroundPrompt(trigger string) string {
+	return "[background autonomous reflection]\n" +
+		"Trigger: " + trigger + "\n" +
+		"Review recent context, device states, calendar and reminders. Decide " +
+		"whether anything is worth proactively telling the owner. If nothing " +
+		"matters, conclude \"nothing to report\". Keep the conclusion to one or " +
+		"two sentences. Do not produce a chat-style message to the user."
+}

internal/services/backthink/trigger.go

@@ -0,0 +1,55 @@
+package backthink
+
+import (
+	"fmt"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
+)
+
+// 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.
+const (
+	// SubjectReminderTrigger is the bus subject on which reminders fire.
+	SubjectReminderTrigger = "reminder.trigger"
+	// SubjectSystemEvent is the bus subject for generic system events.
+	SubjectSystemEvent = "system.event"
+)
+
+// Trigger describes what woke a thinking pass: the source subsystem, a
+// human-readable summary and the event's estimated importance.
+type Trigger struct {
+	Source  string
+	Summary string
+	Import  Importance
+}
+
+// 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}
+}
+
+// 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.
+func triggerForEvent(e bus.Event) Trigger {
+	switch e.Subject {
+	case iot.DeviceUpdateSubject:
+		if d, ok := e.Payload.(iot.DeviceUpdate); ok {
+			return Trigger{
+				Source:  e.Subject,
+				Summary: fmt.Sprintf("device %s (%s) %s", d.Device.ID, d.Device.Type, d.Action),
+				Import:  ImportanceNormal,
+			}
+		}
+		return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal}
+	case SubjectReminderTrigger:
+		return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceHigh}
+	case SubjectSystemEvent:
+		return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal}
+	default:
+		return Trigger{Source: e.Subject, Summary: e.Subject, Import: ImportanceNormal}
+	}
+}