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. The
// proactive-push rate gate of DevDocs/design/11 §6 governs the periodic heartbeat
// only; an event-driven push bypasses it (owner mandate: "该发就发") subject to
// same-event dedup (see event_dedup.go). Every pass is persisted to SQLite via
// Store so the M9/M10 monitor panel can render the background-thinking page
// (design/12 §4).
//
// The package is concurrency-safe: a mutex guards the lifecycle state, the
// injected store/pusher/thinker are required to be concurrency-safe by contract,
// and the internal publish path never waits on a slow consumer.
package backthink

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)

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

// 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
	// Recent supplies the most recent main-session messages so the activity gate
	// can sense whether the owner just chatted (recent-conversation cooldown and
	// idle-window suppression). Nil disables activity sensing.
	Recent recentProvider
	// 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
	// Exporter mirrors a proactive message to the external push outlet (the
	// owner's self-hosted unified push service, then Getui). Optional: nil
	// disables the hook. It never blocks or breaks the Active/IM delivery.
	Exporter PushExporter
	// 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
	gate      PushGate
	now       func() time.Time
	sessionID string

	mu      sync.Mutex
	running bool
	cancel  context.CancelFunc
	subs    []func()
	wg      sync.WaitGroup

	// eventMu guards the same-event dedup guard (event_dedup.go). It is separate
	// from the lifecycle lock so a dedup decision never blocks a sink swap.
	eventMu      sync.Mutex
	lastEventKey string
}

// New builds a Backthink from cfg and deps, applying defaults and validating the
// 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)
	}

	// Resolve the gate up front so the sink setters can rebuild the pusher with it.
	gate := deps.Gate
	if gate == nil {
		gate = newRateGate(cfg.PushLimit, cfg.PushWindow, cfg.PushCooldown, cfg.KindWindow, log)
	}
	pusher := deps.Pusher
	if pusher == nil {
		pusher = newPusher(deps.Active, deps.IM, gate, log).withExporter(deps.Exporter)
	}

	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,
		gate:      gate,
		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. It first
// applies the minimal-disturbance gate (design/12 §6 策略一): a low-value system
// event is downgraded to a registration-only record rather than waking a full
// thinking pass, so platform/system noise cannot flood the thinker.
func (b *Backthink) onEvent(ctx context.Context, e bus.Event) error {
	tr := triggerForEvent(e)
	if !b.shouldThink(tr) {
		b.log.Debug("backthink system event downgraded, no thinking",
			"source", tr.Source, "summary", tr.Summary)
		b.record(Record{
			Trigger:    tr.Source,
			At:         b.now(),
			Conclusion: "system event: low value, not worth waking a thinking pass",
			ShouldPush: false,
			PushResult: PushResult{Route: RouteNone, Detail: "downgraded"},
		})
		return nil
	}
	b.think(ctx, tr)
	return nil
}

// shouldThink gates event-triggered thinking (design/12 §6 策略一). On a public
// platform a generic system event is low value, so only a high-importance
// (owner-relevant) system event wakes a pass. IoT device updates and reminder
// triggers are owner-relevant / high value and are never gated, so the productive
// triggers stay fully functional.
func (b *Backthink) shouldThink(tr Trigger) bool {
	if tr.Source == SubjectSystemEvent {
		return tr.Import >= ImportanceHigh
	}
	return true
}

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

	// P2 (design/20 §2): her own conclusion decides whether to push and to which
	// channel — not the trigger's importance. parseProactive recovers the decision
	// robustly and fails safe (malformed/ambiguous → no push).
	message, channelHint, push := parseProactive(conclusion)

	state := b.currentState(cctx)
	lastActivity, hasActivity := b.lastUserActivity(cctx)
	// The "do not disturb" gate (asleep/working/busy/quiet) still applies as a hard
	// safety; the activity gate (fresh conversation / long idle) softens it; the
	// frequency gate (PushGate) is applied inside the pusher.
	shouldPush := push && b.disturbAllowed(at, state) && b.activityAllowed(tr.Import, lastActivity, hasActivity, at)

	recordConclusion := message
	if recordConclusion == "" {
		recordConclusion = conclusion
	}

	pushRes := PushResult{Route: RouteNone, Detail: b.interruptReason(push, at, state, tr.Import, lastActivity, hasActivity)}
	if shouldPush {
		// Same-event dedup: a repeating identical event is suppressed after the
		// first push (防刷), while a genuinely new event still goes out. Heartbeats
		// are never deduped (they reset the chain) and stay gate-governed.
		if b.eventRedundant(tr) {
			pushRes = PushResult{Route: RouteNone, Detail: "duplicate event push suppressed"}
		} else {
			pushRes = b.currentPusher().Push(cctx, PushRequest{
				Message:     b.buildProactiveMessage(message, channelHint),
				ChannelHint: channelHint,
				TriggerKind: tr.Kind,
			})
		}
	}

	b.record(Record{
		Trigger:    tr.Source,
		At:         at,
		Conclusion: recordConclusion,
		ShouldPush: shouldPush,
		PushResult: pushRes,
	})

	if shouldPush {
		b.log.Info("backthink proactive push", "source", tr.Source, "route", pushRes.Route, "pushed", pushRes.Pushed, "bypass", pushRes.Bypassed, "reason", pushRes.Detail)
	} else {
		b.log.Debug("backthink did not interrupt", "source", tr.Source, "reason", pushRes.Detail)
	}
}

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

// lastUserActivity reports the wall-clock time of the owner's most recent message
// in the main session, and whether any user message was found. It is the activity
// signal for the activity gate (design/12 §6): a nil provider, a failed read or an
// empty tail yield has=false so the activity gate falls through to the current
// behaviour rather than guessing. The most recent user message is the newest
// RoleUser entry in the recent tail (Recent returns oldest-first).
func (b *Backthink) lastUserActivity(ctx context.Context) (time.Time, bool) {
	if b.deps.Recent == nil {
		return time.Time{}, false
	}
	msgs, err := b.deps.Recent.Recent(ctx, history.MainSessionID, recentContextCount)
	if err != nil {
		return time.Time{}, false
	}
	for i := len(msgs) - 1; i >= 0; i-- {
		if msgs[i].Role == history.RoleUser {
			return time.UnixMilli(msgs[i].CreatedAt), true
		}
	}
	return time.Time{}, false
}

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