From 4f516657dab69d3871ca141db6004bf0be450952 Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 4 Aug 2026 05:14:50 +0400 Subject: [PATCH 1/4] loop, store: read a fact family by prefix (V-444) A rule over a key set that only exists at read time cannot declare its keys at wiring time. Kuma has one monitor per service and the names live in the gauge, so the rule declares a prefix and the gatherer resolves the family per tick. ServiceDownRule now fires on any monitor reading down, names it through DownServices, and is edge-triggered: a service that stays down is one nudge, not one per tick with cooldown as the only brake. --- internal/loop/gather.go | 14 ++++++++ internal/loop/rules.go | 68 +++++++++++++++++++++++++++++------- internal/loop/state.go | 28 +++++++++++++++ internal/store/facts.go | 35 +++++++++++++++++++ internal/store/facts_test.go | 62 ++++++++++++++++++++++++++++++++ 5 files changed, 194 insertions(+), 13 deletions(-) create mode 100644 internal/store/facts_test.go diff --git a/internal/loop/gather.go b/internal/loop/gather.go index c55919f..73e164f 100644 --- a/internal/loop/gather.go +++ b/internal/loop/gather.go @@ -91,6 +91,20 @@ func (g *Gatherer) GatherState(ctx context.Context, now time.Time) (State, []sto return State{}, nil, err } + // prefix families — the keys a rule cannot name at wiring time (one fact + // per kuma monitor). Loaded into the same map; State.FactsUnder reads them. + for _, r := range g.rules { + for _, p := range r.WantPrefixes { + fam, err := g.store.LatestFactsByPrefix(ctx, p) + if err != nil { + return State{}, nil, err + } + for _, f := range fam { + facts[f.Key] = f + } + } + } + // last nudge per rule + cooldown-until derived from the active cooldown. // "active" = the feedback tuner's persisted base if one exists, else the // rule's static Base. LatestFactBySource is the trust-by-provenance read diff --git a/internal/loop/rules.go b/internal/loop/rules.go index f144e2d..1fc5ff6 100644 --- a/internal/loop/rules.go +++ b/internal/loop/rules.go @@ -28,6 +28,13 @@ type Rule struct { // that check itself, leave this empty. Otherwise set to the key(s) the rule // needs and the gate will skip the rule when any are missing. InertWhenNoData []string + + // WantPrefixes — key prefixes whose whole family the gatherer must load. + // InertWhenNoData names keys that exist at wiring time; a rule over a key + // set that is only known at read time (one fact per kuma monitor) declares + // the prefix here instead. Prefixes never make a rule inert: an empty + // family is the predicate's own "no data" case. + WantPrefixes []string } // Cooldown — tunable bounded by the envelope so a weird week (auto-tuned) can't @@ -98,23 +105,58 @@ func BreakRule() Rule { } } -// ServiceDownRule — sev4 ops hard: the `service_down` aggregate fact reads -// "down". Source must be poll:uptimekuma — kuma is the source of truth for -// service up/down (mavpoll writes this key). The predicate is provenance-scoped: -// a compromised poller writing under a different source can't forge the trigger. +// ServiceDownPrefix — mavpoll writes one fact per kuma monitor under this +// prefix, `service_down:`. The suffix is the name he hears. +const ServiceDownPrefix = "service_down:" + +// ServiceDownSource — kuma is the source of truth for service up/down. The +// rule is provenance-scoped: a poller writing under a different source cannot +// forge the trigger. +const ServiceDownSource = "poll:uptimekuma" + +// DownServices — the monitors currently reading "down", by name, in key order. +// +// Pure, and the rule and the phraser both call it, so the message can never +// name a service the predicate did not fire on. +func DownServices(s State) []string { + var out []string + for _, f := range s.FactsUnder(ServiceDownPrefix) { + if f.Source == ServiceDownSource && f.Value == `"down"` { + out = append(out, strings.TrimPrefix(f.Key, ServiceDownPrefix)) + } + } + return out +} + +// ServiceDownRule — sev4 ops hard: at least one kuma monitor reads "down". +// +// It used to read one aggregate `service_down` fact, which is why it was +// disabled in deploy: the nudge could say that something on homesrv was down +// but never which thing. Per-monitor facts fix that, and pausing a monitor in +// kuma now silences that monitor rather than nothing. +// +// Edge-triggered — see State.NudgedSince. Without it a service that stays down +// for a day qualifies on every tick and cooldown alone is the only brake. func ServiceDownRule() Rule { return Rule{ - Name: "service_down", - Severity: Sev4, - Cooldown: Cooldown{Base: 15 * time.Minute, Min: 5 * time.Minute, Max: 1 * time.Hour}, - InertWhenNoData: []string{"service_down"}, + Name: "service_down", + Severity: Sev4, + Cooldown: Cooldown{Base: 15 * time.Minute, Min: 5 * time.Minute, Max: 1 * time.Hour}, + WantPrefixes: []string{ServiceDownPrefix}, Predicate: func(s State) bool { - f, ok := s.Fact("service_down") - if !ok || f.Ts.IsZero() { - return false + var newest time.Time + for _, f := range s.FactsUnder(ServiceDownPrefix) { + if f.Source != ServiceDownSource || f.Value != `"down"` { + continue + } + if f.Ts.After(newest) { + newest = f.Ts + } } - // value is json `"down"`; trivial check keyed off source provenance. - return f.Source == "poll:uptimekuma" && f.Value == `"down"` + if newest.IsZero() { + return false // nothing down, or no data at all → shut up + } + return !s.NudgedSince("service_down", newest) }, } } diff --git a/internal/loop/state.go b/internal/loop/state.go index f17b074..1c42f6e 100644 --- a/internal/loop/state.go +++ b/internal/loop/state.go @@ -23,6 +23,8 @@ package loop import ( + "sort" + "strings" "time" "github.com/kami/maven/internal/store" @@ -95,6 +97,32 @@ func (s State) Fact(key string) (store.Fact, bool) { return f, true } +// FactsUnder returns every gathered fact whose key starts with prefix, ordered +// by key so a caller that names them speaks them in a stable order. Facts with +// a zero Ts are skipped, the same "no data" rule Fact applies. +func (s State) FactsUnder(prefix string) []store.Fact { + var out []store.Fact + for k, f := range s.Facts { + if strings.HasPrefix(k, prefix) && !f.Ts.IsZero() { + out = append(out, f) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key }) + return out +} + +// NudgedSince reports whether rule already sent a nudge at or after ts. +// +// It is what makes a rule edge-triggered. A polled fact is written only when +// the value changes, so its Ts is the moment the service went down — but the +// predicate reads the current value, so a service that stays down keeps +// qualifying forever and cooldown alone only slows the repetition. Asking +// whether he was already told about THIS transition stops it. +func (s State) NudgedSince(rule string, ts time.Time) bool { + n, ok := s.LastNudge[rule] + return ok && !n.Ts.Before(ts) +} + // Since returns the duration since the latest fact for key, or (0,false). // "false" ⇒ no data ⇒ shuts up when uncertain. func (s State) Since(key string) (time.Duration, bool) { diff --git a/internal/store/facts.go b/internal/store/facts.go index 6e0a2c6..0644fda 100644 --- a/internal/store/facts.go +++ b/internal/store/facts.go @@ -254,6 +254,41 @@ func (s *Store) LatestFactBySource(ctx context.Context, key, source string) (Fac return scanFact(row) } +// LatestFactsByPrefix — the latest non-voided fact for every key that starts +// with prefix, newest-per-key, ordered by key. +// +// The loop's gatherer loads the keys its rules declare, which works while the +// key set is static. Kuma's monitors are not: one fact per monitor means the +// keys are only known once the gauge is read, so the rule declares the prefix +// and this read resolves it per tick. `_` and `%` are escaped — a monitor name +// is user text and must not act as a LIKE wildcard. +func (s *Store) LatestFactsByPrefix(ctx context.Context, prefix string) ([]Fact, error) { + esc := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(prefix) + rows, err := s.db.QueryContext(ctx, ` + SELECT id, ts, kind, key, value, source, confidence, voids_id + FROM facts f + WHERE key LIKE ? ESCAPE '\' + AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL) + AND id = (SELECT id FROM facts g + WHERE g.key = f.key + AND g.id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL) + ORDER BY g.ts DESC, g.id DESC LIMIT 1) + ORDER BY key`, esc+"%") + if err != nil { + return nil, fmt.Errorf("facts by prefix %q: %w", prefix, err) + } + defer rows.Close() + var out []Fact + for rows.Next() { + f, err := scanFact(rows) + if err != nil { + return nil, err + } + out = append(out, f) + } + return out, rows.Err() +} + // Since returns how long ago the latest non-voided fact for key landed, or // (0, ErrNoFact). Implements the `since(key)==null → don't fire` guard from // the spec — silence on no-data is "shuts up when uncertain". diff --git a/internal/store/facts_test.go b/internal/store/facts_test.go new file mode 100644 index 0000000..47e666e --- /dev/null +++ b/internal/store/facts_test.go @@ -0,0 +1,62 @@ +package store + +import ( + "context" + "testing" + "time" +) + +// One fact per kuma monitor means the loop cannot name its keys at wiring time, +// so it asks for the family by prefix. The read must return the newest row per +// key and stop at the prefix boundary. +func TestLatestFactsByPrefix(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Millisecond) + write := func(key, val string, at time.Time) int64 { + id, err := s.SetValue(ctx, KindEnv, key, "poll:uptimekuma", val, at) + if err != nil { + t.Fatalf("SetValue %s: %v", key, err) + } + return id + } + write("service_down:db", "up", now.Add(-2*time.Hour)) + write("service_down:db", "down", now.Add(-time.Hour)) // newer wins + write("service_down:web", "up", now.Add(-time.Hour)) + write("service_downtime", "irrelevant", now) // no colon, not in the family + write("water", "250ml", now) + + got, err := s.LatestFactsByPrefix(ctx, "service_down:") + if err != nil { + t.Fatalf("LatestFactsByPrefix: %v", err) + } + if len(got) != 2 { + t.Fatalf("got %d facts, want 2: %+v", len(got), got) + } + if got[0].Key != "service_down:db" || got[0].Value != `"down"` { + t.Errorf("first = %s=%s, want the newest db row", got[0].Key, got[0].Value) + } + if got[1].Key != "service_down:web" { + t.Errorf("second = %s, want service_down:web", got[1].Key) + } +} + +// A monitor name is user text. An underscore in it must match itself, not act +// as a LIKE wildcard and drag in every other monitor. +func TestLatestFactsByPrefixEscapesWildcards(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Millisecond) + for _, k := range []string{"a_b:one", "axb:two"} { + if _, err := s.SetValue(ctx, KindEnv, k, "poll:uptimekuma", "down", now); err != nil { + t.Fatalf("SetValue %s: %v", k, err) + } + } + got, err := s.LatestFactsByPrefix(ctx, "a_b:") + if err != nil { + t.Fatalf("LatestFactsByPrefix: %v", err) + } + if len(got) != 1 || got[0].Key != "a_b:one" { + t.Fatalf("got %+v, want only a_b:one", got) + } +} From 09c648b93491fd43ba89aecc7f333eab4089e9f1 Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 4 Aug 2026 05:15:01 +0400 Subject: [PATCH 2/4] mavpoll: write one kuma fact per monitor (V-444) The aggregate could not name the service, which is the whole reason the nudge said 'a service on homesrv is down' and the rule shipped disabled. A monitor deleted in kuma stops appearing in the gauge and its last fact would read down forever, so a vanished monitor is marked unknown. Pending and maintenance are not down: a monitor paused in kuma now silences that monitor rather than nothing. --- cmd/mavpoll/main.go | 92 +++++++++++++++++++++++++++++++--------- cmd/mavpoll/main_test.go | 48 +++++++++++++++------ 2 files changed, 107 insertions(+), 33 deletions(-) diff --git a/cmd/mavpoll/main.go b/cmd/mavpoll/main.go index 69c9eb4..f3f0702 100644 --- a/cmd/mavpoll/main.go +++ b/cmd/mavpoll/main.go @@ -140,6 +140,10 @@ type poller struct { wgIface string wgCmd string + // kumaSeen — monitor name → state as of the last poll, so a monitor that + // disappears from the gauge can be marked unknown instead of staying down. + kumaSeen map[string]string + // zen is nil unless a token file was configured — money tracking is a // capability, off by default like weather and telegram. zen *zenmoney.Client @@ -333,50 +337,96 @@ func maxSeverity(a netdataAlarms) string { return sev } -// ---- kuma: monitor_status gauge → aggregate service_down ------------------- +// ---- kuma: monitor_status gauge → one fact per monitor --------------------- // Kuma exposes Prometheus text: `monitor_status{...,monitor_name="X"} V` where -// V is 1=up 0=down 2=pending 3=maintenance. We reduce to one aggregate the -// existing ServiceDownRule consumes: "down" if ANY monitor reads 0, else "up". -// Per-service granularity is a later add (a fact per monitor) — the MVP nudge -// only needs "something is down". -var kumaLine = regexp.MustCompile(`^monitor_status\{([^}]*)\}\s+([0-9.eE+-]+)`) +// V is 1=up 0=down 2=pending 3=maintenance. We write one fact per monitor, +// keyed `service_down:`, because the nudge has to say WHICH +// service is down. The aggregate this used to write could not, which is why +// the rule shipped disabled. +var ( + kumaLine = regexp.MustCompile(`^monitor_status\{([^}]*)\}\s+([0-9.eE+-]+)`) + kumaName = regexp.MustCompile(`monitor_name="([^"]*)"`) +) func (p *poller) pollKuma(ctx context.Context, now time.Time) error { body, err := p.get(ctx, p.kumaURL, p.kumaKey) if err != nil { return err } - down, seen := kumaAnyDown(body) - if !seen { + states := kumaMonitors(body) + if len(states) == 0 { return fmt.Errorf("no monitor_status metrics (auth/endpoint wrong?)") } - val := "up" - if down { - val = "down" + var firstErr error + for name, val := range states { + if err := p.writeIfChanged(ctx, kumaFactKey(name), kumaSource, val, now); err != nil && firstErr == nil { + firstErr = err // one bad monitor must not blind the rest + } } - return p.writeIfChanged(ctx, "service_down", "poll:uptimekuma", val, now) + // A monitor deleted in kuma stops appearing in the gauge, and its last fact + // would otherwise read "down" forever. Mark it unknown, which no rule fires + // on. The seen-set is in memory, so a restart forgets it — harmless, since + // the next poll that still lacks the monitor says nothing new either. + for name := range p.kumaSeen { + if _, still := states[name]; !still { + if err := p.writeIfChanged(ctx, kumaFactKey(name), kumaSource, "unknown", now); err != nil && firstErr == nil { + firstErr = err + } + } + } + p.kumaSeen = states + return firstErr } -// kumaAnyDown parses kuma's Prometheus text: down=true if any monitor reads 0 -// (pending=2/maintenance=3 are not "down"). seen=false ⇒ no monitor_status -// lines matched at all (wrong endpoint or auth rejected before the body). -func kumaAnyDown(body []byte) (down, seen bool) { +// kumaSource — the provenance the loop rule requires. Written here, checked in +// loop.ServiceDownRule; a poller under any other source cannot fire it. +const kumaSource = "poll:uptimekuma" + +// kumaFactKey — the fact key for one monitor. The suffix is the name he hears, +// so it stays as kuma spells it rather than being slugged into something else. +func kumaFactKey(name string) string { return "service_down:" + name } + +// kumaMonitors parses kuma's Prometheus text into monitor name → state +// ("up"/"down"/"pending"/"maintenance"). An empty map means no monitor_status +// line matched at all (wrong endpoint, or auth rejected before the body). +// A line with no monitor_name label is skipped: a fact nobody can name is +// exactly the thing this replaced. +func kumaMonitors(body []byte) map[string]string { + out := make(map[string]string) for _, line := range strings.Split(string(body), "\n") { m := kumaLine.FindStringSubmatch(strings.TrimSpace(line)) if m == nil { continue } - seen = true + nm := kumaName.FindStringSubmatch(m[1]) + if nm == nil || strings.TrimSpace(nm[1]) == "" { + continue + } v, err := strconv.ParseFloat(m[2], 64) if err != nil { continue } - if v == 0 { - down = true - } + out[strings.TrimSpace(nm[1])] = kumaState(v) + } + return out +} + +// kumaState — the gauge's four values. pending and maintenance are not "down": +// a monitor paused in kuma should silence that monitor, not page him. +func kumaState(v float64) string { + switch v { + case 0: + return "down" + case 1: + return "up" + case 2: + return "pending" + case 3: + return "maintenance" + default: + return "unknown" } - return down, seen } // ---- helpers --------------------------------------------------------------- diff --git a/cmd/mavpoll/main_test.go b/cmd/mavpoll/main_test.go index 7004518..229dd19 100644 --- a/cmd/mavpoll/main_test.go +++ b/cmd/mavpoll/main_test.go @@ -35,22 +35,46 @@ func TestMaxSeverity(t *testing.T) { } } -func TestKumaAnyDown(t *testing.T) { +func TestKumaMonitorsNamesEveryOne(t *testing.T) { cases := []struct { - body string - down, seen bool + name string + body string + want map[string]string }{ - {"", false, false}, - {`monitor_status{monitor_name="web"} 1`, false, true}, - {`monitor_status{monitor_name="web"} 1` + "\n" + `monitor_status{monitor_name="db"} 0`, true, true}, - {`monitor_status{monitor_name="mnt"} 3`, false, true}, // maintenance ≠ down - {`# HELP monitor_status ...`, false, false}, + {"empty body", "", map[string]string{}}, + {"help line only", `# HELP monitor_status ...`, map[string]string{}}, + { + "one up one down", + `monitor_status{monitor_name="web"} 1` + "\n" + `monitor_status{monitor_name="db"} 0`, + map[string]string{"web": "up", "db": "down"}, + }, + {"maintenance is not down", `monitor_status{monitor_name="mnt"} 3`, map[string]string{"mnt": "maintenance"}}, + {"pending is not down", `monitor_status{monitor_name="p"} 2`, map[string]string{"p": "pending"}}, + { + "other labels do not hide the name", + `monitor_status{monitor_type="http",monitor_name="ci",monitor_url="x"} 0`, + map[string]string{"ci": "down"}, + }, + {"a nameless line is skipped", `monitor_status{monitor_type="http"} 0`, map[string]string{}}, } for _, c := range cases { - down, seen := kumaAnyDown([]byte(c.body)) - if down != c.down || seen != c.seen { - t.Errorf("kumaAnyDown(%q) = (%v,%v), want (%v,%v)", c.body, down, seen, c.down, c.seen) - } + t.Run(c.name, func(t *testing.T) { + got := kumaMonitors([]byte(c.body)) + if len(got) != len(c.want) { + t.Fatalf("kumaMonitors = %v, want %v", got, c.want) + } + for k, v := range c.want { + if got[k] != v { + t.Errorf("monitor %q = %q, want %q", k, got[k], v) + } + } + }) + } +} + +func TestKumaFactKeyCarriesTheName(t *testing.T) { + if got := kumaFactKey("nexus db"); got != "service_down:nexus db" { + t.Errorf("kumaFactKey = %q", got) } } From 7e21cd06b34ac464e91ec0ce6f255e0ac3bbed64 Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 4 Aug 2026 05:15:01 +0400 Subject: [PATCH 3/4] phraser: name the service that is down (V-444) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stub and LLM paths both read loop.DownServices, so the message can never name a service the predicate did not fire on. Two down at once are both named — he needs the blast radius. --- cmd/mavend/tick_test.go | 2 +- internal/loop/explain_test.go | 6 +- internal/loop/gate_test.go | 12 ++-- internal/loop/gather_test.go | 47 ++++++++++++++++ internal/loop/loop_test.go | 8 +-- internal/loop/rules_test.go | 95 +++++++++++++++++++++++++++----- internal/phraser/llmphraser.go | 8 +++ internal/phraser/phraser.go | 23 +++++--- internal/phraser/phraser_test.go | 36 +++++++++--- 9 files changed, 192 insertions(+), 45 deletions(-) create mode 100644 internal/loop/gather_test.go diff --git a/cmd/mavend/tick_test.go b/cmd/mavend/tick_test.go index 7290ece..c2c4ac9 100644 --- a/cmd/mavend/tick_test.go +++ b/cmd/mavend/tick_test.go @@ -584,7 +584,7 @@ func TestDigestSev4BypassesQueue(t *testing.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 { + if _, err := st.SetValue(ctx, store.KindSelf, "service_down:db", "poll:uptimekuma", "down", now); err != nil { t.Fatalf("seed service_down: %v", err) } sink := &fakeSink{} diff --git a/internal/loop/explain_test.go b/internal/loop/explain_test.go index 0acbf30..1110106 100644 --- a/internal/loop/explain_test.go +++ b/internal/loop/explain_test.go @@ -103,7 +103,7 @@ func TestExplainGate_PresenceAway(t *testing.T) { func TestExplainGate_PresenceAwayOpsBypass(t *testing.T) { now := refTime() s := State{Now: now, Presence: store.Away, - Facts: map[string]store.Fact{"service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute))}, + Facts: map[string]store.Fact{"service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute))}, } r := ServiceDownRule() // Sev4 ops passed, blocked, d := ExplainGate(s, r) @@ -259,8 +259,8 @@ func TestExplainTick_WinnerRecorded(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), - "service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), + "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), + "service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), }, } cand, trace := ExplainTick(s, DefaultRules()) diff --git a/internal/loop/gate_test.go b/internal/loop/gate_test.go index eb613d4..4ad6e90 100644 --- a/internal/loop/gate_test.go +++ b/internal/loop/gate_test.go @@ -198,12 +198,12 @@ func TestTickNeverDogpilesAndPicksLoudest(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "water": ago("water", "tap:water", `"250ml"`, 5*time.Hour), - "meal": ago("meal", "voice", `"lunch"`, 8*time.Hour), - "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), - "break": ago("break", "voice", `"walk"`, 3*time.Hour), - "service_down": ago("service_down", "poll:uptimekuma", `"down"`, time.Minute), - "netdata_alarm": ago("netdata_alarm", "poll:netdata", `"critical"`, time.Minute), + "water": ago("water", "tap:water", `"250ml"`, 5*time.Hour), + "meal": ago("meal", "voice", `"lunch"`, 8*time.Hour), + "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), + "break": ago("break", "voice", `"walk"`, 3*time.Hour), + "service_down:db": ago("service_down:db", "poll:uptimekuma", `"down"`, time.Minute), + "netdata_alarm": ago("netdata_alarm", "poll:netdata", `"critical"`, time.Minute), }, } // sanity: every rule really does want to fire, so the pick is a real choice. diff --git a/internal/loop/gather_test.go b/internal/loop/gather_test.go new file mode 100644 index 0000000..c854eba --- /dev/null +++ b/internal/loop/gather_test.go @@ -0,0 +1,47 @@ +package loop + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/kami/maven/internal/store" +) + +// The gatherer is the only impure piece, and a rule over a prefix has no keys +// to declare at wiring time. This is the end of that path: mavpoll's per-monitor +// facts reach the snapshot, and the rule fires on the one that is down. +func TestGatherStateLoadsPrefixFamilies(t *testing.T) { + ctx := context.Background() + s, err := store.Open(ctx, filepath.Join(t.TempDir(), "loop_test.db")) + if err != nil { + t.Fatalf("Open: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + + now := time.Now().UTC().Truncate(time.Millisecond) + for key, val := range map[string]string{ + "service_down:db": "down", + "service_down:web": "up", + } { + if _, err := s.SetValue(ctx, store.KindEnv, key, ServiceDownSource, val, now.Add(-time.Minute)); err != nil { + t.Fatalf("SetValue %s: %v", key, err) + } + } + + rules := []Rule{ServiceDownRule()} + st, _, err := NewGatherer(s, rules).GatherState(ctx, now) + if err != nil { + t.Fatalf("GatherState: %v", err) + } + if _, ok := st.Facts["service_down:db"]; !ok { + t.Fatalf("prefix family not gathered: %v", st.Facts) + } + if got := DownServices(st); len(got) != 1 || got[0] != "db" { + t.Fatalf("DownServices = %v, want [db]", got) + } + if !rules[0].Predicate(st) { + t.Fatal("the rule must fire on a gathered per-monitor fact") + } +} diff --git a/internal/loop/loop_test.go b/internal/loop/loop_test.go index d6837c8..0f5cf1a 100644 --- a/internal/loop/loop_test.go +++ b/internal/loop/loop_test.go @@ -82,7 +82,7 @@ func TestTickOpsHardSurvivesAwayAndQuiet(t *testing.T) { Presence: store.Away, QuietHours: true, Facts: map[string]store.Fact{ - "service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), + "service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), }, } got := Tick(s, DefaultRules()) @@ -99,7 +99,7 @@ func TestTickServiceSourceTrustRefusesForgedTrigger(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "service_down": factAt("service_down", "ambient", `"down"`, now.Add(-1*time.Minute)), + "service_down:db": factAt("service_down:db", "ambient", `"down"`, now.Add(-1*time.Minute)), }, } if got := Tick(s, DefaultRules()); got != nil { @@ -115,8 +115,8 @@ func TestTickOneNudgePerTickMaxSeverityWins(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), - "service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), + "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), + "service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), }, } got := Tick(s, DefaultRules()) diff --git a/internal/loop/rules_test.go b/internal/loop/rules_test.go index 3bec077..70cc11c 100644 --- a/internal/loop/rules_test.go +++ b/internal/loop/rules_test.go @@ -193,13 +193,13 @@ func TestOpsRulePredicates(t *testing.T) { { name: "service_down fires on a kuma down fact", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma", `"down"`, time.Minute)}, want: true, }, { name: "service_down quiet when kuma says up", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma", `"up"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma", `"up"`, time.Minute)}, want: false, }, { @@ -211,38 +211,38 @@ func TestOpsRulePredicates(t *testing.T) { { name: "service_down quiet on a zero-timestamp fact", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": {Key: "service_down", Source: "poll:uptimekuma", Value: `"down"`}}, + facts: map[string]store.Fact{"service_down:db": {Key: "service_down:db", Source: "poll:uptimekuma", Value: `"down"`}}, want: false, }, // forgery attempts — right value, wrong writer. { name: "service_down refuses a forgery from the netdata poller", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:netdata", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:netdata", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses a forgery from ambient audio", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "ambient:other", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "ambient:other", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses a forgery from the user's own voice", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "voice", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "voice", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses a source that only looks like kuma", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma-staging", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma-staging", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses an unquoted down value", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma", `down`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma", `down`, time.Minute)}, want: false, }, @@ -312,9 +312,11 @@ func TestOpsRulePredicates(t *testing.T) { // a second no-data backstop, so a rule that forgets it loses the safety net // even if its predicate happens to check. func TestDefaultRulesDeclareInertKeys(t *testing.T) { + // A rule over a key set that only exists at read time declares a prefix + // instead — the gatherer still needs to be told what to load. for _, r := range DefaultRules() { - if len(r.InertWhenNoData) == 0 { - t.Errorf("rule %q declares no InertWhenNoData keys", r.Name) + if len(r.InertWhenNoData) == 0 && len(r.WantPrefixes) == 0 { + t.Errorf("rule %q declares neither InertWhenNoData keys nor WantPrefixes", r.Name) } } } @@ -377,11 +379,11 @@ func TestDefaultRuleCooldownsAreBounded(t *testing.T) { // hidden state, no clock reads. func TestPredicatesArePure(t *testing.T) { s := stateWith(map[string]store.Fact{ - "water": ago("water", "tap:water", `"250ml"`, 4*time.Hour), - "meal": ago("meal", "voice", `"lunch"`, 7*time.Hour), - "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), - "break": ago("break", "voice", `"walk"`, 2*time.Hour), - "service_down": ago("service_down", "poll:uptimekuma", `"down"`, time.Minute), + "water": ago("water", "tap:water", `"250ml"`, 4*time.Hour), + "meal": ago("meal", "voice", `"lunch"`, 7*time.Hour), + "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), + "break": ago("break", "voice", `"walk"`, 2*time.Hour), + "service_down:db": ago("service_down:db", "poll:uptimekuma", `"down"`, time.Minute), }) for _, r := range DefaultRules() { first := r.Predicate(s) @@ -434,3 +436,66 @@ func TestRulesExceptEmptyKeepsEverything(t *testing.T) { t.Errorf("rules = %v, dropped = %v", ruleNames(rules), dropped) } } + +// ---------------------------- per-monitor service_down ----------------------- + +// The rule must name what fired on it, and the phraser reads the same helper, +// so a service that is up can never be spoken as down. +func TestDownServicesNamesOnlyTheDownOnes(t *testing.T) { + s := State{ + Now: refTime(), + Facts: map[string]store.Fact{ + "service_down:web": ago("service_down:web", ServiceDownSource, `"up"`, time.Minute), + "service_down:db": ago("service_down:db", ServiceDownSource, `"down"`, time.Minute), + "service_down:vault": ago("service_down:vault", ServiceDownSource, `"down"`, time.Minute), + "service_down:paused": ago("service_down:paused", ServiceDownSource, `"maintenance"`, time.Minute), + "service_down:forged": ago("service_down:forged", "voice", `"down"`, time.Minute), + "service_down:missing": ago("service_down:missing", ServiceDownSource, `"unknown"`, time.Minute), + }, + } + got := DownServices(s) + want := []string{"db", "vault"} // key order, so speech is stable + if len(got) != len(want) { + t.Fatalf("DownServices = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("DownServices = %v, want %v", got, want) + } + } +} + +// A monitor paused in kuma must silence that monitor. Before per-monitor facts +// the aggregate stayed "down" and pausing achieved nothing. +func TestPausedMonitorSilencesOnlyItself(t *testing.T) { + base := map[string]store.Fact{ + "service_down:db": ago("service_down:db", ServiceDownSource, `"maintenance"`, time.Minute), + "service_down:web": ago("service_down:web", ServiceDownSource, `"down"`, time.Minute), + } + if !ServiceDownRule().Predicate(State{Now: refTime(), Facts: base}) { + t.Fatal("web is still down, the rule must fire") + } + delete(base, "service_down:web") + if ServiceDownRule().Predicate(State{Now: refTime(), Facts: base}) { + t.Fatal("only a paused monitor is left, the rule must be quiet") + } +} + +// Edge-triggered: he is told once per transition. A service that stays down +// for a day used to qualify on every tick, with cooldown as the only brake. +func TestServiceDownFiresOncePerTransition(t *testing.T) { + down := ago("service_down:db", ServiceDownSource, `"down"`, time.Hour) + s := State{Now: refTime(), Facts: map[string]store.Fact{"service_down:db": down}} + if !ServiceDownRule().Predicate(s) { + t.Fatal("first sight of the transition must fire") + } + s.LastNudge = map[string]store.Nudge{"service_down": {Ts: down.Ts.Add(time.Minute)}} + if ServiceDownRule().Predicate(s) { + t.Fatal("already told about this transition, must be quiet") + } + // A second service goes down after that nudge — a new edge, so it fires. + s.Facts["service_down:web"] = ago("service_down:web", ServiceDownSource, `"down"`, time.Minute) + if !ServiceDownRule().Predicate(s) { + t.Fatal("a later transition must fire again") + } +} diff --git a/internal/phraser/llmphraser.go b/internal/phraser/llmphraser.go index 2c5e28d..8d6c04f 100644 --- a/internal/phraser/llmphraser.go +++ b/internal/phraser/llmphraser.go @@ -975,6 +975,9 @@ var fallbackNudges = map[string]string{ } func fallbackNudge(c loop.Candidate) string { + if down := loop.DownServices(c.State); len(down) > 0 { + return "Не отвечает: " + strings.Join(down, ", ") + "." + } if s, ok := fallbackNudges[c.Rule.Name]; ok { return s } @@ -990,6 +993,11 @@ func buildNudgePrompt(c loop.Candidate) string { if f, ok := c.State.Facts[c.Rule.Name]; ok && f.Key != "" && f.Key != c.Rule.Name { ctxParts = append(ctxParts, "Что именно: "+f.Key) } + if down := loop.DownServices(c.State); len(down) > 0 { + // The names come from the same helper the rule fired on, so the model + // is never handed a service that is actually up. + ctxParts = append(ctxParts, "Какие сервисы лежат: "+strings.Join(down, ", ")) + } if d, ok := c.State.Since(c.Rule.Name); ok { ctxParts = append(ctxParts, "Прошло: "+ruDur(d)) } diff --git a/internal/phraser/phraser.go b/internal/phraser/phraser.go index 79488d2..cb3fa9d 100644 --- a/internal/phraser/phraser.go +++ b/internal/phraser/phraser.go @@ -142,16 +142,21 @@ func phraseNudge(c loop.Candidate) (body, summary string) { } return body, "take a break" case "service_down": - // the fact value is json `"down"`; the key carries the service name. - body = "a service on homesrv is down — check journalctl." - summary = "service down on homesrv" - if f, ok := c.State.Fact("service_down"); ok { - if f.Key != "" && f.Key != "service_down" { - body = fmt.Sprintf("%s on homesrv is down — check journalctl.", f.Key) - summary = fmt.Sprintf("%s down on homesrv", f.Key) - } + // One fact per kuma monitor, so the nudge names the service. The rule + // and this share loop.DownServices, so the message cannot name a + // service the predicate did not fire on. + down := loop.DownServices(c.State) + switch len(down) { + case 0: + return "a service on homesrv is down — check journalctl.", "service down on homesrv" + case 1: + return fmt.Sprintf("%s on homesrv is down — check journalctl.", down[0]), + fmt.Sprintf("%s down on homesrv", down[0]) + default: + list := strings.Join(down, ", ") + return fmt.Sprintf("%s on homesrv are down — check journalctl.", list), + fmt.Sprintf("%d services down on homesrv", len(down)) } - return body, summary default: // generic: name the rule + severity; the LLM impl replaces this with // a prompted phrase. the Stub never editorializes beyond the rule name. diff --git a/internal/phraser/phraser_test.go b/internal/phraser/phraser_test.go index 23cddfc..69bc2ea 100644 --- a/internal/phraser/phraser_test.go +++ b/internal/phraser/phraser_test.go @@ -78,12 +78,14 @@ func TestPhraseNudgeBreakDeskDuration(t *testing.T) { } func TestPhraseNudgeServiceDownNamedService(t *testing.T) { - // a service_down fact whose Key is the specific service name → the phrase - // names the service, not just "service down". + // one fact per kuma monitor → the phrase names the monitor that is down. now := time.Now().UTC() st := loop.State{ - Now: now, - Facts: map[string]store.Fact{"service_down": {Key: "nginx", Ts: now, Source: "poll:healthcheck", Value: `"down"`}}, + Now: now, + Facts: map[string]store.Fact{ + "service_down:nginx": {Key: "service_down:nginx", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}, + "service_down:db": {Key: "service_down:db", Ts: now, Source: loop.ServiceDownSource, Value: `"up"`}, + }, } c := loop.Candidate{Rule: loop.ServiceDownRule(), Severity: loop.Sev4, State: st} pn, _ := NewStub().PhraseNudge(context.Background(), c) @@ -96,12 +98,12 @@ func TestPhraseNudgeServiceDownNamedService(t *testing.T) { } func TestPhraseNudgeServiceDownGenericKey(t *testing.T) { - // the rule key itself ("service_down") rather than a specific service → - // the generic phrase, not a phantom "service_down down on homesrv". + // the old aggregate key, still in the store from before the per-monitor + // facts landed → the generic phrase, never a phantom "service_down down". now := time.Now().UTC() st := loop.State{ Now: now, - Facts: map[string]store.Fact{"service_down": {Key: "service_down", Ts: now, Source: "poll:healthcheck", Value: `"down"`}}, + Facts: map[string]store.Fact{"service_down": {Key: "service_down", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}}, } c := loop.Candidate{Rule: loop.ServiceDownRule(), Severity: loop.Sev4, State: st} pn, _ := NewStub().PhraseNudge(context.Background(), c) @@ -110,6 +112,26 @@ func TestPhraseNudgeServiceDownGenericKey(t *testing.T) { } } +// Two monitors down at once must both be named — he needs to know the blast +// radius, and "a service is down" was the whole defect being fixed here. +func TestPhraseNudgeServiceDownNamesEveryDownMonitor(t *testing.T) { + now := time.Now().UTC() + st := loop.State{ + Now: now, + Facts: map[string]store.Fact{ + "service_down:nginx": {Key: "service_down:nginx", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}, + "service_down:db": {Key: "service_down:db", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}, + }, + } + c := loop.Candidate{Rule: loop.ServiceDownRule(), Severity: loop.Sev4, State: st} + pn, _ := NewStub().PhraseNudge(context.Background(), c) + for _, want := range []string{"nginx", "db"} { + if !strings.Contains(pn.Body, want) { + t.Fatalf("body should name %q, got %q", want, pn.Body) + } + } +} + func TestPhraseNudgeUnknownRuleFallsBack(t *testing.T) { // a rule without a dedicated template — generic fallback names the rule + // severity gist. never empty. From 6d8a95095a33ec402c90f2f9814e6ca7526345a4 Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 4 Aug 2026 05:15:02 +0400 Subject: [PATCH 4/4] deploy, docs: turn service_down back on (V-444) It was disabled because it could not say which service. It can now. --- deploy/mavend.json | 10 +++++----- docs/design.md | 23 +++++++++++++++++++++++ internal/config/config.go | 9 +++++---- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/deploy/mavend.json b/deploy/mavend.json index bb67b34..e1d1a9e 100644 --- a/deploy/mavend.json +++ b/deploy/mavend.json @@ -8,12 +8,12 @@ "//disabled_rules": [ "Nudge rules that are not wired at all. Names come from loop.DefaultRules:", "water, meal, break, service_down, netdata_critical.", - "service_down is off because it cannot say WHICH service — mavpoll folds the", - "whole kuma gauge into one boolean, so the nudge is always the generic 'a", - "service on homesrv is down'. Nothing to act on, every fifteen minutes.", - "Turn it back on once Vikunja #444 lands a fact per monitor." + "service_down is back on: mavpoll now writes one fact per kuma monitor", + "(service_down:), so the nudge names the service and pausing a monitor", + "in kuma silences that monitor. It is also edge-triggered, so a service that", + "stays down is one nudge, not one every fifteen minutes." ], - "disabled_rules": ["service_down"], + "disabled_rules": [], "phraser": { "model_path": "/opt/maven/models/llm/qwen3/Qwen3-1.7B-UD-Q4_K_XL.gguf", diff --git a/docs/design.md b/docs/design.md index 9c9e01d..4bf6ba4 100644 --- a/docs/design.md +++ b/docs/design.md @@ -389,6 +389,29 @@ lives in `source`; rules trust provenance. `source=poll:healthcheck`. A compromised poller must not be able to forge a trigger. +#### A fact per monitor, not an aggregate + +mavpoll writes one fact per kuma monitor, keyed `service_down:`. +It used to fold the whole gauge into a single boolean, and the nudge could then +only say that something on homesrv was down. That is not something he can act +on, so the rule shipped disabled. + +Three things follow from the split: + +- The key set is no longer known at wiring time. A rule declares + `WantPrefixes` and the gatherer resolves the family per tick, which is the + only prefix read in the loop. +- Pausing a monitor in kuma silences that monitor. Under the aggregate it + silenced nothing, because some other monitor kept the boolean at "down". +- A monitor deleted in kuma would keep its last fact reading "down" forever, so + mavpoll marks a vanished monitor "unknown". No rule fires on "unknown". + +The rule is also edge-triggered: it fires on a transition it has not already +nudged about (`State.NudgedSince`). A polled fact is written only when the +value changes, but the predicate reads the current value, so without the edge +check a service that stays down qualifies on every tick and cooldown is the +only brake. + ### Presence — concrete scoring **Combiner — noisy-OR, not weighted sum.** These are independent-ish positive diff --git a/internal/config/config.go b/internal/config/config.go index 990ae1e..51297bb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -149,10 +149,11 @@ type Config struct { // // Rules are code, not config (see loop.DefaultRules), and that stays true: // this only subtracts. It exists because a rule can be right in principle - // and useless in practice — kuma's service_down cannot name the service it - // is nudging about (Vikunja #444), so being told "a service on homesrv is - // down" every fifteen minutes is noise with no action attached. Turning it - // off beats learning to ignore her. + // and useless in practice. service_down was the case that forced it: it + // could not name the service it was nudging about, so being told "a service + // on homesrv is down" every fifteen minutes was noise with no action + // attached. That is fixed — one fact per kuma monitor — and the rule ships + // enabled again. The escape hatch stays. // // A disabled rule is never gathered for, never evaluated, and never // delivered on any channel. Unknown names are ignored, so removing a rule