// 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" "errors" "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/morning" "github.com/kami/maven/internal/phraser" "github.com/kami/maven/internal/routine" "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 // routines — operator-declared scheduled behaviors. fired through the // dispatcher when their cron crosses. routineLast tracks the per-routine // last-fire time across ticks (the driver owns it; routine.Due mutates it). routines []routine.Routine routineLast map[string]time.Time // morningRoutines — daily checklists (see internal/morning). morningLast // tracks the per-routine last-nudge day, mirroring routineLast. morningRoutines []morning.Routine morningLast map[string]time.Time // 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, routines []routine.Routine, morningRoutines []morning.Routine, ) *tickLoop { return &tickLoop{ store: st, gatherer: g, dispatcher: d, phraser: p, rules: rules, tickInterval: tickInterval, repeatInterval: repeatInterval, autotuneInterval: autotuneInterval, digestCfg: digestCfg, digestQ: nil, routines: routines, routineLast: make(map[string]time.Time), morningRoutines: morningRoutines, morningLast: make(map[string]time.Time), 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) // routines: operator-declared scheduled behaviors. fire the ones whose cron // crossed since last fire, delivered through the normal routing (voice when // present, away channels otherwise). bodies are literal operator text — not // LLM-phrased — so a routine can't hallucinate. severity comes from config. t.fireRoutines(ctx, now, state) // morning routines: daily checklists (medicine/water/pets/...), nagged at // most once per day per routine, and only for items still unevidenced at // nudge time. See internal/morning for the "why not four timers" rationale. t.fireMorningRoutines(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 } // routinesFromConfig maps the config's routine blocks to the engine type. // Validation (cron parses, name/body present, severity defaulted) already ran // in config.Load, so this is a pure field copy. func routinesFromConfig(rc []config.RoutineConfig) []routine.Routine { if len(rc) == 0 { return nil } out := make([]routine.Routine, len(rc)) for i, r := range rc { out[i] = routine.Routine{Name: r.Name, Cron: r.Cron, Body: r.Body, Severity: r.Severity} } return out } // fireRoutines dispatches the routines whose cron schedule crossed since their // last fire. Each is delivered as a nudge through the normal routing table // (ChannelsFor(severity, presence)) with a "routine:"-prefixed rule name so it // can't collide with a care rule in the feedback autotuner. A dispatch failure // logs and continues — one bad send must not skip the rest, and routine.Due has // already advanced the last-fire time so a transient failure drops that fire // rather than replaying it every tick (a routine is clockwork, not an alarm — // no repeat-til-ack). func (t *tickLoop) fireRoutines(ctx context.Context, now time.Time, state loop.State) { for _, r := range routine.Due(t.routines, t.routineLast, now) { pn := delivery.PhrasedNudge{ Candidate: loop.Candidate{ Rule: loop.Rule{Name: "routine:" + r.Name, Severity: loop.Severity(r.Severity)}, Severity: loop.Severity(r.Severity), State: state, }, Body: r.Body, Summary: r.Body, } if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil { log.Printf("tick: dispatch routine %s: %v", r.Name, err) } } } // fireMorningRoutines checks each configured checklist against today's facts // and dispatches a nag listing exactly what's still missing, at most once per // routine per calendar day. Fact reads happen here (not in loop.Gatherer) // because the item↔fact-key mapping is morning-routine-specific, not a rule // concern — pulling it into the shared gather path would leak that mapping // into loop's "rules declare wanted keys" contract. Bodies are literal // operator text (item labels joined), not LLM-phrased, same rationale as // cron routines: deterministic, can't hallucinate a checklist item. func (t *tickLoop) fireMorningRoutines(ctx context.Context, now time.Time, state loop.State) { if len(t.morningRoutines) == 0 { return } facts := t.gatherMorningFacts(ctx) for _, cand := range morning.Due(t.morningRoutines, facts, t.morningLast, now) { labels := make([]string, len(cand.Missing)) for i, it := range cand.Missing { labels[i] = it.Label } body := fmt.Sprintf("%s: не сделано — %s", cand.Routine.Name, strings.Join(labels, ", ")) pn := delivery.PhrasedNudge{ Candidate: loop.Candidate{ Rule: loop.Rule{Name: "morning:" + cand.Routine.Name, Severity: loop.Severity(cand.Routine.Severity)}, Severity: loop.Severity(cand.Routine.Severity), State: state, }, Body: body, Summary: body, } if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil { log.Printf("tick: dispatch morning routine %s: %v", cand.Routine.Name, err) } } } // gatherMorningFacts reads the latest fact for every item's fact_key across // all configured morning routines. Shared by fireMorningRoutines (nudge // decision) and morningStatus (read-only query) so the two paths can never // disagree about what evidence exists. func (t *tickLoop) gatherMorningFacts(ctx context.Context) map[string]store.Fact { keys := make(map[string]struct{}) for _, r := range t.morningRoutines { for _, it := range r.Items { keys[it.FactKey] = struct{}{} } } facts := make(map[string]store.Fact, len(keys)) for k := range keys { f, err := t.store.LatestFact(ctx, k) if err == nil { facts[k] = f continue } if err != store.ErrNoFact { log.Printf("tick: morning: latest fact %s: %v", k, err) } } return facts } // morningStatus is the read-only "what's missing" query the web UI (and // eventually a voice query) calls. Pure recompute over the current facts — // no dedupe/nudge-time gating, unlike fireMorningRoutines: this answers // "state right now," not "should we nag." func (t *tickLoop) morningStatus(ctx context.Context, now time.Time) []ipc.MorningRoutineStatus { if len(t.morningRoutines) == 0 { return nil } facts := t.gatherMorningFacts(ctx) out := make([]ipc.MorningRoutineStatus, 0, len(t.morningRoutines)) for _, r := range t.morningRoutines { st := morning.Evaluate(r, facts, now) done := make(map[string]bool, len(st.Completed)) for _, it := range st.Completed { done[it.Key] = true } items := make([]ipc.MorningRoutineItem, len(r.Items)) for i, it := range r.Items { items[i] = ipc.MorningRoutineItem{Key: it.Key, Label: it.Label, Done: done[it.Key]} } out = append(out, ipc.MorningRoutineStatus{ Name: r.Name, Active: st.Active, WindowStart: r.WindowStart, WindowEnd: r.WindowEnd, Items: items, }) } return out } // 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:)` // 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 getMorningStatus func(ctx context.Context) []ipc.MorningRoutineStatus chatFn func(ctx context.Context, text string) string } func (d *daemonAPI) Chat(ctx context.Context, text string) (string, error) { if d.chatFn == nil { return "", errors.New("mavend: chat not available") } return d.chatFn(ctx, text), nil } 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 (d *daemonAPI) MorningStatus(ctx context.Context) ([]ipc.MorningRoutineStatus, error) { if d.getMorningStatus == nil { return nil, errors.New("mavend: morning status not available") } return d.getMorningStatus(ctx), 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, } }