diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go new file mode 100644 index 0000000..e39bf28 --- /dev/null +++ b/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)) +} diff --git a/internal/services/backthink/backthink_test.go b/internal/services/backthink/backthink_test.go new file mode 100644 index 0000000..337f970 --- /dev/null +++ b/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) + } +} diff --git a/internal/services/backthink/decide.go b/internal/services/backthink/decide.go new file mode 100644 index 0000000..26ac40c --- /dev/null +++ b/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 +} diff --git a/internal/services/backthink/decide_test.go b/internal/services/backthink/decide_test.go new file mode 100644 index 0000000..e3ec315 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/internal/services/backthink/mocks_test.go b/internal/services/backthink/mocks_test.go new file mode 100644 index 0000000..b65fbdb --- /dev/null +++ b/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) +} diff --git a/internal/services/backthink/push.go b/internal/services/backthink/push.go new file mode 100644 index 0000000..f8b9b74 --- /dev/null +++ b/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(), + } +} diff --git a/internal/services/backthink/push_test.go b/internal/services/backthink/push_test.go new file mode 100644 index 0000000..d2bf7ff --- /dev/null +++ b/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) + } +} diff --git a/internal/services/backthink/state.go b/internal/services/backthink/state.go new file mode 100644 index 0000000..d0793b5 --- /dev/null +++ b/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 +} diff --git a/internal/services/backthink/state_test.go b/internal/services/backthink/state_test.go new file mode 100644 index 0000000..5e7b004 --- /dev/null +++ b/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) + } +} diff --git a/internal/services/backthink/store.go b/internal/services/backthink/store.go new file mode 100644 index 0000000..f60214f --- /dev/null +++ b/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{} +} diff --git a/internal/services/backthink/store_test.go b/internal/services/backthink/store_test.go new file mode 100644 index 0000000..0f7d19b --- /dev/null +++ b/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) + } +} diff --git a/internal/services/backthink/thinker.go b/internal/services/backthink/thinker.go new file mode 100644 index 0000000..f64ace7 --- /dev/null +++ b/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." +} diff --git a/internal/services/backthink/trigger.go b/internal/services/backthink/trigger.go new file mode 100644 index 0000000..23a13c5 --- /dev/null +++ b/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} + } +}