766ca091a7
The digestion tick already runs the pattern detector over all recorded
events (67563ed) and writes a proposed_routines row. What was missing is
the other half of #247: a proposal that nobody is at the mic for reaches
nothing but the /routines page, so a pattern noticed at 03:00 is only
seen if he goes looking.
This wires the tick's proposals into the existing care-delivery path
rather than a second channel: sev1 nudge, loop.Gate, dispatcher, same
routing table as an accepted routine. Restraints, since a feature that
speaks unprompted is the easiest way to turn Maven into a nag:
- off unless configured — the new pattern_proposals block, absent by
default, and deploy/mavend.json ships notify: false;
- at most one announcement per tick however many patterns surfaced;
- at most one per cooldown (24h default) across all pairs;
- sev1, so quiet hours, away and snooze suppress it;
- suppressed means dropped, not queued — /routines still has it;
- once per pair for good, since proposed_routines is
UNIQUE(action, object) and the row survives dismissal.
The body is pattern.PhraseRoutine's literal Russian, not LLM-worded, so
an inferred routine cannot arrive describing something never observed.
Also raises pattern.MinEvents from 3 to 4 — the interval-quality item on
#43. Two intervals with a ±50% band is a coincidence with a mean, not a
pattern, and now that a scan of all history can announce itself the cost
of a false positive is a permanent dismissal of that pair.
931 lines
34 KiB
Go
931 lines
34 KiB
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"
|
|
"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/pattern"
|
|
"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
|
|
|
|
// proposalCfg — announcement policy for routines the tick inferred itself.
|
|
// nil ⇒ detect silently, never announce (the default). lastProposalAt is
|
|
// the cooldown clock, in-memory on purpose: a restart is allowed to permit
|
|
// one more announcement, and a restart-per-day loop is a bigger problem
|
|
// than a duplicate proposal notice.
|
|
proposalCfg *config.PatternProposalConfig
|
|
lastProposalAt 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,
|
|
proposalCfg *config.PatternProposalConfig,
|
|
) *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),
|
|
proposalCfg: proposalCfg,
|
|
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)
|
|
|
|
// gate-suppressed digest (Vikunja #281): rules the restraint gate held
|
|
// back this tick (quiet hours / away / calendar-busy), not because they
|
|
// weren't due, but because it wasn't the moment. Some of those are worth
|
|
// resurfacing later instead of just being lost — loop.DigestEligible
|
|
// draws that line. This is a SEPARATE mechanism from the in-memory
|
|
// digestQ above: that one batches candidates the gate already ALLOWED to
|
|
// fire; this one durably holds candidates the gate BLOCKED.
|
|
t.enqueueSuppressedDigest(ctx, trace, state, now)
|
|
t.expireStaleDigest(ctx, now)
|
|
t.maybeDrainDigest(ctx, state, now)
|
|
|
|
// 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)
|
|
|
|
// accepted routines: patterns the user confirmed. read straight from the
|
|
// store each tick so the schedule survives a restart.
|
|
t.fireAcceptedRoutines(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)
|
|
|
|
// pattern detection: scan every action+object pair with recorded events
|
|
// and propose a routine for any stable one not already decided (Vikunja
|
|
// #43). This used to only run as a side effect of the voice fact-write
|
|
// path, so a pattern already sitting in history went unnoticed until he
|
|
// happened to mention it again by voice. See patterns.go and
|
|
// detectPatterns below for how idempotence and dismissal are respected.
|
|
t.detectPatterns(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
|
|
}
|
|
|
|
// detectPatterns runs the pattern detector proactively over every
|
|
// action+object pair that has ever produced an event, independent of
|
|
// whichever fact write (or channel) last touched it (Vikunja #43). This is
|
|
// what makes pattern inference actually proactive: it fires on the daemon's
|
|
// own schedule reading accumulated history, not only as a side effect of a
|
|
// live voice turn.
|
|
//
|
|
// Idempotence and noise are handled by the store, not here — this function
|
|
// is safe to call every tick:
|
|
// - Same pattern, tick after tick: detectAndPropose's LookupProposedRoutine
|
|
// check plus proposed_routines' UNIQUE(action, object) constraint (with
|
|
// CreateProposedRoutine's ON CONFLICT DO NOTHING) mean a pair that
|
|
// already has a row — in ANY status — produces no second row and no log
|
|
// spam beyond the one line at genuine creation.
|
|
// - A DISMISSED proposal must never come back. DismissProposedRoutine flips
|
|
// status in place; the row is never deleted. So the same Lookup check
|
|
// that stops a duplicate "proposed" also stops a "dismissed" one from
|
|
// resurrecting — there is nothing tick-specific to get right here beyond
|
|
// calling the same shared path the voice route already used.
|
|
//
|
|
// By default this only creates a row for the /routines page to show: it does
|
|
// not notify, ring, or speak. Detection is not the same act as disturbing him
|
|
// about it, and Maven is "not a nag, not autonomous" (CLAUDE.md). Announcing
|
|
// is opt-in through the pattern_proposals config block — see announceProposal
|
|
// for the restraints that apply even then. A proposal only starts producing
|
|
// recurring nudges once he accepts it (fireAcceptedRoutines).
|
|
func (t *tickLoop) detectPatterns(ctx context.Context, now time.Time, state loop.State) {
|
|
pairs, err := t.store.DistinctEventPairs(ctx)
|
|
if err != nil {
|
|
log.Printf("tick: distinct event pairs: %v", err)
|
|
return
|
|
}
|
|
announced := false
|
|
for _, p := range pairs {
|
|
r, _, err := detectAndPropose(ctx, t.store, p.Action, p.Object, now)
|
|
if err != nil {
|
|
log.Printf("tick: detect pattern %s/%s: %v", p.Action, p.Object, err)
|
|
continue
|
|
}
|
|
if r == nil {
|
|
continue // no stable pattern, or already proposed/accepted/dismissed
|
|
}
|
|
log.Printf("tick: proposed routine: %s/%s every %.1f days", r.Action, r.Object, r.IntervalDays)
|
|
// One announcement per tick at most, whatever the scan turned up. The
|
|
// rest are on /routines; they are not lost, they are just not shouted.
|
|
if announced {
|
|
continue
|
|
}
|
|
announced = t.announceProposal(ctx, r, now, state)
|
|
}
|
|
}
|
|
|
|
// announceProposal offers a freshly inferred routine through the ordinary
|
|
// care-delivery path, if announcing is switched on at all. Returns true when
|
|
// something was actually sent.
|
|
//
|
|
// Everything here is restraint. The feature is off unless configured; when on
|
|
// it is sev1 (the lowest severity, so quiet hours, away presence and snooze
|
|
// all suppress it via loop.Gate exactly like a care nudge); it is spaced by
|
|
// proposalCfg.Cooldown across every pair, not per pair; and a suppressed or
|
|
// dropped announcement is NOT retried — the cooldown clock advances only on a
|
|
// real send, but the proposal row already exists, so the next tick will not
|
|
// re-detect it and nothing queues up behind it. A missed announcement means
|
|
// he reads it on /routines instead, which is the whole point of the page.
|
|
//
|
|
// The body is the detector's own literal Russian phrasing (pattern.PhraseRoutine
|
|
// — "ты заправляешь поилку раз в 7 дней — напоминать?"), not LLM-generated, so
|
|
// an inferred routine cannot arrive worded as something Maven never observed.
|
|
func (t *tickLoop) announceProposal(ctx context.Context, r *pattern.ProposedRoutine, now time.Time, state loop.State) bool {
|
|
if !t.proposalCfg.AnnounceProposals() {
|
|
return false
|
|
}
|
|
cooldown := time.Duration(t.proposalCfg.Cooldown)
|
|
if cooldown <= 0 {
|
|
cooldown = config.DefaultProposalCooldown
|
|
}
|
|
if !t.lastProposalAt.IsZero() && now.Sub(t.lastProposalAt) < cooldown {
|
|
return false
|
|
}
|
|
|
|
rule := loop.Rule{Name: "proposal:" + r.Action + " " + r.Object, Severity: loop.Sev1}
|
|
if !loop.Gate(state, rule) {
|
|
return false
|
|
}
|
|
body := pattern.PhraseRoutine(r)
|
|
pn := delivery.PhrasedNudge{
|
|
Candidate: loop.Candidate{Rule: rule, Severity: rule.Severity, State: state},
|
|
Body: body,
|
|
Summary: body,
|
|
}
|
|
sent, err := t.dispatcher.DispatchNudge(ctx, pn, now)
|
|
if err != nil {
|
|
log.Printf("tick: announce proposal %s/%s: %v", r.Action, r.Object, err)
|
|
return false
|
|
}
|
|
if len(sent) == 0 {
|
|
return false // routing dropped it — /routines still has it.
|
|
}
|
|
t.lastProposalAt = now
|
|
return true
|
|
}
|
|
|
|
// digestExpiry — how long a gate-suppressed care nudge stays worth
|
|
// resurfacing. 24h: these are daily-cadence rules (water/meal/break run on
|
|
// hour-scale cooldowns and re-derive from facts that reset every day), so a
|
|
// digest entry that outlives one full day is describing a day that's already
|
|
// over — "you skipped a break yesterday" said tomorrow evening is noise, not
|
|
// news. Bounding at one day also means a digest can never silently span a
|
|
// weekend of quiet hours into an unbounded backlog.
|
|
const digestExpiry = 24 * time.Hour
|
|
|
|
// maxDigestSpokenItems — the bundle read-out is capped so "batched, not
|
|
// dropped" cannot regress into "she dumps twelve things on me the moment I
|
|
// walk in" — a digest that nags in bulk is worse than the drops it replaced.
|
|
// Anything beyond the cap is still marked drained (it did get its moment;
|
|
// the cap limits WORDS, not whether it counted) and folded into a trailing
|
|
// count instead of being spoken in full.
|
|
const maxDigestSpokenItems = 3
|
|
|
|
// enqueueSuppressedDigest scans this tick's trace for care candidates the
|
|
// gate blocked for a genuine restraint reason and durably records the
|
|
// digest-eligible ones (loop.DigestEligible). Phrasing happens once, here,
|
|
// at enqueue time — not re-derived at drain time — the same way queueNudge
|
|
// phrases once and caches, so a rule suppressed for hours isn't re-prompting
|
|
// the LLM every tick it stays blocked (EnqueueDigestEntry's rule+body dedupe
|
|
// makes repeat calls here harmless, but skipping the phrase call entirely
|
|
// when a pending entry already exists avoids the LLM round-trip too).
|
|
func (t *tickLoop) enqueueSuppressedDigest(ctx context.Context, trace *loop.TickTrace, state loop.State, now time.Time) {
|
|
if trace == nil {
|
|
return
|
|
}
|
|
for _, tr := range trace.RuleTraces {
|
|
if !tr.PredicateResult || tr.GateResult {
|
|
continue // didn't want to fire, or wasn't suppressed
|
|
}
|
|
if !loop.DigestEligible(tr.Severity, tr.GateBlockedBy) {
|
|
continue
|
|
}
|
|
rule := loop.Rule{Name: tr.RuleName, Severity: tr.Severity}
|
|
cand := loop.Candidate{Rule: rule, Severity: tr.Severity, State: state}
|
|
pn, err := t.phraser.PhraseNudge(ctx, cand)
|
|
if err != nil {
|
|
log.Printf("tick: phrase digest candidate %s: %v", tr.RuleName, err)
|
|
continue
|
|
}
|
|
expires := now.Add(digestExpiry)
|
|
if _, deduped, err := t.store.EnqueueDigestEntry(ctx, tr.RuleName, int(tr.Severity), pn.Body, now, expires); err != nil {
|
|
log.Printf("tick: enqueue digest entry %s: %v", tr.RuleName, err)
|
|
} else if deduped {
|
|
// same suppressed nudge already pending — nothing new to say.
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
|
|
// expireStaleDigest sweeps entries past their expiry once per tick — cheap
|
|
// bookkeeping, mirrors ReconcileStaleDeliveryAttempts's shape.
|
|
func (t *tickLoop) expireStaleDigest(ctx context.Context, now time.Time) {
|
|
n, err := t.store.ExpireStaleDigestEntries(ctx, now)
|
|
if err != nil {
|
|
log.Printf("tick: expire stale digest entries: %v", err)
|
|
return
|
|
}
|
|
if n > 0 {
|
|
log.Printf("tick: expired %d stale digest entr(y/ies) unspoken", n)
|
|
}
|
|
}
|
|
|
|
// maybeDrainDigest speaks the pending digest bundle once the gate's
|
|
// suppression reasons have actually cleared — quiet hours over, back from
|
|
// away, out of the meeting. Draining while still suppressed would just be a
|
|
// second way to nag through quiet hours; the bundle waits for the same "is
|
|
// it allowed right now" condition a live nudge already waits for.
|
|
func (t *tickLoop) maybeDrainDigest(ctx context.Context, state loop.State, now time.Time) {
|
|
if state.QuietHours || state.CalendarBusy || state.Presence == store.Away {
|
|
return
|
|
}
|
|
entries, err := t.store.PendingDigestEntries(ctx, now)
|
|
if err != nil {
|
|
log.Printf("tick: pending digest entries: %v", err)
|
|
return
|
|
}
|
|
if len(entries) == 0 {
|
|
return
|
|
}
|
|
|
|
spoken := entries
|
|
extra := 0
|
|
if len(spoken) > maxDigestSpokenItems {
|
|
spoken = entries[:maxDigestSpokenItems]
|
|
extra = len(entries) - maxDigestSpokenItems
|
|
}
|
|
var b strings.Builder
|
|
maxSev := 0
|
|
for i, e := range spoken {
|
|
if i > 0 {
|
|
b.WriteString(" · ")
|
|
}
|
|
b.WriteString(e.Body)
|
|
if e.Severity > maxSev {
|
|
maxSev = e.Severity
|
|
}
|
|
}
|
|
if extra > 0 {
|
|
fmt.Fprintf(&b, " · и ещё %d", extra)
|
|
}
|
|
body := b.String()
|
|
summary := fmt.Sprintf("%d отложенных уведомлений", len(entries))
|
|
|
|
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 {
|
|
log.Printf("tick: dispatch digest bundle: %v", err)
|
|
return // leave entries pending; retried next tick
|
|
}
|
|
ids := make([]int64, len(entries))
|
|
for i, e := range entries {
|
|
ids[i] = e.ID
|
|
}
|
|
if err := t.store.DrainDigestEntries(ctx, ids, now); err != nil {
|
|
log.Printf("tick: drain digest entries: %v", err)
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|
|
}
|
|
|
|
// fireAcceptedRoutines nudges about the routines the user accepted, once per
|
|
// interval (Vikunja #366). Accepting used to create a single reminder, so a
|
|
// non-weekly routine fired once and went quiet forever; the schedule lives in
|
|
// the proposed_routines row now and the loop re-reads it every tick.
|
|
//
|
|
// A routine is a care-class nudge and goes through the restraint gate like any
|
|
// other: quiet hours, away presence and snooze all suppress it. Reminders bypass
|
|
// that gate; routines must not. A suppressed nudge is NOT marked fired, so it
|
|
// goes out on the next tick that the gate allows — one nudge, held, not dropped
|
|
// and not repeated.
|
|
//
|
|
// The body is literal text built from the detected action and object, not
|
|
// LLM-phrased, so a routine can't hallucinate. It nudges; it never acts.
|
|
func (t *tickLoop) fireAcceptedRoutines(ctx context.Context, now time.Time, state loop.State) {
|
|
rows, err := t.store.ListAcceptedRoutines(ctx)
|
|
if err != nil {
|
|
log.Printf("tick: list accepted routines: %v", err)
|
|
return
|
|
}
|
|
accepted := make([]routine.Accepted, 0, len(rows))
|
|
for _, r := range rows {
|
|
if r.AcceptedTs == nil {
|
|
continue // accepted before the schedule column existed — no clock to start from.
|
|
}
|
|
accepted = append(accepted, routine.Accepted{
|
|
ID: r.ID,
|
|
Name: r.Action + " " + r.Object,
|
|
IntervalDays: r.IntervalDays,
|
|
Accepted: *r.AcceptedTs,
|
|
LastFired: r.LastFiredTs,
|
|
})
|
|
}
|
|
|
|
for _, a := range routine.DueAccepted(accepted, now) {
|
|
rule := loop.Rule{Name: "routine:" + a.Name, Severity: loop.Sev1}
|
|
if !loop.Gate(state, rule) {
|
|
continue
|
|
}
|
|
body := "пора: " + a.Name
|
|
pn := delivery.PhrasedNudge{
|
|
Candidate: loop.Candidate{Rule: rule, Severity: rule.Severity, State: state},
|
|
Body: body,
|
|
Summary: body,
|
|
}
|
|
sent, err := t.dispatcher.DispatchNudge(ctx, pn, now)
|
|
if err != nil {
|
|
log.Printf("tick: dispatch accepted routine %d: %v", a.ID, err)
|
|
continue
|
|
}
|
|
if len(sent) == 0 {
|
|
continue // routing dropped it — leave it due.
|
|
}
|
|
if err := t.store.MarkRoutineFired(ctx, a.ID, now); err != nil {
|
|
log.Printf("tick: mark routine %d fired: %v", a.ID, 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:<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
|
|
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,
|
|
}
|
|
}
|