From 354990fed0a35c0cbea152ebfbc3112e726f732e Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 5 Jul 2026 13:07:28 +0400 Subject: [PATCH] nudge: add digest/batching mode for care nudges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When enabled, eligible nudges (sev ≤ ceiling) are queued in memory instead of sent immediately. Every 'window' duration (or when 'max_items' reached), the queue is flushed as a single digest notification with concatenated bodies. Config: digest: enabled: true # default false window: 30m # flush window (default 30m) max_items: 5 # flush at this count (default 5) severity_ceiling: 2 # max sev batched (default 2; sev3+ bypass) Changes: - config.go: add DigestConfig struct with defaults - tick.go: QueuedNudge type, digest queue/flush in TickLoop, shouldQueue/maybeFlush/flushDigest helpers - main.go: pass cfg.Digest to newTickLoop - tick_test.go: 6 new tests covering queue, flush, bypass, dedup --- SESSION-05-07-2026.md | 6 +- cmd/mavend/main.go | 2 +- cmd/mavend/tick.go | 129 +++++++++++++++++++- cmd/mavend/tick_test.go | 240 ++++++++++++++++++++++++++++++++++++-- internal/config/config.go | 29 +++++ 5 files changed, 388 insertions(+), 18 deletions(-) diff --git a/SESSION-05-07-2026.md b/SESSION-05-07-2026.md index 70b7786..d45d49f 100644 --- a/SESSION-05-07-2026.md +++ b/SESSION-05-07-2026.md @@ -77,15 +77,15 @@ Notes: | # | Task | Commit | Status | |---|------|--------|--------| -| 17 | **In-process auth gate for /tools** — add WeAuthn session check to POST /tools handler. Currently relies solely on wg+nginx | — | pending | +| 17 | **In-process auth gate for /tools** — local PasskeySession check on POST, returns 403 if unasserted | 5afff00 | done | | 18 | **Digest / notification history UI** — section on `/dash` showing batched notifications | — | pending | | 19 | **Rule trace page** — new `/trace` route showing predicate eval results per rule per tick | — | pending | | 20 | **Command history page** — new `/history` route showing recent facts/commands | f8ba396 | done | -| 21 | **PWA icons** — add proper icon array to `manifest.json` (generate or inline SVG) | — | pending | +| 21 | **PWA icons** — SVG icon + manifest.json icons array | 00a3bba | done | | 22 | **Language unification** — bilingual cheatsheet with RU/EN toggle in nav + ?lang= param | c225ba3 | done | Notes: -- Current `/tools` has no in-process auth — `handlers_test.go` explicitly pins this behavior with `TestEnableTool_NoInProcessAuthGate`. +- In-process auth gate added for POST /tools (5afff00). Digest UI (#18) depends on #14; trace page (#19) depends on #15. - PWA manifest currently has `"icons": []` — no icons, mobile add-to-home-screen shows a blank tile. --- diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 5a6638b..43c42ae 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -175,7 +175,7 @@ func run(args []string) error { tickInterval := time.Duration(cfg.TickInterval) repeatInterval := time.Duration(cfg.RepeatInterval) autotuneInterval := time.Duration(cfg.AutotuneInterval) - loop := newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval) + loop := newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest) // ----- IPC boundary (core ↔ modules) ----- coreAPI := ipc.NewStoreAPI(st) diff --git a/cmd/mavend/tick.go b/cmd/mavend/tick.go index bde17aa..839a354 100644 --- a/cmd/mavend/tick.go +++ b/cmd/mavend/tick.go @@ -14,15 +14,26 @@ import ( "log" "os" "path/filepath" + "strings" "sync" "time" + "github.com/kami/maven/internal/config" "github.com/kami/maven/internal/delivery" "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. @@ -37,6 +48,14 @@ type tickLoop struct { 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 @@ -54,6 +73,7 @@ func newTickLoop( p phraser.Phraser, rules []loop.Rule, tickInterval, repeatInterval, autotuneInterval time.Duration, + digestCfg *config.DigestConfig, ) *tickLoop { return &tickLoop{ store: st, @@ -64,6 +84,8 @@ func newTickLoop( tickInterval: tickInterval, repeatInterval: repeatInterval, autotuneInterval: autotuneInterval, + digestCfg: digestCfg, + digestQ: nil, lastPhrase: make(map[string]delivery.PhrasedNudge), } } @@ -112,17 +134,26 @@ func (t *tickLoop) tick(ctx context.Context, now time.Time) { // proactive: at most one candidate, max severity. if cand := loop.Tick(state, t.rules); cand != nil { - pn, err := t.phraser.PhraseNudge(ctx, *cand) - if err != nil { - log.Printf("tick: phrase nudge %s: %v", cand.Rule.Name, err) + if t.shouldQueue(cand) { + t.queueNudge(ctx, cand, state, now) } 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) + 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. @@ -182,6 +213,92 @@ func (t *tickLoop) repeatPhrase(rule string) (body, summary string) { 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 { + log.Printf("tick: dispatch digest: %v", err) + } + 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: diff --git a/cmd/mavend/tick_test.go b/cmd/mavend/tick_test.go index c4ac858..cd3a10a 100644 --- a/cmd/mavend/tick_test.go +++ b/cmd/mavend/tick_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/kami/maven/internal/config" "github.com/kami/maven/internal/delivery" "github.com/kami/maven/internal/loop" "github.com/kami/maven/internal/phraser" @@ -33,7 +34,7 @@ func newTestStore(t *testing.T) *store.Store { return st } -func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink) *tickLoop { +func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink, digestCfg *config.DigestConfig) *tickLoop { t.Helper() rules := loop.DefaultRules() g := loop.NewGatherer(st, rules) @@ -44,7 +45,7 @@ func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink) *tickLoo Nudges: st, Reminders: st, }) - return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0) + return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0, digestCfg) } // refNow — fixed tick time so presence decay + since durations are deterministic. @@ -69,7 +70,7 @@ func TestTickColdStoreSendsNothing(t *testing.T) { st := newTestStore(t) ctx := context.Background() sink := &fakeSink{} - tl := newTestTickLoop(t, st, sink) + tl := newTestTickLoop(t, st, sink, nil) tl.tick(ctx, refNow()) @@ -90,7 +91,7 @@ func TestTickWaterFiresWhenDueAndPresent(t *testing.T) { t.Fatalf("seed water: %v", err) } sink := &fakeSink{} - tl := newTestTickLoop(t, st, sink) + tl := newTestTickLoop(t, st, sink, nil) tl.tick(ctx, now) @@ -131,7 +132,7 @@ func TestTickCooldownSuppressesSecondSend(t *testing.T) { _ = err } sink := &fakeSink{} - tl := newTestTickLoop(t, st, sink) + tl := newTestTickLoop(t, st, sink, nil) tl.tick(ctx, now) // fires tl.tick(ctx, now.Add(time.Minute)) // still within 30m cooldown ⇒ suppressed @@ -154,7 +155,7 @@ func TestTickReminderFiresOnceAndMarkedFired(t *testing.T) { t.Fatalf("CreateReminder: %v", err) } sink := &fakeSink{} - tl := newTestTickLoop(t, st, sink) + tl := newTestTickLoop(t, st, sink, nil) tl.tick(ctx, now) if got, want := len(sink.sends), 1; got != want { @@ -190,7 +191,7 @@ func TestTuneWritesFeedbackCooldownToStore(t *testing.T) { now := refNow() sink := &fakeSink{} - tl := newTestTickLoop(t, st, sink) + tl := newTestTickLoop(t, st, sink, nil) // seed enough resolved `ignored` nudge outcomes to trip the tuner (above // TuneMinOutcomes). all-ignored ⇒ factor 1.5 ⇒ base × 1.5; clamped to Max. @@ -285,4 +286,227 @@ func TestTuneWritesFeedbackCooldownToStore(t *testing.T) { t.Fatalf("gatherer used tuned base: CooldownUntil want %v, got %v", wantUntil, got) } -} \ No newline at end of file +} + +// ----------------------------- digest / batching ---------------------------- + +func TestDigestQueuesEligibleNudge(t *testing.T) { + // sev1 (water) with digest enabled → queued, not dispatched. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + } + sink := &fakeSink{} + dc := &config.DigestConfig{ + Enabled: true, + Window: config.Duration(30 * time.Minute), + MaxItems: 5, + SeverityCeiling: 2, + } + tl := newTestTickLoop(t, st, sink, dc) + + tl.tick(ctx, now) + + if len(sink.sends) != 0 { + t.Fatalf("digest: eligible nudge should not dispatch immediately, got %d sends", len(sink.sends)) + } + if len(tl.digestQ) != 1 { + t.Fatalf("digestQ length = %d, want 1", len(tl.digestQ)) + } + if tl.digestQ[0].Rule != "water" { + t.Errorf("queued rule = %q, want %q", tl.digestQ[0].Rule, "water") + } +} + +func TestDigestFlushesAfterWindow(t *testing.T) { + // Queue a sev1 nudge, advance past the window, tick again → flush. + // Re-mark presence on the second tick so the flush dispatch has a + // non-away presence route (care nudges drop on away). + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + } + sink := &fakeSink{} + dc := &config.DigestConfig{ + Enabled: true, + Window: config.Duration(30 * time.Minute), + MaxItems: 5, + SeverityCeiling: 2, + } + tl := newTestTickLoop(t, st, sink, dc) + + // Tick 1: water queues, nothing dispatched. + tl.tick(ctx, now) + if len(sink.sends) != 0 { + t.Fatalf("tick 1: want 0 sends (queued), got %d", len(sink.sends)) + } + if len(tl.digestQ) != 1 { + t.Fatalf("tick 1: digestQ = %d, want 1", len(tl.digestQ)) + } + + // Tick 2: window elapsed → flush. Re-mark presence to keep routing + // present (care nudge away → drop). + sink.sends = nil + later := now.Add(31 * time.Minute) + markPresent(t, st, ctx, later) + tl.tick(ctx, later) + + if len(sink.sends) != 1 { + t.Fatalf("tick 2 (flush): sends = %d, want 1", len(sink.sends)) + } + if sink.sends[0].RuleName != "digest" { + t.Errorf("flush rule = %q, want %q", sink.sends[0].RuleName, "digest") + } + if sink.sends[0].Body == "" { + t.Error("digest body must not be empty") + } + if len(tl.digestQ) != 0 { + t.Errorf("digestQ should be empty after flush, got %d", len(tl.digestQ)) + } +} + +func TestDigestFlushesAtMaxItems(t *testing.T) { + // Queue water via tick, then manually push a second item, then tick again + // so the before-candidate maybeFlush sees len >= MaxItems and flushes. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + } + + sink := &fakeSink{} + dc := &config.DigestConfig{ + Enabled: true, + Window: config.Duration(30 * time.Minute), + MaxItems: 2, + SeverityCeiling: 2, + } + tl := newTestTickLoop(t, st, sink, dc) + + // Tick 1: water queues (1 item, below MaxItems). + tl.tick(ctx, now) + if len(tl.digestQ) != 1 { + t.Fatalf("tick 1: digestQ = %d, want 1", len(tl.digestQ)) + } + if len(sink.sends) != 0 { + t.Fatalf("tick 1: sends = %d, want 0", len(sink.sends)) + } + + // Manually push a second item to hit MaxItems. + tl.digestQ = append(tl.digestQ, QueuedNudge{ + Rule: "meal", Severity: 1, Body: "eat something", Key: "meal", QueuedAt: now.Add(time.Second), + }) + + // Tick 2: after-candidate maybeFlush sees len=2 >= MaxItems=2 → flush. + sink.sends = nil + tl.tick(ctx, now.Add(2*time.Second)) + + if len(sink.sends) != 1 { + t.Fatalf("after flush tick: sends = %d, want 1", len(sink.sends)) + } + if sink.sends[0].RuleName != "digest" { + t.Errorf("flush rule = %q, want %q", sink.sends[0].RuleName, "digest") + } + if len(tl.digestQ) != 0 { + t.Errorf("digestQ should be empty after flush, got %d", len(tl.digestQ)) + } +} + +func TestDigestSev4BypassesQueue(t *testing.T) { + // Sev4 (service_down) with digest enabled → dispatches immediately. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "service_down", "poll:uptimekuma", "down", now); err != nil { + t.Fatalf("seed service_down: %v", err) + } + sink := &fakeSink{} + dc := &config.DigestConfig{ + Enabled: true, + Window: config.Duration(30 * time.Minute), + MaxItems: 5, + SeverityCeiling: 2, + } + tl := newTestTickLoop(t, st, sink, dc) + + tl.tick(ctx, now) + + if len(sink.sends) == 0 { + t.Fatal("sev4 nudge must dispatch immediately, bypassing digest queue") + } + if len(tl.digestQ) != 0 { + t.Errorf("digestQ should be empty (sev4 bypassed), got %d", len(tl.digestQ)) + } +} + +func TestDigestDisabledSendsImmediately(t *testing.T) { + // Digest disabled (nil config) → sev1 dispatches immediately. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + } + sink := &fakeSink{} + tl := newTestTickLoop(t, st, sink, nil) + + tl.tick(ctx, now) + + if len(sink.sends) != 1 { + t.Fatalf("digest disabled: sends = %d, want 1", len(sink.sends)) + } + if sink.sends[0].RuleName != "water" { + t.Errorf("send rule = %q, want %q", sink.sends[0].RuleName, "water") + } +} + +func TestDigestDeduplicatesByRule(t *testing.T) { + // Same rule (water) fires in two ticks; only one copy in the queue. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + } + sink := &fakeSink{} + dc := &config.DigestConfig{ + Enabled: true, + Window: config.Duration(30 * time.Minute), + MaxItems: 5, + SeverityCeiling: 2, + } + tl := newTestTickLoop(t, st, sink, dc) + + // Tick 1: water queues. + tl.tick(ctx, now) + if len(tl.digestQ) != 1 { + t.Fatalf("tick 1: digestQ = %d, want 1", len(tl.digestQ)) + } + + // Tick 2: water still in cooldown → gate suppresses, but if cooldown + // weren't active the dedup check would prevent a second copy. We verify + // by making a second direct call to queueNudge with the same rule name. + // We also tick again with a different time to see if water fires again + // through the normal path — but cooldown should suppress it. Instead, + // directly verify dedup by calling queueNudge with a synthetic candidate. + tl.queueNudge(ctx, &loop.Candidate{ + Rule: loop.Rule{Name: "water", Severity: loop.Sev1}, + Severity: loop.Sev1, + }, loop.State{}, now.Add(time.Minute)) + + if len(tl.digestQ) != 1 { + t.Fatalf("after duplicate queue attempt: digestQ = %d, want 1 (dedup)", len(tl.digestQ)) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 8d20da6..56422a6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -115,6 +115,10 @@ type Config struct { // fact independently; both the schedule AND the toggle activate quiet. // nil ⇒ quiet hours only activate via the voice toggle. QuietHours *QuietHoursConfig `json:"quiet_hours,omitempty"` + + // Digest — notification batching / digest mode. nil ⇒ digest disabled + // (every nudge is sent as it fires — legacy behaviour). + Digest *DigestConfig `json:"digest,omitempty"` } // QuietHoursConfig — a recurring daily quiet-window. Times are local to the @@ -185,6 +189,18 @@ type ToolConfig struct { Destructive bool `json:"destructive,omitempty"` } +// DigestConfig — notification batching / digest mode. When enabled, eligible +// nudges (severity ≤ SeverityCeiling) are queued in memory instead of sent +// immediately. Every Window duration (or when MaxItems reached), the queue is +// flushed as a single digest notification. nil ⇒ digest disabled (legacy +// behaviour — every nudge is sent as it fires). +type DigestConfig struct { + Enabled bool `json:"enabled,omitempty"` + Window Duration `json:"window,omitempty"` // e.g. "30m" + MaxItems int `json:"max_items,omitempty"` // flush at this count + SeverityCeiling int `json:"severity_ceiling,omitempty"` // max sev batched +} + // PhraserConfig — the LLM-backed phraser seam. The daemon spawns llama-server // as a managed subprocess and sends chat-completion requests to phrase nudge // and reminder messages. nil ⇒ the template-based Stub is used instead. @@ -315,6 +331,19 @@ func (c *Config) applyDefaults() { } } + if c.Digest == nil { + c.Digest = &DigestConfig{Enabled: false} + } + if c.Digest.Window == 0 { + c.Digest.Window = Duration(30 * time.Minute) + } + if c.Digest.MaxItems == 0 { + c.Digest.MaxItems = 5 + } + if c.Digest.SeverityCeiling == 0 { + c.Digest.SeverityCeiling = 2 + } + if c.Voice != nil { if c.Voice.RouterThreshold <= 0 { c.Voice.RouterThreshold = DefaultRouterThreshold