// loop/gather.go — the ONE impure piece in the loop. // // Gather builds a State snapshot under the store lock at the start of each // tick. From there, every predicate and the gate are pure functions over State. // // Why centralize the I/O: the loop is "dumb + deterministic", the spec // repeatedly enforces a no-I/O contract on predicates. Centralizing read here // makes the contract checkable (anywhere outside gather.go doing I/O is a bug). package loop import ( "context" "encoding/json" "fmt" "time" "github.com/kami/maven/internal/store" ) // Gatherer — holds nothing mutable; the Store is the only dependency. The // daemon runs one Gatherer per tick. type Gatherer struct { store *store.Store rules []Rule quietStart string // "HH:MM" local time, "" = disabled quietEnd string // "HH:MM" local time, "" = disabled } func NewGatherer(s *store.Store, rules []Rule) *Gatherer { return &Gatherer{store: s, rules: rules} } // SetQuietHours enables the time-window schedule check. Times are local. func (g *Gatherer) SetQuietHours(start, end string) { g.quietStart = start g.quietEnd = end } // GatherState — reads the store ONCE and assembles the snapshot the pure Tick // will operate on. // // Reads the loop needs: // - presence: probes (ts per signal key), current bucket, compute score, resolve. // - facts: every key any rule's Predicate OR InertWhenNoData names. // - last nudge per rule (for cooldown). // - due reminders (the loop reuses the loop for reminders; we gather them here). // - env flags QuietHours / CalendarBusy — read as facts (kind=config/env). // // All reads share a single read-only transaction for a consistent snapshot. func (g *Gatherer) GatherState(ctx context.Context, now time.Time) (State, []store.Reminder, error) { // presence first — pure function over probes. the gate checks the bucket; // delivery (later) checks the score. probes, err := g.store.PresenceProbes(ctx) if err != nil { return State{}, nil, err } lastBucket, _, _, err := g.store.LoadPresenceState(ctx) if err != nil { return State{}, nil, err } score := store.PresenceScore(now, probes) bucket := store.Resolve(score, lastBucket) // collect every key any rule references (predicate + inert list). // Daemon rules are small (≤ ~30, per the "revisit at 30 rules" line); // a single map over rules per tick is negligible at 60s cadence. wanted := make(map[string]struct{}) for _, r := range g.rules { // We don't introspect the predicate closure (Go can't); the rule author // declares InertWhenNoData for keys the predicate reads. Reuse that list. for _, k := range r.InertWhenNoData { wanted[k] = struct{}{} } } // presence signal keys live in facts too — included via the probes path, // but also surface via Fact() for rules that want direct access (e.g. break). for _, sig := range store.PresenceSignals { wanted[sig.Key] = struct{}{} } facts := make(map[string]store.Fact, len(wanted)) for k := range wanted { f, err := g.store.LatestFact(ctx, k) if err == nil { facts[k] = f continue } if err == store.ErrNoFact { continue // missing ⇒ shut up; the gate handles it } 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 // (a module that doesn't own the `feedback` source can't poison a rule's // cooldown once the auth source-scope lands — same shape as ServiceDownRule). lastNudge := make(map[string]store.Nudge, len(g.rules)) cooldownUntil := make(map[string]time.Time, len(g.rules)) for _, r := range g.rules { base := r.Cooldown.Base if f, err := g.store.LatestFactBySource(ctx, FeedbackKey(r), FeedbackSource); err == nil { if tuned, ok := ParseCooldownFact(f); ok { base = tuned } } else if err != store.ErrNoFact { return State{}, nil, err } n, err := g.store.LastNudge(ctx, r.Name) if err == nil { lastNudge[r.Name] = n cooldownUntil[r.Name] = CooldownFor(base, n.Ts) continue } if err == store.ErrNudgeNotFound { continue // never fired → no cooldown } return State{}, nil, err } // live snoozes — "leave me alone until X", per rule. The `snoozed` outcome // on the nudges table is the whole record; the store turns it into an // expiry. Absent rules mean "not snoozed", which is what the gate reads. snoozeUntil, err := g.store.SnoozedUntil(ctx, now) if err != nil { return State{}, nil, err } // env flags — QuietHours / CalendarBusy as config facts. // QuietHours: presence != reachability, sleep/quiet-hours handled separately // in the gate. We read a config `quiet_hours` fact for the boolean. var quiet bool if f, ok := readFact(ctx, g.store, "quiet_hours"); ok { quiet = f.Value == "true" || f.Value == `"true"` } // Schedule-based quiet hours: if a time window is configured AND we're // inside it, quiet is true regardless of the config fact. The voice // toggle extends activation independently — both can fire at once. if g.quietStart != "" && g.quietEnd != "" && inQuietWindow(now, g.quietStart, g.quietEnd) { quiet = true } var calBusy bool if f, ok := readFact(ctx, g.store, "calendar_busy"); ok { calBusy = f.Value == "true" || f.Value == `"true"` } // due reminders — gate-bypassing class. read here, the daemon emits them. due, err := g.store.DueReminders(ctx, now) if err != nil { return State{}, nil, err } due = collapseReminders(due) s := State{ Now: now, Presence: bucket, PresenceScore: score, Facts: facts, LastNudge: lastNudge, SnoozeUntil: snoozeUntil, CooldownUntil: cooldownUntil, QuietHours: quiet, CalendarBusy: calBusy, } return s, due, nil } // inQuietWindow returns true when `now` falls within the daily window // [start, end). start/end are "HH:MM" in local time. Windows that cross // midnight (start > end) are handled: "23:00"-"08:00" means quiet from // 23:00 to 08:00 the next day. func inQuietWindow(now time.Time, start, end string) bool { startH, startM, ok1 := parseHHMM(start) endH, endM, ok2 := parseHHMM(end) if !ok1 || !ok2 { return false } nowMin := now.Hour()*60 + now.Minute() startMin := startH*60 + startM endMin := endH*60 + endM if startMin < endMin { return nowMin >= startMin && nowMin < endMin } // Crossing midnight: e.g., 23:00-08:00. return nowMin >= startMin || nowMin < endMin } func parseHHMM(s string) (hour, min int, ok bool) { if len(s) != 5 || s[2] != ':' { return 0, 0, false } h := int(s[0]-'0')*10 + int(s[1]-'0') m := int(s[3]-'0')*10 + int(s[4]-'0') if h < 0 || h > 23 || m < 0 || m > 59 { return 0, 0, false } return h, m, true } func readFact(ctx context.Context, s *store.Store, key string) (store.Fact, bool) { f, err := s.LatestFact(ctx, key) if err != nil { return store.Fact{}, false } return f, true } // collapseReminders — when multiple reminders are due at once (e.g. after // the daemon was offline), collapse them into a single digest reminder to // avoid a burst of individual notifications. The synthetic digest (ID=0) // carries the originals in Collapsed; the dispatcher completes them (mark // fired / reschedule) only after the digest actually delivers, preserving // the "failed send leaves the reminder pending" invariant. // When 0 or 1 reminders are due, returns them unchanged. func collapseReminders(due []store.Reminder) []store.Reminder { if len(due) <= 1 { return due } // Build a summary payload. var items []string earliest := due[0].FireTs for _, r := range due { // Each reminder's Payload is JSON. Try to extract a "text" field; // fall back to the raw payload. var parsed struct { Text string `json:"text"` } if json.Unmarshal([]byte(r.Payload), &parsed) == nil && parsed.Text != "" { items = append(items, parsed.Text) } else { items = append(items, r.Payload) } if r.FireTs.Before(earliest) { earliest = r.FireTs } } summary := fmt.Sprintf("You have %d pending reminders", len(due)) digestPayload, _ := json.Marshal(map[string]any{ "text": summary, "items": items, }) // Return a single synthetic digest reminder. ID=0 signals "digest" to // the dispatcher, which completes the Collapsed originals on success. return []store.Reminder{{ ID: 0, FireTs: earliest, Payload: string(digestPayload), Status: "pending", Collapsed: due, }} }