Cyrene_v2/internal/services/backthink/backthink.go

// 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)
	}
	// Surface the persisted pass on the WS stream so the frontend's backthink
	// consumer (or its default fallback) observes it without polling the store.
	if b.deps.Bus != nil {
		if err := b.deps.Bus.Publish(context.Background(), "ws.backthink", r); err != nil {
			b.log.Warn("backthink bus publish 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))
}