// mavend/tick_digest.go — the digest queue. // // Split out of tick.go, move-only (Vikunja #422). A nudge below the severity // ceiling is held here instead of spoken, flushed as one batch on the window, // and drained by hand when he asks. Everything about batching lives here. package main import ( "context" "fmt" "log" "strings" "time" "github.com/kami/maven/internal/delivery" "github.com/kami/maven/internal/loop" "github.com/kami/maven/internal/say" "github.com/kami/maven/internal/store" ) // 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 } // 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() // The adjective declines with the noun, so the count picks the whole // phrase: 1 отложенное уведомление, 2 отложенных уведомления, 5 // отложенных уведомлений. summary := fmt.Sprintf("%d %s", len(entries), say.CountWord(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) } }