59a4e06615
Two additive proactive/recall features. Routines (internal/routine): a third proactive class beside reminders (user-stated) and care rules (world-state) — operator-declared clockwork. config.routines[] (cron + literal RU body + severity) fire through the normal dispatcher on schedule. Bodies are literal, not LLM-phrased (can't hallucinate); rule name routine:<name> keeps them out of the care autotuner; a cold-start guard seeds on first sight so a restart never replays a missed schedule. Pure routine.Due + config validation, unit- tested; the tick driver holds the last-fired map and calls fireRoutines. Persistent memory (internal/store/memory.go): store.MemoryStore backs the memory.Store interface with the SAME encrypted sqlite db — survives restarts and recall text inherits at-rest encryption (no plaintext sidecar). float32-blob vectors, brute-force cosine (ANN is a later swap behind the interface), upsert-by-id. The daemon wires st.VectorMemory() into wireVoice; the in-memory impl stays the test/no-store floor. Closes the "in-memory only, lost on restart" gap (PROGRESS #8). Gate green: gofmt/vet clean, -race across routine/config/store/mavend. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01U2PNdwDj2Gt8YW294J7oSc
561 lines
19 KiB
Go
561 lines
19 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"path/filepath"
|
||
"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"
|
||
"github.com/kami/maven/internal/routine"
|
||
"github.com/kami/maven/internal/store"
|
||
)
|
||
|
||
// fakeSink — captures every Send for assertion. implements delivery.Sink.
|
||
type fakeSink struct {
|
||
sends []delivery.Sendable
|
||
}
|
||
|
||
func (f *fakeSink) Send(_ context.Context, s delivery.Sendable) error {
|
||
f.sends = append(f.sends, s)
|
||
return nil
|
||
}
|
||
|
||
func newTestStore(t *testing.T) *store.Store {
|
||
t.Helper()
|
||
path := filepath.Join(t.TempDir(), "mavend_test.db")
|
||
st, err := store.Open(context.Background(), path)
|
||
if err != nil {
|
||
t.Fatalf("store.Open: %v", err)
|
||
}
|
||
t.Cleanup(func() { _ = st.Close() })
|
||
return st
|
||
}
|
||
|
||
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)
|
||
d := delivery.NewDispatcher(delivery.Config{
|
||
Voice: sink,
|
||
Ntfy: sink,
|
||
Telegram: sink,
|
||
Nudges: st,
|
||
Reminders: st,
|
||
})
|
||
return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0, digestCfg, nil)
|
||
}
|
||
|
||
func TestTickFiresRoutineWhenScheduleCrosses(t *testing.T) {
|
||
// A routine scheduled for 12:00 daily. On the first tick it seeds (cold-start
|
||
// guard — no fire); on a tick past the next 12:00 crossing it fires through
|
||
// the dispatcher with its literal body and severity.
|
||
st := newTestStore(t)
|
||
ctx := context.Background()
|
||
now := refNow() // 2026-06-30 12:00 UTC
|
||
markPresent(t, st, ctx, now)
|
||
|
||
rules := loop.DefaultRules()
|
||
g := loop.NewGatherer(st, rules)
|
||
sink := &fakeSink{}
|
||
d := delivery.NewDispatcher(delivery.Config{Voice: sink, Ntfy: sink, Telegram: sink, Nudges: st, Reminders: st})
|
||
rs := []routine.Routine{{Name: "morning", Cron: "0 12 * * *", Body: "полдень, время воды", Severity: 1}}
|
||
tl := newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0, nil, rs)
|
||
|
||
// first tick: seeds, does not fire the routine.
|
||
tl.tick(ctx, now)
|
||
for _, s := range sink.sends {
|
||
if s.RuleName == "routine:morning" {
|
||
t.Fatal("routine fired on the seeding tick (cold-start guard failed)")
|
||
}
|
||
}
|
||
|
||
// next day, just past 12:00 — the schedule crossed.
|
||
next := now.Add(24 * time.Hour).Add(time.Minute)
|
||
markPresent(t, st, ctx, next)
|
||
sink.sends = nil
|
||
tl.tick(ctx, next)
|
||
|
||
var got *delivery.Sendable
|
||
for i := range sink.sends {
|
||
if sink.sends[i].RuleName == "routine:morning" {
|
||
got = &sink.sends[i]
|
||
}
|
||
}
|
||
if got == nil {
|
||
t.Fatalf("routine did not fire after its schedule crossed; sends=%+v", sink.sends)
|
||
}
|
||
if got.Body != "полдень, время воды" {
|
||
t.Errorf("routine body = %q, want the literal config body", got.Body)
|
||
}
|
||
if got.Channel != delivery.ChannelVoice {
|
||
t.Errorf("routine channel = %v, want voice (sev1 present)", got.Channel)
|
||
}
|
||
}
|
||
|
||
// refNow — fixed tick time so presence decay + since durations are deterministic.
|
||
func refNow() time.Time { return time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) }
|
||
|
||
// markPresent seeds desk_active + page_heartbeat with fresh ts so presence
|
||
// resolves to Present for the given tick time (cold-start is away; ENTER at
|
||
// 0.55 — a fresh desk_active alone gives 0.90, well over).
|
||
func markPresent(t *testing.T, st *store.Store, ctx context.Context, now time.Time) {
|
||
t.Helper()
|
||
for _, key := range []string{"desk_active", "page_heartbeat"} {
|
||
if _, err := st.SetValue(ctx, store.KindSelf, key, "tap:desk", map[string]bool{key: true}, now); err != nil {
|
||
t.Fatalf("seed %s: %v", key, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
func TestTickColdStoreSendsNothing(t *testing.T) {
|
||
// The "shuts up when uncertain" floor: no facts ⇒ every rule's
|
||
// InertWhenNoData keys are missing ⇒ gate skips them. the loop's silence
|
||
// is the default outcome of a tick on an empty store.
|
||
st := newTestStore(t)
|
||
ctx := context.Background()
|
||
sink := &fakeSink{}
|
||
tl := newTestTickLoop(t, st, sink, nil)
|
||
|
||
tl.tick(ctx, refNow())
|
||
|
||
if len(sink.sends) != 0 {
|
||
t.Fatalf("cold-store tick sent %d; want 0 (shuts up when no data)", len(sink.sends))
|
||
}
|
||
}
|
||
|
||
func TestTickWaterFiresWhenDueAndPresent(t *testing.T) {
|
||
// water fact 4h ago ⇒ since(water)=4h ≥ 3h ⇒ predicate true. presence
|
||
// present ⇒ sev1 care gate holds (no quiet/cal/cooldown). routing for
|
||
// sev1 present is [voice] — one send captured.
|
||
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("tick sends = %d, want 1 (water, voice only)", len(sink.sends))
|
||
}
|
||
if got, want := sink.sends[0].RuleName, "water"; got != want {
|
||
t.Errorf("send rule = %q, want %q", got, want)
|
||
}
|
||
if got, want := sink.sends[0].Channel, delivery.ChannelVoice; got != want {
|
||
t.Errorf("send channel = %v, want voice (sev1 present)", got)
|
||
}
|
||
if sink.sends[0].Body == "" {
|
||
t.Error("phraser Stub produced an empty body for the water nudge")
|
||
}
|
||
|
||
// one nudge row recorded with channel=voice — verify the dispatch path
|
||
// wrote through to the store (the feedback loop's only input). RecentOutcomes
|
||
// filters to resolved rows, so confirm the recorded nudge exists via LastNudge.
|
||
n, err := st.LastNudge(ctx, "water")
|
||
if err != nil {
|
||
t.Fatalf("LastNudge: %v", err)
|
||
}
|
||
if n.Channel != string(delivery.ChannelVoice) {
|
||
t.Errorf("recorded nudge channel = %q, want %q", n.Channel, delivery.ChannelVoice)
|
||
}
|
||
}
|
||
|
||
func TestTickCooldownSuppressesSecondSend(t *testing.T) {
|
||
// After a water nudge, the gate's cooldown (DefaultRules sets water
|
||
// base cooldown = 30m) suppresses the same rule on the next tick.
|
||
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)
|
||
_ = err
|
||
}
|
||
sink := &fakeSink{}
|
||
tl := newTestTickLoop(t, st, sink, nil)
|
||
|
||
tl.tick(ctx, now) // fires
|
||
tl.tick(ctx, now.Add(time.Minute)) // still within 30m cooldown ⇒ suppressed
|
||
|
||
if len(sink.sends) != 1 {
|
||
t.Fatalf("sends after second tick = %d, want 1 (cooldown should suppress)", len(sink.sends))
|
||
}
|
||
}
|
||
|
||
func TestTickReminderFiresOnceAndMarkedFired(t *testing.T) {
|
||
// A due reminder bypasses the gate. away (no presence probes ⇒ cold
|
||
// start away) routes the reminder to [ntfy]. the dispatcher marks the
|
||
// reminder fired only after at least one channel succeeded; verify by
|
||
// re-ticking and confirming it isn't re-dispatched (DueReminders returns
|
||
// only `status == pending AND fire_ts <= now`).
|
||
st := newTestStore(t)
|
||
ctx := context.Background()
|
||
now := refNow()
|
||
if _, err := st.CreateReminder(ctx, now.Add(-time.Minute), `{"text":"stand up"}`, ""); err != nil {
|
||
t.Fatalf("CreateReminder: %v", err)
|
||
}
|
||
sink := &fakeSink{}
|
||
tl := newTestTickLoop(t, st, sink, nil)
|
||
|
||
tl.tick(ctx, now)
|
||
if got, want := len(sink.sends), 1; got != want {
|
||
t.Fatalf("reminder tick sends = %d, want 1 (away → ntfy)", got)
|
||
}
|
||
if sink.sends[0].Channel != delivery.ChannelNtfy {
|
||
t.Errorf("reminder channel = %v, want ntfy (away)", sink.sends[0].Channel)
|
||
}
|
||
if sink.sends[0].Body != "stand up" {
|
||
t.Errorf("reminder body = %q, want %q (router payload text)", sink.sends[0].Body, "stand up")
|
||
}
|
||
if sink.sends[0].Kind != delivery.KindReminder {
|
||
t.Errorf("reminder kind = %v, want %v", sink.sends[0].Kind, delivery.KindReminder)
|
||
}
|
||
|
||
// re-tick: the reminder is no longer pending (marked fired) ⇒ not in
|
||
// DueReminders ⇒ the reminder path is silent.
|
||
sink.sends = nil
|
||
tl.tick(ctx, now.Add(time.Minute))
|
||
if len(sink.sends) != 0 {
|
||
t.Fatalf("second reminder tick sends = %d, want 0 (fired once)", len(sink.sends))
|
||
}
|
||
}
|
||
|
||
// TestTuneWritesFeedbackCooldownToStore — the daemon's impure tune() step end
|
||
// to end: seed a rule with ≥ TuneMinOutcomes resolved `ignored` outcomes,
|
||
// call tune(), assert it wrote a `cooldown:<rule>` (source=feedback) fact that
|
||
// the gatherer then reads back as the active cooldown base. This is the closed
|
||
// feedback loop: outcomes → tune → write → gather → gate sees the tuned base.
|
||
func TestTuneWritesFeedbackCooldownToStore(t *testing.T) {
|
||
st := newTestStore(t)
|
||
ctx := context.Background()
|
||
now := refNow()
|
||
|
||
sink := &fakeSink{}
|
||
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.
|
||
r := loop.WaterRule()
|
||
for i := 0; i < loop.TuneSampleN; i++ {
|
||
id, err := st.RecordNudge(ctx, r.Name, "voice", "drink water", now.Add(-time.Hour))
|
||
if err != nil {
|
||
t.Fatalf("RecordNudge %d: %v", i, err)
|
||
}
|
||
if err := st.ResolveNudge(ctx, id, store.NudgeIgnored, now); err != nil {
|
||
t.Fatalf("ResolveNudge %d: %v", i, err)
|
||
}
|
||
}
|
||
|
||
tl.tune(ctx)
|
||
|
||
// the gatherer should now use the tuned base (clamped to WaterRule.Max =
|
||
// 6h) as the cooldown base for `water`. verify via the cooldown-until
|
||
// field by seeding a water nudge + asserting cooldown = sendTs + Max.
|
||
// (we read the persisted tuned base directly rather than going via gather
|
||
// so the assertion isolates tune()'s write from the gatherer path.)
|
||
fb, err := st.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource)
|
||
if err != nil {
|
||
t.Fatalf("LatestFactBySource: %v (no feedback fact written?)", err)
|
||
}
|
||
tuned, ok := loop.ParseCooldownFact(fb)
|
||
if !ok {
|
||
t.Fatalf("ParseCooldownFact: not ok (value %q)", fb.Value)
|
||
}
|
||
if tuned != 45*time.Minute {
|
||
t.Fatalf("all-ignored: base 30m × 1.5 = 45m (no — under WaterRule.Max 6h so unclamped): want 45m, got %v", tuned)
|
||
}
|
||
|
||
// idempotent: a second tune() with the same outcomes writes nothing new
|
||
// (RecentOutcomes is steady between resolves; the persisted value equals
|
||
// the computed one ⇒ skip).
|
||
fb1 := fb
|
||
tl.tune(ctx)
|
||
fb2, err := st.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource)
|
||
if err != nil {
|
||
t.Fatalf("LatestFactBySource second call: %v", err)
|
||
}
|
||
if fb2.ID != fb1.ID {
|
||
t.Fatalf("tune() re-wrote identical value: fact id %d → %d (should skip when unchanged)",
|
||
fb1.ID, fb2.ID)
|
||
}
|
||
|
||
// flip the outcomes pattern to `acted`: next tune() writes a new value,
|
||
// shorter than Max. RecentOutcomes is sorted DESC ts,id, so re-seeding
|
||
// newer-acted nudges makes them dominate the older-ignored set.
|
||
for i := 0; i < loop.TuneSampleN; i++ {
|
||
id, err := st.RecordNudge(ctx, r.Name, "voice", "drink water", now.Add(time.Minute+time.Duration(i)*time.Second))
|
||
if err != nil {
|
||
t.Fatalf("RecordNudge acted %d: %v", i, err)
|
||
}
|
||
if err := st.ResolveNudge(ctx, id, store.NudgeActed, now); err != nil {
|
||
t.Fatalf("ResolveNudge acted %d: %v", i, err)
|
||
}
|
||
}
|
||
tl.tune(ctx)
|
||
fb3, err := st.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource)
|
||
if err != nil {
|
||
t.Fatalf("LatestFactBySource post-flip: %v", err)
|
||
}
|
||
tuned3, ok := loop.ParseCooldownFact(fb3)
|
||
if !ok {
|
||
t.Fatalf("ParseCooldownFact post-flip: not ok (value %q)", fb3.Value)
|
||
}
|
||
if tuned3 >= r.Cooldown.Max {
|
||
t.Fatalf("acted-dominated outcomes should shrink cooldown below Max: got %v (Max %v)",
|
||
tuned3, r.Cooldown.Max)
|
||
}
|
||
if tuned3 < r.Cooldown.Min {
|
||
t.Fatalf("tuned cooldown below Min envelope: got %v (Min %v) — clamp broken",
|
||
tuned3, r.Cooldown.Min)
|
||
}
|
||
|
||
// the gatherer actually reads it back: assert CooldownUntil for water is
|
||
// derived from the tuned base (not the rule's static Base) when a real
|
||
// nudge exists. seed a water nudge now and gather.
|
||
nudgeTs := now.Add(2 * time.Minute)
|
||
if _, err := st.RecordNudge(ctx, r.Name, "voice", "drink water", nudgeTs); err != nil {
|
||
t.Fatalf("final RecordNudge: %v", err)
|
||
}
|
||
g := loop.NewGatherer(st, loop.DefaultRules())
|
||
snap, _, err := g.GatherState(ctx, now.Add(3*time.Minute))
|
||
if err != nil {
|
||
t.Fatalf("GatherState: %v", err)
|
||
}
|
||
wantUntil := nudgeTs.Add(tuned3)
|
||
if got := snap.CooldownUntil[r.Name]; got != wantUntil {
|
||
t.Fatalf("gatherer used tuned base: CooldownUntil want %v, got %v",
|
||
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))
|
||
}
|
||
}
|