nudge: add digest/batching mode for care nudges
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
This commit is contained in:
@@ -77,15 +77,15 @@ Notes:
|
|||||||
|
|
||||||
| # | Task | Commit | Status |
|
| # | 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 |
|
| 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 |
|
| 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 |
|
| 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 |
|
| 22 | **Language unification** — bilingual cheatsheet with RU/EN toggle in nav + ?lang= param | c225ba3 | done |
|
||||||
|
|
||||||
Notes:
|
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.
|
- PWA manifest currently has `"icons": []` — no icons, mobile add-to-home-screen shows a blank tile.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|||||||
+1
-1
@@ -175,7 +175,7 @@ func run(args []string) error {
|
|||||||
tickInterval := time.Duration(cfg.TickInterval)
|
tickInterval := time.Duration(cfg.TickInterval)
|
||||||
repeatInterval := time.Duration(cfg.RepeatInterval)
|
repeatInterval := time.Duration(cfg.RepeatInterval)
|
||||||
autotuneInterval := time.Duration(cfg.AutotuneInterval)
|
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) -----
|
// ----- IPC boundary (core ↔ modules) -----
|
||||||
coreAPI := ipc.NewStoreAPI(st)
|
coreAPI := ipc.NewStoreAPI(st)
|
||||||
|
|||||||
+123
-6
@@ -14,15 +14,26 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/config"
|
||||||
"github.com/kami/maven/internal/delivery"
|
"github.com/kami/maven/internal/delivery"
|
||||||
"github.com/kami/maven/internal/loop"
|
"github.com/kami/maven/internal/loop"
|
||||||
"github.com/kami/maven/internal/phraser"
|
"github.com/kami/maven/internal/phraser"
|
||||||
"github.com/kami/maven/internal/store"
|
"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
|
// 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
|
// 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.
|
// already captured it, but we keep it for a possible future re-seed path.
|
||||||
@@ -37,6 +48,14 @@ type tickLoop struct {
|
|||||||
repeatInterval time.Duration
|
repeatInterval time.Duration
|
||||||
autotuneInterval time.Duration // 0 ⇒ autotune disabled (gatherer falls back to static Base)
|
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
|
// 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
|
// 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
|
// that re-phrases differently every 5m is hostile; the same terse body
|
||||||
@@ -54,6 +73,7 @@ func newTickLoop(
|
|||||||
p phraser.Phraser,
|
p phraser.Phraser,
|
||||||
rules []loop.Rule,
|
rules []loop.Rule,
|
||||||
tickInterval, repeatInterval, autotuneInterval time.Duration,
|
tickInterval, repeatInterval, autotuneInterval time.Duration,
|
||||||
|
digestCfg *config.DigestConfig,
|
||||||
) *tickLoop {
|
) *tickLoop {
|
||||||
return &tickLoop{
|
return &tickLoop{
|
||||||
store: st,
|
store: st,
|
||||||
@@ -64,6 +84,8 @@ func newTickLoop(
|
|||||||
tickInterval: tickInterval,
|
tickInterval: tickInterval,
|
||||||
repeatInterval: repeatInterval,
|
repeatInterval: repeatInterval,
|
||||||
autotuneInterval: autotuneInterval,
|
autotuneInterval: autotuneInterval,
|
||||||
|
digestCfg: digestCfg,
|
||||||
|
digestQ: nil,
|
||||||
lastPhrase: make(map[string]delivery.PhrasedNudge),
|
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.
|
// proactive: at most one candidate, max severity.
|
||||||
if cand := loop.Tick(state, t.rules); cand != nil {
|
if cand := loop.Tick(state, t.rules); cand != nil {
|
||||||
pn, err := t.phraser.PhraseNudge(ctx, *cand)
|
if t.shouldQueue(cand) {
|
||||||
if err != nil {
|
t.queueNudge(ctx, cand, state, now)
|
||||||
log.Printf("tick: phrase nudge %s: %v", cand.Rule.Name, err)
|
|
||||||
} else {
|
} else {
|
||||||
t.cachePhrase(pn)
|
pn, err := t.phraser.PhraseNudge(ctx, *cand)
|
||||||
if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil {
|
if err != nil {
|
||||||
log.Printf("tick: dispatch nudge %s: %v", cand.Rule.Name, err)
|
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
|
// reminders: gate-bypassing class. fired once, marked after a successful
|
||||||
// delivery. a failed send leaves the reminder pending — the next tick
|
// delivery. a failed send leaves the reminder pending — the next tick
|
||||||
// re-gathers and re-attempts.
|
// re-gathers and re-attempts.
|
||||||
@@ -182,6 +213,92 @@ func (t *tickLoop) repeatPhrase(rule string) (body, summary string) {
|
|||||||
return pn.Body, pn.Summary
|
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
|
// 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
|
// (autotuneInterval, see run) so it doesn't write a fact every tick. for each
|
||||||
// rule:
|
// rule:
|
||||||
|
|||||||
+231
-7
@@ -6,6 +6,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/config"
|
||||||
"github.com/kami/maven/internal/delivery"
|
"github.com/kami/maven/internal/delivery"
|
||||||
"github.com/kami/maven/internal/loop"
|
"github.com/kami/maven/internal/loop"
|
||||||
"github.com/kami/maven/internal/phraser"
|
"github.com/kami/maven/internal/phraser"
|
||||||
@@ -33,7 +34,7 @@ func newTestStore(t *testing.T) *store.Store {
|
|||||||
return st
|
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()
|
t.Helper()
|
||||||
rules := loop.DefaultRules()
|
rules := loop.DefaultRules()
|
||||||
g := loop.NewGatherer(st, rules)
|
g := loop.NewGatherer(st, rules)
|
||||||
@@ -44,7 +45,7 @@ func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink) *tickLoo
|
|||||||
Nudges: st,
|
Nudges: st,
|
||||||
Reminders: 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.
|
// refNow — fixed tick time so presence decay + since durations are deterministic.
|
||||||
@@ -69,7 +70,7 @@ func TestTickColdStoreSendsNothing(t *testing.T) {
|
|||||||
st := newTestStore(t)
|
st := newTestStore(t)
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
sink := &fakeSink{}
|
sink := &fakeSink{}
|
||||||
tl := newTestTickLoop(t, st, sink)
|
tl := newTestTickLoop(t, st, sink, nil)
|
||||||
|
|
||||||
tl.tick(ctx, refNow())
|
tl.tick(ctx, refNow())
|
||||||
|
|
||||||
@@ -90,7 +91,7 @@ func TestTickWaterFiresWhenDueAndPresent(t *testing.T) {
|
|||||||
t.Fatalf("seed water: %v", err)
|
t.Fatalf("seed water: %v", err)
|
||||||
}
|
}
|
||||||
sink := &fakeSink{}
|
sink := &fakeSink{}
|
||||||
tl := newTestTickLoop(t, st, sink)
|
tl := newTestTickLoop(t, st, sink, nil)
|
||||||
|
|
||||||
tl.tick(ctx, now)
|
tl.tick(ctx, now)
|
||||||
|
|
||||||
@@ -131,7 +132,7 @@ func TestTickCooldownSuppressesSecondSend(t *testing.T) {
|
|||||||
_ = err
|
_ = err
|
||||||
}
|
}
|
||||||
sink := &fakeSink{}
|
sink := &fakeSink{}
|
||||||
tl := newTestTickLoop(t, st, sink)
|
tl := newTestTickLoop(t, st, sink, nil)
|
||||||
|
|
||||||
tl.tick(ctx, now) // fires
|
tl.tick(ctx, now) // fires
|
||||||
tl.tick(ctx, now.Add(time.Minute)) // still within 30m cooldown ⇒ suppressed
|
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)
|
t.Fatalf("CreateReminder: %v", err)
|
||||||
}
|
}
|
||||||
sink := &fakeSink{}
|
sink := &fakeSink{}
|
||||||
tl := newTestTickLoop(t, st, sink)
|
tl := newTestTickLoop(t, st, sink, nil)
|
||||||
|
|
||||||
tl.tick(ctx, now)
|
tl.tick(ctx, now)
|
||||||
if got, want := len(sink.sends), 1; got != want {
|
if got, want := len(sink.sends), 1; got != want {
|
||||||
@@ -190,7 +191,7 @@ func TestTuneWritesFeedbackCooldownToStore(t *testing.T) {
|
|||||||
now := refNow()
|
now := refNow()
|
||||||
|
|
||||||
sink := &fakeSink{}
|
sink := &fakeSink{}
|
||||||
tl := newTestTickLoop(t, st, sink)
|
tl := newTestTickLoop(t, st, sink, nil)
|
||||||
|
|
||||||
// seed enough resolved `ignored` nudge outcomes to trip the tuner (above
|
// seed enough resolved `ignored` nudge outcomes to trip the tuner (above
|
||||||
// TuneMinOutcomes). all-ignored ⇒ factor 1.5 ⇒ base × 1.5; clamped to Max.
|
// TuneMinOutcomes). all-ignored ⇒ factor 1.5 ⇒ base × 1.5; clamped to Max.
|
||||||
@@ -286,3 +287,226 @@ func TestTuneWritesFeedbackCooldownToStore(t *testing.T) {
|
|||||||
wantUntil, got)
|
wantUntil, got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ----------------------------- 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))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -115,6 +115,10 @@ type Config struct {
|
|||||||
// fact independently; both the schedule AND the toggle activate quiet.
|
// fact independently; both the schedule AND the toggle activate quiet.
|
||||||
// nil ⇒ quiet hours only activate via the voice toggle.
|
// nil ⇒ quiet hours only activate via the voice toggle.
|
||||||
QuietHours *QuietHoursConfig `json:"quiet_hours,omitempty"`
|
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
|
// QuietHoursConfig — a recurring daily quiet-window. Times are local to the
|
||||||
@@ -185,6 +189,18 @@ type ToolConfig struct {
|
|||||||
Destructive bool `json:"destructive,omitempty"`
|
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
|
// PhraserConfig — the LLM-backed phraser seam. The daemon spawns llama-server
|
||||||
// as a managed subprocess and sends chat-completion requests to phrase nudge
|
// as a managed subprocess and sends chat-completion requests to phrase nudge
|
||||||
// and reminder messages. nil ⇒ the template-based Stub is used instead.
|
// 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 != nil {
|
||||||
if c.Voice.RouterThreshold <= 0 {
|
if c.Voice.RouterThreshold <= 0 {
|
||||||
c.Voice.RouterThreshold = DefaultRouterThreshold
|
c.Voice.RouterThreshold = DefaultRouterThreshold
|
||||||
|
|||||||
Reference in New Issue
Block a user