Maven/cmd/mavend/tick.go

// mavend/tick.go — the proactive loop driver.
//
// Per spec the 60s schedule loop (`for { tick; sleep }`) lives in the daemon
// main, NOT in `internal/loop/` — that keeps the loop package pure + unit-
// testable without time side effects. the driver here is the ONE impure
// orchestrator: it gathers state under the store lock, runs the pure Tick,
// phrases the candidate, dispatches it, then handles reminders + sev4 repeats
// and runs the feedback auto-tuner on its own slow cadence.
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/kami/maven/internal/config"
	"github.com/kami/maven/internal/delivery"
	"github.com/kami/maven/internal/ipc"
	"github.com/kami/maven/internal/loop"
	"github.com/kami/maven/internal/phraser"
	"github.com/kami/maven/internal/store"
)

// QueuedNudge — a nudge held in the digest queue pending batch flush.
type QueuedNudge struct {
	Rule     string
	Severity int
	Body     string
	Key      string
	QueuedAt time.Time
}

// tickLoop — the impure driver. holds everything wired at daemon construction
// that the per-tick path needs. the rules slice is read-only here; the gatherer
// already captured it, but we keep it for a possible future re-seed path.
type tickLoop struct {
	store      *store.Store
	gatherer   *loop.Gatherer
	dispatcher *delivery.Dispatcher
	phraser    phraser.Phraser
	rules      []loop.Rule

	tickInterval   time.Duration
	repeatInterval time.Duration
	autotuneInterval time.Duration // 0 ⇒ autotune disabled (gatherer falls back to static Base)

	// digestCfg — the digest/batching config. nil ⇒ every nudge is sent
	// immediately (legacy behaviour).
	digestCfg *config.DigestConfig

	// digestQ — in-memory queue of eligible nudges waiting for batch flush.
	// populated when digestCfg != nil && digestCfg.Enabled.
	digestQ []QueuedNudge

	// lastPhrase caches the phraser output per rule so the sev4 repeat path
	// can re-send roughly what the user was first alerted with (an alarm
	// that re-phrases differently every 5m is hostile; the same terse body
	// IS the insistence signal). keyed by rule name. nil phrase for a rule
	// = no successful initial dispatch yet (cold-start edge — fall back to
	// a generic body).
	mu         sync.Mutex
	lastPhrase map[string]delivery.PhrasedNudge
	lastTrace  *loop.TickTrace // cached from the most recent tick
}

func newTickLoop(
	st *store.Store,
	g *loop.Gatherer,
	d *delivery.Dispatcher,
	p phraser.Phraser,
	rules []loop.Rule,
	tickInterval, repeatInterval, autotuneInterval time.Duration,
	digestCfg *config.DigestConfig,
) *tickLoop {
	return &tickLoop{
		store:            st,
		gatherer:         g,
		dispatcher:       d,
		phraser:          p,
		rules:            rules,
		tickInterval:     tickInterval,
		repeatInterval:   repeatInterval,
		autotuneInterval: autotuneInterval,
		digestCfg:        digestCfg,
		digestQ:          nil,
		lastPhrase:       make(map[string]delivery.PhrasedNudge),
	}
}

// run drives the loop until ctx is canceled. one tick per tickInterval;
// the first tick fires immediately so a freshly-started daemon doesn't sit
// idle for 60s before its first evaluation (cold-start responsiveness). the
// feedback auto-tuner runs on its own slower ticker (autotuneInterval) so it
// doesn't write a feedback fact every tick — append-only facts would churn.
func (t *tickLoop) run(ctx context.Context) {
	t.tick(ctx, time.Now())
	ticker := time.NewTicker(t.tickInterval)
	defer ticker.Stop()

	var autotune *time.Ticker
	var autotuneC <-chan time.Time
	if t.autotuneInterval > 0 {
		autotune = time.NewTicker(t.autotuneInterval)
		defer autotune.Stop()
		autotuneC = autotune.C
	}

	for {
		select {
		case <-ctx.Done():
			return
		case now := <-ticker.C:
			t.tick(ctx, now)
		case <-autotuneC:
			t.tune(ctx)
		}
	}
}

// tick — one pass of the proactive loop. gathers, decides, phrases, delivers.
// errors at any sub-step are logged and the tick continues / aborts as the
// layer warrants: a gather failure aborts (no consistent snapshot ⇒ no
// decisions); a phrase/dispatch failure logs the failure and continues so a
// transient delivery fault doesn't kill the whole loop.
func (t *tickLoop) tick(ctx context.Context, now time.Time) {
	state, due, err := t.gatherer.GatherState(ctx, now)
	if err != nil {
		log.Printf("tick: gather: %v", err)
		return
	}

	// proactive: at most one candidate, max severity.
	cand, trace := loop.ExplainTick(state, t.rules)
	t.mu.Lock()
	t.lastTrace = trace
	t.mu.Unlock()
	if cand != nil {
		if t.shouldQueue(cand) {
			t.queueNudge(ctx, cand, state, now)
		} else {
			pn, err := t.phraser.PhraseNudge(ctx, *cand)
			if err != nil {
				log.Printf("tick: phrase nudge %s: %v", cand.Rule.Name, err)
			} else {
				t.cachePhrase(pn)
				if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil {
					log.Printf("tick: dispatch nudge %s: %v", cand.Rule.Name, err)
				}
			}
		}
	}

	// digest flush: after candidate processing, flush if the window has
	// elapsed or MaxItems was reached. Processing the candidate first
	// (with dedup) avoids re-queueing the same rule after a flush.
	t.maybeFlush(ctx, now, state)

	// reminders: gate-bypassing class. fired once, marked after a successful
	// delivery.  a failed send leaves the reminder pending — the next tick
	// re-gathers and re-attempts.
	for _, d := range loop.RemindDecisions(state, due) {
		pr, err := t.phraser.PhraseReminder(ctx, d)
		if err != nil {
			log.Printf("tick: phrase reminder %d: %v", d.Reminder.ID, err)
			continue
		}
		if _, err := t.dispatcher.DispatchReminder(ctx, pr, now); err != nil {
			log.Printf("tick: dispatch reminder %d: %v", d.Reminder.ID, err)
		}
	}

	// sev4-away repeats: re-send un-acked telegram nudges per repeatInterval.
	// the source of ack truth IS the nudges table (outcome=pending ⇒ not
	// acked); store.UnackedTelegramRules surfaces the keys. body/summary come
	// from the cached phrase from the initial dispatch — see lastPhrase notes.
	// if not cached (cold-start mid-alarm), fall back to a terse generic body.
	keys, err := t.store.UnackedTelegramRules(ctx)
	if err != nil {
		log.Printf("tick: unacked telegram rules: %v", err)
		return
	}
	if len(keys) == 0 {
		return
	}
	for _, key := range keys {
		body, summary := t.repeatPhrase(key)
		if _, err := t.dispatcher.RepeatUnacked(ctx, []string{key}, now, t.repeatInterval, body, summary); err != nil {
			log.Printf("tick: repeat telegram %s: %v", key, err)
		}
	}
}

// cachePhrase keeps the latest phrased nudge per rule for the sev4-repeat
// path. writing under a mutex; the repeat path reads under the same. the
// cache is bounded by the rule count (≤ ~30 per spec) so eviction is not a
// concern at this scale.
func (t *tickLoop) cachePhrase(pn delivery.PhrasedNudge) {
	t.mu.Lock()
	t.lastPhrase[pn.Candidate.Rule.Name] = pn
	t.mu.Unlock()
}

func (t *tickLoop) repeatPhrase(rule string) (body, summary string) {
	t.mu.Lock()
	pn, ok := t.lastPhrase[rule]
	t.mu.Unlock()
	if !ok || pn.Summary == "" {
		// cold-start mid-alarm: no cached phrase. a deliberately terse generic
		// body — the alarm IS the insistence; the wording repeats, the ring
		// is what changes. the LLM phraser impl will refresh this on its next
		// tick when the rule re-fires through Tick.
		return fmt.Sprintf("maven: %s still active", rule), rule
	}
	return pn.Body, pn.Summary
}

// shouldQueue — true when digest is enabled and the candidate's severity is
// at or below the configured ceiling.
func (t *tickLoop) shouldQueue(cand *loop.Candidate) bool {
	return t.digestCfg != nil && t.digestCfg.Enabled &&
		cand.Severity <= loop.Severity(t.digestCfg.SeverityCeiling)
}

// queueNudge — phrases the candidate and appends it to the digest queue.
// Deduplicates by rule name: if the same rule is already queued, this is a
// no-op (the first fire within the window is the one that counts).
func (t *tickLoop) queueNudge(ctx context.Context, cand *loop.Candidate, _ loop.State, now time.Time) {
	for _, q := range t.digestQ {
		if q.Rule == cand.Rule.Name {
			return // already queued
		}
	}
	pn, err := t.phraser.PhraseNudge(ctx, *cand)
	if err != nil {
		log.Printf("tick: phrase nudge %s: %v", cand.Rule.Name, err)
		return
	}
	t.digestQ = append(t.digestQ, QueuedNudge{
		Rule:     cand.Rule.Name,
		Severity: int(cand.Severity),
		Body:     pn.Body,
		Key:      cand.Rule.Name,
		QueuedAt: now,
	})
	t.cachePhrase(pn)
}

// maybeFlush — flushes the digest queue if the window has elapsed since the
// first item or the queue reached MaxItems.
func (t *tickLoop) maybeFlush(ctx context.Context, now time.Time, state loop.State) {
	if t.digestCfg == nil || !t.digestCfg.Enabled || len(t.digestQ) == 0 {
		return
	}
	first := t.digestQ[0]
	if now.Sub(first.QueuedAt) >= time.Duration(t.digestCfg.Window) ||
		len(t.digestQ) >= t.digestCfg.MaxItems {
		t.flushDigest(ctx, now, state)
	}
}

// flushDigest — concatenates queued nudge bodies into a single digest
// notification and dispatches it. Clears the queue after a successful send.
// The digest uses the max severity among queued items for routing.
func (t *tickLoop) flushDigest(ctx context.Context, now time.Time, state loop.State) {
	if len(t.digestQ) == 0 {
		return
	}

	var b strings.Builder
	maxSev := 0
	for i, q := range t.digestQ {
		if i > 0 {
			b.WriteString(" · ")
		}
		b.WriteString(q.Body)
		if q.Severity > maxSev {
			maxSev = q.Severity
		}
	}
	body := b.String()
	summary := fmt.Sprintf("%d pending notifications", len(t.digestQ))

	cand := loop.Candidate{
		Rule: loop.Rule{
			Name:     "digest",
			Severity: loop.Severity(maxSev),
		},
		Severity: loop.Severity(maxSev),
		State:    state,
	}
	pn := delivery.PhrasedNudge{
		Candidate: cand,
		Body:      body,
		Summary:   summary,
	}
	t.cachePhrase(pn)
	if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil {
		// keep the queue — the next tick's maybeFlush re-attempts.
		log.Printf("tick: dispatch digest: %v", err)
		return
	}
	t.digestQ = nil
}

// tune — the feedback auto-tuner's impure step. runs on a slow cadence
// (autotuneInterval, see run) so it doesn't write a fact every tick. for each
// rule:
//  1. read store.RecentOutcomes for the last TuneSampleN resolved outcomes.
//  2. if there's not enough signal (TuneMinOutcomes), leave Base alone.
//  3. compute the tuned cooldown with loop.TuneCooldown (pure).
//  4. read the currently-persisted feedback fact; if the tuned value equals
//     it, skip the write (RecentOutcomes is itself steady ⇒ no churn).
//  5. else write a `facts (kind=config, source=feedback, key=cooldown:<rule>)`
//     row. the Gatherer reads it next tick.
//
// Error at any step logs + continues to the next rule — a transient store
// fault on one rule must not abort tuning for the rest.
func (t *tickLoop) tune(ctx context.Context) {
	now := time.Now()
	for _, r := range t.rules {
		outcomes, err := t.store.RecentOutcomes(ctx, r.Name, loop.TuneSampleN)
		if err != nil {
			log.Printf("tune: outcomes %s: %v", r.Name, err)
			continue
		}
		if len(outcomes) < loop.TuneMinOutcomes {
			continue // sparse — no signal yet, don't whipsaw on first sight.
		}
		tuned := loop.TuneCooldown(r, outcomes)

		// the currently-persisted tuned base, if any. equal ⇒ skip the write
		// (RecentOutcomes is monotone-steady between resolved outcomes).
		if cur, ok := t.currentTunedBase(ctx, r); ok && cur == tuned {
			continue
		}
		if _, err := t.store.SetValue(
			ctx, store.KindConfig, loop.FeedbackKey(r), loop.FeedbackSource,
			tuned, now,
		); err != nil {
			log.Printf("tune: persist %s: %v", r.Name, err)
			continue
		}
		log.Printf("tune: %s cooldown -> %v", r.Name, tuned)
	}
}

// currentTunedBase — read the persisted feedback cooldown fact back into a
// duration. (dur, false) when no feedback fact exists yet OR it's malformed;
// the caller treats that as "differ from anything we'd write — write."
func (t *tickLoop) currentTunedBase(ctx context.Context, r loop.Rule) (time.Duration, bool) {
	f, err := t.store.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource)
	if err != nil {
		return 0, false
	}
	return loop.ParseCooldownFact(f)
}

// defaultConfigPath — the config file path the daemon loads if -config wasn't
// passed. XDG_CONFIG_HOME/maven/mavend.json, falling back to ~/.config/maven.
func defaultConfigPath() string {
	if x := os.Getenv("XDG_CONFIG_HOME"); x != "" {
		return filepath.Join(x, "maven", "mavend.json")
	}
	home, err := os.UserHomeDir()
	if err != nil || home == "" {
		return "mavend.json"
	}
	return filepath.Join(home, ".config", "maven", "mavend.json")
}

// trace returns the most recent TickTrace, or nil if no tick has run yet.
func (t *tickLoop) trace() *loop.TickTrace {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.lastTrace
}

// daemonAPI wraps a store-backed CoreAPI and overrides TickTrace with the
// daemon's in-memory tick trace cache.
type daemonAPI struct {
	ipc.CoreAPI
	getTrace func() *loop.TickTrace
}

func (d *daemonAPI) TickTrace(ctx context.Context) (ipc.TickTrace, error) {
	trace := d.getTrace()
	if trace == nil {
		return ipc.TickTrace{}, nil
	}
	return toIPCTickTrace(*trace), nil
}

func toIPCTickTrace(t loop.TickTrace) ipc.TickTrace {
	rules := make([]ipc.RuleTrace, len(t.RuleTraces))
	for i, r := range t.RuleTraces {
		rules[i] = toIPCRuleTrace(r)
	}
	return ipc.TickTrace{
		Now:    t.Now,
		Winner: t.Winner,
		Rules:  rules,
	}
}

func toIPCRuleTrace(r loop.RuleTrace) ipc.RuleTrace {
	return ipc.RuleTrace{
		RuleName:        r.RuleName,
		Severity:        int(r.Severity),
		PredicateResult: r.PredicateResult,
		GateResult:      r.GateResult,
		GateBlockedBy:   r.GateBlockedBy,
		GateDetail:      toIPCGateDetail(r.GateDetail),
		WasSelected:     r.WasSelected,
		LostTo:          r.LostTo,
	}
}

func toIPCGateDetail(d loop.GateDetail) ipc.GateDetail {
	return ipc.GateDetail{
		SnoozeUntil:      d.SnoozeUntil,
		CooldownUntil:    d.CooldownUntil,
		QuietHours:       d.QuietHours,
		CalendarBusy:     d.CalendarBusy,
		Presence:         d.Presence,
		InertKeysMissing: d.InertKeysMissing,
	}
}